Spring Kafka Consumer Error Handling, Concurrency and Rebalancing

How listener concurrency, poll timing and error handling fit together

This guide is for engineers running Spring Kafka listeners in production who need to keep throughput, ordering and failure handling under control at the same time.

A ConcurrentMessageListenerContainer with concurrency N creates N child containers. Each one owns a separate KafkaConsumer on a dedicated thread, because KafkaConsumer is not thread-safe. The group coordinator assigns each partition to at most one consumer in the group. With 6 partitions and concurrency 8, two threads receive no partitions and do no work. Partition count therefore sets the ceiling on parallelism.

Adding partitions later raises that ceiling, but it has a cost. The default partitioner maps a key to a partition by hashing it modulo the partition count. When the count changes, many existing keys move to a different partition. During the transition, new records for a key can be consumed before older records still waiting on the old partition, so per-key ordering is lost.

Copy topic: orders (P0..P5)
 P0 P1 ──► thread-0: KafkaConsumer ─┐
 P2 P3 ──► thread-1: KafkaConsumer ─┼─ each thread loops:
 P4 P5 ──► thread-2: KafkaConsumer ─┘   poll() ──► listener(record) ──► [error handler:
                                          ▲          retry + backoff]
                                          └──── next poll() only after the batch finishes

The time between two polls on one thread is roughly max.poll.records × per-record latency, plus any blocking retry backoff. Blocking backoff runs on the consumer thread. If that gap exceeds max.poll.interval.ms, the client leaves the group and its partitions are reassigned.

Illustrative estimate. Assume the following inputs:

• max.poll.records = 500
• 150 ms per record
• max.poll.interval.ms = 300 s
• one failing record retried 6 times with a 30 s backoff

Processing the batch takes 500 × 0.15 s = 75 s. The retries add 6 × 30 s = 180 s. The total is 255 s, which fits under the 300 s limit. A second failing record in the same batch adds another 180 s, bringing the total to 435 s. That exceeds the limit, so the consumer is evicted and a rebalance is triggered.

Copy topic "orders"                  ConcurrentMessageListenerContainer
 (6 partitions)                  concurrency = 8
 +------+                        +--------------------------------------+
 |  P0  |----------------------->| thread-0  KafkaConsumer              |
 |  P1  |----------------------->| thread-1  KafkaConsumer              |
 |  P2  |----------------------->| thread-2  KafkaConsumer              |
 |  P3  |----------------------->| thread-3  KafkaConsumer              |
 |  P4  |----------------------->| thread-4  KafkaConsumer              |
 |  P5  |----------------------->| thread-5  KafkaConsumer              |
 +------+                        | thread-6  KafkaConsumer   (idle)     |
                                 | thread-7  KafkaConsumer   (idle)     |
                                 +--------------------------------------+

  each thread runs its own copy of this loop:
  +-----------+     +------------------------+     +----------------------+
  |  poll()   |---->| listener(record)       |---->| error handler        |
  | returns a |     | repeated for each      | on  | retry + backoff      |
  | batch     |     | record in the batch    |error| (blocks the thread)  |
  +-----------+     +------------------------+     +----------------------+
        ^                                                     |
        +----- next poll() only after the batch finishes -----+

  gap between polls ~ max.poll.records x per-record latency + backoff
  gap > max.poll.interval.ms  ==>  member leaves group, partitions move

Each of the eight container threads owns one KafkaConsumer, but only six partitions exist, so two threads sit idle, and on every active thread the listener and the error handler's blocking retries must finish before the next poll() can run within max.poll.interval.ms.

What happens to offsets during consumer rebalancing

A rebalance happens when a member joins, leaves or is evicted. The group coordinator then reassigns partitions, and each new owner resumes from the partition's last committed offset. It does not know which records the previous owner actually processed. Any record that was handled but not yet committed is therefore delivered again.

Spring's listener container disables Kafka's enable.auto.commit by default and commits according to its ack mode (BATCH by default, or RECORD, MANUAL and others). Kafka's own auto-commit runs on a timer inside poll(), independent of whether your listener succeeded.

With the classic eager protocol, every member revokes all of its partitions, so consumption across the whole group pauses. Cooperative incremental assignment (CooperativeStickyAssignor) moves only the partitions that need to change owner. Switching to it requires a staged rolling migration with both assignors configured during the transition, and it should be tested on its own before production.

A ConsumerRebalanceListener (or Spring's ConsumerAwareRebalanceListener) can commit offsets or flush local state in onPartitionsRevoked. This narrows the duplicate window when a partition changes owner.

To reduce evictions, you can raise max.poll.interval.ms, but a longer limit also lets a stuck consumer hold its partitions longer before the group notices. Lowering max.poll.records shortens each poll loop, at the cost of more fetch round trips and lower throughput.

Copy@Configuration
public class KafkaConsumerConfig {
  // Next offset to commit (last processed + 1) per partition; the @KafkaListener
  // updates it after each record, then calls ack.acknowledge() only periodically.
  static final Map<TopicPartition, OffsetAndMetadata> processed = new ConcurrentHashMap<>();

  @Bean
  ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(
      ConsumerFactory<String, String> consumerFactory) {
    var factory = new ConcurrentKafkaListenerContainerFactory<String, String>();
    factory.setConsumerFactory(consumerFactory);
    ContainerProperties props = factory.getContainerProperties();
    props.setAckMode(ContainerProperties.AckMode.MANUAL); // app decides when to acknowledge

    props.setConsumerRebalanceListener(new ConsumerAwareRebalanceListener() {
      @Override
      public void onPartitionsRevokedBeforeCommit(Consumer<?, ?> consumer,
                                                  Collection<TopicPartition> partitions) {
        // We still own these partitions: commit work done since the last ack
        // so the next owner does not reprocess it.
        Map<TopicPartition, OffsetAndMetadata> toCommit = new HashMap<>();
        for (TopicPartition tp : partitions) {
          OffsetAndMetadata next = processed.remove(tp);
          if (next != null) toCommit.put(tp, next);
        }
        if (!toCommit.isEmpty()) consumer.commitSync(toCommit);
      }

      @Override
      public void onPartitionsLost(Consumer<?, ?> consumer,
                                   Collection<TopicPartition> partitions) {
        // Ownership is already gone (e.g. evicted after max.poll.interval.ms):
        // a commit would fail, so only discard local state; duplicates follow.
        partitions.forEach(processed::remove);
      }
    });
    return factory;
  }
}

A ConsumerAwareRebalanceListener that commits processed-but-unacknowledged offsets when partitions are revoked and only drops local state when partitions are lost, narrowing the duplicate window after a rebalance.

How DefaultErrorHandler retries and recovers failed records

DefaultErrorHandler treats a listener exception as a request to try again. It consults a BackOff to decide whether another attempt is allowed and how long to wait. It then seeks the consumer back so the failed record and any unprocessed records from the same poll are redelivered. When the BackOff reports that attempts are exhausted, the handler passes the record to a ConsumerRecordRecoverer, commits past it, and moves on. With no explicit configuration the recoverer only logs, so the record is silently skipped.

Record listeners fit this model directly. Batch listeners are less precise. If the listener throws BatchListenerFailedException with the index of the bad record, the handler commits the records before it and retries from that point. Any other exception means the handler cannot tell which record failed, so it retries the entire batch and, on exhaustion, recovers every record in it.

FixedBackOff gives predictable waits. ExponentialBackOff spaces attempts out for transient faults such as a database failover. Either way, the waits run on the consumer thread, so budget them against max.poll.interval.ms. Illustrative estimate: initial 1 s, multiplier 2, five retries gives 1 + 2 + 4 + 8 + 16 = 31 s. Add the processing time for the rest of the poll and keep the total under the configured limit with clear headroom.

While the thread waits, nothing else on that partition moves. Ordering is preserved, but every later record is stalled behind the failing one. Keep in-thread retries short, and push waits measured in minutes to a retry topic.

Copy@Configuration
public class KafkaErrorHandlingConfig {

    @Bean
    public DefaultErrorHandler errorHandler(KafkaTemplate<Object, Object> template) {
        // After retries are exhausted, publish to <topic>-dlt instead of the default log-only recoverer
        DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template);

        // 5 retries with waits of 1s, 2s, 4s, 8s, 16s = 31s of blocking per failed record
        // (illustrative estimate). Keep this plus the processing time for the rest of the poll
        // well under max.poll.interval.ms (default 300000 ms).
        ExponentialBackOffWithMaxRetries backOff = new ExponentialBackOffWithMaxRetries(5);
        backOff.setInitialInterval(1_000L);
        backOff.setMultiplier(2.0);
        backOff.setMaxInterval(16_000L); // cap any single wait on the consumer thread

        DefaultErrorHandler handler = new DefaultErrorHandler(recoverer, backOff);

        // Retrying cannot fix bad input, so send these straight to the recoverer
        handler.addNotRetryableExceptions(DeserializationException.class, IllegalArgumentException.class);
        return handler;
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(
            ConsumerFactory<String, String> consumerFactory, DefaultErrorHandler errorHandler) {
        ConcurrentKafkaListenerContainerFactory<String, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.setCommonErrorHandler(errorHandler);
        return factory;
    }
}

// In a batch listener, report the failing index so earlier records are committed
// and only the failed record and later ones are retried:
//   throw new BatchListenerFailedException("bad record", e, index);

Configures a DefaultErrorHandler with a capped exponential backoff of 31 seconds total and a dead-letter recoverer, then registers it on the listener container factory.

Classifying exceptions and stopping poison pills

Retries only help when the next attempt can succeed. A record that fails bean validation, carries an unknown enum value, or cannot be parsed will fail the same way on every attempt. Retrying it burns the full backoff budget and blocks the partition the whole time. DefaultErrorHandler lets you mark such exceptions as not retryable with addNotRetryableExceptions(...). A matching failure skips the backoff and goes straight to the recoverer. The handler already treats several framework exceptions this way, including DeserializationException, MessageConversionException and ClassCastException. Add your own domain types alongside them, such as a ValidationException.

Misclassification costs you in both directions. If you leave too much retryable, you pay for errors that are deterministic: every bad record holds the partition for the full backoff sequence. If you leave too little retryable, a thirty-second database failover sends a batch of perfectly valid records to the dead-letter topic, and someone has to replay them by hand. When the set of transient failures is small and well understood, invert the default. Call defaultFalse() and then list only the retryable exceptions with addRetryableExceptions(...), for example connection and timeout exceptions.

Deserialization failures need separate handling because they happen inside poll(), before the listener runs. With a plain deserializer, the consumer throws, never advances past the bad offset, and fetches the same record again on every poll. That is the poison-pill loop. ErrorHandlingDeserializer wraps your real deserializer, which you configure through the spring.deserializer.value.delegate.class property. It catches the failure and passes the record along with a null value and the exception stored in a header. The container detects that header and sends the failure to the error handler. Because DeserializationException is not retryable, the record goes straight to the recoverer and the partition keeps moving.

Copy@Configuration
public class OrderConsumerConfig {

  @Bean
  public ConsumerFactory<String, Order> consumerFactory() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-1.internal:9092");
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-service");
    // Kafka instantiates the wrapper. A parse failure becomes a null value plus
    // an exception header, so poll() advances instead of looping on the bad offset.
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
    // The real deserializers, read through the spring.deserializer.*.delegate.class properties
    props.put(ErrorHandlingDeserializer.KEY_DESERIALIZER_CLASS, StringDeserializer.class);
    props.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, JsonDeserializer.class);
    props.put(JsonDeserializer.TRUSTED_PACKAGES, "com.example.orders");
    return new DefaultKafkaConsumerFactory<>(props);
  }

  @Bean
  public DefaultErrorHandler errorHandler(KafkaTemplate<Object, Object> template) {
    // Publishes failed records to <topic>-dlt. The template must be able to write the
    // raw byte[] payloads that deserialization failures carry.
    var recoverer = new DeadLetterPublishingRecoverer(template);
    // Transient failures get 3 retries, 2 seconds apart, before recovery
    var handler = new DefaultErrorHandler(recoverer, new FixedBackOff(2000L, 3L));
    // Deterministic failures skip the backoff and go straight to the recoverer.
    // DeserializationException is already non-retryable by default.
    handler.addNotRetryableExceptions(
        jakarta.validation.ValidationException.class,
        IllegalArgumentException.class);
    // Inverted alternative, for when the transient set is small and well understood:
    // handler.defaultFalse();
    // handler.addRetryableExceptions(java.net.SocketTimeoutException.class,
    //     org.springframework.dao.TransientDataAccessException.class);
    return handler;
  }
}

This Spring Kafka configuration (imports omitted) wraps the real deserializers in ErrorHandlingDeserializer so a bad payload cannot cause a poison-pill loop, and it sends validation errors straight to the dead-letter recoverer while transient failures are retried with backoff.

Sending failures to a dead-letter topic with DeadLetterPublishingRecoverer

A recoverer is any function that takes the failed record and the exception. DeadLetterPublishingRecoverer is the stock implementation. You construct it with a KafkaTemplate, then pass it to the DefaultErrorHandler constructor alongside the BackOff. Once retries are exhausted, or immediately for non-retryable exceptions, the handler calls the recoverer and commits the offset so the partition moves on.

Naming and partition mapping

By default the recoverer publishes to a topic named after the source topic with a .DLT suffix, so orders fails over to orders.DLT. It also writes to the same partition number as the original record. This preserves per-key grouping and makes it easy to correlate failures with source partitions. It also means the DLT must have at least as many partitions as the source topic. If it has fewer, publishing to a nonexistent partition fails, and the record is not recovered. If you need different routing, supply a destination resolver that returns a custom TopicPartition.

Diagnostic headers

Each dead-lettered record carries headers, defined as constants on KafkaHeaders. They hold the exception class name, the message and the stack trace. They also record the original topic, partition, offset, timestamp and consumer group. Together these let you trace a failure to its exact source record and decide whether replaying it is safe.

Ownership

A DLT is a holding area, not a fix. Assign an owning team, alert when the DLT receives records or its depth grows, and keep a documented replay path that republishes corrected records to the source topic. Without these, failures accumulate silently, retention eventually deletes them, and the data is lost.

Alternatives: blocking retries, non-blocking retry topics and idempotent handlers

Blocking retries keep the work on the listener thread: DefaultErrorHandler seeks back to the failed offset and waits out its back-off before redelivering. Non-blocking retry topics, configured with @RetryableTopic or a retry topic configuration bean, publish the failed record to a delay topic and commit the original offset, so the main partition moves on.

Use blocking retries for short, transient failures, or wherever key order matters, such as state transitions for one account. Avoid them for long back-offs, because one bad record holds up every record behind it on that partition. Retry topics suit slow-recovering dependencies when records are independent. Avoid them when a later event for the same key must not overtake an earlier one. If an update for order 42 is in retry and a cancellation for order 42 succeeds, the retried update arrives last and wins.

Neither choice removes duplicates. A rebalance after processing but before the commit redelivers the record, and a retry after a partial side effect repeats it. The guarantee has to come from the handler. Record an idempotency key, usually a business ID or the topic-partition-offset, in the same transaction as the write. Alternatively, express writes as upserts so that applying a record twice leaves the same state.

Observing consumer groups, and the impact on cost and engineering time

Once handlers are idempotent, the remaining risk is operational: problems you cannot see. Track three signals per consumer group, and alert on each.

• Lag per partition, not just the total. A healthy total can hide one stuck partition behind a poison pill or a slow key. Alert when lag on any partition keeps growing over several minutes, not on a fixed threshold.
• Rebalance frequency. Count partition assignment and revocation events from the container logs or the consumer's coordinator metrics. A steady rate under constant load usually means max.poll.interval.ms is being exceeded or pods are restarting.
• DLT volume. Alert on any growth in dead-letter topic volume, and assign someone to triage it. A DLT that nobody reads is silent data loss with an audit trail.

For a quick check, run kafka-consumer-groups --bootstrap-server <broker> --describe --group <group>. It lists each partition's current offset, log-end offset, lag, and the consumer that owns it. If a partition shows no owner, or the same partition has high lag across repeated runs, start investigating there.

This design has a cost. You add retry and DLT topics, consumers or tooling to replay them, and tests for duplicate delivery and for cooperative rebalancing during deploys. That work takes days. Diagnosing a rebalance storm or duplicated payments in production typically takes much longer and costs trust as well.

Tuning checklist

• Keep listener concurrency at or below the partition count.
• Make max.poll.records × worst-case processing time well below max.poll.interval.ms.
• Keep the total DefaultErrorHandler backoff inside the poll interval.
• Mark non-retryable exceptions explicitly.
• Route exhausted records to a DLT, monitor it, and test replaying it.
• Use cooperative rebalancing and test it during rolling deploys.

How OneEnv can help

Disclosure: OneEnv is the product behind this blog.

OneEnv (oneenv.io) is a workspace-based governance platform for the things services share — environment variables, database structure, API routes and validation rules. A change to a shared value is staged on a branch, reviewed once, and then every affected service explicitly approves or rejects it before it goes live, so you can see who a change touches before it ships instead of finding out in production.

Applied to spring kafka consumer error handling, this is where it helps:

Saves time

• Every service that inherits a shared value is listed in one place, so nobody searches repos and chats to find out who is affected.
• One review per branch replaces coordinating each team separately; each service still keeps its own approve/reject.
• Changing a shared variable is one edit and one review instead of editing every service's .env by hand.

Saves cost

• Fewer production incidents and rollbacks caused by a shared value, route or column changing under a consumer unannounced.
• Less duplicated configuration to maintain: one shared definition inherited by many services.
• A wrong or stale value is caught at review time by the services it affects, not by a failed deploy.
• OneEnv has a free plan (1 workspace, unlimited services and members, full governance), so trying it costs nothing.

These are qualitative benefits, not measured results — how much you save depends on how many services share the value.

Want to try it? Start at oneenv.io.