Kafka consumer group rebalancing fails during GPU worker restart

Q: Root Cause Analysis

Kafka’s consumer group coordination follows a join‑group / sync‑group protocol. When a member leaves, the coordinator triggers a rebalance and expects all remaining members to send heartbeats within session.timeout.ms. The following conditions commonly break this cycle in GPU workloads:

Q: 2. Use Static Membership

Assign a stable group.instance.id derived from the pod name. This prevents the coordinator from treating a restart as a new member.

Q: 5. Tune Kubernetes termination settings

Increase the grace period to exceed the longest close() call.

In this article

Table of Contents

Problem Description

In a Kubernetes‑based distributed training pipeline, each GPU worker runs a Kafka consumer that streams training batches. When a pod is terminated (pre‑empted, rolling update, or node failure), the consumer group experiences a rebalance that never completes, leading to errors such as:


org.apache.kafka.clients.consumer.CommitFailedException: Commit cannot be completed since the group is rebalancing
java.lang.IllegalStateException: Consumer is not part of a group
org.apache.kafka.common.errors.RebalanceInProgressException
WARN  ConsumerCoordinator - Lost all assigned partitions due to rebalance timeout (session.timeout.ms exceeded)

The training job stalls, checkpoints are not written, and downstream services see a sudden drop in throughput.

Root Cause Analysis

Kafka’s consumer group coordination follows a join‑group / sync‑group protocol. When a member leaves, the coordinator triggers a rebalance and expects all remaining members to send heartbeats within session.timeout.ms. The following conditions commonly break this cycle in GPU workloads:

Abrupt pod termination: Kubernetes sends SIGTERM followed by SIGKILL after terminationGracePeriodSeconds. If the consumer does not close the client before the kill, the coordinator never receives a proper leave request, leaving the member in a Dead state.
Long processing loops: A single training step can exceed max.poll.interval.ms. The consumer stops polling, heartbeats stop, and the coordinator evicts the member, causing a rebalance while the worker is still processing.
Default static membership missing: Without group.instance.id, each restart creates a new logical member. The coordinator must reassign all partitions, leading to a “stop‑the‑world” rebalance (KIP‑429).
Coordinator unavailability: During a rolling restart of the Kafka cluster, the group coordinator may become temporarily unreachable, producing GroupCoordinatorNotAvailableException.

These factors combine to produce the observed “rebalance hangs” and “consumer not part of a group” errors, as reported in the Uber Michelangelo and Lyft ML platform incidents.

Investigation & Debugging Steps

Collect consumer logs around the restart event:


2024-06-20 12:34:56,789 WARN  ConsumerCoordinator - Lost all assigned partitions due to rebalance timeout (session.timeout.ms exceeded)
2024-06-20 12:34:58,102 ERROR RebalanceInProgressException: Rebalance in progress, cannot commit offsets
2024-06-20 12:35:01,001 INFO  ConsumerShutdownHook - Received shutdown signal, closing consumer...

Inspect Kubernetes pod termination timeline:


$ kubectl get pod gpu-trainer-0 -o jsonpath='{.status.containerStatuses[0].state.terminated}'
{
  "reason":"Error",
  "message":"Container terminated due to OOMKilled",
  "startedAt":"2024-06-20T12:33:45Z",
  "finishedAt":"2024-06-20T12:34:55Z",
  "containerID":"docker://..."
}

Check terminationGracePeriodSeconds and verify that the shutdown hook runs before the kill.

Verify consumer configuration values against the official docs (Consumer Configs):
```
session.timeout.ms = 10000
heartbeat.interval.ms = 3000
max.poll.interval.ms = 300000
```
If max.poll.interval.ms is lower than the longest training step, the consumer will be evicted.

Check group membership via the Kafka admin API:


$ kafka-consumer-groups.sh --bootstrap-server broker:9092 --describe --group training-group
GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG  OWNER
training-group  training-data   0          123456          123500          44   gpu-trainer-1
training-group  training-data   1          123450          123500          50   -

Missing owners indicate members that failed to rejoin.

Capture a packet trace during a restart (optional):
```
tcpdump -i eth0 port 9092 -w rebalance.pcap
```
Look for missing HeartbeatRequest frames from the terminated pod.

Solution Implementation

1. Enable Incremental Cooperative Rebalancing (KIP‑429)

Switch the consumer to the cooperative protocol to avoid full group stalls.


// before (default)
props.put("partition.assignment.strategy", "org.apache.kafka.clients.consumer.RangeAssignor");

// after (cooperative)
props.put("partition.assignment.strategy",
          "org.apache.kafka.clients.consumer.CooperativeStickyAssignor");

2. Use Static Membership

Assign a stable group.instance.id derived from the pod name. This prevents the coordinator from treating a restart as a new member.


// before
props.put("group.id", "training-group");

// after
props.put("group.id", "training-group");
props.put("group.instance.id", System.getenv("HOSTNAME")); // e.g., gpu-trainer-0

3. Align Timeouts with GPU workload characteristics

Parameter	Current	Recommended	Rationale
session.timeout.ms	10000	30000	Give the pod extra time to send heartbeats during graceful shutdown.
heartbeat.interval.ms	3000	5000	Matches the increased session timeout.
max.poll.interval.ms	300000	900000	Accommodates long training batches that may take >5 min.


props.put("session.timeout.ms", "30000");
props.put("heartbeat.interval.ms", "5000");
props.put("max.poll.interval.ms", "900000");

4. Add a graceful shutdown hook

Ensure the consumer leaves the group cleanly before the pod is killed.


Runtime.getRuntime().addShutdownHook(new Thread(() -> {
    try {
        consumer.wakeup(); // interrupt poll
        consumer.close(); // sends LeaveGroup request
    } catch (Exception e) {
        // log but ignore – container is terminating
    }
}));

5. Tune Kubernetes termination settings

Increase the grace period to exceed the longest close() call.


apiVersion: v1
kind: Pod
metadata:
  name: gpu-trainer
spec:
  terminationGracePeriodSeconds: 60
  containers:
  - name: trainer
    image: myorg/trainer:latest
    env:
    - name: KAFKA_GROUP_INSTANCE_ID
      valueFrom:
        fieldRef:
          fieldPath: metadata.name

Verification Steps

Deploy the updated consumer image to a staging namespace.

Trigger a rolling restart of a single GPU pod and watch the logs:


2024-06-21 08:12:00,001 INFO  ConsumerShutdownHook - Received shutdown signal, closing consumer...
2024-06-21 08:12:00,105 INFO  ConsumerCoordinator - Leaving group due to consumer close
2024-06-21 08:12:00,210 INFO  ConsumerCoordinator - Successfully left group
2024-06-21 08:12:01,015 INFO  ConsumerCoordinator - Joined group training-group with static member gpu-trainer-2
2024-06-21 08:12:01,120 INFO  ConsumerCoordinator - Assignment received: [partition=0, partition=1]

Run kafka-consumer-groups.sh --describe and confirm all partitions are assigned without - owners.
Monitor training throughput (e.g., samples/sec) before, during, and after the restart to ensure no dip.
Check that no CommitFailedException or RebalanceInProgressException appears in the pod logs for at least 10 minutes after the restart.

Prevention & Operational Best Practices

Monitoring: Export consumer_lag, rebalance_latency_ms, and consumer_coordinator_status to Prometheus. Alert on rebalance latency > 30 s.
Pod Disruption Budgets: Define a PDB that limits simultaneous GPU pod evictions to 1 to avoid multiple concurrent rebalances.
Static Membership Registry: Keep a mapping of group.instance.id to pod names in a ConfigMap for easier debugging.
Graceful Shutdown Policy: Enforce terminationGracePeriodSeconds ≥ max(max.poll.interval.ms, session.timeout.ms) + 10 s.
Version Compatibility: Use Kafka 2.8+ or Confluent Platform 6.2+ where cooperative rebalancing is GA.

Related Topic Hub: Data Infrastructure Troubleshooting Hub

FAQ

Why does the rebalance fail only when a GPU pod restarts?
The restart triggers a member leave without a proper LeaveGroup request. If the consumer is still processing a long batch, heartbeats stop, causing the coordinator to evict the member and start a full rebalance.
Can I keep the default (non‑cooperative) assignor and still avoid stalls?
It is possible by dramatically increasing session.timeout.ms and ensuring graceful shutdown, but cooperative assignors are designed to isolate the impact of a single member change, making them the recommended solution.
Do I need to set group.instance.id for every consumer?
For workloads with frequent pod churn, static membership eliminates the “member removed then added” cycle, preventing unnecessary partition revocations.
What if the Kafka broker itself becomes the coordinator during a restart?
Configure partition.assignment.strategy and enable rebalance.timeout.ms (default 60 s) to give the new coordinator time to stabilize. Also, monitor the broker logs for GroupCoordinatorNotAvailableException and consider increasing the number of broker replicas for the __consumer_offsets__ topic.
How do I verify which assignor is active at runtime?
Run kafka-configs.sh --describe --entity-type clients --entity-name and look for the partition.assignment.strategy property, or query the consumer’s client.metrics() for the assignor metric.