Amazon MSK：用下一代协议优化消费者再均衡

如果你在 Apache Kafka 和 Amazon Managed Streaming for Apache Kafka（Amazon MSK）上运行大型消费者组，你很可能经历过缓慢重平衡带来的痛苦：所有消费者处理停滞、由常规扩缩容事件触发的“重平衡风暴”，以及影响下游应用的漫长恢复时间。使用经典重平衡协议，即使只有一个消费者加入或离开组，也会强制形成全局同步屏障，暂停每一个消费者，无论其分区分配是否发生变化。

该KIP-848 消费者协议在 Apache Kafka 4.0 中引入，从根本上重新设计了消费者组重平衡的工作方式。该协议也被称为“下一代消费者重平衡协议”，KIP-848 将协调逻辑从客户端转移到 broker 端的组协调器。这支持完全增量式、服务器驱动的重平衡，可显著提升大型消费者组的性能。你可以在 Amazon MSK 上的所有 4.x Apache Kafka 版本上使用该消费者协议，MSK Standard 和 Express broker 均支持。

在本文中，我们将解释该消费者协议的工作原理、如何在 Amazon MSK 上启用它，以及如何诊断和解决缓慢重平衡问题，以帮助提升性能。

经典协议与消费者协议的比较

经典协议依赖客户端侧的重平衡逻辑和全局同步屏障。每次重平衡都会导致组内所有消费者同时暂停处理，无论其分区分配是否发生变化。这会在大型消费者组中引发“重平衡风暴”，级联重平衡可能需要数分钟才能解决。CooperativeStickyAssignor 是一种客户端侧分区分配策略，支持增量式、协作式重平衡。它显著提升了重平衡性能，并最大限度地减少重平衡事件期间对组的干扰。然而，随着消费者组规模和分区数量增加，它仍然存在瓶颈。对于大型工作负载，客户端侧重平衡行为可能导致更长的重平衡时间，并且需要在重平衡期间进行大量客户端调优和监控。

消费者协议通过将所有重平衡逻辑移至服务器端来解决这些限制。broker 使用持续心跳机制和服务器驱动的协调过程来处理协调。重平衡期间只有受影响的分区会移动，分配未发生变化的消费者可以继续不间断地处理。与经典协议相比，这带来了更快的恢复速度，并随着工作负载增长提升了可扩展性。

下表从关键维度比较了经典协议与下一代协议。

服务器端配置

使用消费者协议时，关键参数现在在服务器端而非客户端配置：

• group.consumer.heartbeat.interval.ms – 控制消费者心跳间隔（服务器端）。
• group.consumer.session.timeout.ms – 控制会话超时（服务器端）。
• group.consumer.assignors – 指定可用的分配器（默认为 uniform 和 range）。

在 Amazon MSK Express broker 中，这些配置为只读，无法修改。在 Amazon MSK Standard broker 中，这些配置通过 broker 配置进行管理。要在 Amazon MSK Standard broker 中更新这些配置，请参阅 更新 Amazon MSK 集群的配置。

何时使用消费者协议

消费者协议对具有以下需求的工作负载收益最大：

• 大型消费者组： 拥有大量消费者和分区的组由于消除了全局同步屏障，可获得最显著的改进。
• 高可用性应用： 无法承受处理中断的应用可从重平衡期间持续的消息处理中受益。金融服务、实时分析和欺诈检测系统是理想的选择。
• 频繁重平衡的环境： 自动扩缩容部署、频繁重启 Pod 的 Kubernetes，或持续集成与持续交付（CI/CD）环境所受到的干扰会显著减少。
• 动态分区扩缩容： 定期添加分区或主题的工作负载可从增量式、服务器驱动的方法中受益。

先决条件

在开始之前，请确保你具备以下条件：

• 运行 Apache Kafka 4.0 或更高版本的 Amazon MSK 集群（MSK Standard 和 Express broker 均支持）。
• 支持 KIP-848 消费者协议的 Kafka 客户端库（支持的版本见步骤 4）。
• 基本熟悉 Apache Kafka 消费者组和分区分配。
• 拥有具备管理 MSK 集群相应权限的 AWS 账户。

在 Amazon MSK 上启用消费者协议

以下步骤将引导你验证集群版本、配置消费者客户端、移除已弃用的配置，并确认客户端库支持。

步骤 1：验证集群版本

消费者协议需要 Apache Kafka 4.0 或更高版本。要在 Amazon MSK 上使用消费者协议，请验证你的集群运行的是 Apache Kafka 4.0.x 或更高版本。你可以使用 AWS 管理控制台、AWS 命令行界面（AWS CLI）或 AWS SDK 验证集群的 Apache Kafka 版本：

aws kafka describe-cluster-v2 --cluster-arn <your-cluster-arn> \
    --query "ClusterInfo.Provisioned.CurrentBrokerSoftwareInfo.KafkaVersion"

如果你的集群运行的是 Apache Kafka 4.0.x 或更高版本，消费者协议会在服务器端自动启用并可直接使用。无需额外的服务器端功能标志验证。

步骤 2：配置消费者客户端

在您的消费者配置中设置 group.protocol=consumer。该协议默认未启用：

# confluent-kafka-python example
config = {
    'bootstrap.servers': bootstrap_servers,
    'group.id': group_id,
    'group.protocol': 'consumer',  # Required — defaults to 'classic' if omitted
    'auto.offset.reset': 'earliest'
}

消费者协议可以就地变更，适用于现有的消费者组。当您更新 group.protocol 时，请对您的消费者执行滚动重启。代理端组协调器会自动处理向消费者协议的升级，并在处理已升级客户端的同时处理来自旧客户端的经典协议请求。

步骤 3：移除已弃用的客户端配置

启用消费者协议后，以下客户端配置不再受支持，因为它们由代理控制：

• heartbeat.interval.ms。
• session.timeout.ms。
• partition.assignment.strategy。

步骤 4：验证客户端库支持

验证您的 Kafka 客户端版本支持消费者协议：

• Java 客户端： 在 Apache Kafka 4.0+ 中已正式可用（GA）。
• confluent-kafka-python： 版本 2.12.0+（对 KIP-848 的 GA 支持）。请参阅 发布说明。
• 基于 librdkafka 的客户端（Go、.NET、C/C++）： 基于 librdkafka 2.12.0+。

注意： 对于其他 Kafka 客户端库，请在启用下一代协议之前，查阅您的客户端库文档以了解 group.protocol=consumer 支持情况。如果您的客户端不支持 KIP-848，它将继续使用经典协议。

使用消费者协议诊断缓慢的消费者组再平衡

即使在启用消费者协议之后，您也可能遇到消费者组再平衡耗时超过预期的情况。以下各节可帮助您诊断并解决这些问题。

常见症状

• 尽管设置了 group.protocol=consumer，消费者组再平衡仍然耗时超过预期。
• 消费者在再平衡期间暂停处理。
• 代理日志显示“member session expired”或“fenced”消息。
• 在滚动部署或 Pod 重启期间触发频繁的再平衡。

步骤 1：使用代理日志确认消费者协议处于活动状态

在排查性能问题之前，请验证您的消费者实际使用的是哪种协议。请检查 代理日志，位于 Amazon CloudWatch Logs Insights 中。两种协议的日志模式差异显著。

消费者协议预期的日志：

关键指标：“consumer protocol”、“epoch”术语、带有服务端分配器的“target assignment”、用于成员移除的“fenced”。

[GroupCoordinator id=X] [GroupId <group-id>] Member <member-id> joins the consumer group using the consumer protocol.
[GroupCoordinator id=X] [GroupId <group-id>] Bumped group epoch to 309 with metadata hash 4064309670987706693.
[GroupCoordinator id=X] [GroupId <group-id>] Computed a new target assignment for epoch 309 with 'uniform' assignor in 0ms.
[GroupCoordinator id=X] [GroupId <group-id>] Member <member-id> fenced from the group because the member session expired.

经典协议预期的日志：

关键指标：“PreparingRebalance”状态、“old generation”术语、“Assignment received from leader”。

如果您看到经典协议日志，则消费者协议未处于活动状态。请继续执行步骤 2 以排查原因。

[GroupCoordinator id=X] Preparing to rebalance group <group-id> in state PreparingRebalance with old generation X
[GroupCoordinator id=X] Stabilized group <group-id> with X members
[GroupCoordinator id=X] Assignment received from leader for group <group-id>

步骤 2：排查消费者协议为何未处于活动状态

如前文步骤 1 到步骤 4 所述，验证您的客户端配置、客户端库版本和集群版本是否支持消费者协议。

步骤 3：在 KIP-848 处于活动状态时解决缓慢的再平衡

在您确认消费者协议已处于活动状态但再平衡仍然缓慢后，请调查以下原因：

A. 消费者会话超时导致成员被过早移除

使用消费者协议时，会话超时由服务端通过 group.consumer.session.timeout.ms 控制（默认值：45 秒）。诊断路径取决于您是否使用静态组成员身份。下表概述了每种场景的诊断路径和建议操作。

B. 缺少优雅关闭处理

当消费者在未调用 consumer.close() 的情况下终止时，协调器在移除成员之前会等待整个会话超时。这是容器化环境中导致再平衡缓慢的最常见原因。

解决方案：实现适当的 SIGTERM 处理以触发立即离开组：

import signal
import sys
from confluent_kafka import Consumer

class GracefulKafkaConsumer:
    def __init__(self, config):
        self.running = True
        self.consumer = Consumer(config)
        signal.signal(signal.SIGTERM, self.shutdown_handler)
        signal.signal(signal.SIGINT, self.shutdown_handler)

    def shutdown_handler(self, signum, frame):
        print(f"Received signal {signum}, initiating graceful shutdown...")
        self.running = False

    def consume_loop(self):
        self.consumer.subscribe(['your-topic'])
        while self.running:
            msg = self.consumer.poll(timeout=1.0)
            if msg is None:
                continue
            # Process message

        print("Closing consumer gracefully...")
        self.consumer.close()  # Sends LeaveGroup — triggers immediate rebalance
        sys.exit(0)

对于 Kubernetes，请验证 terminationGracePeriodSeconds 是否留出足够时间让 consumer.close() 完成：

spec:
  terminationGracePeriodSeconds: 60
  containers:
    - name: kafka-consumer

C. 不稳定的消费者导致频繁再平衡

如果消费者反复加入和离开（内存不足（OOM）被杀、CrashLoopBackOff 或短生命周期任务），每个事件都会触发新的再平衡纪元。

解决方案：通过分配唯一的 group.instance.id 来使用静态成员身份：

config = {
    'bootstrap.servers': bootstrap_servers,
    'group.id': 'my-group',
    'group.protocol': 'consumer',
    'group.instance.id': f'consumer-{unique_identifier}'  # Unique per consumer
}

使用静态成员身份时：

• 在会话超时内的短暂重启不会触发再平衡。
• 消费者以相同的分区分配重新加入。
• 扩容（添加新消费者）仍然有效。新的 group.instance.id 值仅会触发未分配分区的分配。

监控与验证

应用更改后，确认改进效果：

• 检查代理日志： 确认在正常操作或部署期间不再出现“member session expired”消息。
• 监控消费者延迟： 使用 SumOffsetLag 和 EstimatedMaxTimeLag Amazon CloudWatch 指标 验证再平衡后延迟是否迅速恢复到零。
• 描述消费者组： 使用 kafka-consumer-groups.sh --describe 验证所有成员均处于活动状态且稳定。

结论

实现消费者协议后，对于消费者组重平衡，您应观察到以下行为：

• 与经典协议相比，重平衡时间始终更快。
• 与会话超时相关的重平衡更少。
• 消费者组成员关系更稳定。
• 平滑扩缩容操作，不会中断现有消费者。
• 使用静态成员资格时，消费者重启期间不必要的重平衡更少。
• 使用优雅关闭时，消费者干净退出，无需等待超时到期。

要开始使用，请在非生产工作负载中尝试消费者协议，并在扩大和缩小工作负载规模时观察重平衡的改进。

要了解有关 Amazon MSK 和消费者重平衡协议的更多信息，请参阅以下资源：

• Amazon Managed Streaming for Apache Kafka（Amazon MSK）
• Amazon MSK 开发人员指南
• 消费者重平衡协议 – Apache Kafka 文档
• 排查 Amazon MSK 消费者组的持续重平衡问题