2.1 Consumer Groups & Coordination

Module Overview

Goal: Explain why consumers scale the way they do and how coordination works.

What You'll Learn:

• Why consumer groups exist
• How group.id determines membership
• The one-consumer-per-partition rule
• Consumer Group Coordinator role
• Heartbeats and liveness vs progress
• How consumers join and leave groups
• Static vs dynamic membership (conceptual)

What's NOT Covered Here:

• Partition assignment strategies (covered in Module 3.1)
• Rebalancing protocol details (covered in Module 4.1)
• Offset storage and commits (covered in Module 5)

Key Insight: This module answers "why coordination exists", not the mechanics of how it changes.

1. The Scalability Problem

Single Consumer Limitation

Topic: "orders" (3 partitions)

┌─────────────┐
│  CONSUMER   │ ←─── reads from ───┐
└─────────────┘                    │
                                   │
    Partition 0: [R][R][R][R]  ───┘
    Partition 1: [R][R][R][R]  ───┐
    Partition 2: [R][R][R][R]  ───┤
                                   │
                        Must read all 3 partitions

Problem:

• One consumer must read all partitions
• Throughput limited by single consumer capacity
• No fault tolerance (if consumer dies, consumption stops)
• Can't scale horizontally

The Solution: Consumer Groups

Topic: "orders" (3 partitions)

Consumer Group: "order-processors"

┌─────────────┐
│ CONSUMER 1  │ ←─── Partition 0: [R][R][R][R]
└─────────────┘

┌─────────────┐
│ CONSUMER 2  │ ←─── Partition 1: [R][R][R][R]
└─────────────┘

┌─────────────┐
│ CONSUMER 3  │ ←─── Partition 2: [R][R][R][R]
└─────────────┘

Each consumer reads from ONE partition
Load distributed across 3 consumers

Benefits:

• ✅ Parallel processing — Multiple consumers work simultaneously
• ✅ Scalability — Add more consumers to handle more load
• ✅ Fault tolerance — If one consumer fails, others take over
• ✅ Load balancing — Work distributed evenly

2. What is a Consumer Group?

Definition

A Consumer Group is a set of consumers that:

• Share the same group.id
• Cooperate to consume from a topic
• Each consume from different partitions
• Coordinate through a Group Coordinator

Think of a consumer group as a team of workers where:

• Each worker handles specific tasks (partitions)
• No two workers do the same task
• If a worker leaves, others pick up their tasks
• The manager (coordinator) oversees the team

The group.id Configuration

props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-processors");

The group.id determines:

• Which consumers belong to the same group
• Which offsets to use (each group tracks its own offsets)
• Who coordinates with whom

Consumer Group: "order-processors"
├─ Consumer A (reads partitions 0, 1)
├─ Consumer B (reads partitions 2, 3)
└─ Consumer C (reads partitions 4, 5)

Consumer Group: "analytics-processors" (different group!)
├─ Consumer X (reads partitions 0, 1, 2)
└─ Consumer Y (reads partitions 3, 4, 5)

Both groups read the SAME topic independently!

Multiple Groups Reading the Same Topic

Topic: "orders"

┌─────────────────────────────────────────┐
│  Group: "order-processors"              │
│  Consumer 1 → Partition 0               │
│  Consumer 2 → Partition 1               │
└─────────────────────────────────────────┘
            ↓
    [All orders processed]

┌─────────────────────────────────────────┐
│  Group: "analytics"                     │
│  Consumer 1 → Partitions 0, 1           │
└─────────────────────────────────────────┘
            ↓
    [Analytics generated]

┌─────────────────────────────────────────┐
│  Group: "fraud-detection"               │
│  Consumer 1 → Partitions 0, 1           │
└─────────────────────────────────────────┘
            ↓
    [Fraud checks performed]

Key Point: Each consumer group has:

• Independent consumption progress
• Separate offset tracking
• No impact on other groups

3. The One-Consumer-Per-Partition Rule

The Golden Rule

A partition can be consumed by only ONE consumer within a group at a time.

✅ VALID: Each partition assigned to one consumer

Consumer Group: "processors"
Consumer 1 → Partition 0
Consumer 2 → Partition 1
Consumer 3 → Partition 2

❌ INVALID: Two consumers reading same partition

Consumer 1 → Partition 0  ┐
Consumer 2 → Partition 0  ┘ ← NOT ALLOWED!

Why this rule exists:

• Ensures ordering within partition
• Prevents duplicate processing (within group)
• Simplifies offset management

Scaling Limits

Topic: 3 partitions

┌──────────────────────────────────────────────┐
│  1 Consumer: All 3 partitions               │
│  Consumer 1 → Partitions 0, 1, 2            │
└──────────────────────────────────────────────┘

┌──────────────────────────────────────────────┐
│  2 Consumers: Partitions split               │
│  Consumer 1 → Partitions 0, 1               │
│  Consumer 2 → Partition 2                    │
└──────────────────────────────────────────────┘

┌──────────────────────────────────────────────┐
│  3 Consumers: Ideal (1:1 mapping)           │
│  Consumer 1 → Partition 0                    │
│  Consumer 2 → Partition 1                    │
│  Consumer 3 → Partition 2                    │
└──────────────────────────────────────────────┘

┌──────────────────────────────────────────────┐
│  4 Consumers: One is IDLE!                  │
│  Consumer 1 → Partition 0                    │
│  Consumer 2 → Partition 1                    │
│  Consumer 3 → Partition 2                    │
│  Consumer 4 → (no partitions assigned)       │
└──────────────────────────────────────────────┘

Key Insight: Max consumers = number of partitions

Adding more consumers than partitions is wasteful — extra consumers sit idle.

Optimal Consumer Count

Topic partitions: P
Consumer count: C

Best performance:  C = P  (1:1 mapping)
Acceptable:        C < P  (some consumers handle multiple partitions)
Wasteful:          C > P  (idle consumers)

Example:

• Topic has 12 partitions
• Optimal: 12 consumers
• Acceptable: 6 consumers (each handles 2 partitions)
• Wasteful: 15 consumers (3 sit idle)

4. The Consumer Group Coordinator

What is the Group Coordinator?

The Group Coordinator is a Kafka broker that:

• Manages consumer group membership
• Orchestrates partition assignment
• Tracks heartbeats from consumers
• Triggers rebalances when needed
• Stores group metadata

Think of it as the team manager who:

• Knows who's on the team
• Assigns work to team members
• Checks if team members are alive
• Reorganizes when someone joins/leaves

How Coordinator is Selected

Consumer Group: "order-processors"
↓
Hash(group.id) % number_of_brokers
↓
Coordinator: Broker 2

Key Points:

• Each consumer group has one coordinator
• Coordinator is a regular Kafka broker
• Selected deterministically (not random)
• If coordinator broker fails, another takes over

Coordinator Responsibilities

┌────────────────────────────────────────────┐
│       CONSUMER GROUP COORDINATOR           │
├────────────────────────────────────────────┤
│                                            │
│  1. Track group membership                 │
│     - Which consumers are in the group     │
│     - Consumer metadata and capabilities   │
│                                            │
│  2. Monitor consumer health                │
│     - Receive heartbeats                   │
│     - Detect dead consumers                │
│                                            │
│  3. Manage partition assignment            │
│     - Coordinate assignment protocol       │
│     - Distribute assignment to consumers   │
│                                            │
│  4. Store group offsets                    │
│     - Receive offset commits               │
│     - Provide offsets on recovery          │
│                                            │
│  5. Trigger rebalances                     │
│     - When consumers join/leave            │
│     - When partition count changes         │
│                                            │
└────────────────────────────────────────────┘

5. Consumer Registration and Group Join

The Join Process

When a consumer starts, it goes through this process:

1. Consumer starts
   ├─ Reads config: bootstrap.servers, group.id
   └─ Calls consumer.subscribe(["orders"])

2. Find Group Coordinator
   ├─ Send FindCoordinatorRequest
   └─ Receive coordinator broker address

3. Join Group
   ├─ Send JoinGroupRequest to coordinator
   ├─ Coordinator waits for all consumers
   └─ Coordinator selects group leader

4. Sync Group
   ├─ Group leader computes partition assignment
   ├─ Leader sends assignment to coordinator
   ├─ Coordinator distributes to all consumers
   └─ Consumers receive their partitions

5. Start Consuming
   └─ Consumers begin polling from assigned partitions

Join Group Timeline

Time  Consumer              Coordinator
────┼─────────────────────┼──────────────────────────────
  1 │ Start up            │
  2 │ FindCoordinator ──→ │
  3 │ ←─ Broker 2         │
  4 │ JoinGroup ────────→ │
  5 │                     │ Wait for other consumers...
  6 │                     │ (session.timeout.ms)
  7 │                     │ Select leader (first consumer)
  8 │ ←── You are leader  │
  9 │ Compute assignment  │
 10 │ SyncGroup ────────→ │ (with assignment)
 11 │ ←── Your partitions │ (0, 1, 2)
 12 │ Start polling       │

Group Leader Role

Important: One consumer in each group is designated the group leader.

Consumer Group: "order-processors"
├─ Consumer 1 (GROUP LEADER ⭐)
├─ Consumer 2
└─ Consumer 3

Group Leader Responsibilities:

• Receives list of all consumers in group
• Computes partition assignment (using assignment strategy)
• Sends assignment back to coordinator
• Coordinator distributes to all consumers

Key Points:

• Group leader is just another consumer
• It's not a separate process
• Leadership can change between rebalances
• Leader doesn't have special privileges (just extra work)

6. Heartbeats: Liveness vs Progress

What are Heartbeats?

Heartbeats are periodic signals from consumers to the coordinator saying:

"I'm alive and part of the group."

Consumer                           Coordinator
   │                                   │
   ├──────── heartbeat ──────────────→ │
   │                                   │
   ├──────── heartbeat ──────────────→ │
   │                                   │
   ├──────── heartbeat ──────────────→ │
   │                                   │
   ├──────── heartbeat ──────────────→ │

If heartbeats stop: Coordinator assumes consumer is dead and triggers rebalance.

Two Dimensions of Consumer Health

1. Liveness (Heartbeat-Based)

props.put("session.timeout.ms", 10000);      // 10 seconds
props.put("heartbeat.interval.ms", 3000);    // 3 seconds

Question: Is the consumer process alive?

Detection:

• Consumer sends heartbeat every heartbeat.interval.ms
• If coordinator doesn't receive heartbeat for session.timeout.ms, consumer is considered dead

Consumer alive:    ❤️ ──→ ❤️ ──→ ❤️ ──→ ❤️
Consumer dead:     ❤️ ──→ ❤️ ──→ ⚫ (no heartbeat)
                              ↓
                         Rebalance triggered

2. Progress (Poll-Based)

props.put("max.poll.interval.ms", 300000);   // 5 minutes

Question: Is the consumer making progress (calling poll)?

Detection:

• Consumer must call poll() within max.poll.interval.ms
• If poll is not called in time, consumer is considered stuck

Consumer progressing:  poll() → poll() → poll() → poll()
Consumer stuck:        poll() → 😴😴😴😴😴😴 (long processing)
                                     ↓
                              Rebalance triggered

Why Both Matter

Scenario 1: Process alive but stuck
- Heartbeats still sent (background thread)
- But poll() not called (application thread stuck)
- After max.poll.interval.ms → Rebalance
- Prevents zombie consumers

Scenario 2: Process dead
- No heartbeats sent
- After session.timeout.ms → Rebalance
- Fast detection of failures

Key Insight:

• Heartbeats detect process crashes
• Poll interval detects application hangs

Configuration Relationships

heartbeat.interval.ms < session.timeout.ms < max.poll.interval.ms

Example (typical):
heartbeat.interval.ms    = 3 seconds
session.timeout.ms       = 10 seconds
max.poll.interval.ms     = 300 seconds (5 minutes)

Rules:

• heartbeat.interval.ms should be 1/3 of session.timeout.ms
• session.timeout.ms should be short enough to detect failures quickly
• max.poll.interval.ms should be long enough for record processing

7. Consumer Group States

Group Lifecycle States

┌─────────────────────────────────────────────┐
│         CONSUMER GROUP STATES               │
├─────────────────────────────────────────────┤
│                                             │
│  Empty                                      │
│    ├─ No consumers in group                │
│    └─ Metadata may still exist             │
│                                             │
│  PreparingRebalance                         │
│    ├─ Waiting for consumers to join        │
│    └─ Old generation being closed           │
│                                             │
│  CompletingRebalance                        │
│    ├─ Assignment computed                   │
│    └─ Waiting for consumers to sync         │
│                                             │
│  Stable                                     │
│    ├─ All consumers consuming               │
│    └─ No rebalance in progress              │
│                                             │
│  Dead                                       │
│    └─ Group being removed                   │
│                                             │
└─────────────────────────────────────────────┘

State Transitions

Empty ──→ PreparingRebalance (consumer joins)
    ↑           ↓
    │   CompletingRebalance (assignment ready)
    │           ↓
    └───── Stable (consumers consuming)
                │
                ↓
         PreparingRebalance (consumer leaves/dies)

8. Static vs Dynamic Membership

Dynamic Membership (Default)

// No special configuration needed
props.put("group.id", "order-processors");

Behavior:

• Consumer gets new member ID on every restart
• Rebalance triggered on every join/leave
• Fast failover (session.timeout.ms)

Use when:

• Consumers restart frequently
• Fast failure detection needed
• Default for most applications

Static Membership

props.put("group.id", "order-processors");
props.put("group.instance.id", "consumer-1"); // Static ID

Behavior:

• Consumer has persistent identity
• Rebalance not triggered if consumer rejoins within session.timeout.ms
• Partitions retained if consumer restarts quickly

Use when:

• Consumers have persistent state (local caches, RocksDB)
• Minimizing rebalances is critical
• Controlled deployments (rolling restarts)

Comparison

Static Membership: Conflict Prevention Mechanism

Problem: What happens if two consumers try to use the same group.instance.id?

Kafka has a built-in fencing mechanism to prevent duplicate static members from causing chaos.

How Fencing Works

Scenario: Two consumers with same group.instance.id

1. Consumer A (original) is running with group.instance.id="consumer-1"
   ├─ Registered with coordinator
   ├─ Has epoch number: 5
   └─ Consuming partitions 0, 1, 2

2. Consumer B (duplicate) starts with group.instance.id="consumer-1"
   ├─ Sends JoinGroupRequest to coordinator
   └─ Coordinator checks existing registrations

3. Coordinator detects conflict:
   ├─ Same group.instance.id already exists
   ├─ Consumer B gets REJECTED
   └─ Error: org.apache.kafka.common.errors.FencedInstanceIdException

4. Result:
   ├─ Consumer A (original) remains active and keeps consuming
   └─ Consumer B (duplicate) cannot join the group

Epoch-Based Protection

Each static member has an epoch number that increments on each rejoin:

Consumer "consumer-1" lifecycle:

First start:     Epoch = 1
First restart:   Epoch = 2
Second restart:  Epoch = 3
...and so on

Coordinator Rejection Rules:

// Pseudo-code for coordinator logic
if (incomingRequest.groupInstanceId == existingMember.groupInstanceId) {
    if (incomingRequest.epoch <= existingMember.epoch) {
        // Reject: duplicate or stale consumer
        throw new FencedInstanceIdException(
            "Consumer with same group.instance.id already exists"
        );
    } else {
        // Accept: this is a legitimate restart
        // (epoch is higher, meaning consumer restarted)
        allowJoin();
    }
}

What This Means for You

✅ Safety Guarantee:

• You cannot accidentally run two consumers with the same group.instance.id
• The coordinator protects against configuration errors
• Only one instance can be active at a time

❌ Error Handling:

try {
    consumer.subscribe(Collections.singletonList("orders"));
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        // ... process records ...
    }
} catch (FencedInstanceIdException e) {
    // Another consumer with same group.instance.id exists!
    // This is a fatal error - you must fix your configuration
    log.error("Duplicate group.instance.id detected: {}", e.getMessage());
    // Typically: exit and alert ops team
    System.exit(1);
}

When This Happens:

Best Practices:

1. Unique Static IDs: Ensure each consumer has a unique group.instance.id

   // Good: Use hostname or pod name
   props.put("group.instance.id", System.getenv("HOSTNAME"));
   
   // Bad: Hardcoded same ID for all instances
   props.put("group.instance.id", "consumer-1");
   

2. Monitor for Fencing: Alert on FencedInstanceIdException — it indicates a configuration problem

3. Graceful Shutdown: Always close consumers cleanly so epoch updates properly

9. Consumer Group Example

Three Consumers in One Group

// Consumer 1
Properties props1 = new Properties();
props1.put("bootstrap.servers", "localhost:9092");
props1.put("group.id", "order-processors");  // Same group
props1.put("client.id", "consumer-1");       // Unique ID
// ... serializers ...

KafkaConsumer<String, String> consumer1 = new KafkaConsumer<>(props1);
consumer1.subscribe(Collections.singletonList("orders"));

// Consumer 2
Properties props2 = new Properties();
props2.put("bootstrap.servers", "localhost:9092");
props2.put("group.id", "order-processors");  // Same group
props2.put("client.id", "consumer-2");       // Unique ID
// ... serializers ...

KafkaConsumer<String, String> consumer2 = new KafkaConsumer<>(props2);
consumer2.subscribe(Collections.singletonList("orders"));

// Consumer 3
Properties props3 = new Properties();
props3.put("bootstrap.servers", "localhost:9092");
props3.put("group.id", "order-processors");  // Same group
props3.put("client.id", "consumer-3");       // Unique ID
// ... serializers ...

KafkaConsumer<String, String> consumer3 = new KafkaConsumer<>(props3);
consumer3.subscribe(Collections.singletonList("orders"));

Result:

Topic: "orders" (3 partitions)

Consumer 1 → Partition 0
Consumer 2 → Partition 1
Consumer 3 → Partition 2

All in group: "order-processors"
All read same topic: "orders"
Each gets different partitions

10. Mental Model Summary

Key Concepts

CONSUMER GROUP = Team of cooperating consumers
    │
    ├─ Identified by group.id
    │
    ├─ Managed by Group Coordinator (a broker)
    │
    ├─ One partition → One consumer (within group)
    │
    ├─ Heartbeats ensure liveness
    │
    └─ Rebalances handle changes

The Coordination Flow

1. Consumers join group (via coordinator)
2. Partitions assigned to consumers
3. Consumers start polling
4. Consumers send heartbeats
5. If changes occur → Rebalance → Go to 1

Questions You Should Be Able to Answer

✅ Why do consumer groups exist?
✅ What does group.id determine?
✅ What is the one-consumer-per-partition rule?
✅ What is the Group Coordinator's role?
✅ What are heartbeats for?
✅ What's the difference between liveness and progress?
✅ What's the difference between static and dynamic membership?

Questions We Haven't Answered Yet

❌ How are partitions assigned? → Module 3.1
❌ What exactly happens during a rebalance? → Module 4.1
❌ How are offsets managed? → Module 5
❌ How does the poll loop work internally? → Module 6.1

11. What's Next?

Now that you understand why consumers coordinate, the next modules will cover:

• Module 3.1: How partitions are assigned to consumers (assignment strategies)
• Module 4.1: What happens during rebalances (the mechanics of change)
• Module 5: How offsets are managed and committed
• Module 6.1: The internal mechanics of the poll loop

Summary

You learned:

• ✅ Consumer groups enable parallel consumption and scalability
• ✅ group.id determines group membership
• ✅ One partition per consumer (within a group) is the golden rule
• ✅ Group Coordinator manages group membership and assignments
• ✅ Heartbeats detect dead consumers, poll interval detects stuck consumers
• ✅ Static membership reduces rebalances for stateful consumers

Next up: Partition Assignment Strategies (Module 3.1) — Learn how partitions are distributed to consumers.