Apache Kafka with Spring Boot

Kafka is the standard choice for event streaming between services. This page shows how Spring Boot connects to Kafka, how the producer and consumer work internally, and how to build reliable, retrying, idempotent consumers with a dead letter topic.

Kafka in 60 seconds

Producer ──► Topic "orders" (3 partitions, replicated over brokers) ──► Consumer group "payment-service"
                P0 [0][1][2]                                                      consumer A ◄─ P0, P1
                P1 [0][1]                                                         consumer B ◄─ P2
                P2 [0][1][2][3]
Records with the same KEY go to the same partition (order is kept). A partition is read by ONE consumer per group.
Each group keeps its own OFFSET per partition (stored in the topic __consumer_offsets).

Dependency and configuration

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>
<!-- Boot 4 also offers spring-boot-starter-kafka -->

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      acks: all
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      properties:
        enable.idempotence: true
        linger.ms: 10
        compression.type: lz4
    consumer:
      group-id: payment-service
      auto-offset-reset: earliest
      enable-auto-commit: false
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring.json.trusted.packages: com.example.events
        spring.json.value.default.type: com.example.events.OrderPlaced
    listener:
      ack-mode: record
      concurrency: 3

Internal workflow of sending

kafkaTemplate.send("orders", key, event)
   ├─ 1. serialize key and value (JsonSerializer)
   ├─ 2. choose the partition: hash(key) % numberOfPartitions
   ├─ 3. add the record to a per-partition BATCH in the RecordAccumulator (memory buffer)
   ├─ 4. Sender thread (background) sends the batch when it is full (batch.size) or linger.ms passed
   ├─ 5. leader broker appends it to the log, followers replicate it
   ├─ 6. leader replies after acks condition (acks=all: all in-sync replicas)
   └─ 7. the CompletableFuture completes: metadata (partition, offset) or exception (retried automatically)

Producer code

public record OrderPlaced(String orderId, String customerId, BigDecimal total) { }

@Configuration
public class KafkaTopics {
    @Bean
    NewTopic ordersTopic() {
        return TopicBuilder.name("orders").partitions(3).replicas(1).build();     // replicas(3) in production
    }
}

@Service
public class OrderEventPublisher {

    private static final Logger log = LoggerFactory.getLogger(OrderEventPublisher.class);
    private final KafkaTemplate<String, OrderPlaced> kafka;

    public OrderEventPublisher(KafkaTemplate<String, OrderPlaced> kafka) { this.kafka = kafka; }

    public void publish(OrderPlaced event) {
        kafka.send("orders", event.customerId(), event)          // key = customerId keeps one customer's events in order
             .whenComplete((result, ex) -> {
                 if (ex != null) log.error("Send failed for order {}", event.orderId(), ex);
                 else log.info("Sent to partition {} offset {}", result.getRecordMetadata().partition(),
                               result.getRecordMetadata().offset());
             });
    }
}

Consumer code

@Component
public class PaymentListener {

    private static final Logger log = LoggerFactory.getLogger(PaymentListener.class);
    private final PaymentService payments;

    public PaymentListener(PaymentService payments) { this.payments = payments; }

    @KafkaListener(topics = "orders", groupId = "payment-service")
    public void onOrder(ConsumerRecord<String, OrderPlaced> record) {
        OrderPlaced event = record.value();
        log.info("Received {} from partition {} offset {}", event.orderId(), record.partition(), record.offset());
        payments.charge(event);                                     // must be idempotent
    }
}

Internal workflow of consuming

ConcurrentMessageListenerContainer (concurrency = 3)  ─► 3 KafkaMessageListenerContainer threads
   each thread runs a loop:
     1. consumer.poll(timeout)  ─► fetches a batch of records from assigned partitions
     2. for each record: deserialize ─► call your @KafkaListener method
     3. if the method returns normally  ─► commit the offset (by ack-mode: record, batch, manual ...)
     4. if it throws ─► DefaultErrorHandler: retry with back off ─► after the last retry: Dead Letter Topic
   Group coordinator (a broker): heartbeats, partition assignment, rebalance when consumers join or leave

Errors, retries and dead letter topic

@Configuration
public class KafkaErrorConfig {

    @Bean
    DefaultErrorHandler errorHandler(KafkaTemplate<Object, Object> template) {
        // failed records go to "<topic>.DLT"
        DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template);
        DefaultErrorHandler handler = new DefaultErrorHandler(recoverer, new FixedBackOff(1000L, 3L));   // 3 retries, 1 s apart
        handler.addNotRetryableExceptions(IllegalArgumentException.class);                                // bad data will never succeed
        return handler;
    }
}

@Component
class DeadLetterListener {
    @KafkaListener(topics = "orders.DLT", groupId = "dlt-monitor")
    void onDead(ConsumerRecord<String, String> record) {
        // alert, store for manual review, or replay later
    }
}

Alternative: @RetryableTopic creates retry topics (orders-retry-0, orders-retry-1) with delays, so a failing message does not block the partition. This is called non-blocking retry.

@RetryableTopic(attempts = "4", backoff = @Backoff(delay = 2000, multiplier = 2.0), dltStrategy = DltStrategy.FAIL_ON_ERROR)
@KafkaListener(topics = "orders", groupId = "payment-service")
public void onOrder(OrderPlaced event) { payments.charge(event); }

@DltHandler
public void dlt(OrderPlaced event) { /* last chance handling */ }

Delivery guarantees you must design for

@Transactional
public void charge(OrderPlaced event) {
    if (processedRepo.existsById(event.orderId())) return;         // duplicate: ignore
    // ... business work ...
    processedRepo.save(new ProcessedEvent(event.orderId()));        // same transaction as the work
}

Testing

@SpringBootTest
@Testcontainers
class OrderFlowIT {
    @Container
    @ServiceConnection
    static KafkaContainer kafka = new KafkaContainer(DockerImageName.parse("apache/kafka:4.3.1"));

    @Autowired OrderEventPublisher publisher;

    @Test
    void publishesAndConsumes() {
        publisher.publish(new OrderPlaced("o-1", "c-1", BigDecimal.TEN));
        // await() from Awaitility: wait until the listener side effect is visible
    }
}

Useful commands

bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic orders
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group payment-service   # shows LAG
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders --from-beginning --property print.key=true

Common mistake: Assuming a message is processed exactly once. With retries, rebalances and restarts, duplicates will happen. Design consumers to be idempotent, and use an outbox for reliable publishing after database changes.

Interview questions

How many consumers can read one topic in parallel?

As many as the number of partitions inside one consumer group. More consumers than partitions stay idle.

What causes a rebalance?

A consumer joins or leaves, a consumer misses the heartbeat or max.poll.interval.ms (processing too slow), or partitions change. During a rebalance consumption pauses for the group.

Why is my consumer lag growing?

The consumer processes slower than the producer writes. Increase consumers (up to the partition count), speed up processing, or use batch listeners. Check for slow downstream calls.