分配不是消费者自己决定的事
一个 topic 有 3 个分区，先起一个消费者，它拿全部 3 个；再起第二个，理想状态是 2 和 1。这个从「3-0」到「2-1」的调整过程就叫再平衡（rebalance，也译重平衡）。
很多人对再平衡的印象停留在「会卡一下」。这一节要把这一下拆开：卡在哪个协议阶段、卡在客户端哪一行、eager 和 cooperative 卡的范围有什么不同。全程用两个真实消费者进程的时间线说话。
先给结论，后面再用数据支撑：
- 再平衡由协调器（group coordinator）发起，但分配方案由组内的 leader 成员在客户端算出来。
- 协议分三段：JoinGroup 收集成员与订阅，SyncGroup 下发分配，Heartbeat 维持存活。
- 应用线程的停顿发生在
poll()内部等待 JoinGroup/SyncGroup 完成的那段，不是在收消息的时候。 - eager 撤销全部分区，cooperative 只撤销要搬走的那部分，代价是多一轮协议。
三个阶段各做什么
JoinGroup：把人凑齐
协调器把组状态改成 PreparingRebalance，然后等所有成员重新发 JoinGroup。每个成员在请求里带上：
memberId：第一次为空，协调器分配一个；之后带上，表示「还是我」。protocols：自己支持的分配器列表，例如range、cooperative-sticky。sessionTimeoutMs、rebalanceTimeoutMs：会话超时与本次再平衡的等待上限。- 订阅信息：订阅了哪些 topic。
协调器选出一个所有成员都支持的协议，选出 leader（通常是第一个加入的成员），把成员列表放进 JoinGroup 响应发给 leader。普通成员拿到的响应里没有成员列表，它们只需要等 SyncGroup。
SyncGroup：leader 算，协调器发
leader 成员在 onLeaderElected 回调里调用分配器，算出「谁拿哪些分区」，然后把完整方案放进 SyncGroup 请求发给协调器。其他成员发的是空方案的 SyncGroup。
协调器把 leader 给的方案分发给每个成员。成员收到响应后，在 onJoinComplete 里安装分配、恢复拉取。到这里一轮再平衡结束，组状态变成 Stable。
Heartbeat：活着，并且知道组变了
加入成功后，客户端会启动一个独立的心跳线程，按 heartbeat.interval.ms 给协调器发 Heartbeat。它的返回值不只是「你还活着」：
- 正常：什么都不用做。
REBALANCE_IN_PROGRESS：组开始新一轮再平衡，主线程下一次poll()要重新加入。UNKNOWN_MEMBER_ID/ILLEGAL_GENERATION：成员身份失效，必须清掉 memberId 重新加入。
心跳线程独立于应用线程，这一点很关键：应用线程在 poll() 里等 JoinGroup 时，心跳仍然在发，所以成员不会因为「正在再平衡」而超时。
谁在算分配
这是经典协议和 KIP-848 最大的区别之一：经典协议的分配器跑在客户端。
读 4.3.0 的源码可以确认这条链路。ConsumerCoordinator.poll() 在发现 rejoinNeededOrPending() 后调用 ensureActiveGroup，后者进入 AbstractCoordinator.joinGroupIfNeeded，里面有一行决定停顿的代码：
// org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
final RequestFuture<ByteBuffer> future = initiateJoinGroup();
client.poll(future, timer);
if (!future.isDone()) {
// we ran out of time
return false;
}
client.poll(future, timer) 是一个阻塞等待：网络线程收响应，应用线程停在这里。JoinGroup 响应回来后，leader 还要算分配、再发 SyncGroup，然后继续等 SyncGroup 响应。这两段等待都发生在应用线程调用 poll() 的栈里。
还有一个常被忽略的事实：源码 GroupRebalanceConfig 里，消费者协议的 rebalanceTimeoutMs 直接取 max.poll.interval.ms。默认 300000 意味着，一轮再平衡理论上可以等 5 分钟。这也解释了常见现象：某个消费者卡住不 poll，其他成员会一直等在 PreparingRebalance，直到它被踢掉或超时。
三种触发条件
再平衡不只由「消费者上下线」触发。经典的触发条件有三类：
| 触发 | 例子 | 谁先知道 |
|---|---|---|
| 成员变化 | 新消费者加入、消费者关闭、成员被判定超时 | 协调器 |
| 订阅变化 | subscribe() 换了 topic 列表、正则匹配到新 topic | 消费者自己，下一次心跳或 poll 时上报 |
| 分区数变化 | topic 增加分区 | 消费者刷新元数据后发现，主动请求重新加入 |
第二类常被低估：用正则订阅时，新建一个匹配的 topic 会让整个组再平衡一次；如果各成员刷新元数据的时间不一致，还可能连续触发多轮。源码在加入前强制刷新一次元数据来缓解这个问题。
eager 与 cooperative 的区别
两者的差别只在撤销范围，但影响很大。
eager（RangeAssignor、RoundRobinAssignor 等）：所有成员先撤销全部分区，再重新分配。分配方案里 p2 从 A 搬到 B，A 也会把 p0、p1 一起停掉再拿回来。停的范围是「全部」。
cooperative（CooperativeStickyAssignor）：成员只撤销「新方案里不再属于自己」的分区。p2 要搬走，A 只停 p2，p0、p1 继续消费。代价是协议要多走一轮：第一轮 A 撤销 p2，第二轮 B 才拿到 p2。
partition.assignment.strategy 的默认值是一个列表，本机配置日志里是：
partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor, class org.apache.kafka.clients.consumer.CooperativeStickyAssignor]
实际生效的是「所有成员都支持的第一个」。两个消费者都用默认值时，选中的是 Range，也就是 eager。我们可以在组元数据里看到协商结果：
protocolType":"consumer","generation":1,"protocol":"range"
要让组走 cooperative，必须显式把 partition.assignment.strategy 设成 CooperativeStickyAssignor，并且组内所有成员都这么配。混配的后果是协商失败或退回 eager。
stop-the-world 到底停在哪一步
把一次 eager 再平衡按时间顺序摊开：
- 协调器收到成员变化，组进入
PreparingRebalance，通过心跳响应告诉现有成员。 - 现有成员在下一次
poll()里发现要重新加入。 onJoinPrepare被调用：先做一次自动提交（如果开了自动提交），再调用用户的onPartitionsRevoked，此时分区已经被撤销，拉取停止。- 发 JoinGroup，应用线程在
client.poll(future, timer)上等。 - leader 收到成员列表，算分配，发 SyncGroup。
- 所有成员收到 SyncGroup 响应，
onJoinComplete安装新分配，调用onPartitionsAssigned。 poll()返回，拉取恢复。
所以「停」的准确位置是 第 3 步到第 6 步之间：从撤销回调开始，到分配回调结束。对 eager 来说，这段时间内这个消费者一个分区都不消费；对 cooperative 来说，只停被撤销的那些分区，其余分区不受影响。
停的时长由最慢的成员决定，不是由本地算分配的速度决定。协调器要等所有成员都重新 JoinGroup，任何一个成员卡住，其他人都得等，直到它被踢出或 rebalanceTimeoutMs 到期。这就是「一个消费者卡住，全组停摆」的机制来源。
三个时间参数的关系
| 参数 | 默认值 | 谁在检查 | 作用 |
|---|---|---|---|
session.timeout.ms | 45000 | 协调器 | 多久收不到心跳就判定成员死亡 |
heartbeat.interval.ms | 3000 | 消费者心跳线程 | 多久发一次心跳 |
max.poll.interval.ms | 300000 | 消费者心跳线程 | 两次 poll() 间隔超过多久就主动退组 |
heartbeat.interval.ms必须明显小于session.timeout.ms，官方建议 1/3。否则一次网络抖动就可能误判死亡。broker 端还限制了会话超时的上下界，本机实测group.min.session.timeout.ms=6000、group.max.session.timeout.ms=1800000。max.poll.interval.ms要大于「处理一批消息的最长时间」。它同时是再平衡的等待上限，配得太小会把正常的长耗时处理误判成卡死。session.timeout.ms和max.poll.interval.ms管的是两种不同的「不响应」：前者是进程没了/网络断了，心跳发不出去；后者是进程活着、心跳正常，但应用线程没调用poll()。
本机还有一个影响首次分配的参数：group.initial.rebalance.delay.ms=3000。组第一次建立时协调器多等 3 秒，给其他成员加入的机会；下面实验里第一个消费者 3.4 秒才拿到分配，主要就是这 3 秒。
实验一：两个消费者加入同一个组
程序
一个可复用的观察程序，打印每次撤销/分配的时间戳和分区：
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.*;
public class RebalanceObserver {
private static final long T0 = System.currentTimeMillis();
private static void log(String s) {
System.out.printf("[%6d ms] %s%n", System.currentTimeMillis() - T0, s);
}
public static void main(String[] args) throws Exception {
String name = args[0];
long runMs = Long.parseLong(args[1]);
String mode = args.length > 2 ? args[2] : "normal";
String strategy = args.length > 3 ? args[3] : "range";
Properties p = new Properties();
p.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:19211");
p.put(ConsumerConfig.GROUP_ID_CONFIG, "b1.rebal.group");
p.put(ConsumerConfig.CLIENT_ID_CONFIG, "b1-rebal-" + name);
p.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
p.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
p.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
p.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
p.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "10000");
p.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, "3000");
p.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, "10000");
p.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
strategy.equals("cooperative")
? "org.apache.kafka.clients.consumer.CooperativeStickyAssignor"
: "org.apache.kafka.clients.consumer.RangeAssignor");
log(name + " start mode=" + mode + " strategy=" + strategy);
try (KafkaConsumer<String, String> c = new KafkaConsumer<>(p)) {
c.subscribe(List.of("b1.offsets.demo"), new ConsumerRebalanceListener() {
public void onPartitionsRevoked(Collection<TopicPartition> ps) {
log(name + " REVOKED " + ps);
}
public void onPartitionsAssigned(Collection<TopicPartition> ps) {
log(name + " ASSIGNED " + ps);
}
});
long start = System.currentTimeMillis();
boolean stalled = false;
while (System.currentTimeMillis() - start < runMs) {
if (mode.equals("stall") && !stalled && System.currentTimeMillis() - start > 2000) {
log(name + " STALL: sleep 20s without poll");
stalled = true;
Thread.sleep(20000);
log(name + " STALL over, polling again");
}
c.poll(Duration.ofMillis(500));
}
}
log(name + " closed");
}
}
运行
A 先跑 42 秒，第 8 秒启动 B 跑 22 秒，第 18 秒抓一次成员表：
java -cp ... RebalanceObserver A 42000 normal range > eager_a.log 2>&1 &
sleep 8
java -cp ... RebalanceObserver B 22000 normal range > eager_b.log 2>&1 &
sleep 10
docker exec mq-b1 /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092 \
--describe --group b1.rebal.group --members --verbose > eager_members.log
wait
A 的时间线
本机实测（Temurin 25.0.5+8 / Kafka 4.3.0 KRaft 单节点 / 2026-10-29；单次运行，随调度波动，不作基准）：
[ 7 ms] A start mode=normal strategy=range
[ 3447 ms] A ASSIGNED [b1.offsets.demo-0, b1.offsets.demo-1, b1.offsets.demo-2]
[ 9442 ms] A REVOKED [b1.offsets.demo-0, b1.offsets.demo-1, b1.offsets.demo-2]
[ 9450 ms] A ASSIGNED [b1.offsets.demo-0, b1.offsets.demo-1]
[ 33471 ms] A REVOKED [b1.offsets.demo-0, b1.offsets.demo-1]
[ 33489 ms] A ASSIGNED [b1.offsets.demo-0, b1.offsets.demo-1, b1.offsets.demo-2]
[ 42469 ms] A end, leaving group
B 的时间线
[ 6 ms] B start mode=normal strategy=range
[ 1437 ms] B ASSIGNED [b1.offsets.demo-2]
[ 22568 ms] B end, leaving group
三段时间可以直接读出来：
- 首次加入用了 3.4 秒：不是网络慢，是
group.initial.rebalance.delay.ms=3000的启动延迟。 - B 加入触发的再平衡：B 从启动到拿到 p2 用了 1437 ms；A 在 9442 ms 撤销全部三个分区，9450 ms 拿回两个，本地停顿只有 8 ms。B 的 1.4 秒大部分花在协调器等 A 重新加入上。A 的下一次 poll 在 500 ms 内，心跳间隔 3 秒，所以最坏要等一轮心跳。
- B 退出触发的再平衡：B 在约 30.5 秒关闭，A 在 33.47 秒拿回全部三个分区，间隔约 3 秒，正好是一个
heartbeat.interval.ms。协调器通过心跳响应把「组变了」告诉 A，A 才知道要重新加入。
成员表也能看到这一刻的分配：
GROUP CONSUMER-ID HOST CLIENT-ID #PARTITIONS CURRENT-ASSIGNMENT
b1.rebal.group b1-rebal-A-... /172.17.0.1 b1-rebal-A 2 b1.offsets.demo:0,1
b1.rebal.group b1-rebal-B-... /172.17.0.1 b1-rebal-B 1 b1.offsets.demo:2
A 两个、B 一个，和日志里的分配回调一致。--describe 的普通输出同样能看到每个分区归谁；这里 CURRENT-OFFSET 是 -，因为程序关掉了自动提交也没手动提交，不影响看分配归属。
实验二：同样的两个人，换成 cooperative
只把分配器换成 CooperativeStickyAssignor，时间参数不变。本机实测：
=== A ===
[ 3598 ms] A ASSIGNED [b1.offsets.demo-0, b1.offsets.demo-1, b1.offsets.demo-2]
[ 9631 ms] A REVOKED [b1.offsets.demo-2]
[ 9639 ms] A ASSIGNED []
[ 12621 ms] A ASSIGNED []
[ 33635 ms] A ASSIGNED [b1.offsets.demo-2]
=== B ===
[ 1030 ms] B ASSIGNED []
[ 4030 ms] B ASSIGNED [b1.offsets.demo-2]
对比 eager 的 A 日志，差别一目了然：
- A 只撤销了
p2，p0、p1 没有出现在撤销列表里。这两个分区在整个再平衡期间继续被 A 消费。 - A 在 9639 ms 拿到空分配
[]，12621 ms 又收到一次，这是 cooperative 两轮协议完成的信号。 - B 先拿到空分配，到 4030 ms 才拿到 p2。B 从启动到真正持有分区用了约 4 秒，比 eager 的 1.4 秒长。
cooperative 的取舍很清楚：用多一轮协议换更小的停顿范围。分区多、单分区流量大的场景，这个交换通常值得；分区很少、再平衡本来就只有几毫秒的场景，eager 的简单性更划算。
实验三：停 poll 超过 max.poll.interval.ms
让 A 正常运行 60 秒，B 启动后先正常 poll，2 秒后停 20 秒不 poll。B 的 max.poll.interval.ms 设成 10000，心跳仍然每 3 秒发一次。
本机实测 A 的时间线：
[ 3464 ms] A ASSIGNED [p0, p1, p2]
[ 9447 ms] A REVOKED [p0, p1, p2]
[ 9458 ms] A ASSIGNED [p0, p1] <- B 加入，A 让出 p2
[ 21460 ms] A REVOKED [p0, p1]
[ 21470 ms] A ASSIGNED [p0, p1, p2] <- B 被踢，A 收回全部
[ 30476 ms] A REVOKED [p0, p1, p2]
[ 30484 ms] A ASSIGNED [p0, p1] <- B 重新加入，A 再让出 p2
B 的时间线：
[ 7 ms] B start mode=stall strategy=range
[ 2345 ms] B STALL: sleep 20s without poll (max.poll.interval.ms=10000)
[ 22356 ms] B STALL over, polling again
[ 24300 ms] B ASSIGNED [b1.offsets.demo-2]
关键时间点：B 在约 8.35 秒（全局）停止 poll，A 在 21.46 秒收回全部三个分区，间隔约 13 秒。这个数字由两段组成：max.poll.interval.ms=10000 到期后心跳线程发起退组，加上最多一个 heartbeat.interval.ms=3000 的传播延迟。
在 B 停 poll 期间抓成员表，本机输出：
Warning: Consumer group 'b1.rebal.group' is rebalancing.
GROUP CONSUMER-ID HOST CLIENT-ID #PARTITIONS CURRENT-ASSIGNMENT
b1.rebal.group b1-rebal-A-... /172.17.0.1 b1-rebal-A 2 b1.offsets.demo:0,1
B 已经不在成员列表里，组处于 rebalancing。B 的 20 秒睡眠结束后再次 poll，重新加入，又触发一轮再平衡，A 再次让出 p2。
「被踢出」这条路径的判定在客户端心跳线程里，不在协调器。源码 AbstractCoordinator 的心跳循环：
} else if (heartbeat.pollTimeoutExpired(now)) {
// the poll timeout has expired, which means that the foreground thread has stalled
// in between calls to poll().
handlePollTimeoutExpiry();
}
handlePollTimeoutExpiry 会打一条明确的 WARN，然后主动发 LeaveGroup：
consumer poll timeout has expired. This means the time between subsequent calls to poll()
was longer than the configured max.poll.interval.ms ...
这条日志是排查「消费者莫名被踢」的第一线索。它和 session.timeout.ms 超时的区别是：进程还活着，心跳也正常，只是应用线程没回来 poll。
把三条路径放在一张表里
| 现象 | 检测方 | 判定依据 | 本机实测间隔 |
|---|---|---|---|
| 新成员加入 | 协调器 | JoinGroup | B 启动到拿到分区约 1.4 s（eager） |
| 成员主动关闭 | 协调器 | LeaveGroup | B 关闭到 A 收回约 3 s（一轮心跳） |
| 进程卡住但心跳在发 | 客户端心跳线程 | max.poll.interval.ms | 约 13 s（10 s + 心跳传播） |
| 进程冻结/网络断开 | 协调器 | session.timeout.ms | 未在本机执行；预期约 10 s（本实验配置） |
最后一行未在本机执行：本机没有冻结进程或断网。做法可以是 docker pause 消费者所在环境，预期会话超时后成员被移除、客户端恢复后收到 UNKNOWN_MEMBER_ID；只给方法和预期形态，不给编造的数字。
配置建议
session.timeout.ms 不要低于 10 秒。 本机设成 10 s / 3 s / 10 s 是为了缩短实验；生产环境默认 45 s / 3 s 更稳，除非能确定网络抖动很小。
max.poll.interval.ms 按一批消息的最坏处理时间定。 它同时是再平衡等待上限，配小会误杀慢处理，配大会让卡住的成员拖住全组。默认 5 分钟对多数同步处理偏宽松。
分配器优先 cooperative，但要求全组同时支持。 混配会协商失败或退回 eager。group.initial.rebalance.delay.ms=3000 是启动期保护，不是故障，不要为了「启动快」随手改成 0。
本节判据
- JoinGroup 和 SyncGroup 分别传什么？ 答：JoinGroup 传成员身份、订阅、支持的协议；SyncGroup 由 leader 传分配方案，协调器分发给所有成员。
- 分配是谁算的？ 答：经典协议里是 leader 成员在客户端算的，协调器只负责收集和转发。
- stop-the-world 停在哪？ 答：从
onPartitionsRevoked开始，到onJoinComplete安装新分配为止；应用线程阻塞在poll()内的client.poll(future, timer)。 - eager 和 cooperative 的核心差别？ 答：eager 撤销全部分区再重分；cooperative 只撤销要搬走的分区，代价是多一轮协议。
- 停 poll 超时和会话超时的区别？ 答：前者由客户端心跳线程按
max.poll.interval.ms主动退组；后者由协调器按session.timeout.ms判定成员死亡。
下一节换掉这套客户端主导的协议：KIP-848 把分配搬到服务端，客户端只上报成员状态和订阅。我们会用同一个程序分别跑 group.protocol=classic 和 consumer，打印真正被实例化的实现类，再从 __consumer_offsets 里读出新协议多出来的五种记录类型。
阅读导航：上一节 6.1 位移与 __consumer_offsets ｜ 下一节 6.3 KIP-848 新一代消费组协议
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。