Kafka Producer & Consumer Deep Dive: acks, Retries, Batch Processing in Spring Boot

The first Kafka integration you build usually works great in the demo. The producer sends, the consumer receives, everything looks smooth. Then your system hits its first real incident: the broker has a leadership election mid-deployment, your consumer receives a malformed message from a third-party service, a downstream database goes down for 45 seconds, and suddenly you realize your Kafka setup has no durability guarantees, no retry logic, and no way to isolate bad messages from good ones.

Producer acknowledgment modes are the first line of defense. acks=0 is performance theater — you're broadcasting messages into the void and hoping for the best. acks=1 is a false comfort — the leader acknowledges but hasn't replicated, so a leader failover between write and replication loses your data permanently. acks=all with min.insync.replicas=2 is the production standard: the message is durable on multiple brokers before the producer considers the send complete.

On the consumer side, offset management is where most production bugs live. Auto-commit increments the offset on a timer, not after successful processing — so a consumer crash mid-processing auto-commits the offset, and those messages are gone forever. AckMode.MANUAL_IMMEDIATE gives you exact control: commit only when your database write succeeds.

Retry logic with @RetryableTopic addresses the class of transient failures — downstream database timeouts, external API blips, temporary lock contention — where the message is valid but the processing environment wasn't ready. Spring Kafka creates retry topics automatically, routes failed messages through configurable backoff periods, and finally sends to a dead-letter topic if all attempts fail. No manual retry infrastructure required.

Batch consumers unlock an order-of-magnitude throughput improvement for high-volume processing. Instead of one database call per message, you receive List<ConsumerRecord>, process 500 records at once, and bulk-insert in a single transaction. The operational complexity increases, but so does your system's ability to keep up under load.

Producer Acknowledgment Modes: acks=0, acks=1, acks=all

The acks producer configuration is the most impactful durability setting in Kafka. It determines how many broker replicas must confirm message receipt before the KafkaProducer.send() future completes. Each value represents a distinct point on the latency-durability trade-off spectrum.

acks=0 (fire-and-forget): The producer sends the message and immediately considers it delivered, without waiting for any broker response. No acknowledgment packet traverses the network for this message. This maximizes throughput and minimizes latency but provides zero durability — if the broker is unavailable or rejects the message, the producer is unaware. Use cases are narrow: high-volume metrics, IoT sensor readings, or application logs where occasional loss is acceptable and the monitoring overhead of tracking every message exceeds the business value of any individual one.

acks=1 (leader-only): The partition leader writes the message to its local log and acknowledges the producer. This is faster than acks=all and provides protection against temporary broker unavailability. However, there is a critical gap: if the leader fails between writing the message and replicating it to followers, the message is lost permanently. The new elected leader (a follower that never received the message) has no record of it, and the producer already received an ACK. This failure scenario is not hypothetical — it occurs during every broker restart and leadership election in a rolling update.

acks=all (or acks=-1): The leader writes the message and waits for all in-sync replicas (ISRs) to confirm they have written it to their logs before acknowledging the producer. The ISR list is maintained by the broker — replicas that fall behind (haven't fetched from the leader within replica.lag.time.max.ms) are removed from the ISR. Setting min.insync.replicas=2 on the topic ensures that even if replicas lag out, at least 2 (leader + 1 follower) must confirm the write. If fewer than min.insync.replicas are in sync, the broker rejects the write with NotEnoughReplicasException — the producer should retry until replicas catch up.

In Spring Boot, set spring.kafka.producer.acks=all and configure min.insync.replicas on your NewTopic bean. Pair with enable.idempotence=true to get duplicate-free retries — the broker assigns a producer ID and sequence number to each message, deduplicating retransmissions that occur during retry after an ambiguous failure.

AckMode.MANUAL_IMMEDIATE and Offset Management

Spring Kafka's AckMode controls when consumed message offsets are committed back to the Kafka broker. The default AckMode.BATCH commits offsets after each poll loop completes — after the listener method returns for all records in the batch, regardless of whether processing succeeded. This is convenient but dangerous: if your database write fails inside the listener after the poll completes, the offset is already committed and the message is silently lost.

AckMode.MANUAL_IMMEDIATE transfers offset commit control entirely to your code. The listener container injects an Acknowledgment object into your listener method. Calling ack.acknowledge() commits the current record's offset to the broker immediately. If you do not call acknowledge() — because an exception was thrown before reaching that line — the offset stays uncommitted and the message is redelivered on the next poll.

The `IMMEDIATE` suffix is important: AckMode.MANUAL batches acknowledgments and flushes them at the end of the poll loop, while MANUAL_IMMEDIATE commits the offset synchronously as soon as acknowledge() is called. For high-reliability consumers where you need to know that each record's offset is committed before proceeding to the next, use MANUAL_IMMEDIATE.

Pairing MANUAL_IMMEDIATE with error handling requires careful design. A naive implementation that calls ack.acknowledge() inside a try-catch that also catches the failure will acknowledge even on error — defeating the purpose. The correct pattern: call acknowledge() only in the success path. Configure SeekToCurrentErrorHandler on the container so that when an exception escapes the listener, the container seeks the consumer back to the failed record's offset, ensuring it is redelivered rather than skipped.

SeekToCurrentErrorHandler (renamed DefaultErrorHandler in Spring Kafka 2.8+) accepts a BackOff configuration for retry timing. With ExponentialBackOffWithMaxRetries(5), the handler retries the failed record 5 times with exponential backoff before routing to the dead-letter topic via DeadLetterPublishingRecoverer. This gives transient failures (DB timeout, external API blip) multiple chances to succeed before the message is quarantined.

@RetryableTopic: Declarative Retry with Exponential Backoff

Before @RetryableTopic, implementing retry logic for Kafka consumers required building custom retry infrastructure: separate retry topics, a scheduler to re-enqueue messages after a delay, a DLQ writer, and manual tracking of retry counts. Spring Kafka 2.7+ collapses all of this into a single annotation.

@RetryableTopic(attempts = 3, backoff = @Backoff(delay = 1000, multiplier = 2)) on a @KafkaListener method automatically creates two retry topics (orders-retry-0, orders-retry-1) and one dead-letter topic (orders-dlt). When the listener method throws a retryable exception, the framework publishes the message to orders-retry-0 with headers recording the original topic, the attempt number, and the timestamp after which the message should be reprocessed. A background listener on the retry topic waits until the delay has elapsed, then re-delivers the message to the main listener. After the configured number of attempts, the message routes to the DLT.

The retry delay strategy is configurable. @Backoff(delay=1000, multiplier=2, maxDelay=30000) implements exponential backoff: attempt 1 waits 1s, attempt 2 waits 2s, attempt 3 waits 4s, capped at 30s. This gives transient failures (DB connection timeouts typically last 1-5s) time to resolve while not holding retry topics open indefinitely.

Exception classification is granular. notRetryOn = {DataIntegrityViolationException.class, DeserializationException.class} sends these exceptions directly to the DLT without retrying — they will fail on every attempt, so retrying wastes resources and delays DLT processing. include = {OptimisticLockingFailureException.class} restricts retrying to only the listed exceptions, sending everything else to DLT.

dltStrategy = DltStrategy.FAIL_ON_ERROR (the default) routes to DLT when retries are exhausted. dltStrategy = DltStrategy.NO_DLT drops the message after exhaustion — dangerous, but useful for truly ephemeral events. The @DltHandler annotation on a method in the same class receives DLT messages for specialized handling: logging, alerting, inserting into a database for manual review, or triggering a compensation event.

By default, @RetryableTopic uses the same consumer group for retry and main topic consumption. This means retry processing uses the same concurrency settings as main topic processing. For high-volume retry scenarios, configure sameIntervalTopicReuseStrategy = SameIntervalTopicReuseStrategy.SINGLE_TOPIC to consolidate retries or override the container factory for retry topics separately.

RecordFilterStrategy: Filtering Messages Before Processing

Not every message in a topic is meant for every consumer. A RecordFilterStrategy allows you to declaratively exclude messages that match a filter condition, without the consumer ever invoking the listener method. The filter runs after deserialization but before listener invocation — the consumer polls the record, deserializes it, evaluates the filter, and if the filter returns true (filtered), the record is acknowledged and the listener is skipped.

Common use cases: filtering by message type header (one topic carries multiple event types, but each consumer only handles one), filtering by tenant ID in a multi-tenant system, or excluding test messages (produced by integration tests) from production consumers.

The filter function receives the deserialized ConsumerRecord<K, V>. You have full access to the record's key, value, headers, topic, partition, and offset. The return value is a boolean: true means discard (filter out), false means pass through to the listener.

When a record is filtered, its offset still needs to be committed so the consumer doesn't re-read it endlessly. Set factory.setAckDiscarded(true) to auto-acknowledge filtered records. Without this, filtered records accumulate and the consumer's committed offset falls behind the end offset, making lag monitoring misleading — you'll see apparent lag from filtered records that will never be processed.

For batch listeners, BatchRecordFilterStrategy filters at the batch level. The default behavior is to filter individual records within the batch and deliver only the non-filtered subset to the listener. If all records in a batch are filtered, the listener is not called at all. Configure factory.setBatchToRecordFilterStrategy(strategy) for fine-grained control.

Batch Consumers with List

Individual record processing — one listener invocation per Kafka message — is clean and simple but inefficient at scale. For each record, you pay the overhead of listener dispatch, Spring proxy invocation, and crucially, one database round-trip. At 10,000 messages per second, that's 10,000 individual INSERT statements per second — far more than most databases sustain comfortably.

Batch consumers receive a List<ConsumerRecord<K, V>> (or List<V> for value-only access) representing all records returned by a single KafkaConsumer.poll() call. Instead of 500 individual database calls, you execute one bulk INSERT with 500 rows — a 50-100x reduction in database round-trips. Spring Boot's spring.kafka.listener.type=batch enables batch mode globally, or you can configure it per-factory.

Batch size is controlled by max.poll.records (how many records per poll, default 500) and the consumer's fetch.min.bytes / fetch.max.wait.ms settings which control how full the fetch buffer must be before returning. Setting max.poll.records=1000 with fetch.min.bytes=1048576 (1MB) creates large, efficient batches at the cost of slightly increased latency.

Error handling in batch mode is more complex than single-record mode. If the listener throws, the entire batch is the unit of retry — all records in the batch will be redelivered. This makes idempotent processing essential. The BatchToRecordAdapter bridges batch delivery with single-record error handling, allowing DefaultErrorHandler (which operates at the record level) to function within a batch context.

For partial batch commits — acknowledging records 0-249 but retrying records 250-499 — use Acknowledgment.nack(int index, Duration duration). This seeks the consumer back to record at index within the batch, ensuring records after the failure point are redelivered while records before it are committed. This is the correct approach when batch processing involves heterogeneous records that may fail independently.

Batch listeners integrate naturally with Spring's @Transactional — wrap the entire batch in one database transaction for atomic commit/rollback. This is significantly more efficient than one transaction per record. Be aware of transaction size limits and timeout settings when batches are very large.

SeekToCurrentErrorHandler and DefaultErrorHandler Patterns

The error handler sits between the listener container and your listener method. When an unhandled exception escapes your @KafkaListener method, the error handler decides what to do next: retry immediately, seek back to retry after backoff, skip the record, or route to a dead-letter topic.

Pre-2.8 Spring Kafka used SeekToCurrentErrorHandler. From 2.8+, DefaultErrorHandler is the unified replacement that handles both single-record and batch consumers. The behavior is identical: on exception, the container seeks the consumer back to the failed record's offset so the next poll re-reads it. Without this seek, the container would advance past the failed record (as if it were processed), losing the message.

The DefaultErrorHandler constructor accepts a BackOff and a ConsumerRecordRecoverer. The BackOff determines retry timing. FixedBackOff(1000L, 3) retries 3 times with 1-second delays between each. ExponentialBackOffWithMaxRetries(5) retries 5 times with exponential delays. After exhausting retries, the ConsumerRecordRecoverer handles the failed record. DeadLetterPublishingRecoverer publishes the failed record to a .DLT topic, including headers identifying the original topic, partition, offset, and exception details.

Exception classification controls which exceptions trigger retries and which go directly to the DLT. errorHandler.addNotRetryableExceptions(DeserializationException.class) ensures that malformed messages (which will always fail deserialization) bypass retries and go straight to the DLT. errorHandler.addRetryableExceptions(DataAccessException.class) forces retry for transient database exceptions. Without classification, all exceptions retry, wasting time on failures that are guaranteed to repeat.

The ExceptionClassifier interface provides fine-grained control, allowing a lambda or class to inspect the exception (including its cause chain) and return whether it should be retried. This handles wrapped exceptions where the root cause is what matters, not the wrapper type thrown by the listener.