Build a Docker‑Based Kafka Cluster and Integrate It with Spring Boot

Version

JDK 14

Zookeeper

Kafka

Install Zookeeper and Kafka

Kafka depends on Zookeeper, so you need Zookeeper before installing Kafka. Prepare the following docker-compose.yaml file and replace the host address 192.168.1.100 with your own environment’s address.

version: "3"
services:
  zookeeper:
    image: zookeeper
    build:
      context: ./
    container_name: zookeeper
    ports:
      - 2181:2181
    volumes:
      - ./data/zookeeper/data:/data
      - ./data/zookeeper/datalog:/datalog
      - ./data/zookeeper/logs:/logs
    restart: always
  kafka_node_0:
    depends_on:
      - zookeeper
    build:
      context: ./
    container_name: kafka-node-0
    image: wurstmeister/kafka
    environment:
      KAFKA_BROKER_ID: 0
      KAFKA_ZOOKEEPER_CONNECT: 192.168.1.100:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://192.168.1.100:9092
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092
      KAFKA_NUM_PARTITIONS: 3
      KAFKA_DEFAULT_REPLICATION_FACTOR: 2
    ports:
      - 9092:9092
    volumes:
      - ./data/kafka/node_0:/kafka
    restart: unless-stopped
  kafka_node_1:
    depends_on:
      - kafka_node_0
    build:
      context: ./
    container_name: kafka-node-1
    image: wurstmeister/kafka
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: 192.168.1.100:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://192.168.1.100:9093
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9093
      KAFKA_NUM_PARTITIONS: 3
      KAFKA_DEFAULT_REPLICATION_FACTOR: 2
    ports:
      - 9093:9093
    volumes:
      - ./data/kafka/node_1:/kafka
    restart: unless-stopped
  kafka_node_2:
    depends_on:
      - kafka_node_1
    build:
      context: ./
    container_name: kafka-node-2
    image: wurstmeister/kafka
    environment:
      KAFKA_BROKER_ID: 2
      KAFKA_ZOOKEEPER_CONNECT: 192.168.1.100:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://192.168.1.100:9094
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9094
      KAFKA_NUM_PARTITIONS: 3
      KAFKA_DEFAULT_REPLICATION_FACTOR: 2
    ports:
      - 9094:9094
    volumes:
      - ./data/kafka/node_2:/kafka
    restart: unless-stopped

Run the script with docker-compose up -d to build the cluster. After a short wait, you will see logs indicating successful startup.

SpringBoot Integration with Kafka Cluster

Create a new SpringBoot project and add the following dependencies to build.gradle:

dependencies {
    // other dependencies ...
    implementation 'org.springframework.kafka:spring-kafka:2.5.2.RELEASE'
    implementation 'com.alibaba:fastjson:1.2.71'
}

Configure Kafka parameters in application.properties:

spring.kafka.bootstrap-servers=192.168.1.100:9092,192.168.1.100:9093,192.168.1.100:9094
spring.kafka.producer.retries=0
spring.kafka.producer.batch-size=16384
spring.kafka.producer.buffer-memory=33554432
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.consumer.auto-offset-reset=latest
spring.kafka.consumer.enable-auto-commit=true
spring.kafka.consumer.auto-commit-interval=100

Create a simple message POJO:

public class Message {
    private Long id;
    private String message;
    private Date sendAt;
}

Implement a producer:

public class Sender {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    public void send() {
        Message message = new Message();
        message.setId(System.currentTimeMillis());
        message.setMessage(UUID.randomUUID().toString());
        message.setSendAt(new Date());
        log.info("message = {}", JSON.toJSONString(message));
        kafkaTemplate.send("test", JSON.toJSONString(message));
    }
}

Implement a consumer:

public class Receiver {
    @KafkaListener(topics = {"test"}, groupId = "test")
    public void listen(ConsumerRecord<?,?> record) {
        Optional<?> message = Optional.ofNullable(record.value());
        if (message.isPresent()) {
            log.info("receiver record = " + record);
            log.info("receiver message = " + message.get());
        }
    }
}

Expose a REST endpoint to test the queue:

public class QueueController {
    @Autowired
    private Sender sender;
    @PostMapping("/test")
    public void testQueue() {
        sender.send();
        sender.send();
        sender.send();
    }
}

When you invoke /test, the messages are sent to Kafka and the consumer logs show the received records, confirming successful integration.