feat(java): add TCP/TLS integration tests and examples

.github/actions/java-gradle/pre-merge/action.yml

@@ -146,6 +146,34 @@ runs:
         pid-file: ${{ steps.iggy.outputs.pid_file }}
         log-file: ${{ steps.iggy.outputs.log_file }}
 
+    - name: Start Iggy server (TLS)
+      id: iggy-tls
+      if: inputs.task == 'test'
+      uses: ./.github/actions/utils/server-start
+      with:
+        pid-file: ${{ runner.temp }}/iggy-server-tls.pid
+        log-file: ${{ runner.temp }}/iggy-server-tls.log
+      env:
+        IGGY_TCP_TLS_ENABLED: "true"
+        IGGY_TCP_TLS_CERT_FILE: core/certs/iggy_cert.pem
+        IGGY_TCP_TLS_KEY_FILE: core/certs/iggy_key.pem
+
+    - name: TLS integration tests
+      if: inputs.task == 'test'
+      shell: bash
+      working-directory: foreign/java
+      env:
+        USE_EXTERNAL_SERVER: true
+        IGGY_TCP_TLS_ENABLED: "true"
+      run: ./gradlew :iggy:test --tests "*TlsConnectionTest*"
+
+    - name: Stop Iggy server (TLS)
+      if: always() && inputs.task == 'test'
+      uses: ./.github/actions/utils/server-stop
+      with:
+        pid-file: ${{ steps.iggy-tls.outputs.pid_file }}
+        log-file: ${{ steps.iggy-tls.outputs.log_file }}
+
     - name: Test Summary
       uses: test-summary/action@v2
       with:

examples/java/README.md

@@ -174,6 +174,28 @@ The async client uses Netty's event loop threads for I/O operations. **NEVER** b
 
 If your message processing involves blocking operations, offload to a separate thread pool using `thenApplyAsync(fn, executor)`.
 
+## Security Examples
+
+### TCP/TLS
+
+Demonstrates secure TLS-encrypted TCP connections:
+
+```bash
+./gradlew runTcpTlsProducer
+./gradlew runTcpTlsConsumer
+```
+
+These examples require a TLS-enabled Iggy server. Start the server with:
+
+```bash
+IGGY_TCP_TLS_ENABLED=true \
+IGGY_TCP_TLS_CERT_FILE=core/certs/iggy_cert.pem \
+IGGY_TCP_TLS_KEY_FILE=core/certs/iggy_key.pem \
+cargo r --bin iggy-server
+```
+
+Uses `IggyTcpClientBuilder` with TLS options (`enableTls`, `tlsDomain`, `tlsCaCertPath`) to establish TLS-encrypted TCP connections with CA certificate verification.
+
 ## Blocking vs. Async - When to Use Each
 
 The Iggy Java SDK provides two client types: **blocking (synchronous)** and **async (non-blocking)**. Choose based on your use case:

examples/java/build.gradle.kts

@@ -110,3 +110,12 @@ tasks.register<JavaExec>("runAsyncConsumer") {
     mainClass.set("org.apache.iggy.examples.async.AsyncConsumer")
 }
 
+tasks.register<JavaExec>("runTcpTlsProducer") {
+    classpath = sourceSets["main"].runtimeClasspath
+    mainClass.set("org.apache.iggy.examples.tcptls.producer.TcpTlsProducer")
+}
+
+tasks.register<JavaExec>("runTcpTlsConsumer") {
+    classpath = sourceSets["main"].runtimeClasspath
+    mainClass.set("org.apache.iggy.examples.tcptls.consumer.TcpTlsConsumer")
+}

examples/java/src/main/java/org/apache/iggy/examples/tcptls/consumer/TcpTlsConsumer.java

@@ -0,0 +1,141 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iggy.examples.tcptls.consumer;
+
+import org.apache.iggy.client.blocking.tcp.IggyTcpClient;
+import org.apache.iggy.consumergroup.Consumer;
+import org.apache.iggy.identifier.StreamId;
+import org.apache.iggy.identifier.TopicId;
+import org.apache.iggy.message.Message;
+import org.apache.iggy.message.PolledMessages;
+import org.apache.iggy.message.PollingStrategy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.math.BigInteger;
+import java.nio.charset.StandardCharsets;
+import java.util.Optional;
+
+/**
+ * TCP/TLS Consumer Example
+ *
+ * <p>Demonstrates consuming messages over a TLS-encrypted TCP connection
+ * using custom certificates from core/certs/.
+ *
+ * <p>Prerequisites: Start the Iggy server with TLS enabled:
+ * <pre>
+ *   IGGY_TCP_TLS_ENABLED=true \
+ *   IGGY_TCP_TLS_CERT_FILE=core/certs/iggy_cert.pem \
+ *   IGGY_TCP_TLS_KEY_FILE=core/certs/iggy_key.pem \
+ *   cargo r --bin iggy-server
+ * </pre>
+ */
+public final class TcpTlsConsumer {
+
+    private static final StreamId STREAM_ID = StreamId.of("tls-stream");
+    private static final TopicId TOPIC_ID = TopicId.of("tls-topic");
+
+    private static final long PARTITION_ID = 0L;
+
+    private static final int BATCHES_LIMIT = 5;
+
+    private static final long MESSAGES_PER_BATCH = 10L;
+    private static final long INTERVAL_MS = 500;
+
+    private static final Logger log = LoggerFactory.getLogger(TcpTlsConsumer.class);
+
+    private TcpTlsConsumer() {}
+
+    public static void main(String[] args) {
+        // Build a TCP client with TLS enabled.
+        // enableTls()           activates TLS on the TCP transport
+        // tlsCertificate(...)   points to the CA certificate used to verify the server cert
+        var client = IggyTcpClient.builder()
+                .host("localhost")
+                .port(8090)
+                .enableTls()
+                .tlsCertificate("../../core/certs/iggy_ca_cert.pem")
+                .credentials("iggy", "iggy")
+                .buildAndLogin();
+
+        consumeMessages(client);
+    }
+
+    private static void consumeMessages(IggyTcpClient client) {
+        log.info(
+                "Messages will be consumed from stream: {}, topic: {}, partition: {} with interval {}ms.",
+                STREAM_ID,
+                TOPIC_ID,
+                PARTITION_ID,
+                INTERVAL_MS);
+
+        BigInteger offset = BigInteger.ZERO;
+        int consumedBatches = 0;
+
+        Consumer consumer = Consumer.of(0L);
+
+        while (true) {
+            if (consumedBatches == BATCHES_LIMIT) {
+                log.info("Consumed {} batches of messages, exiting.", consumedBatches);
+                return;
+            }
+
+            try {
+                PolledMessages polledMessages = client.messages()
+                        .pollMessages(
+                                STREAM_ID,
+                                TOPIC_ID,
+                                Optional.of(PARTITION_ID),
+                                consumer,
+                                PollingStrategy.offset(offset),
+                                MESSAGES_PER_BATCH,
+                                false);
+
+                if (polledMessages.messages().isEmpty()) {
+                    log.info("No messages found.");
+                    Thread.sleep(INTERVAL_MS);
+                    continue;
+                }
+
+                for (Message message : polledMessages.messages()) {
+                    handleMessage(message, offset);
+                }
+
+                consumedBatches++;
+
+                offset = offset.add(BigInteger.valueOf(polledMessages.messages().size()));
+
+                Thread.sleep(INTERVAL_MS);
+
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                break;
+            } catch (Exception e) {
+                log.error("Error polling messages", e);
+                break;
+            }
+        }
+    }
+
+    private static void handleMessage(Message message, BigInteger offset) {
+        String payload = new String(message.payload(), StandardCharsets.UTF_8);
+        log.info("Handling message at offset {}, payload: {}...", offset, payload);
+    }
+}

examples/java/src/main/java/org/apache/iggy/examples/tcptls/producer/TcpTlsProducer.java

@@ -0,0 +1,152 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iggy.examples.tcptls.producer;
+
+import org.apache.iggy.client.blocking.tcp.IggyTcpClient;
+import org.apache.iggy.identifier.StreamId;
+import org.apache.iggy.identifier.TopicId;
+import org.apache.iggy.message.Message;
+import org.apache.iggy.message.Partitioning;
+import org.apache.iggy.stream.StreamDetails;
+import org.apache.iggy.topic.CompressionAlgorithm;
+import org.apache.iggy.topic.TopicDetails;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.math.BigInteger;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+
+import static java.util.Optional.empty;
+
+/**
+ * TCP/TLS Producer Example
+ *
+ * <p>Demonstrates producing messages over a TLS-encrypted TCP connection
+ * using custom certificates from core/certs/.
+ *
+ * <p>Prerequisites: Start the Iggy server with TLS enabled:
+ * <pre>
+ *   IGGY_TCP_TLS_ENABLED=true \
+ *   IGGY_TCP_TLS_CERT_FILE=core/certs/iggy_cert.pem \
+ *   IGGY_TCP_TLS_KEY_FILE=core/certs/iggy_key.pem \
+ *   cargo r --bin iggy-server
+ * </pre>
+ */
+public final class TcpTlsProducer {
+
+    private static final String STREAM_NAME = "tls-stream";
+    private static final StreamId STREAM_ID = StreamId.of(STREAM_NAME);
+
+    private static final String TOPIC_NAME = "tls-topic";
+    private static final TopicId TOPIC_ID = TopicId.of(TOPIC_NAME);
+
+    private static final long PARTITION_ID = 0L;
+
+    private static final int BATCHES_LIMIT = 5;
+
+    private static final int MESSAGES_PER_BATCH = 10;
+    private static final long INTERVAL_MS = 500;
+
+    private static final Logger log = LoggerFactory.getLogger(TcpTlsProducer.class);
+
+    private TcpTlsProducer() {}
+
+    public static void main(String[] args) {
+        // Build a TCP client with TLS enabled.
+        // enableTls()           activates TLS on the TCP transport
+        // tlsCertificate(...)   points to the CA certificate used to verify the server cert
+        var client = IggyTcpClient.builder()
+                .host("localhost")
+                .port(8090)
+                .enableTls()
+                .tlsCertificate("../../core/certs/iggy_ca_cert.pem")
+                .credentials("iggy", "iggy")
+                .buildAndLogin();
+
+        createStream(client);
+        createTopic(client);
+        produceMessages(client);
+    }
+
+    private static void produceMessages(IggyTcpClient client) {
+        log.info(
+                "Messages will be sent to stream: {}, topic: {}, partition: {} with interval {}ms.",
+                STREAM_NAME,
+                TOPIC_NAME,
+                PARTITION_ID,
+                INTERVAL_MS);
+
+        int currentId = 0;
+        int sentBatches = 0;
+
+        Partitioning partitioning = Partitioning.partitionId(PARTITION_ID);
+
+        while (sentBatches < BATCHES_LIMIT) {
+            try {
+                Thread.sleep(INTERVAL_MS);
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                break;
+            }
+
+            List<Message> messages = new ArrayList<>();
+            for (int i = 0; i < MESSAGES_PER_BATCH; i++) {
+                currentId++;
+                String payload = "message-" + currentId;
+                messages.add(Message.of(payload));
+            }
+
+            client.messages().sendMessages(STREAM_ID, TOPIC_ID, partitioning, messages);
+            sentBatches++;
+            log.info("Sent {} message(s).", MESSAGES_PER_BATCH);
+        }
+
+        log.info("Sent {} batches of messages, exiting.", sentBatches);
+    }
+
+    private static void createStream(IggyTcpClient client) {
+        Optional<StreamDetails> stream = client.streams().getStream(STREAM_ID);
+        if (stream.isPresent()) {
+            return;
+        }
+        client.streams().createStream(STREAM_NAME);
+        log.info("Stream {} was created.", STREAM_NAME);
+    }
+
+    private static void createTopic(IggyTcpClient client) {
+        Optional<TopicDetails> topic = client.topics().getTopic(STREAM_ID, TOPIC_ID);
+        if (topic.isPresent()) {
+            log.warn("Topic already exists and will not be created again.");
+            return;
+        }
+        client.topics()
+                .createTopic(
+                        STREAM_ID,
+                        1L,
+                        CompressionAlgorithm.None,
+                        BigInteger.ZERO,
+                        BigInteger.ZERO,
+                        empty(),
+                        TOPIC_NAME);
+        log.info("Topic {} was created.", TOPIC_NAME);
+    }
+}