Messaging & Streaming

Messaging & Streaming

Kafka Integration

Spring Boot integrates Kafka via spring-kafka, which wraps the native Kafka Producer/Consumer clients with Spring abstractions (KafkaTemplate, @KafkaListener, listener container factories) — same philosophy as JDBC → Spring Data: remove boilerplate around connection/serialization management while leaving the underlying client semantics intact.

spring:
  kafka:
    bootstrap-servers: kafka-broker:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      acks: all                        # wait for all in-sync replicas — strongest durability guarantee
      properties:
        enable.idempotence: true       # see Idempotency section below
        max.in.flight.requests.per.connection: 5
    consumer:
      group-id: order-service
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      auto-offset-reset: earliest
      enable-auto-commit: false        # manual ack — see Error Handling section
    listener:
      ack-mode: manual_immediate

Producing:

@Service
@RequiredArgsConstructor
public class OrderEventPublisher {

    private final KafkaTemplate<String, OrderEvent> kafkaTemplate;

    public void publishOrderCreated(Order order) {
        OrderEvent event = OrderEvent.created(order);
        kafkaTemplate.send("order-events", order.getId().toString(), event)
            .whenComplete((result, ex) -> {
                if (ex != null) {
                    log.error("Failed to publish order event for {}", order.getId(), ex);
                } else {
                    log.debug("Published to partition {} offset {}",
                        result.getRecordMetadata().partition(), result.getRecordMetadata().offset());
                }
            });
    }
}

Key choice matters architecturally: using order.getId() as the message key guarantees all events for that order land on the same partition, preserving per-order ordering — a frequent design requirement (e.g., “created” must always be processed before “shipped” for the same order) that’s easy to overlook until ordering bugs show up downstream.

@KafkaListener

@Component
@RequiredArgsConstructor
public class OrderEventListener {

    @KafkaListener(
        topics = "order-events",
        groupId = "inventory-service",
        containerFactory = "kafkaListenerContainerFactory")
    public void handleOrderEvent(
            @Payload OrderEvent event,
            @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
            @Header(KafkaHeaders.OFFSET) long offset,
            Acknowledgment ack) {

        try {
            inventoryService.processOrderEvent(event);
            ack.acknowledge(); // manual commit — only after successful processing
        } catch (Exception e) {
            log.error("Failed processing offset {} partition {}", offset, partition, e);
            throw e; // let the error handler / retry mechanism decide what happens next
        }
    }
}

Container factory configuration — where concurrency, error handling, and ack mode actually get wired:

@Bean
public ConcurrentKafkaListenerContainerFactory<String, OrderEvent> kafkaListenerContainerFactory(
        ConsumerFactory<String, OrderEvent> consumerFactory,
        DefaultErrorHandler errorHandler) {

    ConcurrentKafkaListenerContainerFactory<String, OrderEvent> factory =
        new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory);
    factory.setConcurrency(3); // 3 consumer threads in this instance — capped by partition count
    factory.setCommonErrorHandler(errorHandler);
    factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
    return factory;
}

Architect-level point — concurrency vs partition count: setConcurrency(3) spins up 3 consumer threads within this single application instance, but actual parallelism is capped by the topic’s partition count — if the topic has only 2 partitions, the 3rd thread sits idle (Kafka assigns at most one consumer per partition per consumer group). Partition count is therefore the real upper bound on consumption parallelism across an entire consumer group, not just within one instance — a frequent point of confusion when teams scale up instance count or thread concurrency without first checking whether the topic has enough partitions to actually use that capacity.

Consumer Groups

A consumer group is Kafka’s mechanism for both load-balancing (partitions spread across group members) and independent consumption (different groups each get their own full copy of the stream, unaffected by other groups’ offsets).

Topic: order-events (4 partitions)

Consumer Group "inventory-service" (3 instances):
  Instance A → partition 0, 1
  Instance B → partition 2
  Instance C → partition 3

Consumer Group "notification-service" (1 instance):
  Instance X → partitions 0, 1, 2, 3  (independent offset tracking from inventory-service)

Key properties to be precise about in an interview:

  • Each partition is consumed by exactly one instance within a given group at a time — never split across two instances of the same group simultaneously.
  • A group can have more instances than partitions (extras sit idle, ready for failover) or fewer instances than partitions (some instances handle multiple partitions) — but never more active consumers-per-partition than 1 within the same group.
  • Rebalancing occurs when group membership changes (instance joins/leaves/crashes) — partitions get reassigned, and depending on the rebalance strategy, this can briefly pause consumption across the whole group (classic “stop-the-world” rebalance) unless using incremental cooperative rebalancing (CooperativeStickyAssignor, default-recommended since Kafka 2.4+) which reassigns only the affected partitions.
spring:
  kafka:
    consumer:
      properties:
        partition.assignment.strategy: org.apache.kafka.clients.consumer.CooperativeStickyAssignor

Architect-level point — scaling guidance frequently asked as a system-design follow-up: the number of partitions set on topic creation is the hard ceiling on horizontal consumer scaling for that topic — repartitioning an existing topic with live data is disruptive (changes key-to-partition mapping, breaking ordering guarantees for existing keys) and generally avoided in production. The practical takeaway for architecture reviews: over-provision partition count modestly at topic creation based on expected peak consumer scale-out, since under-provisioning is expensive to fix later and slight over-provisioning has minimal downside.

Idempotency

Kafka’s at-least-once delivery default means consumers must expect duplicate message delivery — from producer retries, consumer rebalances reprocessing uncommitted offsets, or broker failover. Idempotency is the consumer-side discipline of ensuring reprocessing the same message doesn’t corrupt state.

Producer-side idempotence (prevents duplicate sends from producer retries, not the same as consumer idempotency):

spring:
  kafka:
    producer:
      properties:
        enable.idempotence: true   # broker deduplicates retried sends using a producer ID + sequence number

This solves a narrower problem — a network blip causing the producer to retry a send that actually succeeded — and does not protect against duplicate consumption further downstream, which is the more common real-world concern.

Consumer-side idempotency — the actual architectural responsibility, via a dedup table or natural idempotent operations:

@Service
@RequiredArgsConstructor
public class InventoryService {

    private final ProcessedEventRepository processedEventRepository;

    @Transactional
    public void processOrderEvent(OrderEvent event) {
        if (processedEventRepository.existsByEventId(event.getEventId())) {
            log.info("Duplicate event {} — skipping", event.getEventId());
            return; // already processed, safe no-op
        }
        // business logic
        inventoryRepository.decrementStock(event.getSku(), event.getQuantity());
        processedEventRepository.save(new ProcessedEvent(event.getEventId()));
        // both writes in the SAME transaction — atomic: either both happen or neither does
    }
}

Alternative — naturally idempotent operations avoid needing a dedup table at all: UPDATE inventory SET stock = 100 WHERE sku = ? (absolute set) is idempotent by construction; UPDATE inventory SET stock = stock - 1 WHERE sku = ? (relative decrement) is not — reprocessing the same decrement event twice double-decrements. Designing event payloads/handlers to favor absolute-state operations over relative deltas, where business semantics allow it, sidesteps the dedup-table cost entirely — worth raising as a design-level alternative when this topic comes up, rather than treating the dedup-table pattern as the only answer.

Architect-level point: the dedup-check-and-write must be atomic with the business write (same DB transaction, as shown above) — checking for duplicates and writing business state as two separate non-transactional steps reopens exactly the race condition idempotency was meant to close (two near-simultaneous duplicate deliveries both pass the “not yet processed” check before either commits).

Error Handling & Retries

@Bean
public DefaultErrorHandler errorHandler(KafkaTemplate<String, Object> kafkaTemplate) {

    // Retry 3 times with exponential backoff before giving up
    ExponentialBackOff backOff = new ExponentialBackOff(1000L, 2.0);
    backOff.setMaxAttempts(3);

    DefaultErrorHandler errorHandler = new DefaultErrorHandler(
        new DeadLetterPublishingRecoverer(kafkaTemplate), // after retries exhausted, publish to DLT
        backOff
    );

    // Don't retry on deserialization or business-validation failures — fail fast to DLT immediately
    errorHandler.addNotRetryableExceptions(
        DeserializationException.class,
        OrderValidationException.class
    );

    return errorHandler;
}

Dead Letter Topic (DLT) pattern — DeadLetterPublishingRecoverer automatically republishes the failed record to <original-topic>.DLT after retries are exhausted, preserving the original headers/payload for later inspection or manual reprocessing, rather than silently dropping the message or blocking the partition indefinitely.

# Naming convention spring-kafka uses by default
order-events.DLT   # original topic: order-events

Retryable vs non-retryable — a key design decision per exception type:

Exception type Retry? Reasoning
Transient (DB connection timeout, network blip) Yes Likely to succeed on a subsequent attempt
Deserialization failure (malformed payload) No Will never succeed no matter how many retries — straight to DLT
Business validation failure (e.g., order already cancelled) No Not a transient condition — retrying doesn’t change the outcome
Downstream service temporarily unavailable Yes, often combined with circuit breaker Same dependency-resilience concerns as the Resilience4j section — Kafka retry and circuit breaker patterns frequently get layered together for the actual downstream call inside the listener

Architect-level pitfall — blocking retry inside the listener holds up the partition: retrying synchronously inside @KafkaListener (e.g., a while loop with Thread.sleep) blocks that consumer thread from polling further messages on its assigned partition, and if the retry loop runs long enough, triggers a consumer-group rebalance (the broker assumes the consumer died, since it stopped polling within max.poll.interval.ms) — compounding the original problem with a rebalance storm. The DefaultErrorHandler + ExponentialBackOff configuration shown above is specifically designed to retry between poll cycles correctly, respecting max.poll.interval.ms, rather than blocking inside a single poll — this is the detail that separates “I’ve configured Kafka error handling” from “I’ve configured it correctly under load,” and is worth stating explicitly if asked how you’d implement retry logic for a Kafka consumer.

Manual ack placement, tying back to idempotency: acknowledging after successful processing (as shown in the @KafkaListener example) rather than auto-commit-on-receipt means a crash mid-processing causes the message to be redelivered on restart/rebalance — which is exactly why consumer-side idempotency (previous section) is not optional, it’s the direct consequence of choosing at-least-once delivery semantics over the operational complexity of true exactly-once (Kafka transactions/read_process_write pattern), which most teams reserve for narrower, higher-stakes flows rather than applying everywhere by default.