Kafka with Spring

Spring Kafka wraps the client in the shapes a Spring service already uses: a template to send, an annotation to listen, a container that owns the threads, and an error handler that decides what a failure means. Every setting from the previous lessons has a property here, and most of the defaults are the client's defaults, which means a @KafkaListener with no configuration is at-least-once with auto-commit semantics you may not want. This lesson is the producer side, the listener and its container, concurrency, error handling, and the tests that prove any of it.

#Producing with KafkaTemplate

@Service
public class OrderEvents {
    private final KafkaTemplate<String, OrderPlaced> template;
    OrderEvents(KafkaTemplate<String, OrderPlaced> template) { this.template = template; }
 
    public CompletableFuture<SendResult<String, OrderPlaced>> placed(Order o) {
        return template.send("orders", o.getId().toString(), OrderPlaced.from(o))
            .whenComplete((result, ex) -> {
                if (ex != null) metrics.counter("kafka.send.failed", "topic", "orders").increment();
            });
    }
}

send() returns a CompletableFuture (Spring Kafka 3; earlier versions returned a ListenableFuture), which is the callback from the Producers lesson in a Java shape. The key is the order id, so all of an order's events share a partition. Boot builds the KafkaTemplate from spring.kafka.producer.* properties, and the settings that matter are the ones from that lesson:

spring.kafka:
  bootstrap-servers: kafka-1:9092,kafka-2:9092,kafka-3:9092
  producer:
    acks: all
    properties:
      enable.idempotence: true
      linger.ms: 10
      compression.type: lz4
      delivery.timeout.ms: 120000

A send inside a @Transactional database method is the outbox problem from the Producers lesson: the database can commit and the send can fail, or the reverse. template.executeInTransaction and @Transactional with a KafkaTransactionManager exist for Kafka-only transactions; for Kafka plus a database the answer is the outbox, or at minimum @TransactionalEventListener(phase = AFTER_COMMIT) so the send happens only after the commit and a failure is at least visible.

#Listening with @KafkaListener

@Component
public class OrderProjection {
    @KafkaListener(topics = "orders", groupId = "order-projection", concurrency = "3")
    public void on(ConsumerRecord<String, OrderPlaced> record, Acknowledgment ack) {
        projections.upsert(record.key(), record.value());   // idempotent: an upsert keyed on the order
        ack.acknowledge();                                   // commit after the effect
    }
}

The annotation registers a listener container, which owns the KafkaConsumer and the thread it polls on, dispatches each record to the method, and commits according to its ack mode. The container's defaults are what decide the delivery guarantee:

spring.kafka:
  consumer:
    group-id: order-projection
    auto-offset-reset: earliest
    enable-auto-commit: false           # the container commits, not the client
    max-poll-records: 200
    properties:
      partition.assignment.strategy: org.apache.kafka.clients.consumer.CooperativeStickyAssignor
      max.poll.interval.ms: 300000
      session.timeout.ms: 45000
  listener:
    ack-mode: manual_immediate          # commit when acknowledge() is called

The default BATCH with the client's auto-commit off is already at-least-once: the container commits after the whole poll's records have returned from the method, so a crash mid-batch replays the batch. MANUAL_IMMEDIATE gives the method the decision and is what a handler with conditional processing wants. Either way the handler is idempotent, as the Offsets lesson insists.

Batch listeners (@KafkaListener(batch = "true") with a List<ConsumerRecord<...>> parameter) process a poll at a time, which is right for bulk inserts and wrong for anything where one bad record should not fail the batch.

#Concurrency

concurrency = "3" runs three KafkaConsumer instances on three threads inside this process, each a member of the group, each owning a share of the partitions. Three instances of the service with concurrency 3 is nine members; the topic needs at least nine partitions for all of them to work. KafkaConsumer is single-threaded by contract, so the container never shares one across threads; the listener method runs on the consumer's thread, and everything it calls (JPA, RestClient, the outbox) runs there too. A slow call in the method is a slow poll loop, and the max.poll.interval.ms eviction from the Consumer groups lesson is one slow database away.

With spring.threads.virtual.enabled=true on Java 21, the container's consumer threads become virtual threads. The poll loop is still one loop per consumer; what changes is that hundreds of listeners cost nothing to keep, which matters for a service with many topics, not for one hot topic.

#Error handling

The container's CommonErrorHandler decides what an exception from the listener means. Boot's default is a DefaultErrorHandler with no recoverer: it retries a record in place ten times with no backoff and then logs it and skips it, which is a silent data loss dressed as robustness. Replace it on day one:

@Bean
DefaultErrorHandler errorHandler(KafkaOperations<Object, Object> template) {
    var handler = new DefaultErrorHandler(
        new DeadLetterPublishingRecoverer(template),             // exhausted → <topic>.DLT with headers
        new ExponentialBackOffWithMaxRetries(3) {{ setInitialInterval(1_000); setMultiplier(3); }});
    handler.addNotRetryableExceptions(ValidationException.class, DeserializationException.class);
    return handler;
}

Boot picks up a CommonErrorHandler bean automatically. For non-blocking retries, @RetryableTopic on the listener as the Retries lesson showed; the two are alternatives per listener, not layers. And ErrorHandlingDeserializer on every consumer, always, or a poison pill takes the container into a restart loop before any of this runs.

#Testing

Two levels, and both are worth having.

The unit level does not need Kafka: the listener method is a method, and a test calls it with a ConsumerRecord and a stub Acknowledgment. Idempotency is tested here by calling it twice.

The container level runs a real broker. @EmbeddedKafka starts one in the JVM, quickly, with the real client protocol; it is enough for "the listener receives what the template sends". A Testcontainers KafkaContainer (or ConfluentKafkaContainer) starts the real broker image, which is what production runs, and with Spring Boot's @ServiceConnection the bootstrap servers are wired in automatically:

@SpringBootTest
@Testcontainers
class OrderFlowTest {
    @Container @ServiceConnection
    static KafkaContainer kafka = new KafkaContainer("apache/kafka:3.9.0");
 
    @Autowired KafkaTemplate<String, OrderPlaced> template;
    @Autowired ProjectionRepository projections;
 
    @Test void anOrderPlacedEventIsProjected() {
        template.send("orders", "42", new OrderPlaced("42", 1999)).join();
        Awaitility.await().atMost(Duration.ofSeconds(10))
            .untilAsserted(() -> assertThat(projections.findById("42")).isPresent());
    }
}

The assertion waits, because delivery is asynchronous; Awaitility is the honest way to write that. Test the failure paths too: send a record that fails validation and assert it appears on the DLT with the exception header. Those tests are the ones that catch the default error handler skipping records.

#Under the hood: the container's loop, and what concurrency really creates

@KafkaListener is processed by KafkaListenerAnnotationBeanPostProcessor, which builds a MethodKafkaListenerEndpoint per annotated method and registers it with a ConcurrentKafkaListenerContainerFactory. The factory creates a ConcurrentMessageListenerContainer, which is a holder for concurrency KafkaMessageListenerContainer children, each owning one KafkaConsumer and one thread from the container's TaskExecutor (a SimpleAsyncTaskExecutor by default, so a new platform thread, or a virtual one with the Boot property). Each child runs ListenerConsumer.run(): poll() with pollTimeout (5 s), then either hand the whole ConsumerRecords to a batch listener or iterate records to a single-record listener, invoking your method through a RecordMessagingMessageListenerAdapter that resolves the parameters (ConsumerRecord, the payload converted by the RecordMessageConverter, @Headers, Acknowledgment), then commit per the ack mode, then check for pause/resume/seek requests and rebalance callbacks. Everything your method calls runs on that thread, inside that loop, and the loop cannot poll again until you return.

Ack modes are commit timing on top of the client with enable.auto.commit=false (the container forces it off and logs a warning if you set it). BATCH calls commitSync (or async, per syncCommits) after the last record of the poll returns; RECORD after each record; MANUAL records the acknowledgement and commits with the next poll's batch; MANUAL_IMMEDIATE commits at the acknowledge() call. Acknowledgment.nack(sleep) is the escape hatch that seeks back and pauses, the same mechanism the error handler uses. The container also installs a ConsumerRebalanceListener that commits pending offsets on onPartitionsRevoked, which is why a clean rebalance does not replay a finished batch under BATCH mode; a crash still does.

KafkaTemplate wraps one Producer from a DefaultKafkaProducerFactory, which caches a single shared producer for the non-transactional case (thread-safe, so every @Service shares it) and a pool keyed by transactional.id for the transactional case. With Spring's transaction machinery, @Transactional on a method with a KafkaTransactionManager binds a transactional producer to the thread for the method's duration; the container can also start a Kafka transaction per poll (transactionManager on the container) so that offsets are sent with sendOffsetsToTransaction and outputs commit atomically, which is the consume-transform-produce shape from the Offsets lesson, in Spring form. A JPA @Transactional and a Kafka transaction do not compose atomically; ChainedKafkaTransactionManager was deprecated for exactly that reason, and the outbox remains the answer.

#Walkthrough: the listener that stopped and reported healthy

A projection service's Kafka listener stopped consuming one Tuesday; the pod was healthy, the readiness probe green, and lag climbed for three hours before a dashboard was noticed.

@KafkaListener(topics = "orders", groupId = "projection", concurrency = "2")
public void on(OrderPlaced e) {
    var enriched = catalog.lookup(e.sku());          // RestClient, no timeout
    projections.upsert(e.orderId(), enriched);
}

1. The catalogue service hung on one request: TCP accepted, no response ever. RestClient had no read timeout. The listener thread blocked forever inside catalog.lookup, inside the container loop, holding one of the two consumers.
2. The consumer's heartbeat thread kept heartbeating, so the group saw the member as alive. After max.poll.interval.ms (5 min) the heartbeat thread sent LeaveGroup; the partitions moved to the other consumer and to other pods. The stuck thread stayed stuck, and the pod's other consumer picked up extra partitions and kept working, more slowly.
3. Nothing was unhealthy by any probe's definition: the process was up, the database reachable, and Boot's Kafka health indicator (which checks the cluster, not the listeners) reported UP. records.lag.max for the stuck listener was not on a dashboard.
4. The fix was three things. A RestClient with connectTimeout and readTimeout of 2 s. A listener-level watchdog: Spring Kafka's container publishes ListenerContainerIdleEvent (after idleEventInterval) and NonResponsiveConsumerEvent (when poll() has not returned within noPollThreshold × pollTimeout); an ApplicationListener on the second one now marks a custom health indicator DOWN, which flips readiness and pages. And the kafka.consumer.fetch.manager.records.lag.max metric, per listener id, went on the dashboard with an alert.
5. The thread dump from the incident showed the listener thread in SocketInputStream.socketRead0 under RestClient, which is the shape of every "consumer stopped" that is not a rebalance: something the handler called did not come back.

The container cannot know your handler is stuck; it only knows poll() has not been called. The process's health and the listener's health are different questions, and only the second one tells you records are moving.

#Try it yourself

#Misconceptions

• "concurrency adds threads to one consumer." It creates that many consumers, each single-threaded, each a group member; the partition count bounds how many do work.
• "The listener runs asynchronously from polling." It runs inside the poll loop on the consumer's thread; a blocked handler is a consumer that stops polling and is eventually evicted.
• "Boot's Kafka health indicator covers the listeners." It checks the cluster connection. Listener health is NonResponsiveConsumerEvent and lag metrics, which you wire yourself.
• "@Transactional on a listener makes the database write and the offset commit atomic." The offset commit is Kafka's and the write is JPA's; they are two systems. Idempotent handlers or an outbox close the gap.
• "Embedded Kafka tests prove production behaviour." They prove the protocol; Testcontainers with the production image proves the broker's configuration and version.

#Going deeper

• Spring for Apache Kafka reference: "Message Listener Containers" (ack modes, ContainerProperties), "Transactions", "Testing".
• KafkaMessageListenerContainer.ListenerConsumer in the spring-kafka source: the loop is one method, and reading it settles most questions.
• Spring Boot reference, "Messaging → Apache Kafka", for every spring.kafka.* property and the auto-configured beans.
• Gary Russell's answers on Stack Overflow under the spring-kafka tag, which document behaviour the reference does not.
• Testcontainers, KafkaContainer / ConfluentKafkaContainer documentation and Boot's @ServiceConnection support.