60-Day Kafka 4 Learning Plan · Week 3 — Day 17 of 60

60-Day Kafka 4 Learning Plan · Week 3 — Spring Boot Integration Sources: Kafka: The Definitive Guide Ch.4 · docs.spring.io/spring-kafka/reference

Goal

Configure Spring Kafka consumer concurrency for parallel processing, understand polling properties and their interaction, choose the right offset commit mode, implement batch listeners for high-throughput bulk processing, handle partial batch failures, and apply backpressure safely.

1. Concurrency — parallel consumer threads

Each thread in a ConcurrentKafkaListenerContainerFactory owns one or more partitions. The maximum useful concurrency equals the number of partitions assigned to the consumer group.

Topic: orders (6 partitions)
concurrency=2 → Thread 1: P-0, P-1, P-2 | Thread 2: P-3, P-4, P-5
concurrency=6 → Thread 1: P-0 | Thread 2: P-1 | ... | Thread 6: P-5
concurrency=8 → 6 threads busy, 2 idle (wasted)

Configuration

# application.yml — global for all listeners
spring:
  kafka:
    listener:
      concurrency: 3

// Or override on a specific listener
@KafkaListener(topics = "orders", concurrency = "3")
public void consume(ConsumerRecord<String, String> record) {
    // Each invocation runs on one of 3 dedicated threads
}

Programmatic factory config

@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(
        ConsumerFactory<String, String> consumerFactory) {
    var factory = new ConcurrentKafkaListenerContainerFactory<String, String>();
    factory.setConsumerFactory(consumerFactory);
    factory.setConcurrency(3);
    return factory;
}

2. Polling configuration

⚠️ Critical: max.poll.records × processing_time_per_record must fit within max.poll.interval.ms. If processing 500 records takes > 5 minutes, the consumer will be evicted from the group, triggering a rebalance.

Solutions:

Reduce max.poll.records to a smaller batch
Increase max.poll.interval.ms
Use batch processing with async I/O

spring:
  kafka:
    consumer:
      max-poll-records: 50        # safer for slow processing
      properties:
        max.poll.interval.ms: 600000   # 10 min
        fetch.min.bytes: 1024          # wait for at least 1KB
        fetch.max.wait.ms: 500

As covered in Day 6 §2, heartbeats run on a separate background thread from poll() — so session.timeout.ms (heartbeat health) and max.poll.interval.ms (processing speed) are two independent failure modes worth watching separately, not one combined timeout.

3. Offset commit modes

AUTO (default) — `ack-mode: BATCH`

Spring commits offsets after the listener method returns. Simple to use. Risk: if an unhandled exception occurs and the listener swallows it, the offset is committed and the message is lost.

MANUAL — `ack-mode: MANUAL`

Call ack.acknowledge() when ready. The commit is deferred until the next poll() cycle. Fine-grained control, slightly batched commits.

MANUAL_IMMEDIATE ★ — `ack-mode: MANUAL_IMMEDIATE`

Commits immediately when ack.acknowledge() is called. Best for exactly-once processing patterns where you need to be certain the offset is committed before continuing.

4. Manual offset commit in Spring Boot

spring:
  kafka:
    listener:
      ack-mode: MANUAL_IMMEDIATE

@KafkaListener(topics = "orders")
public void consume(ConsumerRecord<String, String> rec, Acknowledgment ack) {
    try {
        processOrder(rec.value());
        ack.acknowledge();          // commit only on success
    } catch (Exception e) {
        log.error("Processing failed for key={}", rec.key(), e);
        // No ack → offset not committed → message will be reprocessed
        // Pair with error handler (Day 18) to avoid infinite retry loops
    }
}

The golden rule: only call ack.acknowledge() after the business logic succeeds. No ack = no commit = the message will be re-delivered after a restart or rebalance.

5. Batch listener — process records in bulk

Batch mode receives the entire poll() result as a List, avoiding per-record method call overhead. Ideal for bulk DB inserts, batched HTTP calls, or stream aggregations.

spring:
  kafka:
    listener:
      type: batch
    consumer:
      max-poll-records: 100       # tune batch size

@KafkaListener(topics = "orders")
public void consumeBatch(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {
    log.info("Processing batch of {} records", records.size());

    // Bulk write — one DB round-trip for the whole batch
    orderRepository.saveAll(
        records.stream().map(this::toOrder).toList()
    );

    ack.acknowledge();  // commit after entire batch succeeds
}

Batch with per-record error handling

@KafkaListener(topics = "orders")
public void consumeBatch(List<ConsumerRecord<String, String>> records) {
    for (ConsumerRecord<String, String> record : records) {
        try {
            processOrder(record.value());
        } catch (Exception e) {
            log.error("Failed offset={} key={}", record.offset(), record.key(), e);
            // Send to DLT here (Day 18)
        }
    }
}

6. Partial batch failure — recovering from a specific record

The simple per-record try/catch above silently absorbs failures without telling Spring which record failed — so on a crash mid-batch, Spring doesn’t know where to resume. BatchListenerFailedException fixes this: throw it naming the exact failed record, and Spring commits offsets for everything before it, then redelivers from that record onward on the next poll (instead of redelivering or silently dropping the whole batch).

@KafkaListener(topics = "orders", containerFactory = "batchFactory")
public void consumeBatch(List<ConsumerRecord<String, String>> records) {
    for (ConsumerRecord<String, String> record : records) {
        try {
            processOrder(record.value());
        } catch (Exception e) {
            // Names the exact failed record — Spring commits everything before it
            // and redelivers from here onward, instead of the whole batch
            throw new BatchListenerFailedException("Processing failed", e, record);
        }
    }
}

Pair this with a DefaultErrorHandler (Day 18) configured for batch listeners — it uses the failed record’s position to drive retry/backoff and, eventually, DLT routing for just that record and everything after it, not the whole batch.

7. Backpressure — pausing and resuming containers

Sometimes a consumer needs to temporarily stop pulling more work — a downstream dependency is degraded, or an internal queue is full — without triggering a rebalance from an idle poll() loop.

@Autowired
private KafkaListenerEndpointRegistry registry;

public void pauseOrdersConsumer() {
    registry.getListenerContainer("orderListener").pause();
}

public void resumeOrdersConsumer() {
    registry.getListenerContainer("orderListener").resume();
}

@KafkaListener(id = "orderListener", topics = "orders")
public void consume(ConsumerRecord<String, String> record) {
    if (downstreamService.isOverloaded()) {
        registry.getListenerContainer("orderListener").pause();
        // Container stops polling for new records; existing heartbeats continue,
        // so no rebalance is triggered while paused
    }
    processOrder(record.value());
}

Why this is safer than just slowing down processing: a paused container still sends heartbeats and stays a healthy group member — it simply stops calling poll() for new records. Deliberately slow processing without pausing risks breaching max.poll.interval.ms and triggering an unwanted rebalance instead of a clean, controlled backpressure signal.

8. Thread safety with concurrency

Each concurrency thread runs an independent KafkaConsumer, but they typically share the same Spring-managed beans (repositories, services, @Autowired collaborators) — those beans must be thread-safe if concurrency > 1.

Stateless @Service/@Repository beans (the Spring default) are safe as-is.
Any mutable instance field on a listener bean (a Map used as a cache, a counter, a buffer) is shared across all concurrent threads and needs explicit synchronization or a concurrent collection.
KafkaTemplate is thread-safe and can be safely shared/injected across concurrently-invoked listener methods.

@Component
public class OrderConsumer {
    // ⚠ NOT thread-safe if concurrency > 1 — shared mutable state across threads
    private final Map<String, Integer> retryCounts = new HashMap<>();

    // ✓ Thread-safe alternative
    private final Map<String, Integer> retryCountsSafe = new ConcurrentHashMap<>();
}

9. Monitoring concurrency & per-thread lag

# Confirm actual per-partition assignment matches expected concurrency
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group order-service

Sizing check: if --describe shows some threads holding multiple partitions and others holding one, and lag is skewed toward the multi-partition threads, that’s a signal to either increase concurrency (up to the partition count) or address a hot partition, not just add more application instances blindly.

10. Testing batch listeners

@SpringBootTest
@EmbeddedKafka(partitions = 3, topics = "orders")
class BatchListenerIntegrationTest {

    @Autowired KafkaTemplate<String, String> kafkaTemplate;
    @SpyBean OrderConsumer orderConsumer;

    @Test
    void processesFullBatch() {
        IntStream.range(0, 10).forEach(i ->
            kafkaTemplate.send("orders", "key" + i, "payload" + i));

        await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
            ArgumentCaptor<List<ConsumerRecord<String, String>>> captor = ArgumentCaptor.forClass(List.class);
            verify(orderConsumer, atLeastOnce()).consumeBatch(captor.capture(), any());
            int totalReceived = captor.getAllValues().stream().mapToInt(List::size).sum();
            assertThat(totalReceived).isEqualTo(10);
        });
    }
}

Batches from @EmbeddedKafka won’t necessarily arrive as one single batch of 10 — Kafka may split them across multiple poll() cycles depending on timing. Assert on the total records received across all invocations, not the size of any single batch.

11. Common pitfalls

Setting concurrency higher than the partition count — extra threads sit permanently idle, same waste pattern as Day 6 §8
Using per-record try/catch in a batch listener without BatchListenerFailedException — Spring has no way to know which record failed, so recovery/redelivery can’t target just the bad record
Slowing down processing instead of pausing the container for backpressure — risks breaching max.poll.interval.ms and causing an unwanted rebalance instead of a clean pause
Mutable instance fields on listener beans with concurrency > 1 — silent race conditions that only surface under real concurrent load, not in single-threaded local testing
Auto commit (ack-mode: BATCH, the default) with exception-swallowing code — an exception caught and logged but not rethrown still lets the batch commit, silently losing the failed record

Key Takeaways

concurrency = number of consumer threads; max useful = number of partitions in the topic
max.poll.records × processing_time must fit within max.poll.interval.ms — or the consumer is evicted
MANUAL_IMMEDIATE: commit only after successful processing — the safest offset strategy
Batch listeners receive a full poll() result — ideal for bulk DB writes
BatchListenerFailedException lets Spring commit offsets up to the failed record and redeliver only from there, instead of the whole batch
Pause/resume via KafkaListenerEndpointRegistry is the safe way to apply backpressure — heartbeats continue, no rebalance triggered
Concurrent listener threads share Spring beans — mutable instance state needs explicit thread safety
auto-offset-reset: earliest reprocesses from the beginning; latest skips old messages
Each concurrency thread is an independent KafkaConsumer instance in the same consumer group

Support me through GitHub Sponsors.

➡️ Day 18: Error handling — dead-letter topic & retry templates

Resources

📘 Kafka: The Definitive Guide — Chapter 4 (Kafka Consumers)
🌐 docs.spring.io/spring-kafka/reference — Listener container

👉 Link to Medium blog