Consumer Group #

When we build large-scale real-time data processing architectures, producing high-throughput data to Apache Kafka is only half the battle. The real challenge lies in how we can consume and process that data at a comparable speed. If our producer sends tens of thousands of messages per second to a topic, a single consumer instance running on one single thread will gradually fall behind in reading (consumer lag). This is where the Consumer Group concept becomes the main pillar of read scalability in the Kafka ecosystem. By grouping several consumer instances under one same group identity (Group ID), Kafka automatically distributes the partition read load of a topic dynamically among group members. This mechanism not only provides instant horizontal scalability, but also offers robust fault tolerance guarantees if one consumer experiences a system failure.

Basic Consumer Group Concepts and Horizontal Scalability #

In the traditional messaging world (like RabbitMQ or ActiveMQ), there’s a clear dichotomy between two main messaging models: Queueing (Point-to-Point) and Publish-Subscribe.

  • Queueing Model: Several consumers read from the same queue, but each message is only processed by exactly one consumer. This model is excellent for processing scalability, but doesn’t support multiple independent applications reading the same data simultaneously.
  • Publish-Subscribe Model: Every message is broadcast to all subscribed consumers. This model allows many applications to read the same data, but every application instance must process all the data, so it doesn’t support horizontal processing scalability within one application cluster.

Apache Kafka unifies both model advantages through the Consumer Group abstraction. In Kafka, every consumer that wants to read data must declare its group name using the group.id property.

flowchart TD
    subgraph Cluster["Kafka Cluster"]
        TopicA["Topic: order-events (4 Partitions)"]
        T0["Partition 0"]
        T1["Partition 1"]
        T2["Partition 2"]
        T3["Partition 3"]
        TopicA --> T0
        TopicA --> T1
        TopicA --> T2
        TopicA --> T3
    end

    subgraph GroupA["Consumer Group: billing-service"]
        C1["Consumer 1"]
        C2["Consumer 2"]
    end

    subgraph GroupB["Consumer Group: inventory-service"]
        C3["Consumer 3"]
    end

    T0 -->|"Read"| C1
    T1 -->|"Read"| C1
    T2 -->|"Read"| C2
    T3 -->|"Read"| C2

    T0 -->|"Broadcast Read"| C3
    T1 -->|"Broadcast Read"| C3
    T2 -->|"Broadcast Read"| C3
    T3 -->|"Broadcast Read"| C3

    style GroupA stroke:#0288d1,stroke-width:2px
    style GroupB stroke:#5d4037,stroke-width:2px
    style Cluster stroke:#333,stroke-width:2px

Through the visualization above, we can see how a Consumer Group solves two problems at once:

  1. Scalability Within One Group (billing-service): The topic workload is divided dynamically. Consumer 1 reads Partitions 0 and 1, while Consumer 2 reads Partitions 2 and 3. If we add a third consumer to this group, the partition division adjusts automatically.
  2. Independence Between Applications: The billing-service group and the inventory-service group act as two isolated logical consumers. Both receive the same data copy from the topic without disturbing each other’s read offsets.

Partition Assignment Mechanism #

The most fundamental rule in Kafka consumer group management is: One partition of a topic may only be allocated to exactly one consumer instance within one Consumer Group at a time.

This rule is crucial for providing the Ordering Guarantee. If Kafka allowed two consumers in the same group to read the same partition in parallel, we’d lose the chronological message processing order due to CPU scheduling uncertainty on each consumer machine. By limiting one partition to one consumer, Kafka ensures all messages in one partition are processed sequentially according to their write time.

This division rule produces three ratio relationship scenarios between the number of partitions and the number of consumer instances in a group:

Scenario 1: Fewer Consumers Than Partitions (Under-provisioned) #

If a topic has 4 partitions (P0, P1, P2, P3) and our group only has 2 consumers (C1, C2), Kafka divides the load evenly.

  • Division: C1 gets responsibility for reading P0 and P1. C2 gets responsibility for reading P2 and P3.
  • Impact: The system runs normally, but the CPU and network I/O load on each consumer is higher because they must manage more than one partition alternately.

Scenario 2: Consumers Equal to Partitions (Ideal/Optimized) #

If a topic has 4 partitions and our group has 4 consumers (C1, C2, C3, C4).

  • Division: Each consumer gets exactly one partition (e.g., C1 reads P0, C2 reads P1, etc.).
  • Impact: This is the most optimal design scenario in production. Each consumer has a single focus, maximizing efficient parallel processing throughput.

Scenario 3: More Consumers Than Partitions (Over-provisioned / Standby) #

If a topic has 4 partitions and our group has 5 consumers (C1, C2, C3, C4, C5).

  • Division: C1 to C4 each read one partition from P0 to P3. The fifth consumer (C5) gets no partition allocation at all.
  • Impact: C5 sits in idle/standby status. It receives no messages from the Kafka broker. Although it seems like wasted compute resources, this scenario is often used deliberately to provide instant fault tolerance. If C1 suddenly dies, C5 can immediately take over partition P0 without waiting for a new application container provisioning process from scratch.
flowchart TD
    subgraph SkenarioA["Scenario A: Consumers < Partitions"]
        direction LR
        AP0["P0"]
        AP1["P1"]
        AP2["P2"]
        AP3["P3"]
        AC1["C1"]
        AC2["C2"]
        AP0 --> AC1
        AP1 --> AC1
        AP2 --> AC2
        AP3 --> AC2
    end

    subgraph SkenarioB["Scenario B: Consumers = Partitions (Ideal)"]
        direction LR
        BP0["P0"]
        BP1["P1"]
        BP2["P2"]
        BP3["P3"]
        BC1["C1"]
        BC2["C2"]
        BC3["C3"]
        BC4["C4"]
        BP0 --> BC1
        BP1 --> BC2
        BP2 --> BC3
        BP3 --> BC4
    end

    subgraph SkenarioC["Scenario C: Consumers > Partitions"]
        direction LR
        CP0["P0"]
        CP1["P1"]
        CP2["P2"]
        CP3["P3"]
        CC1["C1"]
        CC2["C2"]
        CC3["C3"]
        CC4["C4"]
        CC5["C5 (Idle / Standby)"]
        CP0 --> CC1
        CP1 --> CC2
        CP2 --> CC3
        CP3 --> CC4
        style CC5 stroke-dasharray:5,5,stroke:#c62828,stroke-width:2px
    end

    style SkenarioA stroke:#558b2f,stroke-width:2px
    style SkenarioB stroke:#2e7d32,stroke-width:2px
    style SkenarioC stroke:#37474f,stroke-width:2px

The Role of the Broker Coordinator and Group Leadership #

Coordinating many consumers running distributedly across various servers so they always agree on who reads which partition isn’t an easy task. To solve this coordination problem without needing additional external coordination systems (like clients directly using ZooKeeper), Kafka implements the Group Coordinator and Consumer Group Leader mechanisms.

1. Group Coordinator (Broker-Side Role) #

Every Consumer Group in a Kafka cluster is assigned to one broker acting as the Group Coordinator. This Coordinator is automatically chosen by Kafka based on the hash function of the group.id name mapped to a partition of the internal __consumer_offsets topic.

The Group Coordinator’s main tasks include:

  • Accepting join requests from new consumer instances.
  • Monitoring each consumer’s health through heartbeat signal exchange.
  • Detecting if a group member is dead or unresponsive.
  • Triggering the rebalance process when group membership changes.

2. Group Leader (Consumer Client-Side Role) #

When several consumers send JoinGroup requests to the Group Coordinator, the coordinator broker designates one of those consumers as the Group Leader (usually the first consumer that successfully connects).

This role distinction is very unique in Kafka: the Group Coordinator (broker) only manages group membership, while the task of calculating the partition assignment algorithm is fully delegated to the Consumer Group Leader (client).

  • Why is this so? This design decision was made so the partition assignment process stays flexible. If the partition assignment algorithm ran on the broker side, we’d have to upgrade all Kafka brokers every time we wanted to implement a new partition assignment strategy. By delegating this calculation to the client side, developers can easily swap partition assignment strategies (like RangeAssignor, RoundRobinAssignor, or StickyAssignor) just by changing the library configuration in our application code.

Group Membership Communication Protocol Cycle #

The flow below illustrates how a group is initialized until partitions are successfully distributed to members:

sequenceDiagram
    participant C1 as Consumer 1 (Leader)
    participant C2 as Consumer 2
    participant GC as Group Coordinator (Broker)

    Note over C1,C2: Phase 1: Sending JoinGroup Requests
    C1->>GC: JoinGroup (Group ID: billing-service)
    C2->>GC: JoinGroup (Group ID: billing-service)
    
    Note over GC: Waiting for all members to register<br/>Designating Consumer 1 as the Group Leader
    GC-->>C1: JoinGroup Response (Designated as Leader + Member List)
    GC-->>C2: JoinGroup Response (Designated as Member)

    Note over C1: Running the Assignor Algorithm<br/>(C1 to P0, P1, C2 to P2, P3)

    Note over C1,C2: Phase 2: Assignment Synchronization (SyncGroup)
    C1->>GC: SyncGroup (Attaching the Partition Assignment Result)
    C2->>GC: SyncGroup (Empty request / Waiting for instructions)

    Note over GC: Storing the assignment state to __consumer_offsets
    GC-->>C1: SyncGroup Response (Receives allocation: P0, P1)
    GC-->>C2: SyncGroup Response (Receives allocation: P2, P3)

Health Detection and Timeout Configuration #

After the consumer group is formed and actively processing data, every consumer must periodically assure the Group Coordinator that they’re still alive and working well. This coordination is governed by three important configurations:

  1. heartbeat.interval.ms
    • Description: The time interval for how often a consumer sends heartbeat signals to the Group Coordinator.
    • Default Value: 3,000 ms (3 seconds). This signal is sent automatically in the background by the client’s internal heartbeat thread.
  2. session.timeout.ms
    • Description: The maximum time limit for the Group Coordinator to wait for a consumer heartbeat signal before considering that consumer dead.
    • Default Value: 45,000 ms (45 seconds) in modern Kafka.
    • Rule: This value must be larger than heartbeat.interval.ms (usually set with a 1:3 ratio). If the Group Coordinator receives no heartbeat from a consumer during this duration, the broker removes that consumer from the group and triggers a rebalance.
  3. max.poll.interval.ms
    • Description: The maximum allowed gap between .poll() function calls on our application’s main thread.
    • Default Value: 300,000 ms (5 minutes).
    • Important: This is the application logic failure detection mechanism (livelock). If our application’s main thread is busy processing very heavy data or experiences a deadlock, it won’t call the next .poll() on time even though the background heartbeat thread is still actively sending live signals to the broker. If this 5-minute limit is exceeded, the consumer is considered stuck, it’s forcibly removed from the group, and its held partitions are transferred to other consumers.

Dynamic Scenarios: Scale-Up, Scale-Down, and Fault Tolerance #

Let’s dissect chronologically what happens behind the scenes in a Kafka cluster when dynamic changes happen to our Consumer Group membership.

Scenario 1: Adding a New Consumer (Scale-Up) #

Imagine we start with one consumer C1 reading from a topic with 3 partitions (P0, P1, P2).

  1. Initial State: C1 holds all partitions: P0, P1, and P2.
  2. New Consumer Joins: We start a second application instance, C2, with the same group.id.
  3. Rebalance Notification: C2 sends a JoinGroup request to the Group Coordinator. The Coordinator notices a new member and marks the group in rebalance status. On C1’s next .poll() call, the broker returns a special instruction code telling C1 to release the partitions it holds.
  4. Partition Revocation: C1 stops reading new data, commits its last offsets to the broker, and releases ownership of partitions P0, P1, and P2.
  5. Reassignment: Both consumers send new JoinGroup requests. The Coordinator selects C1 as the leader to distribute partitions using the chosen algorithm. For example, the division result is: C1 holds P0 and P1, while C2 holds P2.
  6. Done: After the SyncGroup phase, each consumer starts reading from their new partitions from the last valid committed offset.

Scenario 2: Consumer Failure (Crash / Scale-Down) #

Imagine we have 3 consumers (C1, C2, C3) each reading one partition (P0, P1, P2).

  1. Failure Happens: The server running consumer C3 suddenly dies completely from a hardware failure. Heartbeat signals from C3 stop being sent to the Group Coordinator.
  2. Timeout Detected: The Group Coordinator waits until the session.timeout.ms limit (e.g., 45 seconds) passes. During these 45 seconds, the data on partition P2 is completely unprocessed (experiencing lag).
  3. Rebalance Triggered: After 45 seconds pass without a heartbeat, the Coordinator officially declares C3 as leaving the group. The Coordinator triggers a rebalance process to redistribute partition P2.
  4. Reallocation: Consumers C1 and C2 re-coordinate with the Coordinator. One of them (e.g., C1) is assigned to take over partition P2 alongside partition P0 it already holds.
  5. Recovery: C1 reads the last committed offset of partition P2 from the __consumer_offsets topic and continues processing pending data. The pipeline network returns to normal operation with reduced processing capacity of 2 consumers.

Implementation Code and Detecting Wild Rebalance Problems #

The most common problem developers experience in production is Excessive / Unexpected Rebalances. This problem is usually triggered by a misalignment between application data processing speed and the max.poll.interval.ms configuration.

Java SDK Anti-Pattern: Locking the Main Thread #

The following code shows a fatal error where the consumer processes data slowly on the main poll loop thread, triggering forced consumer removal from the group.

// ANTI-PATTERN: Running slow processing directly on the main poll loop thread
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "payment-processor");
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, "300000"); // 5 Minute limit
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("payment-events"));

try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) {
            // ✗ DON'T: Make slow external API calls on the main thread.
            // If one record takes 10 seconds and poll() returns 50 records,
            // total processing time is 500 seconds (8.3 minutes).
            // This exceeds max.poll.interval.ms (5 minutes), triggering CommitFailedException
            // and causing the consumer to be repeatedly removed from the group (Rebalance Loop).
            processPaymentWithExternalGateway(record.value());
            consumer.commitSync();
        }
    }
} finally {
    consumer.close();
}

Java SDK Solution: Batch Size Limiting & Processing Time Optimization #

To solve the rebalance loop problem above, we can use the approach of limiting the number of records fetched in one poll through the max.poll.records configuration, or delegate processing to an external worker thread pool while carefully managing offset commits.

Here’s a safe and recommended solution example by limiting batch size so processing is guaranteed to finish before the timeout limit:

// CORRECT: Controlling data volume per poll and aligning timeout limit parameters
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "payment-processor");
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

// ✓ Limiting to fetch at most 10 records in one poll
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "10");

// ✓ Giving processing tolerance time up to 10 minutes
props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, "600000"); 

// ✓ Setting faster application node loss detection (15 seconds)
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000");
props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, "5000");

props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("payment-events"));

try {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) {
            // ✓ Safe: Maximum 10 records * 10 seconds = 100 seconds (1.6 minutes).
            // This is far below the max.poll.interval.ms threshold of 10 minutes (600,000 ms).
            processPaymentWithExternalGateway(record.value());
        }
        // ✓ Committing per batch after all records in that poll are successfully processed
        consumer.commitSync();
    }
} catch (Exception e) {
    log.error("A message processing failure occurred", e);
} finally {
    // ✓ Always close the consumer cleanly to inform the Group Coordinator
    // that this consumer is intentionally leaving, cutting the session timeout wait.
    consumer.close();
}

When to Avoid Using Consumer Groups? #

Although Consumer Groups are very useful for data load balancing, there are certain scenarios where we should not use them.

STILL use Consumer Groups if:
  ✓ We want to distribute message processing load to many application instances evenly.
  ✓ We want the system to automatically manage partition recovery when an application node crashes.
  ✓ We need the guarantee that every message is only processed by one instance within the group.

DON'T use Consumer Groups (Use Direct Assignment / Simple Consumer) if:
  ✗ We want every application instance to receive all messages from all partitions (for example, to synchronously update local memory caches on each microservices server).
  ✗ We want to manually control the assignment of specific partitions to specific instances statically without any automatic rebalance process.

If we’re in the second scenario above, instead of calling consumer.subscribe(), we must use the direct assignment function:

// CORRECT: Using direct assign for full control without automatic rebalance processes
TopicPartition partition0 = new TopicPartition("payment-events", 0);
// ✓ Assigning this consumer specifically to only read partition 0
consumer.assign(Collections.singletonList(partition0));

Summary #

  • Horizontal Scalability — Consumer Groups allow several application instances with the same Group ID to collaborate in dynamically dividing a topic’s partition processing load.
  • One Partition One Consumer Guarantee — The basic Kafka rule limits one partition to being consumed by one instance in a group at a time to keep message order chronological.
  • Group Coordinator & Leader — The coordinator broker manages group membership status, while the Group Leader consumer calculates and distributes partition assignment.
  • Active Failure Detection — The combination of the heartbeat.interval.ms and session.timeout.ms parameters is used by the coordinator to quickly detect lost consumer instance responses.
  • Livelock & Deadlock Protection — The max.poll.interval.ms property acts as a safeguard to remove consumers from the group if the main thread gets stuck processing slow business data.
  • Rebalance Overhead — Every time a group member is added or removed, Kafka triggers a rebalance process that temporarily stops data reading (stop-the-world).
  • Rebalance Loop Rescue — Recurring rebalance problems are solved by shrinking max.poll.records or increasing the max.poll.interval.ms parameter value to match business logic duration.

← Previous: Offset Management Next: Partition Assignment Strategy →

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