Merge pull request #18253 from saikatcse03/BAEL6420

BAEL-6420 Implement kafka with SASLAuthentication with GSSAPI

Author: Maiklins

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

@@ -0,0 +1,23 @@
+package com.baeldung.sasl;
+
+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/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/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/src/test/java/com/baeldung/sasl/SprintContextTest.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/src/test/resources/sasl/Dockerfile

@@ -0,0 +1,15 @@
+# Use a minimal base image
+FROM debian:bullseye
+
+RUN apt-get update && \
+    apt-get install -y krb5-kdc krb5-admin-server krb5-user && \
+    rm -rf /var/lib/apt/lists/*
+
+COPY config/krb5.conf /etc/krb5.conf
+COPY setup_kdc.sh /setup_kdc.sh
+
+RUN chmod +x /setup_kdc.sh
+
+EXPOSE 88 749
+
+CMD ["/setup_kdc.sh"]

spring-kafka/src/test/resources/sasl/config/kadm5.acl

@@ -0,0 +1 @@
+*/[email protected] *

spring-kafka/src/test/resources/sasl/config/kafka_server_jaas.conf

@@ -0,0 +1,17 @@
+KafkaServer {
+    com.sun.security.auth.module.Krb5LoginModule required
+    useKeyTab=true
+    storeKey=true
+    keyTab="/etc/kafka/keytabs/kafka.keytab"
+    principal="kafka/[email protected]"
+    serviceName="kafka";
+};
+
+Client {
+  com.sun.security.auth.module.Krb5LoginModule required
+    useKeyTab=true
+    storeKey=true
+    keyTab="/etc/kafka/keytabs/client.keytab"
+    principal="[email protected]"
+    serviceName="kafka";
+};

spring-kafka/src/test/resources/sasl/config/krb5.conf

@@ -0,0 +1,17 @@
+[libdefaults]
+  default_realm = BAELDUNG.COM
+  dns_lookup_realm = false
+  dns_lookup_kdc = false
+  forwardable = true
+  rdns = true
+
+[realms]
+  BAELDUNG.COM = {
+    kdc = kdc
+    admin_server = kdc
+  }
+
+[logging]
+    default = FILE:/var/log/krb5libs.log
+    kdc = FILE:/var/log/krb5kdc.log
+    admin_server = FILE:/var/log/kadmind.log

spring-kafka/src/test/resources/sasl/config/zookeeper_jaas.conf

@@ -0,0 +1,7 @@
+Server {
+    com.sun.security.auth.module.Krb5LoginModule required
+    useKeyTab=true
+    storeKey=true
+    keyTab="/etc/kafka/keytabs/zookeeper.keytab"
+    principal="zookeeper/[email protected]";
+};

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

@@ -0,0 +1,50 @@
+services:
+
+  kdc:
+    build:
+      context: .
+      dockerfile: Dockerfile
+    volumes:
+      - ./config:/etc/krb5kdc
+      - ./keytabs:/etc/krb5kdc/keytabs
+      - ./config/krb5.conf:/etc/krb5.conf
+    ports:
+      - "88:88/udp"
+
+  zookeeper:
+    image: confluentinc/cp-zookeeper:latest
+    container_name: zookeeper
+    hostname: localhost
+    environment:
+      ZOOKEEPER_CLIENT_PORT: 2181
+      ZOOKEEPER_TICK_TIME: 2000
+      KAFKA_OPTS: "-Djava.security.auth.login.config=/etc/kafka/zookeeper_jaas.conf"
+    volumes:
+      - ./config/zookeeper_jaas.conf:/etc/kafka/zookeeper_jaas.conf
+      - ./keytabs:/etc/kafka/keytabs
+      - ./config/krb5.conf:/etc/krb5.conf
+    ports:
+      - "2181:2181"
+
+  kafka:
+    image: confluentinc/cp-kafka:latest
+    container_name: kafka
+    environment:
+      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
+      KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL: GSSAPI
+      KAFKA_SASL_ENABLED_MECHANISMS: GSSAPI
+      KAFKA_LISTENERS: SASL_PLAINTEXT://:9092
+      KAFKA_ADVERTISED_LISTENERS: SASL_PLAINTEXT://localhost:9092
+      KAFKA_INTER_BROKER_LISTENER_NAME: SASL_PLAINTEXT
+      KAFKA_OPTS: "-Djava.security.auth.login.config=/etc/kafka/kafka_server_jaas.conf"
+      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
+    volumes:
+      - ./config/kafka_server_jaas.conf:/etc/kafka/kafka_server_jaas.conf
+      - ./keytabs:/etc/kafka/keytabs
+      - ./config/krb5.conf:/etc/krb5.conf
+    depends_on:
+      - zookeeper
+      - kdc
+    ports:
+      - 9092:9092