To enable SASL/PLAIN authentication for Apache Kafka running in Docker, you must configure both the broker and client components with appropriate security settings. Below is a streamlined guide to set up and validate this configuration.

Docker Container Setup

docker run --name secure-kafka \
  --restart=always \
  --net=host \
  --volume /data/kafka:/data \
  --volume /path/to/server_jaas.conf:/opt/kafka/config/kafka_server_jaas.conf \
  -e KAFKA_BROKER_ID=1 \
  -e KAFKA_LISTENERS=SASL_PLAINTEXT://0.0.0.0:9092 \
  -e KAFKA_ADVERTISED_LISTENERS=SASL_PLAINTEXT://kafka-host:9092 \
  -e KAFKA_LOG_DIRS=/data/kafka \
  -e KAFKA_ZOOKEEPER_CONNECT="localhost:2181" \
  -e KAFKA_SASL_ENABLED_MECHANISMS=PLAIN \
  -e KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL=PLAIN \
  -e KAFKA_SECURITY_INTER_BROKER_PROTOCOL=SASL_PLAINTEXT \
  -d wurstmeister/kafka:2.12-2.1.1

Broker JAAS Configuration

File: server_jaas.conf

KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="admin-secret"
    user_admin="admin-secret"
    user_client="client-secret";
};

Minimal Server Properties

File: server.properties

broker.id=1
listeners=SASL_PLAINTEXT://0.0.0.0:9092
advertised.listeners=SASL_PLAINTEXT://kafka-host:9092
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=PLAIN

log.dirs=/tmp/kafka-logs
num.partitions=3
zookeeper.connect=localhost:2181
zookeeper.connection.timeout.ms=6000

Startup Script Modification

In kafka-server-start.sh, ensure the JAAS config is loaded:

export KAFKA_OPTS="-Djava.security.auth.login.config=/opt/kafka/config/kafka_server_jaas.conf"
exec $base_dir/kafka-run-class.sh $EXTRA_ARGS kafka.Kafka "$@"

Client Configuration Files

Producer config (producer.conf):

security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="admin-secret";

Consumer config (consumer.conf):

security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="client" password="client-secret";
group.id=test-group
auto.offset.reset=earliest

Commannd-Line Validation

Create topic:

bin/kafka-topics.sh --create --bootstrap-server kafka-host:9092 --topic secure-test --partitions 3 --replication-factor 1

Produce message:

bin/kafka-console-producer.sh --bootstrap-server kafka-host:9092 --topic secure-test --producer.config ./producer.conf

Consume message:

bin/kafka-console-consumer.sh --bootstrap-server kafka-host:9092 --topic secure-test --from-beginning --consumer.config ./consumer.conf

Java Producer Example

import org.apache.kafka.clients.producer.*;
import java.util.Properties;

public class SecureKafkaProducer {
    public static void main(String[] args) {
        System.setProperty("java.security.auth.login.config", "/path/to/client_jaas.conf");

        Properties props = new Properties();
        props.put("bootstrap.servers", "kafka-host:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("security.protocol", "SASL_PLAINTEXT");
        props.put("sasl.mechanism", "PLAIN");

        try (Producer<String, String> producer = new KafkaProducer<>(props)) {
            ProducerRecord<String, String> record = new ProducerRecord<>("secure-test", "Hello, authenticated Kafka!");
            producer.send(record);
            System.out.println("Message sent successfully.");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

Java Consumer Example

import org.apache.kafka.clients.consumer.*;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class SecureKafkaConsumer {
    public static void main(String[] args) {
        System.setProperty("java.security.auth.login.config", "/path/to/client_jaas.conf");

        Properties props = new Properties();
        props.put("bootstrap.servers", "kafka-host:9092");
        props.put("group.id", "secure-group");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("security.protocol", "SASL_PLAINTEXT");
        props.put("sasl.mechanism", "PLAIN");
        props.put("auto.offset.reset", "earliest");

        try (Consumer<String, String> consumer = new KafkaConsumer<>(props)) {
            consumer.subscribe(Collections.singletonList("secure-test"));

            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("Consumed: key=%s, value=%s, partition=%d, offset=%d%n",
                        record.key(), record.value(), record.partition(), record.offset());
                }
            }
        }
    }
}

Client JAAS File

File: client_jaas.conf

KafkaClient {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="client"
    password="client-secret";
};