books

《Java 消息中间件·Kafka 内核与客户端》6.3 KIP-848 新一代消费组协议

group.protocol 一个取值就换掉整套消费组实现：classic 走客户端分配和 JoinGroup/SyncGroup，consumer 走服务端分配和 ConsumerGroupHeartbeat。本节用反射打印真实实现类，核实 4.3.0 源码里的分支，再从 __consumer_offsets 读出 KIP-848 新增的五种记录，并给出迁移时必须先改的配置。

Leeting Yan 2026-10-29 8 分钟阅读 3950 字

经典协议的天花板

6.2 节把经典协议拆开之后，问题很清楚：

分配方案由 leader 客户端计算，协调器只转发。组越大，leader 压力越大。
每次成员变化都要全体重新加入，PreparingRebalance 期间等最慢的成员。
eager 分配器要撤销全部分区再重分，分区越多，停顿越明显。
分配器必须组内所有客户端都支持，升级一个分配器要协调整个组。

KIP-848 的改法是把分配搬到服务端：协调器根据每个成员上报的订阅和状态，自己算出增量分配，成员只需要接收。协议名字也从 classic 变成了 consumer。

这一节不背 KIP-848 的提案文本，只做三件事：

用同一个程序、同一个门面类，证明 group.protocol 真的换掉了底层实现。
核实 4.3.0 源码里这个分支写在哪。
从 __consumer_offsets 的段文件里，读出新旧协议在存储层留下的不同记录。

一个配置，两套实现

group.protocol 的取值只有两个。源码 ConsumerConfig 里写得非常直白：

// org/apache/kafka/clients/consumer/ConsumerConfig.java
public static final String GROUP_PROTOCOL_CONFIG = "group.protocol";
public static final String DEFAULT_GROUP_PROTOCOL = GroupProtocol.CLASSIC.name().toLowerCase(Locale.ROOT);
public static final String GROUP_PROTOCOL_DOC = "The group protocol that the consumer uses. The " +
    "supported values are <code>classic</code> or <code>consumer</code>. The default value is <code>classic</code>.";

4.3.0 的默认值仍然是 classic，新协议要显式打开。这一点很容易被「4.x 已经默认新协议」的说法带偏，本机配置日志可以直接反驳：

group.protocol = classic

真正决定实现类的是 ConsumerDelegateCreator.create：

// org/apache/kafka/clients/consumer/internals/ConsumerDelegateCreator.java
GroupProtocol groupProtocol = GroupProtocol.valueOf(
        config.getString(ConsumerConfig.GROUP_PROTOCOL_CONFIG).toUpperCase(Locale.ROOT));

if (groupProtocol == GroupProtocol.CONSUMER)
    return new AsyncKafkaConsumer<>(config, keyDeserializer, valueDeserializer, Optional.empty());
else
    return new ClassicKafkaConsumer<>(config, keyDeserializer, valueDeserializer);

KafkaConsumer 只是一个门面，它把几乎所有方法都委托给 delegate 字段。AsyncKafkaConsumer 不是「classic 的异步包装」，而是一套独立的实现：它有自己的请求管理器、成员管理器和事件队列。

实测：打印真实实现类

门面类的 getClass() 永远返回 KafkaConsumer，看不出底层。要证明分支真的生效，得反射读私有字段 delegate。

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.lang.reflect.Field;
import java.time.Duration;
import java.util.List;
import java.util.Properties;

public class ProtocolProbe {
    public static void main(String[] args) throws Exception {
        String protocol = args[0];
        Properties p = new Properties();
        p.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:19211");
        p.put(ConsumerConfig.GROUP_ID_CONFIG, "b1.proto." + protocol + ".group");
        p.put(ConsumerConfig.CLIENT_ID_CONFIG, "b1-proto-" + protocol);
        p.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        p.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        p.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        p.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, protocol);

        try (KafkaConsumer<String, String> c = new KafkaConsumer<>(p)) {
            Field f = KafkaConsumer.class.getDeclaredField("delegate");
            f.setAccessible(true);
            System.out.println("KafkaConsumer.getClass() = " + c.getClass().getName());
            System.out.println("delegate.getClass()     = " + f.get(c).getClass().getName());

            c.subscribe(List.of("b1.offsets.demo"));
            long start = System.currentTimeMillis();
            while (System.currentTimeMillis() - start < 12000) {
                c.poll(Duration.ofMillis(500));
            }
            System.out.println("final assignment = " + c.assignment());
        }
    }
}

本机实测（Temurin 25.0.5+8 / Kafka 4.3.0 KRaft 单节点 / 2026-10-29）：

########## classic ##########
requested group.protocol = classic
KafkaConsumer.getClass() = org.apache.kafka.clients.consumer.KafkaConsumer
delegate.getClass()     = org.apache.kafka.clients.consumer.internals.ClassicKafkaConsumer
final assignment = [b1.offsets.demo-1, b1.offsets.demo-0, b1.offsets.demo-2]

########## consumer ##########
requested group.protocol = consumer
KafkaConsumer.getClass() = org.apache.kafka.clients.consumer.KafkaConsumer
delegate.getClass()     = org.apache.kafka.clients.consumer.internals.AsyncKafkaConsumer
final assignment = [b1.offsets.demo-1, b1.offsets.demo-0, b1.offsets.demo-2]

门面相同，委托对象不同。两个组都拿到了全部分区，说明新协议在单成员场景下行为一致；差异在成员变化时才显现。

另外，控制台消费者在 4.3.0 启动时会主动打印一条提示：

The consumer rebalance protocol (KIP-848) is production-ready!
Set group.protocol=consumer to try it out.

这条提示本身也是版本证据：4.3.0 认为新协议可用于生产，但默认仍然不打开。

服务端主导分配

打开 group.protocol=consumer 后，客户端不再发送 JoinGroup/SyncGroup，而是周期发送 ConsumerGroupHeartbeat。RPC 名称可以在源码里核实：

// org/apache/kafka/clients/consumer/internals/ConsumerHeartbeatRequestManager.java
import org.apache.kafka.common.requests.ConsumerGroupHeartbeatRequest;
import org.apache.kafka.common.requests.ConsumerGroupHeartbeatResponse;

这个请求同时承担两件事：上报成员状态与订阅，取回服务端算好的分配。没有单独的 SyncGroup 阶段。

服务端的分配行为由 broker 配置控制。本机实测默认值：

docker exec mq-b1 /opt/kafka/bin/kafka-configs.sh --bootstrap-server 127.0.0.1:9092 \
  --describe --entity-type brokers --entity-name 1 --all | grep "group.consumer"

  group.consumer.assignment.interval.ms=1000
  group.consumer.assignors=uniform,range
  group.consumer.heartbeat.interval.ms=5000
  group.consumer.max.assignment.interval.ms=15000
  group.consumer.max.session.timeout.ms=60000
  group.consumer.migration.policy=bidirectional
  group.consumer.min.session.timeout.ms=45000
  group.consumer.session.timeout.ms=45000
  group.coordinator.rebalance.protocols=classic,consumer,streams

几个关键点：

分配器在服务端。group.consumer.assignors 是服务端支持的分配器列表，默认 uniform,range，协调器选第一个可用的，除非客户端用 group.remote.assignor 指定。
会话超时由服务端定。默认 45 秒，上下界 45–60 秒。客户端不能自己设 session.timeout.ms。
心跳间隔也由服务端定。默认 5 秒，客户端不能设 heartbeat.interval.ms。
迁移策略默认 bidirectional。这个枚举的取值可以在 broker 的 kafka-group-coordinator jar 里核实，共四个：BIDIRECTIONAL、UPGRADE、DOWNGRADE、DISABLED。

group.remote.assignor 的效果可以实测。给客户端加上 group.remote.assignor=range、用一个新组名跑一次，然后 dump 这个组的成员记录：

key: {"type":"5","data":{"groupId":"b1.proto.range.group","memberId":"WgurxJpyR2KojPfl7s4dSA"}}
payload: {"version":"0","data":{"...","serverAssignor":"range","classicMemberMetadata":null}}

serverAssignor 确实是 range，说明客户端的选择被写进了成员元数据。

但这里有一个需要如实记录的差异：同一个组用 kafka-consumer-groups.sh --describe --state 看，ASSIGNMENT-STRATEGY 列显示的却是 uniform：

GROUP                COORDINATOR (ID)      ASSIGNMENT-STRATEGY  STATE   #MEMBERS
b1.proto.range.group 127.0.0.1:9092  (1)   uniform              Empty   0

本机没有继续追这个显示差异的实现，所以不下「是 bug」的结论。工程上的结论只有一条：不要只凭 --describe --state 这一列判断服务端实际用了哪个分配器。要确认就看成员记录里的 serverAssignor，或者直接看分配结果。

__consumer_offsets 里多了五种记录

6.1 节看到经典组只有两种 key：type=1 的位移和 type=2 的组元数据。新协议打开后，同一个 topic 里出现了更多类型。

用 kafka-dump-log.sh --offsets-decoder 解 b1.proto.consumer.group 所在的 18 号分区，本机实测（截取成员相关记录）：

| offset: 18 ... key: {"type":"5","data":{"groupId":"b1.proto.consumer.group","memberId":"rG3L_GErSLKFp3Vdw9MvcQ"}} payload: {"version":"0","data":{"instanceId":null,"rackId":null,"clientId":"b1-proto-consumer","clientHost":"/172.17.0.1","subscribedTopicNames":["b1.offsets.demo"],"subscribedTopicRegex":"","rebalanceTimeoutMs":300000,"serverAssignor":null,"classicMemberMetadata":null}}
| offset: 19 ... key: {"type":"3","data":{"groupId":"b1.proto.consumer.group"}} payload: {"version":"0","data":{"epoch":2,"metadataHash":-8376696857400379205}}
| offset: 20 ... key: {"type":"7","data":{"groupId":"b1.proto.consumer.group","memberId":"rG3L_GErSLKFp3Vdw9MvcQ"}} payload: {"version":"0","data":{"topicPartitions":[{"topicId":"Rfl4WzuKSVOyZ2jXqNldDQ","partitions":[0,1,2]}]}}
| offset: 21 ... key: {"type":"6","data":{"groupId":"b1.proto.consumer.group"}} payload: {"version":"0","data":{"assignmentEpoch":2,"assignmentTimestamp":1793248006150}}
| offset: 22 ... key: {"type":"8","data":{"groupId":"b1.proto.consumer.group","memberId":"rG3L_GErSLKFp3Vdw9MvcQ"}} payload: {"version":"0","data":{"memberEpoch":2,"previousMemberEpoch":0,"state":0,"assignedPartitions":[{"topicId":"Rfl4WzuKSVOyZ2jXqNldDQ","partitions":[0,1,2],"assignmentEpochs":[2,2,2]}],"partitionsPendingRevocation":[]}}
| offset: 23 ... key: {"type":"1","data":{"group":"b1.proto.consumer.group","topic":"b1.offsets.demo","partition":1}} payload: {"version":"4","data":{"offset":12,...}}

逐条对照：

`type`	key 字段	值里是什么	作用
`3`	`groupId`	`epoch`、`metadataHash`	组级元数据版本
`5`	`groupId` + `memberId`	clientId、订阅、rebalanceTimeout、serverAssignor	成员元数据
`6`	`groupId`	`assignmentEpoch`、`assignmentTimestamp`	目标分配的版本
`7`	`groupId` + `memberId`	`topicPartitions`	目标分配：这个成员应该拿哪些分区
`8`	`groupId` + `memberId`	`memberEpoch`、`state`、`assignedPartitions`、`partitionsPendingRevocation`	当前分配与待撤销分区

两个细节值得记住：

新类型的 key 用 groupId，经典 type=2 用 group。同一个 topic 里两套 schema 并存，靠 type 区分。
位移记录没变，仍然是 type=1，value 仍是 version 4。换协议不影响已提交位移的存储格式，这也是迁移可以保留进度的基础。

type=8 里的 partitionsPendingRevocation 是新协议实现增量再平衡的关键字段：服务端可以先把「要收回的分区」标出来，等成员确认撤销后再分配给别人，不需要全组停一次。

与 classic 的逐项差异

维度	classic	consumer（KIP-848）
分配计算位置	leader 客户端	协调器（服务端）
协议 RPC	JoinGroup + SyncGroup + Heartbeat	ConsumerGroupHeartbeat
分配粒度	整轮替换	增量，带 epoch
`session.timeout.ms`	客户端可配，默认 45 s	服务端固定，默认 45 s，范围 45–60 s
`heartbeat.interval.ms`	客户端可配，默认 3 s	服务端固定，默认 5 s
`partition.assignment.strategy`	客户端必填，生效于成员协商	禁止设置，用服务端分配器
再平衡等待上限	`max.poll.interval.ms`	服务端 `group.consumer.*` 控制
分配器升级	全组客户端同时支持	改服务端配置即可

这张表里最容易被忽略的是最后一行：新协议把「换分配器」从一次客户端全员升级，变成了服务端的一次配置变更。这是运维上的实质差别。

不能设置的配置

ConsumerConfig 里有一份显式的黑名单，group.protocol=consumer 时这些配置只要非空就抛异常：

// org/apache/kafka/clients/consumer/ConsumerConfig.java
private static final List<String> CONSUMER_PROTOCOL_UNSUPPORTED_CONFIGS = List.of(
        PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
        HEARTBEAT_INTERVAL_MS_CONFIG,
        SESSION_TIMEOUT_MS_CONFIG,
        SHARE_ACKNOWLEDGEMENT_MODE_CONFIG,
        SHARE_ACQUIRE_MODE_CONFIG
);

本机实测，把 session.timeout.ms 和 group.protocol=consumer 一起设置：

EXCEPTION org.apache.kafka.common.config.ConfigException: session.timeout.ms cannot be set when group.protocol=CONSUMER

这不是警告，是构造 KafkaConsumer 时直接失败。迁移时如果配置文件里留着一份通用的 session.timeout.ms，切协议的第一步就会炸。常见做法是按协议拆两套配置，或者在代码里判断协议后剔除这些键。

反过来，classic 也有一份不支持清单：group.remote.assignor、share.acknowledgement.mode、share.acquire.mode。设置 group.remote.assignor 却忘了切协议，同样会报错。

迁移注意事项

第一，确认 broker 版本。 新协议需要服务端支持。本机 4.3.0 的 banner 明确写 production-ready；更老的 broker 即使客户端支持也协商不了。升级顺序应该是 broker 先升，再升客户端。

第二，整组一起切。 同一个组里不能一半 classic、一半 consumer。切换时要么停掉全组消费者再统一改配置，要么依赖 group.consumer.migration.policy 的迁移流程；本机默认是 bidirectional，但本书没有在单节点上演练完整的在线迁移，所以不给出「零停机切换」的承诺。

第三，清理不兼容配置。 上面那五个键必须先删掉，尤其是 partition.assignment.strategy。很多团队的公共配置里都写着 CooperativeStickyAssignor，直接切会构造失败。

第四，改监控口径。 新协议的组状态、成员 epoch、目标分配都能在 __consumer_offsets 里看到，但 kafka-consumer-groups.sh --describe --members --verbose 的列含义和 classic 不完全一样。CURRENT-EPOCH、TARGET-EPOCH 这些列在 classic 组里是 -，在 consumer 组里才有值。做告警规则时不要假设两套协议输出同构。

第五，回归再平衡行为。 单成员场景两套协议看起来一样，差异在多成员增删时才出现。至少要用两个消费者做一次加入、退出、停 poll 的回归，确认分配变化符合预期。

本机实验命令汇总

# 1. 确认默认协议（配置日志里找 group.protocol）
java -cp kafka-clients-4.3.0.jar:... ProtocolProbe classic

# 2. 打印新协议的委托实现类
java -cp kafka-clients-4.3.0.jar:... ProtocolProbe consumer

# 3. 触发不兼容配置错误
java -cp ... ProtocolProbe consumer extra

# 4. 指定服务端分配器
java -Dremote=range -Dgroup=b1.proto.range.group -cp ... ProtocolProbe consumer

# 5. 看组状态
docker exec mq-b1 /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092 \
  --describe --group b1.proto.consumer.group --state

# 6. 解出新协议的记录
docker exec mq-b1 sh -c 'f=$(ls /tmp/kafka-logs/__consumer_offsets-18/*.log | head -1);
  /opt/kafka/bin/kafka-dump-log.sh --offsets-decoder --print-data-log --files $f'

其中第 1、2、3、4、5、6 步都在本机跑过，输出已贴在正文对应位置。没有执行的是多成员在线迁移和多副本场景，后者在单节点上做不了。

本节判据

group.protocol 的默认值是什么？ 答：classic，4.3.0 源码 DEFAULT_GROUP_PROTOCOL 与本机配置日志都指向它。
怎么证明底层实现换了？ 答：反射 KafkaConsumer.delegate，classic 得到 ClassicKafkaConsumer，consumer 得到 AsyncKafkaConsumer。
新协议的分配是谁算的？ 答：服务端协调器，分配器由 group.consumer.assignors 和 group.remote.assignor 决定。
新协议在 __consumer_offsets 里多了哪些记录？ 答：type=3/5/6/7/8，分别对应组元数据、成员元数据、目标分配版本、目标分配、当前分配；位移仍是 type=1。
迁移时最先会撞到什么？ 答：session.timeout.ms、heartbeat.interval.ms、partition.assignment.strategy 在 consumer 协议下禁止设置，构造时直接抛 ConfigException。

第 6 章到这里结束。下一章进入生产者内部：KafkaProducer.send() 到底在调用线程里做了什么，为什么它是线程安全的，以及那条「拦截器→序列化→分区→缓冲」的链路在源码里的真实顺序。

阅读导航：上一节 6.2 经典再平衡协议｜下一节 7.1 KafkaProducer 的线程模型

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章返回首页