The consumer threading model

Key points

• KafkaConsumer is not thread-safe. One consumer belongs to one thread, and calling it from two is a runtime error rather than a race you might get away with.
• Two independent timeouts decide whether you are still in the group. session.timeout.ms is missed heartbeats - the consumer process is gone. max.poll.interval.ms is a gap between poll calls - the process is alive but stuck.
• Since the heartbeat moved to a background thread, a consumer busy in slow processing keeps heartbeating happily. It is the poll interval that catches it, which is why the symptom is so confusing: the consumer looks healthy right up until it is evicted.
• The result is a rebalance storm: eviction reassigns the work, the new owner is equally slow, and the group thrashes without ever making progress.
• The correct fix is pause and resume. Hand the batch to a worker pool, pause those partitions, keep calling poll to stay alive, and resume when the work finishes.
• Offloading without pausing is worse than not offloading: the poll loop races ahead and commits offsets for records the workers have not finished, so a crash loses them silently. Track the lowest incomplete offset per partition and commit only up to it.

Example

// The pattern for processing that takes longer than max.poll.interval.ms.
while (running) {
    ConsumerRecords<String, Order> records = consumer.poll(Duration.ofMillis(500));

    if (!records.isEmpty()) {
        // 1. Stop being given more work while these are in flight.
        consumer.pause(consumer.assignment());

        for (TopicPartition tp : records.partitions()) {
            List<ConsumerRecord<String, Order>> batch = records.records(tp);
            inFlight.put(tp, batch.get(batch.size() - 1).offset());
            workers.submit(() -> process(batch));   // slow work, off the poll thread
        }
    }

    // 2. Keep polling regardless. A paused consumer returns no records but
    //    still heartbeats and still resets the poll-interval clock.
    if (workersIdle()) {
        consumer.resume(consumer.assignment());
    }

    // 3. Commit only what is genuinely finished - never what was merely handed out.
    Map<TopicPartition, OffsetAndMetadata> safe = completedOffsets();
    if (!safe.isEmpty()) {
        consumer.commitSync(safe);
    }
}

/*
 * The two timeouts, side by side:
 *
 *   session.timeout.ms      45000   background heartbeat stopped -> process is dead
 *   heartbeat.interval.ms    3000   how often that heartbeat is sent
 *   max.poll.interval.ms   300000   gap between poll() calls -> process is stuck
 *   max.poll.records          500   lower this first when processing is slow
 *
 * Symptom: "the consumer keeps leaving the group but the process is clearly
 * running and logging" -> it is max.poll.interval.ms, every time.
 */

Heartbeats prove the process is alive; only poll() proves it is keeping up - pause the partitions rather than letting the loop run ahead of the work.