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

---

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

Goal

Wire two Spring Boot microservices together via Kafka 4: labs-api publishes domain events, labs-socket consumes them and pushes to connected browsers over WebSocket — zero direct HTTP coupling between the services — and understand what actually breaks when labs-socket needs to scale to multiple instances.

1. End-to-end architecture

Client
  → POST /api/events
  → labs-api (KafkaTemplate.send)
  → Kafka 4 [labs.events topic]
  → labs-socket (@KafkaListener)
  → SimpMessagingTemplate
  → Browser (WebSocket /topic/user/{userId})

labs-api owns event creation. labs-socket owns real-time delivery. Kafka decouples them completely — neither service knows the other exists.

2. Topic design

Why key by userId? All events for a given user land on the same partition, guaranteeing ordered delivery per user. If ordering doesn’t matter, use a random or round-robin key instead.

Why 6 partitions? Allows up to 6 concurrent consumer threads per group. Start with 2–3× your expected concurrency and scale partitions later (note: you can only add, never remove).

3. labs-api — event producer

// EventProducer.java (labs-api)
@Service
@RequiredArgsConstructor
public class EventProducer {

    private final KafkaTemplate<String, LabsEvent> kafkaTemplate;

    public void publish(LabsEvent event) {
        kafkaTemplate.send(
            "labs.events",
            event.getUserId(),   // partition key → same partition per user
            event
        ).whenComplete((result, ex) -> {
            if (ex != null) {
                log.error("Failed to publish event {}", event.getId(), ex);
            }
        });
    }
}

// EventController.java (labs-api)
@RestController
@RequiredArgsConstructor
@RequestMapping("/api/events")
public class EventController {

    private final EventProducer producer;

    @PostMapping
    public ResponseEntity<Void> create(@RequestBody LabsEvent event) {
        producer.publish(event);
        return ResponseEntity.accepted().build();
    }
}

4. labs-socket — Kafka consumer → WebSocket push

// EventSocketListener.java (labs-socket)
@Component
@RequiredArgsConstructor
@Slf4j
public class EventSocketListener {

    private final SimpMessagingTemplate ws;

    @KafkaListener(
        topics = "labs.events",
        groupId = "socket-service",
        concurrency = "3"           // 3 threads × 2 partitions each = 6 partitions
    )
    public void onEvent(ConsumerRecord<String, LabsEvent> rec) {
        String destination = "/topic/user/" + rec.key();
        ws.convertAndSend(destination, rec.value());
        log.info("Pushed event to WS {}", destination);
    }
}

// WebSocketConfig.java (labs-socket)
@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {

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

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

Client-side subscription (JavaScript):

const client = new StompJs.Client({ brokerURL: 'ws://localhost:8081/ws' });
client.onConnect = () => {
    client.subscribe(`/topic/user/${userId}`, (msg) => {
        console.log('Event received:', JSON.parse(msg.body));
    });
};
client.activate();

5. Shared Kafka config (both services)

# application.yml (shared base — both services)
spring:
  kafka:
    bootstrap-servers: kafka-1:9092,kafka-2:9092,kafka-3:9092
    producer:
      acks: all
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
    consumer:
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring.json.trusted.packages: "com.labs.events"
        auto.offset.reset: earliest

# application.yml (labs-api only)
spring:
  kafka:
    consumer:
      group-id: api-service

# application.yml (labs-socket only)
spring:
  kafka:
    consumer:
      group-id: socket-service

6. LabsEvent model (shared library or duplicated DTO)

@Data
@NoArgsConstructor
@AllArgsConstructor
public class LabsEvent {
    private String id;
    private String userId;
    private String type;       // e.g. "ORDER_CREATED", "PAYMENT_CONFIRMED"
    private Object payload;
    private Instant timestamp;
}

Consider extracting to a shared Maven module (labs-events-model) so both services use the same class. For production, use Avro + Schema Registry (Day 22–28).

7. The multi-instance WebSocket problem — the gap this lab hides

The setup in §4 works perfectly with one labs-socket instance. It silently breaks the moment you scale to two or more.

Why: registry.enableSimpleBroker("/topic") is an in-memory, per-JVM broker. If a user’s browser is connected via WebSocket to labs-socket instance A, but the Kafka partition holding that user’s events gets consumed by instance B (a completely normal outcome of consumer group partition assignment), instance B’s SimpMessagingTemplate.convertAndSend() has no way to reach a client connected to instance A. The event is consumed successfully, logged as “pushed,” and the browser never receives it.

User connects WebSocket → Load balancer → labs-socket instance A (holds the connection)

Kafka rebalances → user's partition now assigned to → labs-socket instance B

Instance B consumes the event → calls ws.convertAndSend() → sends into ITS OWN
in-memory broker → no client connected to instance B for this user → message silently droppedKafka rebalances → user's partition now assigned to → labs-socket instance B

The fix — external STOMP relay broker: replace the simple in-memory broker with a real message broker (RabbitMQ’s STOMP plugin is the most common pairing) that all labs-socket instances connect to. Now any instance can publish to /topic/user/{userId} and the relay broker fans it out to whichever instance actually holds that user’s WebSocket connection.

@Override
public void configureMessageBroker(MessageBrokerRegistry registry) {
    registry.enableStompBrokerRelay("/topic")
        .setRelayHost("rabbitmq-host")
        .setRelayPort(61613)          // STOMP port
        .setClientLogin("guest")
        .setClientPasscode("guest");
    registry.setApplicationDestinationPrefixes("/app");
}

This is not optional for production. A single-instance labs-socket is fine for a demo or low-traffic internal tool, but any deployment expecting horizontal scaling or high availability needs the relay broker from day one — retrofitting it after users start reporting “I randomly don’t get notifications” is a much worse position to debug from.

8. Securing the WebSocket connection

The lab’s /ws endpoint and /topic/user/{userId} destination as shown have no authentication — any connected client can subscribe to any userId‘s topic and read another user’s events.

@Override
public void registerStompEndpoints(StompEndpointRegistry registry) {
    registry.addEndpoint("/ws")
        .setHandshakeHandler(new DefaultHandshakeHandler() {
            @Override
            protected Principal determineUser(ServerHttpRequest request, WebSocketHandler wsHandler,
                                               Map<String, Object> attributes) {
                String token = extractBearerToken(request);
                return jwtService.validateAndGetPrincipal(token); // rejects handshake if invalid
            }
        })
        .withSockJS();
}

// Route to the authenticated user's own destination, not a client-supplied userId
@KafkaListener(topics = "labs.events", groupId = "socket-service")
public void onEvent(ConsumerRecord<String, LabsEvent> rec) {
    // convertAndSendToUser resolves to the authenticated Principal's session(s) only
    ws.convertAndSendToUser(rec.key(), "/queue/events", rec.value());
}

Prefer convertAndSendToUser over a client-guessable /topic/user/{userId} path — with proper authentication wired in, Spring routes the message only to sessions belonging to that authenticated principal, removing the need to trust that clients only subscribe to their own topic.

9. Error handling — WS push failures & dead sessions

ws.convertAndSend() can fail (relay broker down, serialization issue) or succeed while the underlying session is actually already dead (client disconnected but cleanup hasn’t run yet). Neither failure mode is visible from the @KafkaListener method alone unless you handle it explicitly.

@KafkaListener(topics = "labs.events", groupId = "socket-service")
public void onEvent(ConsumerRecord<String, LabsEvent> rec) {
    try {
        ws.convertAndSendToUser(rec.key(), "/queue/events", rec.value());
    } catch (MessageDeliveryException e) {
        log.warn("WS push failed for user={}, will not block Kafka offset commit", rec.key(), e);
        // Decide deliberately: is a missed real-time push acceptable (fire-and-forget UX),
        // or does this user need a fallback (store-and-forward, push notification)?
    }
}

Design decision to make explicitly, not by default: should a WebSocket push failure cause the Kafka offset to NOT commit (triggering redelivery)? For most real-time UX (a live notification), the answer is no — the event already happened, redelivering it later doesn’t restore “real-time,” so log-and-continue is usually right. Store the missed event elsewhere (DB, or a fallback push channel) instead of treating it as a retriable Kafka failure per Day 18’s guidance.

10. End-to-end monitoring

@KafkaListener(topics = "labs.events", groupId = "socket-service")
public void onEvent(ConsumerRecord<String, LabsEvent> rec) {
    long endToEndMs = System.currentTimeMillis() - rec.value().getTimestamp().toEpochMilli();
    meterRegistry.timer("labs.event.end_to_end_latency").record(endToEndMs, TimeUnit.MILLISECONDS);
    ws.convertAndSendToUser(rec.key(), "/queue/events", rec.value());
}

Why this metric matters more than any single-hop metric: producer latency and consumer lag can both look perfectly healthy while end-to-end latency quietly degrades — e.g. an overloaded relay broker (§7) adds delay after Kafka’s part of the pipeline is already done. Only an explicit end-to-end measurement catches that.

11. Testing the full pipeline

@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
@EmbeddedKafka(partitions = 6, topics = "labs.events")
class EventPipelineIntegrationTest {

    @LocalServerPort int port;
    @Autowired KafkaTemplate<String, LabsEvent> kafkaTemplate;

    @Test
    void eventPublishedToKafkaArrivesOverWebSocket() throws Exception {
        var stompClient = new WebSocketStompClient(new StandardWebSocketClient());
        var session = stompClient.connectAsync("ws://localhost:" + port + "/ws", new StompSessionHandlerAdapter() {})
            .get(5, TimeUnit.SECONDS);

        CompletableFuture<LabsEvent> received = new CompletableFuture<>();
        session.subscribe("/topic/user/u1", new StompFrameHandlerAdapter(LabsEvent.class, received::complete));

        kafkaTemplate.send("labs.events", "u1", new LabsEvent("e1", "u1", "ORDER_CREATED", null, Instant.now()));

        LabsEvent event = received.get(10, TimeUnit.SECONDS);
        assertThat(event.getUserId()).isEqualTo("u1");
    }
}

This test only proves the single-instance happy path. It cannot catch the §7 multi-instance gap by construction — testing that requires either running two actual service instances behind a relay broker in an integration environment, or trusting the architectural fix (§7) rather than expecting a unit/integration test to surface it.

12. Common pitfalls

• Deploying labs-socket with more than one instance without a relay broker — the single biggest gap in this lab; see §7, it fails silently rather than loudly
• Trusting a client-supplied userId in the WebSocket subscription path — always resolve destination from the authenticated principal, not a value the client controls (§8)
• Treating a WS push failure as a Kafka-retriable error by default — usually the wrong call for real-time UX; make the decision deliberately (§9)
• Measuring only Kafka-side metrics (producer latency, consumer lag) and assuming that covers the pipeline — the consume-to-WS-delivery hop is invisible unless explicitly instrumented (§10)
• Forgetting concurrency must not exceed partition count here either — same rule as Day 6/17, still applies to socket-service‘s listener

Key Takeaways

• Kafka decouples labs-api (producer) from labs-socket (consumer) — no direct HTTP call
• Key by userId: all events for a user route to the same partition → ordered delivery per user
• concurrency=3: 3 consumer threads, each handling 2 of 6 partitions
• SimpMessagingTemplate‘s default simple broker is per-instance — scaling labs-socket horizontally requires an external STOMP relay broker or messages silently fail to reach clients connected to a different instance
• Secure WebSocket subscriptions with authentication and convertAndSendToUser — don’t trust a client-supplied userId path
• A WS push failure shouldn’t automatically block the Kafka offset commit — decide the fallback strategy deliberately
• Instrument end-to-end latency explicitly — Kafka-only metrics don’t cover the consume-to-delivery hop
• Multiple consumer groups: socket-service and analytics-service each get all events independently
• Adding a new consumer (e.g. push-service) requires zero changes to labs-api

---

Support me through GitHub Sponsors.

Next

➡️ Day 21: Microservice project — end-to-end async pipeline

Resources

• 📘 Kafka: The Definitive Guide — Chapter 3 & 4 (Producers & Consumers)
• 🌐 docs.spring.io/spring-kafka/reference
• 🌐 docs.spring.io/spring-framework — WebSocket
• 🌐 Spring — STOMP broker relay

👉 Link to Medium blog