skip to content

What is a batch @KafkaListener, how do you enable it, and how does acknowledgment differ from record listeners?

level: seniorimportance: should knowfreq 58%

answer

  1. setBatchListener(true) / listener.type=batch
  2. one call per poll(); List<...> or ConsumerRecords
  3. batch size <= max.poll.records
  4. one Acknowledgment for whole batch; nack(index, Duration)
  5. BatchListenerFailedException pinpoints failed record

basics

~20 s

A batch listener receives a whole List of records from one poll() in a single method call instead of one record at a time. You enable it on the container factory (setBatchListener(true)) and the method takes a List; with manual ack, one Acknowledgment covers the entire batch.

solid answer

~40 s

Set `factory.setBatchListener(true)` (or Boot's spring.kafka.listener.type=batch). The listener method then accepts a `List<T>` payloads, or `List<ConsumerRecord<K,V>>`, or `ConsumerRecords<K,V>`, receiving everything returned by a single poll() at once — batch size is bounded by max.poll.records. Benefits: fewer method invocations and the ability to do bulk operations (batch DB insert, bulk index). Acknowledgment differs: in manual mode a single Acknowledgment.acknowledge() commits the whole batch; you can't ack individual records, though ack.nack(index, Duration) negatively acks from a given index onward for redelivery. Error handling also differs — you use a batch-aware error handler (DefaultErrorHandler with a recoverer, or throw BatchListenerFailedException to pinpoint the failed record so only it and after are retried). Ordering within the batch follows partition/offset order.

code

java · 23 lines
java
@Bean
public ConcurrentKafkaListenerContainerFactory<String, Order> batchFactory(
        ConsumerFactory<String, Order> cf) {
    var factory = new ConcurrentKafkaListenerContainerFactory<String, Order>();
    factory.setConsumerFactory(cf);
    factory.setBatchListener(true);
    factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
    return factory;
}

@KafkaListener(topics = "orders", containerFactory = "batchFactory")
public void handle(List<Order> orders, Acknowledgment ack) {
    for (int i = 0; i < orders.size(); i++) {
        try {
            repository.save(orders.get(i));
        } catch (TransientException e) {
            // commit everything before i; redeliver i.. after 2s
            ack.nack(i, Duration.ofSeconds(2));
            return;
        }
    }
    ack.acknowledge(); // commit the whole batch
}

go deeper

for a junior

Know a batch listener takes a List of records instead of one at a time.

for a middle

Enable it via setBatchListener(true), understand batch size ~ max.poll.records and single-Acknowledgment semantics.

for a senior

Handle partial failure with nack(index) / BatchListenerFailedException and idempotency; watch max.poll.interval.ms.

for a principal

Design bulk-processing pipelines with batch DLT recovery, throughput vs redelivery-window trade-offs, and rebalance-safe processing times.

**Record vs batch listeners** A normal (record) listener is invoked **once per ConsumerRecord**. A **batch listener** is invoked **once per poll()**, handed the **whole list** of records that poll returned. The number of records per batch is bounded by the consumer property **`max.poll.records`** (and available data). **Enabling batch mode** - Factory: `factory.setBatchListener(true)`. - Spring Boot: `spring.kafka.listener.type=batch`. With batch mode on, valid method signatures include: - `void handle(List<MyPayload> payloads)` - `void handle(List<ConsumerRecord<K,V>> records)` - `void handle(ConsumerRecords<K,V> records)` - plus optional `Acknowledgment` and/or `Consumer<K,V>` parameters, and parallel header lists (`@Header(...) List<...>`). **Why use it** - **Throughput / efficiency**: one method call and one bulk operation (e.g., JDBC batch insert, Elasticsearch bulk API, one downstream call) instead of N. - **Amortized overhead**: fewer proxy invocations, fewer commits when combined with BATCH ack. **Acknowledgment semantics** In manual ack mode, a batch listener gets **one Acknowledgment for the entire batch**. Calling `acknowledge()` commits offsets for **all** records in the batch. You cannot selectively commit a subset with plain `acknowledge()`. For partial handling you use **`Acknowledgment.nack(int index, Duration sleep)`** — records **before** `index` are considered processed (committed), and the record at `index` and everything after it are **redelivered** after the sleep. **Error handling** Batch error handling is trickier because an exception aborts the whole batch. Options: - **DefaultErrorHandler** (batch-aware): by default retries the entire batch; combined with a **BatchListenerFailedException** thrown from your listener carrying the failing record's index, it can retry/recover only from that record onward instead of reprocessing the whole batch. - A **recoverer** (e.g., DeadLetterPublishingRecoverer) to route poison records to a DLT after retries are exhausted. **Gotchas** - Without `BatchListenerFailedException`, any failure reprocesses the **entire** batch → duplicate side effects for already-succeeded records; make handlers idempotent or use the indexed exception. - Batch size is not fixed — it's whatever poll returned (0..max.poll.records); handle empty/variable sizes. - Long batch processing must stay within **`max.poll.interval.ms`** or the consumer is considered dead and a rebalance ejects it. - You can't mix a batch listener signature with a single-record parameter — the container validates the signature at startup. - Conversion: for typed `List<T>` payloads you need an appropriate deserializer/converter (e.g., JsonDeserializer / BatchMessagingMessageConverter).

  • With a batch listener, can you acknowledge individual records?
    Not with plain acknowledge() — it commits the whole batch. Use ack.nack(index, Duration): records before the index are committed, and from the index onward are redelivered after the sleep. That's the closest to partial acknowledgment.
  • How do you avoid reprocessing an entire batch when only one record fails?
    Throw a BatchListenerFailedException carrying the failing record (or its index) and use a batch-aware DefaultErrorHandler; it commits the records before the failure and retries/recovers only from the failed one onward.
  • What controls how many records land in one batch?
    The consumer property max.poll.records (plus how much data is actually available). The batch is whatever poll() returned, so sizes vary from call to call.

saying these in an interview costs you the question

  • Thinking each record in a batch gets its own Acknowledgment
  • Assuming batch size is fixed/configurable to an exact number
  • Believing a single failure only reprocesses that one record by default
  • Ignoring max.poll.interval.ms for long batch processing

context

resolving links…