Apache Kafka Consumer: Mechanics, Rebalances & Offset Management

Most message brokers push data directly to consumers. If your workers are struggling under load, the broker keeps pushing anyway. Kafka flips this dynamic completely: consumers pull data on their own schedule.

That pull model gives you full control over backpressure, batching, and replaying history. But it also introduces some of the trickiest bugs you'll run into in distributed systems: unexpected rebalances when a database call takes too long, duplicate records when an instance dies mid-batch, or consumers stalling because of improper threading.

In this guide, we'll break down how Kafka consumers actually work under the hood: starting from raw, standalone partition consumption, moving to consumer groups, understanding heartbeat and poll timeouts, and looking at production patterns in Java, Python, and TypeScript.

1. Consumer Groups & The Pull-Based Architecture

Traditional message queues like RabbitMQ or JMS typically push messages down to connected workers. Kafka uses a client-driven pull model instead.

flowchart TD
    subgraph BROKER["Kafka Cluster (Brokers)"]
        P0["Partition 0 (Log)"]
        P1["Partition 1 (Log)"]
        P2["Partition 2 (Log)"]
    end

    subgraph CG["Consumer Group: order-fulfillment"]
        C1["Consumer 1\n(Reading P0, P1)"]
        C2["Consumer 2\n(Reading P2)"]
    end

    C1 -->|"1. poll() pulls records at its own pace"| P0
    C1 -->|"1. poll() pulls records at its own pace"| P1
    C2 -->|"1. poll() pulls records at its own pace"| P2

    C1 -.->|"2. commits offset progress to __consumer_offsets"| BROKER
    C2 -.->|"2. commits offset progress to __consumer_offsets"| BROKER

Why Pulling Beats Pushing at Scale

• Built-in Backpressure: If your downstream service slows down (say your database connection pool is saturated), you simply pull fewer records or wait until processing finishes before calling poll() again. The broker never overwhelms your heap memory.
• Efficient Batch Fetching: Rather than chatting over TCP for every individual record, consumers configure fetch.min.bytes (e.g., 64 KB) and fetch.max.wait.ms (e.g., 500 ms). If traffic is slow, the broker waits until 64 KB accumulates before responding, cutting network round-trips dramatically.
• Replaying History: Because Kafka doesn't delete messages once consumed, you can rewind a consumer's offset back in time to reprocess historical data—great for fixing data corruption bugs or backfilling analytics.

Direct Partition Reading: 10 Partitions with 2 Standalone Python Workers

Before looking at consumer groups, it helps to understand what a consumer does at the bare metal level. A Kafka topic is just a set of partitioned log files on disk. You don't actually need groups or coordination protocols just to read them.

If you want complete, manual control, a consumer can connect directly to the brokers and ask for specific partition numbers using the assign() API.

Splitting Partitions Manually

Say you have a topic named my-kafka-topic with 10 partitions (0 through 9), and you want to split the work across two Python worker threads:

• Worker 1 explicitly reads partitions 0 through 4.
• Worker 2 explicitly reads partitions 5 through 9.

from kafka import KafkaConsumer, TopicPartition
import threading

TOPIC_NAME = "my-kafka-topic"
BOOTSTRAP_SERVERS = ['localhost:9092']

def run_worker(worker_name, partition_range):
    # Turn topic name and partition numbers into TopicPartition instances
    assigned_partitions = [TopicPartition(TOPIC_NAME, p) for p in partition_range]
    
    # Standalone consumer: no group_id specified
    consumer = KafkaConsumer(
        bootstrap_servers=BOOTSTRAP_SERVERS,
        auto_offset_reset='earliest'
    )
    
    # Point directly to this partition range
    consumer.assign(assigned_partitions)
    print(f"[{worker_name}] Assigned to partitions: {list(partition_range)}")

    try:
        for message in consumer:
            print(f"[{worker_name}] Partition {message.partition} | Offset {message.offset}: {message.value.decode('utf-8')}")
    finally:
        consumer.close()

if __name__ == "__main__":
    # Start Worker 1 handling partitions 0-4
    t1 = threading.Thread(target=run_worker, args=("Worker-1", range(0, 5)))
    
    # Start Worker 2 handling partitions 5-9
    t2 = threading.Thread(target=run_worker, args=("Worker-2", range(5, 10)))

    t1.start()
    t2.start()
    t1.join()
    t2.join()

The Trade-offs of Going Standalone

Manual assignment is straightforward and predictable, but in production it quickly becomes a maintenance headache:

• Zero Failover: If Worker 1 crashes, partitions 0–4 stop getting processed entirely. Worker 2 has no idea Worker 1 died and will never pick them up. Someone has to notice the outage and restart or reconfigure the worker.
• Hardcoded Scaling: Want to add a 3rd worker? You have to recalculate the partition math in code, change your config, and redeploy everything.
• No Offset Persistence: Because there is no consumer group, Kafka won't record your reading progress in its internal offset registry. If a worker restarts, it starts over at whatever auto_offset_reset dictates, unless you track offsets yourself in Redis or Postgres.

How Consumer Groups Solve These Problems

When you switch from manual assignment (assign()) to a Consumer Group (subscribe()), Kafka takes over the coordination work. One of the brokers is designated as the Group Coordinator, handling partition distribution automatically.

Here is how consumer groups solve the three major headaches of standalone consumers:

1. Dynamic Partition Assignment

With standalone consumers, you had to manually partition your workload in code (e.g., Worker 1 reads 0–4, Worker 2 reads 5–9). In a consumer group, you simply call .subscribe(["my-kafka-topic"]) and assign every worker the same group_id. The Group Coordinator automatically inspects the cluster, counts the active members, and distributes the partitions evenly across them.

2. Automatic Offset Tracking

Standalone consumers have no built-in memory; on restart, they lose their reading position unless you build custom state tracking. In contrast, consumer group members regularly commit their processed offsets back to Kafka's internal, compacted __consumer_offsets topic. When a crashed or restarted worker comes back online, it queries the coordinator for its last committed offset and picks up right where it left off.

3. Automatic Failover (Rebalancing)

If an independent consumer crashes, its assigned partitions stall completely until a human intervenes. Within a consumer group, instances send periodic background heartbeats to the coordinator. If an instance drops offline or stops responding, the coordinator detects the failure, initiates a rebalance, and immediately reassigns the orphaned partitions to the remaining healthy consumers in the group.

Consumer Group in Python

With consumer groups, your code becomes much simpler—no partition math required:

from kafka import KafkaConsumer

consumer = KafkaConsumer(
    'my-kafka-topic',
    bootstrap_servers=['localhost:9092'],
    group_id='order-processing-group', # <--- Joining a consumer group!
    auto_offset_reset='earliest',
    enable_auto_commit=True
)

# Kafka manages partition assignment, failover, and offsets automatically
for message in consumer:
    print(f"Partition {message.partition} | Offset {message.offset}: {message.value.decode('utf-8')}")

Enterprise Java: Scaling Consumer Groups with Spring for Kafka

In standard Java, running multiple consumer threads means writing your own ExecutorService or thread loop wrappers. If you use Spring for Kafka, you can configure multi-threaded consumption declaratively using the concurrency property.

If your topic has 10 partitions and you set concurrency = "10", Spring starts 10 consumer threads under the same consumer group, assigning one partition per thread.

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class OrderEventListener {

    // Spring spawns 10 concurrent consumer instances under this group,
    // matching a 10-partition topic 1:1
    @KafkaListener(
        topics = "my-kafka-topic", 
        groupId = "order-processing-group", 
        concurrency = "10"
    )
    public void handleOrder(String message) {
        System.out.println("Processing on thread " + Thread.currentThread().getName() + ": " + message);
    }
}

What Spring Does Under the Hood

1. Container Management: Spring creates a ConcurrentMessageListenerContainer.
2. Thread Provisioning: It spins up 10 worker threads, each running an independent KafkaConsumer instance.
3. Coordinator Handshake: All 10 instances join order-processing-group. Kafka's coordinator sees 10 members for 10 partitions and assigns exactly one partition to each thread.

2. The Poll Loop: Threading, Heartbeats & Timeouts

One fundamental rule about the official Java KafkaConsumer: it is strictly single-threaded. If two threads try to call methods on the same consumer instance at the same time, it immediately throws a ConcurrentModificationException.

To keep the client responsive while running your processing logic, modern Kafka clients run two distinct loops:

1. The Main Processing Thread: Calls poll(), deserializes payloads, runs your business logic, and commits offsets.
2. The Background Heartbeat Thread: Runs quietly in the background, pinging the broker's Group Coordinator at regular intervals (defined by heartbeat.interval.ms, usually every 3 seconds) to prove the JVM process is still alive.

sequenceDiagram
    autonumber
    participant App as App Thread (poll loop)
    participant HB as Background Heartbeat Thread
    participant Coord as Group Coordinator (Broker)

    loop Every heartbeat.interval.ms (e.g. 3s)
        HB->>Coord: HeartbeatRequest
        activate Coord
        Coord-->>HB: HeartbeatResponse (Node Healthy)
        deactivate Coord
    end

    Note over App: App invokes consumer.poll()
    App->>Coord: FetchRequest
    activate Coord
    Coord-->>App: Returns Batch of 500 Records
    deactivate Coord

    Note over App: App processes records (DB calls, APIs).
Must complete within max.poll.interval.ms!

The Two Timeouts You Need to Know

3. Partition Assignment & Cooperative Rebalance

Whenever consumers join, leave, or crash, the coordinator reassigns topic partitions across the surviving instances. This process is called a rebalance.

Eager vs. Cooperative Sticky Rebalancing

• Eager Rebalancing (Old way): When a rebalance occurred, every single consumer in the group had to drop all assigned partitions, stop consuming completely, rejoin the group, and wait for a fresh assignment. This caused noticeable "stop-the-world" latency spikes across entire clusters.
• Cooperative Sticky Rebalancing (Modern standard): Introduced via CooperativeStickyAssignor. Instead of a blanket reset, consumers keep reading from unaffected partitions without interruption. Only the specific partitions that need to migrate to a new consumer are revoked and transferred.

sequenceDiagram
    autonumber
    participant C1 as Consumer 1 (Holds P0, P1)
    participant C2 as Consumer 2 (Holds P2, P3)
    participant C3 as New Consumer 3 (Joining)
    participant Coord as Group Coordinator

    Note over C1,C2: Steady state: Reading records normally
    C3->>Coord: JoinGroupRequest (New Member)
    activate Coord
    Coord-->>C1: Rebalance Notification (Cooperative)
    Coord-->>C2: Rebalance Notification (Cooperative)
    Coord-->>C3: JoinGroup Accepted
    deactivate Coord

    Note over C1: C1 keeps processing P0 uninterrupted!
Only revokes P1 for C3.
    Note over C2: C2 keeps processing P2 uninterrupted!
Only revokes P3 for C3.

    C1->>Coord: SyncGroup (Revoked P1)
    C2->>Coord: SyncGroup (Revoked P3)
    activate Coord
    Coord-->>C3: Assigns P1 & P3 to Consumer 3
    deactivate Coord
    Note over C3: Consumer 3 starts consuming P1 & P3.
P0 and P2 were never paused!

4. Offset Commit Strategies: At-Least-Once vs At-Most-Once

Consumers track their position in each partition using offsets, stored in the __consumer_offsets topic. How and when you commit these offsets dictates your delivery semantics.

5. Production Java & TypeScript Implementations

Java: Manual Offsets with Graceful Shutdown

This implementation runs a dedicated consumer thread, disables auto-commit, uses cooperative sticky assignment, and handles clean shutdown via consumer.wakeup().

package me.thelp.kafka.consumer;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.errors.WakeupException;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class OrderConsumerWorker implements Runnable {
    private static final Logger log = LoggerFactory.getLogger(OrderConsumerWorker.class);
    private final KafkaConsumer<String, String> consumer;
    private volatile boolean running = true;

    public OrderConsumerWorker(String bootstrapServers, String groupId) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        // Disable auto-commit to prevent committing before processing succeeds
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        // Use modern cooperative sticky rebalancing
        props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, 
            CooperativeStickyAssignor.class.getName());

        // Protect against the slow DB trap: keep batch sizes reasonable
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 45000);
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 300000); // 5 min
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100);        // Max 100 per poll

        this.consumer = new KafkaConsumer<>(props);
    }

    @Override
    public void run() {
        try {
            consumer.subscribe(Collections.singletonList("orders.events.v1"));

            while (running) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));

                for (ConsumerRecord<String, String> record : records) {
                    processOrder(record.key(), record.value());
                }

                // Asynchronous commit on the hot path so we don't block next poll
                if (!records.isEmpty()) {
                    consumer.commitAsync((offsets, ex) -> {
                        if (ex != null) {
                            log.warn("Async offset commit failed: {}", ex.getMessage());
                        }
                    });
                }
            }
        } catch (WakeupException e) {
            // Normal shutdown triggered via wakeup()
            if (running) throw e;
        } catch (Exception e) {
            log.error("Fatal error in poll loop", e);
        } finally {
            try {
                // Final synchronous commit before releasing partitions
                consumer.commitSync();
            } finally {
                consumer.close();
                log.info("Consumer closed cleanly.");
            }
        }
    }

    private void processOrder(String key, String value) {
        // Business logic, DB persist, or payment service call
    }

    // Called from shutdown hook or separate thread
    public void stop() {
        this.running = false;
        this.consumer.wakeup(); // Safely interrupts blocking poll()
    }
}

TypeScript: KafkaJS with Batch Processing and Heartbeats

When processing batches in Node.js, long asynchronous operations can cause the broker to think your consumer is unresponsive. KafkaJS provides heartbeat() inside eachBatch to keep the connection alive while you process records.

import { Kafka, Consumer, EachBatchPayload } from 'kafkajs';

export class OrderConsumerService {
  private consumer: Consumer;

  constructor(brokers: string[], groupId: string) {
    const kafka = new Kafka({
      clientId: 'order-service-worker',
      brokers,
    });

    this.consumer = kafka.consumer({
      groupId,
      sessionTimeout: 45000,
      heartbeatInterval: 3000,
      maxBytesPerPartition: 1048576, // 1 MB buffer per partition
    });
  }

  async start(): Promise<void> {
    await this.consumer.connect();
    await this.consumer.subscribe({
      topic: 'orders.events.v1',
      fromBeginning: false,
    });

    await this.consumer.run({
      autoCommit: false,
      eachBatchAutoResolve: false,
      eachBatch: async ({ batch, resolveOffset, heartbeat, isRunning, isStale }: EachBatchPayload) => {
        for (const message of batch.messages) {
          if (!isRunning() || isStale()) break;

          const orderId = message.key?.toString();
          const orderData = message.value ? JSON.parse(message.value.toString()) : null;

          await this.processOrder(orderId, orderData);

          // Mark individual message offset resolved and ping broker
          resolveOffset(message.offset);
          await heartbeat();
        }

        // Commit all resolved offsets at the end of the batch
        await this.consumer.commitOffsets([
          {
            topic: batch.topic,
            partition: batch.partition,
            offset: (BigInt(batch.lastOffset()) + 1n).toString(),
          },
        ]);
      },
    });
  }

  private async processOrder(orderId?: string, data?: object): Promise<void> {
    // Database write or webhook call
  }

  async stop(): Promise<void> {
    await this.consumer.disconnect();
  }
}

6. Production Troubleshooting Matrix