[BAEL-4847] Kafka SSL with Spring Boot client

spring-kafka/pom.xml

@@ -28,6 +28,10 @@
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-databind</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.springframework.kafka</groupId>
             <artifactId>spring-kafka-test</artifactId>
@@ -41,9 +45,15 @@
             <scope>test</scope>
         </dependency>
         <dependency>
-            <groupId>commons-collections</groupId>
+            <groupId>org.testcontainers</groupId>
-            <artifactId>commons-collections</artifactId>
+            <artifactId>junit-jupiter</artifactId>
-            <version>3.2.1</version>
+            <version>${testcontainers-kafka.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.awaitility</groupId>
+            <artifactId>awaitility</artifactId>
+            <scope>test</scope>
         </dependency>
     </dependencies>
 

spring-kafka/src/main/java/com/baeldung/kafka/ssl/KafkaConsumer.java

@@ -0,0 +1,25 @@
+package com.baeldung.kafka.ssl;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Component;
+
+import java.util.ArrayList;
+import java.util.List;
+
+@Component
+@Slf4j
+public class KafkaConsumer {
+
+    public static final String TOPIC = "test-topic";
+
+    public final List<String> messages = new ArrayList<>();
+
+    @KafkaListener(topics = TOPIC)
+    public void receive(ConsumerRecord<String, String> consumerRecord) {
+        log.info("Received payload: '{}'", consumerRecord.toString());
+        messages.add(consumerRecord.value());
+    }
+
+}

spring-kafka/src/main/java/com/baeldung/kafka/ssl/KafkaProducer.java

@@ -0,0 +1,23 @@
+package com.baeldung.kafka.ssl;
+
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@AllArgsConstructor
+@Component
+public class KafkaProducer {
+
+    private final KafkaTemplate<String, String> kafkaTemplate;
+
+    public void sendMessage(String message, String topic) {
+        log.info("Producing message: {}", message);
+        kafkaTemplate.send(topic, "key", message)
+          .addCallback(
+            result -> log.info("Message sent to topic: {}", message),
+            ex -> log.error("Failed to send message", ex)
+          );
+    }
+}

spring-kafka/src/main/java/com/baeldung/kafka/ssl/KafkaSslApplication.java

@@ -0,0 +1,15 @@
+package com.baeldung.kafka.ssl;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+@SpringBootApplication
+@EnableAutoConfiguration
+public class KafkaSslApplication {
+
+    public static void main(String[] args) {
+        SpringApplication.run(KafkaSslApplication.class, args);
+    }
+
+}

spring-kafka/src/main/resources/application-ssl.yml

@@ -0,0 +1,18 @@
+spring:
+  kafka:
+    security:
+      protocol: "SSL"
+    bootstrap-servers: localhost:9093
+    ssl:
+      trust-store-location: classpath:/client-certs/kafka.client.truststore.jks
+      trust-store-password: password
+      key-store-location:  classpath:/client-certs/kafka.client.keystore.jks
+      key-store-password: password
+    consumer:
+      group-id: demo-group-id
+      auto-offset-reset: earliest
+      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
+      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
+    producer:
+      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
+      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

spring-kafka/src/test/java/com/baeldung/kafka/ssl/KafkaSslApplicationLiveTest.java

@@ -0,0 +1,54 @@
+package com.baeldung.kafka.ssl;
+
+import lombok.extern.slf4j.Slf4j;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.ActiveProfiles;
+import org.testcontainers.containers.DockerComposeContainer;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+import java.io.File;
+import java.time.Duration;
+import java.util.UUID;
+
+import static com.baeldung.kafka.ssl.KafkaConsumer.TOPIC;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+@Slf4j
+@ActiveProfiles("ssl")
+@Testcontainers
+@SpringBootTest(classes = KafkaSslApplication.class)
+class KafkaSslApplicationLiveTest {
+
+    private static final File KAFKA_COMPOSE_FILE = new File("src/test/resources/docker/docker-compose.yml");
+    private static final String KAFKA_SERVICE = "kafka";
+    private static final int SSL_PORT = 9093;
+
+    @Container
+    public DockerComposeContainer<?> container =
+      new DockerComposeContainer<>(KAFKA_COMPOSE_FILE)
+        .withExposedService(KAFKA_SERVICE, SSL_PORT, Wait.forListeningPort());
+
+    @Autowired
+    private KafkaProducer kafkaProducer;
+
+    @Autowired
+    private KafkaConsumer kafkaConsumer;
+
+    @Test
+    void givenSslIsConfigured_whenProducerSendsMessageOverSsl_thenConsumerReceivesOverSsl() {
+        String message = generateSampleMessage();
+        kafkaProducer.sendMessage(message, TOPIC);
+
+        await().atMost(Duration.ofMinutes(2))
+          .untilAsserted(() -> assertThat(kafkaConsumer.messages).containsExactly(message));
+    }
+
+    private static String generateSampleMessage() {
+        return UUID.randomUUID().toString();
+    }
+}

spring-kafka/src/test/resources/docker/certs/kafka_keystore_credentials

@@ -0,0 +1 @@
+password

spring-kafka/src/test/resources/docker/certs/kafka_sslkey_credentials

@@ -0,0 +1 @@
+password

spring-kafka/src/test/resources/docker/certs/kafka_truststore_credentials

@@ -0,0 +1 @@
+password

spring-kafka/src/test/resources/docker/docker-compose.yml

@@ -0,0 +1,30 @@
+---
+version: '2'
+services:
+  zookeeper:
+    image: confluentinc/cp-zookeeper:6.2.0
+    environment:
+      ZOOKEEPER_CLIENT_PORT: 2181
+      ZOOKEEPER_TICK_TIME: 2000
+
+  kafka:
+    image: confluentinc/cp-kafka:6.2.0
+    depends_on:
+      - zookeeper
+    ports:
+      - 9092:9092
+      - 9093:9093
+    environment:
+      KAFKA_BROKER_ID: 1
+      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092,SSL://localhost:9093
+      KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'true'
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+      KAFKA_SSL_CLIENT_AUTH: "required"
+      KAFKA_SSL_KEYSTORE_FILENAME: '/certs/kafka.server.keystore.jks'
+      KAFKA_SSL_KEYSTORE_CREDENTIALS: '/certs/kafka_keystore_credentials'
+      KAFKA_SSL_KEY_CREDENTIALS: '/certs/kafka_sslkey_credentials'
+      KAFKA_SSL_TRUSTSTORE_FILENAME: '/certs/kafka.server.truststore.jks'
+      KAFKA_SSL_TRUSTSTORE_CREDENTIALS: '/certs/kafka_truststore_credentials'
+    volumes:
+      - ./certs/:/etc/kafka/secrets/certs

spring-kafka/src/test/resources/logback-test.xml

@@ -10,4 +10,10 @@
     <root level="INFO">
         <appender-ref ref="STDOUT" />
     </root>
+
+    <!-- Reduce the noise as the consumer keeps trying to connect until the Kafka instance is available -->
+    <springProfile name="ssl">
+        <logger name="org.apache.kafka.clients.NetworkClient" level="ERROR" additivity="false"/>
+    </springProfile>
+
 </configuration>