60-Day Kafka 4 Learning Plan · Week 3 — Day 21 of 60

60-Day Kafka 4 Learning Plan · Week 3 — Spring Boot Integration (Lab) Sources: Kafka: The Definitive Guide Ch.3, 4 & 9 · docs.spring.io/spring-kafka

Goal

Assemble everything from Week 3 into a runnable two-service project: labs-api publishes events to Kafka 4, labs-socket consumes them and pushes to browsers over WebSocket. Run it with Docker Compose and verify end-to-end with a real integration test — including fixing a gap in the naive test setup and hardening the compose file for anything beyond a laptop demo.

Architecture

Browser
  │  WebSocket (STOMP over SockJS)
  ▼
labs-socket :8081
  │  @KafkaListener (ConcurrentKafkaListenerContainerFactory)
  ▼
Kafka (KRaft, single-node)
  │  KafkaTemplate.send()
  ▼
labs-api :8080
  │  POST /api/events
  ▼
Browser (curl / any REST client)

The two services share no code and no database. Their only coupling is the Kafka topic lab-events. You could swap either service independently — Kafka is the contract boundary.

Event flow:

1. Client POSTs  →  labs-api
2. labs-api builds LabEvent, calls kafkaTemplate.send("lab-events", eventId, event)
3. Kafka durably stores the event in a partition
4. labs-socket @KafkaListener polls the partition, deserializes LabEvent
5. labs-socket calls messagingTemplate.convertAndSend("/topic/events", event)
6. Browser receives the WebSocket message, renders it in the live feed

Project Structure

spring-kafka-websocket-pipeline-demo/
├── docker-compose.yml
├── labs-api/
│   ├── Dockerfile
│   ├── pom.xml
│   └── src/main/java/com/boottechsolutions/labsapi/
│       ├── LabsApiApplication.java
│       ├── config/
│       │   ├── KafkaProducerConfig.java
│       │   └── TopicConfig.java
│       ├── controller/EventController.java
│       ├── dto/
│       │   ├── EventRequest.java
│       │   └── EventResponse.java
│       ├── exception/GlobalExceptionHandler.java
│       ├── model/
│       │   ├── EventType.java
│       │   └── LabEvent.java
│       └── service/EventPublisherService.java
├── labs-socket/
│   ├── Dockerfile
│   ├── pom.xml
│   └── src/main/java/com/boottechsolutions/labssocket/
│       ├── LabsSocketApplication.java
│       ├── config/
│       │   ├── KafkaConsumerConfig.java
│       │   └── WebSocketConfig.java
│       ├── consumer/EventConsumer.java
│       ├── model/LabEvent.java
│       └── service/WebSocketBroadcastService.java
└── client/
    └── index.html

Step 1 — Kafka with Docker Compose

Kafka 4 runs in KRaft mode — no ZooKeeper required. A single-node setup is sufficient for development.

services:
  kafka:
    image: apache/kafka:4.0.0
    container_name: kafka
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"
      CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk
    ports:
      - "9092:9092"
    healthcheck:
      test: ["CMD-SHELL", "/opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --list"]
      interval: 15s
      timeout: 10s
      retries: 10
      start_period: 30s

  labs-api:
    build:
      context: ./labs-api
    ports:
      - "8080:8080"
    environment:
      SPRING_KAFKA_BOOTSTRAP_SERVERS: kafka:9092
    depends_on:
      kafka:
        condition: service_healthy

  labs-socket:
    build:
      context: ./labs-socket
    ports:
      - "8081:8081"
    environment:
      SPRING_KAFKA_BOOTSTRAP_SERVERS: kafka:9092
    depends_on:
      kafka:
        condition: service_healthy

service_healthy ensures both Spring Boot services only start after Kafka’s health check passes. Without this, both services will fail their initial broker connection and the auto-reconnect will eventually succeed — but you’ll see noisy error logs during startup.

CLUSTER_ID is a 22-character base64 UUID required by KRaft to initialize the metadata log. You can generate a fresh one with /opt/kafka/bin/kafka-storage.sh random-uuid inside the container, or use the hardcoded value above for deterministic local environments.

Step 2 — Spring Kafka 3.x Autoconfiguration

When spring-kafka is on the classpath, Spring Boot’s KafkaAutoConfiguration registers:

These beans read their configuration from spring.kafka.* properties in application.yml.

When you override: defining your own ProducerFactory bean suppresses the auto-configured one. This lets you control serializer types, idempotence settings, and batching at the Java level without spring.kafka.producer.properties.* workarounds.

In labs-api, we define ProducerFactory<String, LabEvent> explicitly. Spring Boot’s auto-configured ProducerFactory<Object, Object> is NOT created. The only KafkaTemplate in labs-api is KafkaTemplate<String, LabEvent>.

In labs-socket, we define only a custom ConsumerFactory and kafkaListenerContainerFactory. The auto-configured KafkaTemplate<Object, Object> IS created (no conflict) and is injected into KafkaConsumerConfig for DLT publishing.

Step 3 — labs-api: The Producer

Dependencies

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-validation</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

Domain model

public enum EventType {
    USER_REGISTERED, ORDER_CREATED, PAYMENT_PROCESSED,
    INVENTORY_UPDATED, NOTIFICATION_SENT
}

public record LabEvent(
    String eventId,
    EventType type,
    String payload,
    Instant timestamp,
    String source
) {}

Topic management

KafkaAdmin (auto-configured) creates topics declared as NewTopic beans on startup:

@Configuration
public class TopicConfig {

    @Value("${kafka.topic.lab-events}")
    private String labEventsTopic;

    @Bean
    public NewTopic labEventsTopic() {
        return TopicBuilder.name(labEventsTopic)
                .partitions(3)
                .replicas(1)
                .build();
    }

    @Bean
    public NewTopic labEventsDltTopic() {
        return TopicBuilder.name(labEventsTopic + ".DLT")
                .partitions(1)
                .replicas(1)
                .build();
    }
}

Three partitions on lab-events allows three consumer threads in labs-socket to run in parallel. The DLT has one partition — throughput requirements are lower, since it only receives failed records.

In production, manage topics via Terraform or kafka-topics.sh in CI/CD pipelines. Application-managed topic creation creates implicit dependencies between startup order and broker state.

Producer configuration: acks, retries, idempotence

@Configuration
public class KafkaProducerConfig {

    @Bean
    public ProducerFactory<String, LabEvent> producerFactory(KafkaProperties properties) {
        Map<String, Object> props = new HashMap<>(properties.buildProducerProperties(null));

        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);

        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        props.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);
        props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);

        props.put(ProducerConfig.LINGER_MS_CONFIG, 5);
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);

        props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 120_000);
        props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 30_000);

        props.put(JsonSerializer.ADD_TYPE_INFO_HEADERS, false);

        return new DefaultKafkaProducerFactory<>(props);
    }

    @Bean
    public KafkaTemplate<String, LabEvent> kafkaTemplate(ProducerFactory<String, LabEvent> producerFactory) {
        return new KafkaTemplate<>(producerFactory);
    }
}

acks=all: The broker returns success only after the partition leader AND all in-sync replicas (ISR) have written the record. On a single-node broker (as in this demo), the ISR is just the leader, so acks=all and acks=1 behave identically. On a three-broker cluster, acks=all guarantees the record survives a single broker failure.

enable.idempotence=true: The producer is assigned a ProducerID (PID) by the broker and attaches an incrementing sequence number to each record per partition. On retry, the broker detects the duplicate (PID, sequence) pair and discards the re-send silently. This gives exactly-once delivery at the producer side — no duplicates even under network retries. Requires: acks=all, max.in.flight.requests.per.connection ≤ 5.

retries=Integer.MAX_VALUE with delivery.timeout.ms=120_000: Retry indefinitely until the overall delivery timeout (120 seconds) is reached. This is the recommended pattern for idempotent producers — the timeout, not the retry count, bounds how long the producer tries. Setting retries to a small number (e.g., 3) can cause premature failure when the broker is temporarily unavailable during a rolling restart.

linger.ms=5: The producer waits up to 5ms for additional records before sending a batch. If your workload sends one event per REST request and each event is time-sensitive, set this to 0 to send immediately. For bulk ingestion pipelines, increase to 50–100ms for higher throughput.

ADD_TYPE_INFO_HEADERS=false: Prevents spring-kafka from embedding the Java class name in a __TypeId__ Kafka header. Without this, consumers are forced to have the same class name on the classpath, coupling two independent services. The consumer uses VALUE_DEFAULT_TYPE instead to determine the deserialization target class.

Service and controller

@Service
@RequiredArgsConstructor
@Slf4j
public class EventPublisherService {

    private final KafkaTemplate<String, LabEvent> kafkaTemplate;

    @Value("${kafka.topic.lab-events}")
    private String topic;

    public String publish(EventRequest request) {
        String eventId = UUID.randomUUID().toString();

        LabEvent event = new LabEvent(
                eventId, request.type(), request.payload(),
                Instant.now(), request.source() != null ? request.source() : "labs-api");

        CompletableFuture<SendResult<String, LabEvent>> future =
                kafkaTemplate.send(topic, eventId, event);

        future.whenComplete((result, ex) -> {
            if (ex != null) {
                log.error("Failed to publish: eventId={}, error={}", eventId, ex.getMessage());
            } else {
                log.info("Published: eventId={}, partition={}, offset={}",
                        eventId,
                        result.getRecordMetadata().partition(),
                        result.getRecordMetadata().offset());
            }
        });

        return eventId;
    }
}

kafkaTemplate.send() returns a CompletableFuture. The REST endpoint returns 202 Accepted immediately — it does not block waiting for Kafka acknowledgment. The whenComplete callback logs the result asynchronously. If you need the caller to know whether the event was durably stored before returning, .get() on the future and return 200 OK on success.

@RestController
@RequestMapping("/api/events")
@RequiredArgsConstructor
public class EventController {

    private final EventPublisherService publisherService;

    @PostMapping
    public ResponseEntity<EventResponse> publish(@Valid @RequestBody EventRequest request) {
        String eventId = publisherService.publish(request);
        return ResponseEntity.accepted()
                .body(new EventResponse(eventId, "ACCEPTED", "Event queued for processing"));
    }
}

Step 4 — labs-socket: Consumer + WebSocket

Dependencies

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-websocket</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

spring-boot-starter-websocket includes spring-messaging, spring-websocket, and the embedded Tomcat WebSocket support. STOMP runs over the WebSocket transport.

Event model in labs-socket

public record LabEvent(
    String eventId,
    String type,      // String, not EventType enum
    String payload,
    Instant timestamp,
    String source
) {}

type is String here, not EventType. If labs-api adds a new EventType variant, labs-socket continues to deserialize and forward it without a redeployment. Declaring type as the EventType enum would throw a JsonMappingException on unknown values — a deserialization error the retry loop cannot resolve (it goes straight to DLT). Model the consumer side loosely; validate schema on the producer side.

Consumer configuration: concurrency, offsets, batch

@Configuration
@EnableKafka
public class KafkaConsumerConfig {

    @Bean
    public ConsumerFactory<String, LabEvent> consumerFactory(KafkaProperties properties) {
        Map<String, Object> props = new HashMap<>(properties.buildConsumerProperties(null));

        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 50);
        props.put(JsonDeserializer.TRUSTED_PACKAGES, "com.boottechsolutions.*");
        props.put(JsonDeserializer.VALUE_DEFAULT_TYPE, LabEvent.class.getName());
        props.put(JsonDeserializer.USE_TYPE_INFO_HEADERS, false);

        return new DefaultKafkaConsumerFactory<>(
                props, new StringDeserializer(), new JsonDeserializer<>(LabEvent.class, false));
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, LabEvent> kafkaListenerContainerFactory(
            ConsumerFactory<String, LabEvent> consumerFactory,
            KafkaTemplate<Object, Object> kafkaTemplate) {

        var factory = new ConcurrentKafkaListenerContainerFactory<String, LabEvent>();
        factory.setConsumerFactory(consumerFactory);
        factory.setConcurrency(3);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);

        var recoverer = new DeadLetterPublishingRecoverer(kafkaTemplate);
        var errorHandler = new DefaultErrorHandler(recoverer, new FixedBackOff(1_000L, 3L));
        errorHandler.addNotRetryableExceptions(IllegalArgumentException.class);
        factory.setCommonErrorHandler(errorHandler);

        return factory;
    }
}

setConcurrency(3): Creates 3 KafkaConsumer instances, each running in its own thread. Kafka assigns partitions to consumers in a group — with 3 partitions and concurrency 3, each thread owns one partition. If concurrency exceeds the partition count, the extra threads sit idle. If partition count exceeds concurrency, threads handle multiple partitions each.

AckMode.MANUAL_IMMEDIATE: The listener calls ack.acknowledge() explicitly after successful processing. The offset is committed immediately (not batched). This prevents a crash between poll and processing from losing the message — if the app dies before acknowledge(), the uncommitted offset is re-delivered to the next consumer in the group.

MAX_POLL_RECORDS_CONFIG=50: Each poll() call returns at most 50 records. Tune this together with max.poll.interval.ms (default 5 minutes). If processing 50 records takes longer than max.poll.interval.ms, the broker considers the consumer dead and triggers a rebalance.

Batch mode (alternative): To process multiple records in a single listener invocation, add factory.setBatchListener(true) and change the listener signature:

@KafkaListener(topics = "${kafka.topic.lab-events}", ...)
public void consumeBatch(List<ConsumerRecord<String, LabEvent>> records, Acknowledgment ack) {
    records.forEach(r -> broadcastService.broadcast(r.value()));
    ack.acknowledge();  // commits all offsets in the batch in one call
}

Batch mode reduces per-record overhead and is appropriate when the processing cost per record is low (e.g., a simple in-memory transform). For operations with per-record side effects (WebSocket push, database write), single-record mode gives finer-grained error handling.

Step 5 — Error Handling: DLT and Retry Templates

DefaultErrorHandler

Spring Kafka 2.8 introduced DefaultErrorHandler as a unified replacement for both SeekToCurrentErrorHandler (record-level) and SeekToCurrentBatchErrorHandler (batch-level). It works for both modes.

var recoverer = new DeadLetterPublishingRecoverer(kafkaTemplate);
var errorHandler = new DefaultErrorHandler(recoverer, new FixedBackOff(1_000L, 3L));
errorHandler.addNotRetryableExceptions(IllegalArgumentException.class);
factory.setCommonErrorHandler(errorHandler);

FixedBackOff(1_000L, 3L): Wait 1 second between retries, attempt at most 3 retries. Total handling time for a persistently failing record: 3 seconds. After the third retry, DeadLetterPublishingRecoverer is invoked.

addNotRetryableExceptions: Some failures are structural — retrying won’t help. IllegalArgumentException indicates bad input; DeserializationException (already non-retryable by default) means the bytes cannot be parsed as LabEvent. Both are forwarded to the DLT immediately without retries.

DeadLetterPublishingRecoverer: Publishes the failed record to lab-events.DLT and then commits the original offset. This is critical — the consumer does NOT stall at the bad record forever. Processing continues with the next record. The DLT record includes exception headers for debugging:

kafka_dlt-exception-fqcn       → com.example.SomeException
kafka_dlt-exception-message    → error detail
kafka_dlt-original-topic       → lab-events
kafka_dlt-original-partition   → 2
kafka_dlt-original-offset      → 147

Error handling sequence:

Record arrives at @KafkaListener
     │
     ├── Processing succeeds → ack.acknowledge() → offset committed
     │
     └── Processing fails
           │
           ├── Retry 1 (1s wait) → fail
           ├── Retry 2 (1s wait) → fail
           ├── Retry 3 (1s wait) → fail
           │
           └── DeadLetterPublishingRecoverer
                 ├── Publishes to lab-events.DLT
                 └── Commits original offset (consumer moves on)

ExponentialBackOff for production: FixedBackOff hammers the downstream service at a constant rate during degradation. Use ExponentialBackOff to progressively back off:

var backOff = new ExponentialBackOff(1_000L, 2.0);
backOff.setMaxInterval(30_000L);  // cap at 30s
backOff.setMaxElapsedTime(120_000L);  // stop after 2 minutes total
var errorHandler = new DefaultErrorHandler(recoverer, backOff);

Step 6 — WebSocket Configuration

@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {

    @Override
    public void configureMessageBroker(MessageBrokerRegistry config) {
        config.enableSimpleBroker("/topic");
        config.setApplicationDestinationPrefixes("/app");
    }

    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        registry.addEndpoint("/ws")
                .setAllowedOriginPatterns("*")
                .withSockJS();
    }
}

enableSimpleBroker("/topic") registers an in-process message broker that delivers messages to subscribers of /topic/* destinations. SimpMessagingTemplate.convertAndSend("/topic/events", event) serializes event to JSON and delivers it to every WebSocket client subscribed to /topic/events.

withSockJS(): SockJS provides a WebSocket emulation layer. If the browser or network does not support WebSocket, SockJS falls back to HTTP long-polling or server-sent events automatically. Remove it if you only target modern browsers and modern networks.

Multi-node limitation: enableSimpleBroker is in-process. If you run multiple instances of labs-socket, a browser connected to instance A will not receive messages published by instance B. To fix this, replace the simple broker with a full STOMP relay:

config.enableStompBrokerRelay("/topic")
      .setRelayHost("rabbitmq")
      .setRelayPort(61613);

This forwards all /topic/* traffic through RabbitMQ’s STOMP plugin, which fan-outs to all connected instances.

Broadcast service

@Service
@RequiredArgsConstructor
@Slf4j
public class WebSocketBroadcastService {

    private static final String EVENT_TOPIC = "/topic/events";
    private final SimpMessagingTemplate messagingTemplate;

    public void broadcast(LabEvent event) {
        log.debug("Broadcasting: id={}, type={}", event.eventId(), event.type());
        messagingTemplate.convertAndSend(EVENT_TOPIC, event);
    }
}

Consumer (wiring it all together)

@Component
@RequiredArgsConstructor
@Slf4j
public class EventConsumer {

    private final WebSocketBroadcastService broadcastService;

    @KafkaListener(
            topics = "${kafka.topic.lab-events}",
            groupId = "${spring.kafka.consumer.group-id}",
            containerFactory = "kafkaListenerContainerFactory"
    )
    public void consume(ConsumerRecord<String, LabEvent> record, Acknowledgment ack) {
        log.info("Received: key={}, partition={}, offset={}, type={}",
                record.key(), record.partition(), record.offset(),
                record.value() != null ? record.value().type() : "null");

        broadcastService.broadcast(record.value());
        ack.acknowledge();
    }
}

If broadcast() throws, the exception propagates out of consume(). DefaultErrorHandler intercepts it, retries, and eventually routes to the DLT. ack.acknowledge() is only reached on the success path.

Step 7 — JavaScript Client

client/index.html uses SockJS and the @stomp/stompjs client library to connect to labs-socket and subscribe to /topic/events:

const client = new StompJs.Client({
    webSocketFactory: () => new SockJS('http://localhost:8081/ws'),
    reconnectDelay: 5000,
    onConnect: () => {
        client.subscribe('/topic/events', msg => {
            const event = JSON.parse(msg.body);
            addEvent(event);
        });
    }
});
client.activate();

The reconnectDelay: 5000 means the client automatically reconnects after a 5-second gap if labs-socket restarts. Events published while the client is disconnected will not be re-delivered (WebSocket is a push channel, not a durable queue). If you need gap-fill on reconnect, store events in Redis or a database and expose a REST API for the client to fetch missed events on reconnect.

Open client/index.html directly in a browser — it has no build step. For production, serve it from a CDN or your frontend build pipeline, and change the WebSocket URL to your load-balanced labs-socket endpoint.

Step 8 — Running the Pipeline

cd spring-kafka-websocket-pipeline-demo
docker compose up --build

Wait for all three services to report healthy. Then:

Publish an event:

Open the live feed:

Open client/index.html in a browser. Events appear within milliseconds of publishing.

Publish several types to see the colored feed:

for TYPE in USER_REGISTERED ORDER_CREATED PAYMENT_PROCESSED INVENTORY_UPDATED NOTIFICATION_SENT; do
  curl -s -X POST http://localhost:8080/api/events \
    -H "Content-Type: application/json" \
    -d "{\"type\":\"$TYPE\",\"payload\":\"Demo event for $TYPE\"}" > /dev/null
  sleep 0.5
done

Verify Kafka delivery in labs-api logs:

labs-api     |  INFO 1 --- [labs-api] [-api-producer-1] o.a.k.c.p.internals.TransactionManager   : [Producer clientId=labs-api-producer-1] ProducerId set to 0 with epoch 0
labs-api     |  INFO 1 --- [labs-api] [-api-producer-1] c.b.l.service.EventPublisherService      : Published: eventId=664e2366-4f3c-4b0b-95d5-54d24cd60d0e, topic=lab-events, partition=0, offset=0
labs-api     |  INFO 1 --- [labs-api] [-api-producer-1] c.b.l.service.EventPublisherService      : Published: eventId=b8f11705-520f-44ed-9919-8055771f36a1, topic=lab-events, partition=0, offset=1
labs-api     |  INFO 1 --- [labs-api] [-api-producer-1] c.b.l.service.EventPublisherService      : Published: eventId=ccbcbaa4-2dd1-4693-aed7-83024d8a6415, topic=lab-events, partition=1, offset=0
labs-api     |  INFO 1 --- [labs-api] [-api-producer-1] c.b.l.service.EventPublisherService      : Published: eventId=f0274ead-ac1c-4f48-8759-21d0f76f12bd, topic=lab-events, partition=2, offset=0
labs-api     |  INFO 1 --- [labs-api] [-api-producer-1] c.b.l.service.EventPublisherService      : Published: eventId=bd7e7e66-4eb0-4daf-96e8-0e9b94e87c94, topic=lab-events, partition=1, offset=1
labs-api     |  INFO 1 --- [labs-api] [-api-producer-1] c.b.l.service.EventPublisherService      : Published: eventId=5305ba3a-9cea-4fed-a5cb-bd46a6986270, topic=lab-events, partition=2, offset=1

Verify consumption in labs-socket logs:

labs-socket  | INFO 1 --- [labs-socket] [ntainer#0-0-C-1] c.b.labssocket.consumer.EventConsumer    : Received: key=664e2366-4f3c-4b0b-95d5-54d24cd60d0e, partition=0, offset=0, type=ORDER_CREATED
labs-socket  | INFO 1 --- [labs-socket] [ntainer#0-0-C-1] c.b.labssocket.consumer.EventConsumer    : Received: key=b8f11705-520f-44ed-9919-8055771f36a1, partition=0, offset=1, type=USER_REGISTERED
labs-socket  | INFO 1 --- [labs-socket] [ntainer#0-1-C-1] c.b.labssocket.consumer.EventConsumer    : Received: key=ccbcbaa4-2dd1-4693-aed7-83024d8a6415, partition=1, offset=0, type=ORDER_CREATED
labs-socket  | INFO 1 --- [labs-socket] [ntainer#0-2-C-1] c.b.labssocket.consumer.EventConsumer    : Received: key=f0274ead-ac1c-4f48-8759-21d0f76f12bd, partition=2, offset=0, type=PAYMENT_PROCESSED
labs-socket  | INFO 1 --- [labs-socket] [ntainer#0-1-C-1] c.b.labssocket.consumer.EventConsumer    : Received: key=bd7e7e66-4eb0-4daf-96e8-0e9b94e87c94, partition=1, offset=1, type=INVENTORY_UPDATED
labs-socket  | INFO 1 --- [labs-socket] [ntainer#0-2-C-1] c.b.labssocket.consumer.EventConsumer    : Received: key=5305ba3a-9cea-4fed-a5cb-bd46a6986270, partition=2, offset=1, type=NOTIFICATION_SENT

Consume from DLT directly (after a deliberate failure — e.g., stopping labs-socket mid-retry):

docker exec kafka /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic lab-events.DLT \
  --from-beginning \
  --property print.headers=true

Testing

Integration tests use Spring Kafka’s @EmbeddedKafka — an in-process broker with no Docker dependency.

labs-api: producer integration test

@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
@EmbeddedKafka(
        partitions = 3,
        topics = {"lab-events", "lab-events.DLT"},
        bootstrapServersProperty = "spring.kafka.bootstrap-servers"
)
@DirtiesContext
class EventControllerIntegrationTest {

    @Autowired
    private TestRestTemplate restTemplate;

    @Test
    void publishEvent_returns_accepted_with_event_id() {
        var request = new EventRequest(EventType.ORDER_CREATED, "order payload", "test");
        var response = restTemplate.postForEntity("/api/events", request, EventResponse.class);

        assertThat(response.getStatusCode()).isEqualTo(HttpStatus.ACCEPTED);
        assertThat(response.getBody().status()).isEqualTo("ACCEPTED");
        assertThat(response.getBody().eventId()).isNotBlank();
    }
}

labs-socket: consumer integration test

@SpringBootTest
@EmbeddedKafka(
        partitions = 3,
        topics = {"lab-events", "lab-events.DLT"},
        bootstrapServersProperty = "spring.kafka.bootstrap-servers"
)
@DirtiesContext
class EventConsumerIntegrationTest {

    @Autowired
    private KafkaTemplate<Object, Object> kafkaTemplate;

    @MockitoBean
    private WebSocketBroadcastService broadcastService;

    @Test
    void consume_calls_broadcast_on_incoming_event() {
        LabEvent event = new LabEvent("evt-001", "ORDER_CREATED", "test", Instant.now(), "test");
        kafkaTemplate.send("lab-events", event.eventId(), event);

        verify(broadcastService, timeout(5_000).atLeastOnce()).broadcast(any(LabEvent.class));
    }
}

@MockitoBean (Spring Boot 3.4+) replaces WebSocketBroadcastService in the application context with a Mockito mock. Mockito’s timeout(5_000) waits up to 5 seconds for the async consumer thread to invoke broadcast(). This is the cleanest way to test the consumer without sleeping or using CountDownLatch.

bootstrapServersProperty = "spring.kafka.bootstrap-servers" ensures @EmbeddedKafka writes the embedded broker’s address into the same property that KafkaProperties reads, so consumerFactory and kafkaListenerContainerFactory pick up the embedded broker automatically.

Performance Considerations

WebSocket broadcast is synchronous in SimpMessagingTemplate. If convertAndSend() blocks — for example, because a slow subscriber causes the server-side message queue to fill — the Kafka consumer thread is blocked too. Bounded WebSocket session queues (configured via WebSocketMessageBrokerStats or the STOMP relay) prevent a lagging browser from slowing down Kafka consumption.

Consumer lag. Three consumer threads with one partition each means each thread’s max throughput is roughly max.poll.records × processing_time_per_record. If the producer publishes faster than consumers can broadcast, lag accumulates. Monitor with:

docker exec kafka /opt/kafka/bin/kafka-consumer-groups.sh \
  --bootstrap-server localhost:9092 \
  --describe --group labs-socket-group

A growing LAG column means labs-socket is falling behind. Solutions: increase concurrency (and add partitions), optimize broadcast(), or add a second labs-socket instance (add a STOMP broker relay first).

MAX_POLL_RECORDS and max.poll.interval.ms. The default max.poll.interval.ms is 5 minutes. If processing 50 records takes longer than 5 minutes (unlikely here, but common in database write workloads), the broker marks the consumer dead and triggers a rebalance. Either reduce MAX_POLL_RECORDS or increase max.poll.interval.ms.

DLT monitoring. A growing DLT consumer offset without a corresponding consumer means failed events accumulate silently. Set up an alert on lab-events.DLT consumer lag (or produce rate). A dead-letter queue with no consumer is a data graveyard.

Common Pitfalls

Not disabling type headers on the producer. By default, JsonSerializer adds a __TypeId__ header containing the producer’s fully-qualified class name. If labs-socket doesn’t have com.boottechsolutions.labsapi.model.LabEvent on its classpath (it doesn’t — these are separate services), JsonDeserializer throws a DeserializationException on every message. Set ADD_TYPE_INFO_HEADERS=false on the producer and USE_TYPE_INFO_HEADERS=false + VALUE_DEFAULT_TYPE on the consumer.

Auto-commit with manual ack mode. Setting AckMode.MANUAL_IMMEDIATE but forgetting to set enable.auto.commit=false creates a conflict: auto-commit may commit offsets before acknowledge() is called. Always set both.

Concurrency without matching partitions. setConcurrency(3) with a single-partition topic means only one thread works; the other two idle indefinitely. Partition count must be >= concurrency for full utilization.

Blocking the consumer thread in consume(). If broadcastService.broadcast() makes an HTTP call, waits for a database write, or sleeps, the Kafka consumer thread is blocked for that duration. Kafka has no independent heartbeat thread — if the consumer doesn’t call poll() within max.poll.interval.ms, the broker triggers a rebalance. For slow downstream operations, process records in a thread pool and use AckMode.MANUAL (batched commit) to commit when the async work is done.

No DLT consumer. Publishing to a DLT is only useful if someone reads it. Wire a monitoring consumer that alerts on new DLT records, or set up Kafka Connect to stream the DLT into a database or alerting system.

withSockJS() and HTTP/2. SockJS’s XHR polling transport sends many rapid HTTP requests. If your reverse proxy terminates HTTP/2, configure it to allow many concurrent streams or use WebSocket-only mode (remove withSockJS()) on HTTP/2-capable networks.

Key Takeaways

The producer’s acks=all + enable.idempotence=true combination gives exactly-once delivery at the Kafka layer — retries on network failures do not duplicate messages
Manual offset commit (AckMode.MANUAL_IMMEDIATE) combined with DefaultErrorHandler + DeadLetterPublishingRecoverer ensures no event is silently lost: it either reaches the WebSocket client or ends up in the DLT with full exception metadata
enableSimpleBroker is single-instance only — replace it with a STOMP broker relay (RabbitMQ or ActiveMQ) before scaling labs-socket to more than one replica

Week 3 recap

The complete source code is available on GitHub.

Support me through GitHub Sponsors.

Thank you for Reading !! See you in the next story.

➡️ Week 4 · Day 22: JSON vs Avro — why binary serialization matters

References

📘 Kafka: The Definitive Guide — Chapter 3, 4 & 9
🌐 docs.spring.io/spring-kafka/reference
🌐 Docker Compose — healthcheck & depends_on conditions
Spring Kafka Reference Documentation
Spring WebSocket Reference Documentation
Apache Kafka Producer Configuration
Apache Kafka Consumer Configuration
Spring Boot Auto-configuration for Kafka
STOMP over WebSocket
SockJS Protocol
Apache Kafka Docker Image

👉 Link to Medium blog

60-Day Kafka 4 Learning Plan · Week 3 — Day 21 of 60

Goal

Architecture

Project Structure

Step 1 — Kafka with Docker Compose

Step 2 — Spring Kafka 3.x Autoconfiguration

Step 3 — labs-api: The Producer

Dependencies

Domain model

Topic management

Producer configuration: acks, retries, idempotence

Service and controller

Step 4 — labs-socket: Consumer + WebSocket

Dependencies

Event model in labs-socket

Consumer configuration: concurrency, offsets, batch

Step 5 — Error Handling: DLT and Retry Templates

DefaultErrorHandler

Step 6 — WebSocket Configuration

Broadcast service

Consumer (wiring it all together)

Step 7 — JavaScript Client

Step 8 — Running the Pipeline

Testing

labs-api: producer integration test

labs-socket: consumer integration test

Performance Considerations

Common Pitfalls

Key Takeaways

Week 3 recap

Next

References

Related Posts

60-Day Kafka 4 Learning Plan · Week 7 — Day 46 of 60

60-Day Kafka 4 Learning Plan · Week 7 — Day 45 of 60

Subscribe to my Newsletter