企业官网建设流程全解析

Kafka 如何用 kafka-share-groups.sh 从 consumer group 已提交 offset 文件初始化 Share Group

【免费下载链接】KafkaApache Kafka - A distributed event streaming platform项目地址: https://gitcode.com/GitHub_Trending/kafka4/kafka

把消费任务从 classic/consumer group 迁移到 Share Group（Kafka Queues）时，一个常见诉求是：让新的 share group 从原 consumer group 已经提交的位置开始消费，而不是从头重放或跳到日志末尾。Kafka 4.4.0 开始，kafka-share-groups.sh工具支持用--from-file从 CSV 文件初始化 share group 的 offset（见 升级说明 中 4.4.0 的 Notable changes，对应 KIP-1323）。整个操作路径是两步：先用kafka-consumer-groups.sh把 consumer group 的已提交 offset 导出为 CSV，再用kafka-share-groups.sh --reset-offsets --from-file写入 share group。

下面以 Basic Kafka Operations 文档 的 "Managing share groups" 一节为主线，给出完整可执行步骤。

准备条件

在执行之前，确认以下几点都成立，否则命令会直接报错退出：

• 已有一个运行中的 Kafka 集群，工具来自 Kafka 发行版的bin/目录（所有工具不带参数运行时都会打印完整的命令行选项说明）。
• 源 consumer group 已存在且有已提交 offset，并且处于 inactive 状态（没有活跃成员）。文档明确给出的是 "export the current offsets from an inactive consumer group"。
• 目标 share group没有活跃成员。工具在重置前会检查 share group 状态，只要状态不是 EMPTY/DEAD 就会报错退出，提示Share group '<group>' is not empty.（见 ShareGroupCommand.java 的resetOffsets()）。
• admin client 需要对组内用到的所有 topic 拥有 DESCRIBE 访问权限，这是文档对 share group describe/offset 操作的明确要求。

可以先用下面命令确认两个组的状态（my-group替换为你的 consumer group，my-share-group替换为你的 share group，localhost:9092替换为你的 bootstrap server）：

$ bin/kafka-groups.sh --bootstrap-server localhost:9092 --list GROUP TYPE PROTOCOL my-consumer-group Consumer consumer my-share-group Share share

consumer group 是否 inactive，可以用 describe 的--state选项查看：

$ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my-group --state COORDINATOR (ID) ASSIGNMENT-STRATEGY STATE #MEMBERS localhost:9092 (0) range Stable 4

（文档示例）STATE为Empty或Dead、#MEMBERS为 0 时表示无活跃成员，满足"已停止消费"的前提。

第一步：导出 consumer group 的已提交 offset 到 CSV

文档给出的导出命令是：

$ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --reset-offsets --group my-group --all-topics --to-current --dry-run --export > FILE.CSV

各选项在文档中的含义：

• --all-topics：对该组所有订阅 topic 的分区生效（--reset-offsets的 scope 之一，与--topic二选一）；
• --to-current：以当前已提交 offset 作为重置目标值，也就是"导出消费者现在的位置"；
• --dry-run+--export：只生成 CSV、不修改任何 offset。--export的文档描述是 "to generate offset reset information in CSV format for export to a file"，配合 shell 重定向把输出写入文件。

FILE.CSV是你在本机指定的输出文件名，可以自行替换。生成后 CSV 每行三列，格式为topic,partition,offset（对应 CsvUtils.java 中CsvRecordNoGroup.FIELDS = {"topic", "partition", "offset"}）。

第二步：用 CSV 初始化 share group 的 offset

确认 CSV 内容无误后，把它喂给kafka-share-groups.sh：

$ bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --reset-offsets --group my-share-group --from-file FILE.CSV --execute GROUP TOPIC PARTITION NEW-OFFSET my-share-group topic1 0 10

（文档示例）输出的NEW-OFFSET行显示每个分区将被设置到的目标 offset。

关于这条命令的两个执行细节：

• --from-file本身就是--reset-offsets的一个 scenario，且不需要也不可以再搭配--topic或--all-topics这类 scope 选项，范围由文件内容决定（见 ShareGroupCommandOptions.java 的参数校验）。
• share group 的--reset-offsets有 3 个执行选项：--dry-run只显示将被重置的 offset，--execute真正执行，--export导出 CSV。不显式指定时默认按 dry-run 处理（见 ShareGroupCommand.java 中dryRun = has(--dry-run) || !has(--execute)），所以可以先去掉--execute跑一遍预览，确认 NEW-OFFSET 符合预期后再加--execute正式执行。

验证结果：用 --describe 查看 START-OFFSET

执行成功后，用 describe 检查 share group 的 start offset 是否落在预期位置：

$ bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --describe --group my-share-group GROUP TOPIC PARTITION START-OFFSET LAG my-share-group topic1 0 4 0

（文档示例）START-OFFSET即 share group 每个分区当前的起始 offset，应与你 CSV 文件中该分区写入的 offset 一致；LAG为 0 表示没有 in-flight 积压。文档对 start offset 的解释是：它是正在等待投递给 share consumer 的 in-flight 记录中最早的 offset，start offset 之后的一些记录可能已经投递完成。

如果 CSV 里某个分区的 offset 超出日志当前范围，工具会将其调整到可用边界：高于日志末尾时设为末尾 offset，低于最早可用 offset 时设为最早 offset，并在日志中输出类似New offset (n) is higher than latest offset for topic partition ... . Value will be set to ...的告警（见 GroupOffsetsResetter.java 的checkOffsetsRange）。因此验证时如果 START-OFFSET 与 CSV 值不一致，先检查是否触发了这种边界调整。

限制与注意事项

• 版本要求：从文件初始化 share group offset 是 4.4.0 引入的能力，低版本集群上kafka-share-groups.sh没有这条路径，需要先把 broker 升级到 4.4.0 再执行。

• 两边都要先停下来：源 consumer group 必须 inactive 才能"导出当前进度"；目标 share group 必须无活跃成员才能重置。两个组里有活跃消费者时先停止消费端再操作。

• 只影响 share group，不影响 consumer group：整条路径对 consumer group 只做只读导出（--dry-run保证不改 offset），consumer group 的已提交 offset 不会被修改。

• 如果之后想撤销初始化，可以对单个 topic 用--delete-offsets删除 share group 的 offset：

  $ bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --delete-offsets --group my-share-group --topic topic1 TOPIC STATUS topic1 Successful

  （文档示例）删除后该 topic 的分区在 share group 中不再保留 offset。

更多选项（--describe --members、--state、--delete等）的用法见 docs/operations/basic-kafka-operations.md 的 "Managing share groups" 一节，工具实现可参考 tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java。

【免费下载链接】KafkaApache Kafka - A distributed event streaming platform项目地址: https://gitcode.com/GitHub_Trending/kafka4/kafka