BAEL-9070 Implement SASL PLAIN Authentication in Kafka (#18639)

* opentelemtry for spring boot 3

* refactoring

* fixed lint issues

* refactoring

* remove all deprecated code related to spring cloud sleuth

* include version property

* include version property

* move spring-kafka article code to this module

* remove code related to SASL article

* sasl plaintext implemented

* refactor the sasl config and test

* removed unrelated code

* removed unrelated code

* removed unrelated code

* implement SASL Plain

* refactor the configs

* remove the lombok dependency and spring test

* refactoring error fixes

* fix indentation in jaas configs

* fix method typo

---------

Co-authored-by: Liam Williams <[email protected]>

Author: saikatcse03

spring-kafka-4/pom.xml

@@ -31,7 +31,6 @@
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-actuator</artifactId>
         </dependency>
-
         <dependency>
             <groupId>org.apache.avro</groupId>
             <artifactId>avro</artifactId>

spring-kafka-4/src/main/java/com/baeldung/sasl/KafkaConsumer.java

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

spring-kafka-4/src/main/java/com/baeldung/sasl/KafkaSaslApplication.java

@@ -0,0 +1,13 @@
+package com.baeldung.sasl;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+@SpringBootApplication
+public class KafkaSaslApplication {
+
+    public static void main(String[] args) {
+        System.setProperty("spring.config.name", "application-sasl");
+        SpringApplication.run(KafkaSaslApplication.class, args);
+    }
+}

spring-kafka-4/src/main/java/com/baeldung/saslplaintext/KafkaConsumer.java

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

spring-kafka-4/src/main/java/com/baeldung/saslplaintext/KafkaProducer.java

@@ -0,0 +1,29 @@
+package com.baeldung.saslplaintext;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Component;
+
+@Component
+public class KafkaProducer {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(KafkaProducer.class);
+    private final KafkaTemplate<String, String> kafkaTemplate;
+
+    public KafkaProducer(KafkaTemplate<String, String> kafkaTemplate) {
+        this.kafkaTemplate = kafkaTemplate;
+    }
+
+    public void sendMessage(String message, String topic) {
+        LOGGER.info("Producing message: {}", message);
+        kafkaTemplate.send(topic, "key", message)
+            .whenComplete((result, ex) -> {
+                if (ex == null) {
+                    LOGGER.info("Message sent to topic: {}", message);
+                } else {
+                    LOGGER.error("Failed to send message", ex);
+                }
+            });
+    }
+}

spring-kafka-4/src/main/java/com/baeldung/saslplaintext/KafkaSaslPlaintextApplication.java

@@ -0,0 +1,13 @@
+package com.baeldung.saslplaintext;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+@SpringBootApplication
+public class KafkaSaslPlaintextApplication {
+
+    public static void main(String[] args) {
+        System.setProperty("spring.config.name", "application-sasl-plaintext");
+        SpringApplication.run(KafkaSaslPlaintextApplication.class, args);
+    }
+}

spring-kafka-4/src/main/resources/application-sasl-plaintext.yml

@@ -0,0 +1,16 @@
+spring:
+  kafka:
+    bootstrap-servers: localhost:9092
+    properties:
+      sasl.mechanism: PLAIN
+      sasl.jaas.config: >
+        org.apache.kafka.common.security.plain.PlainLoginModule required
+        username="user1"
+        password="user1-secret";
+    security:
+      protocol: SASL_PLAINTEXT
+    consumer:
+      group-id: test-group
+      auto-offset-reset: earliest
+      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
+      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

spring-kafka-4/src/main/resources/application-sasl.yml

@@ -0,0 +1,19 @@
+spring:
+  kafka:
+    bootstrap-servers: localhost:9092
+    properties:
+      sasl.mechanism: GSSAPI
+      sasl.jaas.config: >
+        com.sun.security.auth.module.Krb5LoginModule required
+        useKeyTab=true
+        storeKey=true
+        keyTab="./src/test/resources/sasl/keytabs/client.keytab"
+        principal="[email protected]"
+        serviceName="kafka";
+    security:
+      protocol: "SASL_PLAINTEXT"
+    consumer:
+      group-id: test
+      auto-offset-reset: earliest
+      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
+      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

spring-kafka-4/src/test/java/com/baeldung/sasl/SpringContextTest.java

@@ -0,0 +1,15 @@
+package com.baeldung.sasl;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.junit.jupiter.SpringExtension;
+
+@ExtendWith(SpringExtension.class)
+@SpringBootTest(classes = KafkaSaslApplication.class)
+class SpringContextTest {
+
+    @Test
+    void whenSpringContextIsBootstrapped_thenNoExceptions() {
+    }
+}

spring-kafka-4/src/test/java/com/baeldung/saslplaintext/KafkaSaslPlaintextApplicationLiveTest.java

@@ -0,0 +1,48 @@
+package com.baeldung.saslplaintext;
+
+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.saslplaintext.KafkaConsumer.TOPIC;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+@Testcontainers
+@ActiveProfiles("sasl-plaintext")
+@SpringBootTest(classes = KafkaSaslPlaintextApplication.class)
+class KafkaSaslPlaintextApplicationLiveTest {
+
+    private static final File KAFKA_COMPOSE_FILE = new File("src/test/resources/sasl-plaintext/docker-compose.yml");
+    private static final String KAFKA_SERVICE = "kafka";
+    private static final int SASL_PORT = 9092;
+
+    @Container
+    public DockerComposeContainer<?> container =
+      new DockerComposeContainer<>(KAFKA_COMPOSE_FILE)
+        .withExposedService(KAFKA_SERVICE, SASL_PORT, Wait.forListeningPort());
+
+    @Autowired
+    private KafkaProducer kafkaProducer;
+
+    @Autowired
+    private KafkaConsumer kafkaConsumer;
+
+    @Test
+    void givenSaslIsConfigured_whenProducerSendsMessageOverSasl_thenConsumerReceivesOverSasl() {
+        String message = UUID.randomUUID().toString();
+        kafkaProducer.sendMessage(message, TOPIC);
+
+        await().atMost(Duration.ofMinutes(2))
+          .untilAsserted(() -> assertThat(kafkaConsumer.messages).containsExactly(message));
+    }
+}