Consumer #

If producers are responsible for writing events into the Apache Kafka cluster, then Consumer is the client application that subscribes to one or several topics and reads that data for processing. Unlike the push-based model of traditional brokers, Kafka adopts a unique pull-based model. This design gives consumers full control over their own processing speed. Understanding how the consumer poll loop works, the group coordination mechanism (Consumer Group), and read position commit management (offset commit) is essential for building distributed data pipelines that are safe, efficient, and free from data duplication or message loss.

Basic Concept: Pull vs Push Model #

In distributed messaging architecture, there are two main methods for delivering data from broker servers to receiving applications: Push (pushed by the broker) and Pull (pulled by the consumer).

  • Push Model (Traditional): Message brokers like RabbitMQ actively monitor queues and immediately send (push) data to consumer applications as soon as new messages arrive. The biggest problem with this model is the Overload threat on the consumer side. If a producer suddenly sends 100,000 messages per second while the consumer can only process 1,000 messages per second, the broker floods the consumer with data until consumer memory runs out and the application crashes (Out Of Memory). To prevent this, the system must implement a complex speed negotiation mechanism (backpressure).
  • Pull Model (Kafka): Kafka brokers are passive. Consumer applications actively send pull requests to the broker when they have compute capacity ready. If a consumer is busy processing heavy transactions, it simply delays sending the next request. The Kafka broker never forces data on anyone. This model naturally solves the backpressure problem because consumers themselves determine their workflow speed based on their local memory and CPU capacity.

Consumer Work Cycle: Poll Loop #

Inside Kafka consumer application code, the heart of all data reading activity is an endless loop called the Poll Loop. The consumer SDK is designed to run on a single thread (single-thread execution model) that repeatedly calls the poll() method inside a while(true) block.

Let’s visualize the poll loop lifecycle inside a Kafka consumer thread:

flowchart TD
    subgraph Loop["Consumer Poll Loop Cycle"]
        direction TB
        CallPoll["1. Call poll(timeout)"] --> FetchData["2. Pull Events from Broker <br/> (Fetch message batch)"]
        FetchData --> ProcessData["3. Main Business Processing <br/> (Save to DB, compute data, etc.)"]
        ProcessData --> SendHeartbeat["4. Send Background Heartbeat <br/> (Tell broker the consumer is alive)"]
        SendHeartbeat --> CommitOffset["5. Commit Read Offset <br/> (Record progress to __consumer_offsets)"]
        CommitOffset --> CallPoll
    end

The cycle above runs continuously. The poll(timeout) call has two main roles:

  1. Fetching Data: This call fetches new message batches from the partitions allocated to that consumer. If there are no new messages on the broker, the thread waits for the timeout period (e.g., 100 ms) before continuing to the next iteration so it doesn’t waste CPU.
  2. Maintaining Group Membership: Behind the scenes, the poll() call periodically sends heartbeat signals to the broker coordinator server. This is an important mechanism to tell the cluster that our consumer application is still healthy and hasn’t died.

Consumer Group & Read Scalability #

To handle large-volume data streams in real-time, a single consumer application is often not enough. We need to run several consumer instances in parallel to divide the workload. Kafka facilitates this natively through the Consumer Group concept.

All consumer instances configured with the same group.id property automatically join the same group. Kafka then divides all active partitions in that topic among group members fairly and evenly.

Let’s study three physical partition allocation scenarios based on the number of members in one consumer group:

flowchart TD
    subgraph Skenario1["Scenario A: Consumers < Partitions (Ideal)"]
        direction LR
        P0_A[("Partition 0")] ---> C1_A["Consumer 1"]
        P1_A[("Partition 1")] ---> C1_A
        P2_A[("Partition 2")] ---> C2_A["Consumer 2"]
    end

    subgraph Skenario2["Scenario B: Consumers = Partitions (Maximum)"]
        direction LR
        P0_B[("Partition 0")] ---> C1_B["Consumer 1"]
        P1_B[("Partition 1")] ---> C2_B["Consumer 2"]
        P2_B[("Partition 2")] ---> C3_B["Consumer 3"]
    end

    subgraph Skenario3["Scenario C: Consumers > Partitions (Wasteful)"]
        direction LR
        P0_C[("Partition 0")] ---> C1_C["Consumer 1"]
        P1_C[("Partition 1")] ---> C2_C["Consumer 2"]
        P2_C[("Partition 2")] ---> C3_C["Consumer 3"]
        P3_Idle["(No partitions)"] -.-> C4_C["Consumer 4 <br/> (Idle)"]
    end

Partition Allocation Rules: #

  • One Partition, One Consumer: Within the same Consumer Group, each partition may only be allocated to at most one consumer. This is absolutely necessary to prevent data contention and preserve per-partition message ordering.
  • Excess Instances Sit Idle: If the number of consumer instances exceeds the total number of topic partitions (like Scenario C above), the extra instances sit idle without receiving any data. They act as a hot standby ready to instantly take over partitions if one of the active consumers crashes.

The Rebalance Mechanism #

The process of reassigning partition ownership among consumers in a group is called Rebalance. This process is coordinated automatically by one Kafka broker designated as the Group Coordinator.

Rebalance is automatically triggered by the cluster when the group’s membership structure changes:

  • A new consumer instance joins the group.
  • An active consumer instance leaves the group intentionally (application shut down cleanly).
  • A consumer instance is considered dead by the broker for failing to send heartbeats within a certain time limit.
  • A change happens to the topic itself (for example, the topic’s partition count increases).

Critical Parameters Controlling Rebalance #

To avoid unnecessary wild rebalances (which can stop data processing temporarily/stop-the-world), we must set the following configuration parameters very carefully:

  • session.timeout.ms: The maximum time limit the broker waits for a consumer heartbeat before declaring it dead (default 45,000 ms or 45 seconds). If the network connection drops past this limit, the broker kicks the consumer out of the group and triggers a rebalance.
  • heartbeat.interval.ms: How often the consumer sends heartbeats to the broker (default 3,000 ms or 3 seconds). The rule of thumb is to set this to one third of session.timeout.ms.
  • max.poll.interval.ms: The maximum tolerated time gap between poll() calls on the consumer side (default 300,000 ms or 5 minutes). If our consumer application needs more than 5 minutes to process one batch (for example, blocked by a slow database query), the consumer is intentionally kicked out of the group because it’s considered stuck, triggering endless repeated rebalances.

Rebalance Protocol: Eager vs Cooperative (Incremental) #

Historically, Kafka used the Eager Rebalance protocol (which is destructive). Under this protocol, when a rebalance happens, all consumers in the group must stop reading data (stop-the-world), release all partitions they’re managing, and wait for the Coordinator broker to assign a fresh partition allocation from scratch. This drastically degrades system performance when rebalance happens on large-scale clusters.

Since Kafka 2.4, the Cooperative Rebalancing protocol (or Incremental Cooperative Rebalancing) was introduced. Under this modern protocol, only partitions that genuinely need to move to another consumer are released, while partitions still allocated to the old consumers keep processing data asynchronously without stopping. This greatly saves rebalancing wait time and minimizes production system performance disruption.

Thread Separation Architecture: Heartbeat vs Poll #

Many developers wonder why the session.timeout.ms and max.poll.interval.ms parameters are separated. Since Kafka 0.10.1, the consumer SDK uses a two-thread internal architecture:

  1. Poll Thread: The main thread executing our business logic code and calling poll(). If this thread gets stuck (e.g., from a deadlock or slow database I/O), it exceeds the max.poll.interval.ms limit.
  2. Heartbeat Thread: A dedicated background thread running automatically behind the scenes to send periodic heartbeats to the broker as long as the poll thread is still alive (not considered dead at the OS level). If our entire application process hard crashes (e.g., JVM crash), the heartbeat thread dies, and the broker detects it after session.timeout.ms passes.

This separation lets us set a heartbeat timeout sensitive to physical server failures (e.g., 10-30 seconds) without worrying about being disrupted by long local batch processing times (e.g., 5 minutes).

Offset Management & Commit #

When a consumer application reads data from partitions, it must periodically record its read progress so that if the application crashes and restarts, it doesn’t need to re-read all data from the beginning. Recording this progress is called Offset Commit.

Kafka stores this offset commit data in a special internal cluster topic named __consumer_offsets. There are two main ways to manage offset commits:

1. Automatic Commit (Auto Commit) #

Enabled by setting enable.auto.commit = true. The consumer automatically sends the last received offset commit to the broker every certain time interval (set by auto.commit.interval.ms, default 5 seconds).

  • Advantage: Very easy to use because it requires no extra code lines.
  • Disadvantage: High risk of data loss or duplication. If our application calls poll(), receives data, then the broker auto-commits that offset 5 seconds later even though our application crashed midway before successfully completing the business calculation, that data is considered successfully processed by Kafka and won’t be redelivered when the application comes back.

2. Manual Commit #

Enabled by setting enable.auto.commit = false. We as developers hold full control over when to tell Kafka that data has been successfully processed through application code. The Kafka SDK provides two methods:

  • commitSync(): Blocks the execution thread until the broker finishes responding with the offset storage status. This is very safe but lowers throughput because it triggers synchronous wait time.
  • commitAsync(): Sends offset commits asynchronously without blocking the thread. This is very fast, but has a risk: if a network failure occurs, an older offset can overwrite a newer one if response order is reversed.

Detecting and Handling Consumer Lag #

In monitoring Apache Kafka data pipeline health, the most important metric we must watch in real-time is Consumer Lag.

Consumer Lag is defined as the difference (gap) between the newest message offset successfully written by the producer on the broker disk (Log End Offset) and the last message offset successfully processed and committed by our consumer group (Current Offset):

$$\text{Consumer Lag} = \text{Log End Offset} - \text{Current Offset}$$

If the lag value keeps increasing constantly over time, it’s a red alarm indicating that our consumers can’t keep up with the producer’s data delivery speed.

How to Handle High Consumer Lag: #

  1. Increase Parallelism (Scale Up): Add more consumer instances to the consumer group. Make sure the topic’s partition count is enough to accommodate the new consumers (if there are only 3 partitions and already 3 consumers, adding a 4th consumer is pointless; we must increase the topic’s partition count first).
  2. Batch Tuning: Increase the max.poll.records value (the maximum number of messages fetched in one poll call) so consumers can process data in larger batch volumes at once.
  3. Offload Blocking I/O (Worker Thread Pattern): If data processing requires slow I/O calls (like calling external REST APIs), separate the data-pulling thread from the data-processing thread using an internal in-memory queue at the application level.

Common Mistakes (Anti-patterns) in Consumer Usage #

Here are consumer-side implementation mistakes that often prove fatal to cluster stability:

1. Doing Heavy Business Processing Directly on the Main Poll Thread #

Developers write time-consuming logic, like processing video conversion, downloading large files, or running complex SQL report queries directly inside the data-pulling while(true) loop.

Consequences: The thread blocks too long and fails to call the next poll() method before the max.poll.interval.ms (5 minutes) limit expires. The Coordinator broker considers that consumer dead, kicks it out of the group, and triggers a rebalance that stops other consumers’ read activity. When the blocked consumer finally finishes processing and tries to call poll(), it’s shocked to find its position already taken over, triggering a CommitFailedException error and an endless rebalance cycle.

# ANTI-PATTERN: Processing heavy logic that blocks the main poll thread
# This triggers a kick-out from the Consumer Group because max.poll.interval.ms is exceeded.
def jalankan_consumer_salah(consumer):
    while True:
        records = consumer.poll(timeout_ms=100)
        for record in records:
            # PROCESSING HEAVY LOGIC THAT TAKES 10 MINUTES SYNCHRONOUSLY
            # DON'T DO THIS! The main poll thread will be blocked.
            proses_analisis_laporan_berat(record.value)
            consumer.commit()

# The CORRECT solution: Worker Thread Pool pattern (Asynchronous)
# The main poll thread only pulls data quickly and hands it to an executor thread pool.
from concurrent.futures import ThreadPoolExecutor
import queue

worker_pool = ThreadPoolExecutor(max_workers=10)

def jalankan_consumer_benar(consumer):
    while True:
        # poll() calls always run fast and smooth, non-blocking
        records = consumer.poll(timeout_ms=100)
        for record in records:
            # Handing the heavy task to an asynchronous thread pool
            worker_pool.submit(proses_analisis_laporan_berat, record.value)
            
            # Offset is manually committed after status coordination is done
            consumer.commit_async()

Summary #

  • Pull Model — Kafka consumers actively pull data (pull) from the broker according to their own local capacity readiness, naturally preventing backpressure overload.
  • Poll Cycle — Consumers run on a single thread in a poll loop cycle that continuously calls poll() to fetch new data while also sending group coordination heartbeats.
  • Consumer Group — The consumer group concept allows partitions of one topic to be divided evenly among several consumer instances to achieve high read parallelism.
  • Rebalance Mechanism — The partition reassignment process triggered when a new consumer joins, a consumer leaves, or a consumer is considered dead for failing to send heartbeats past the session.timeout.ms limit.
  • Offset Management — Read progress is recorded in the internal __consumer_offsets topic. Avoid Auto Commit for important data; always use Manual Commit after successful business processing for data safety.
  • Lag Detection — Monitor consumer health by watching the Consumer Lag metric. Continuously rising lag demands scaling consumer instances or offloading heavy processing to a separate Worker Thread Pool.

← Previous: Producer Next: Broker →

About | Author | Content Scope | Editorial Policy | Privacy Policy | Disclaimer | Contact