Kafka中的消费者组和重新平衡：消费者停止的原因

1. 概述

您有一个从Kafka读取消息的服务，并且运行稳定了许多天。有一天消费者滞后突然增加。日志中充满了“撤销之前分配的分区”和“CommitFailedException”的行。罪魁祸首通常是消费者组的重新平衡。

本文解释了消费者组如何分配分区、何时发生重新平衡以及为什么它会使消费者“卡住”。阅读后，您将了解如何配置Spring Kafka以减少重新平衡的次数和每次停止的时间。

2. 消费者组如何运作

一个消费者组是共享相同group.id的消费者集合。Kafka将主题的分区分配给组中的成员。每个分区在任何给定时间只分配给一个消费者。

假设主题orders有6个分区。您运行3个order-service的实例：

order-service-1  -> orders-0, orders-1
order-service-2  -> orders-2, orders-3
order-service-3  -> orders-4, orders-5

如果运行7个实例，将有一个实例闲置，因为没有更多分区可分配。

首先，添加依赖项：

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

Spring Boot 3.x中的一个简单监听器：

@Component
public class OrderListener {

    private static final Logger log = LoggerFactory.getLogger(OrderListener.class);

    @KafkaListener(topics = "orders", groupId = "order-service")
    public void onOrder(ConsumerRecord<String, String> record) {
        log.info("收到来自分区{}的订单{}", record.key(), record.partition());
    }
}

注解@KafkaListener创建一个监听器容器。在容器内是一个运行poll()循环的线程。使用groupId = "order-service"的每个实例都将加入同一个组。一个broker充当组协调器的角色，管理成员列表和分区分配。

3. 什么是重新平衡以及何时发生

重新平衡是组协调器将分区重新分配给成员的过程。导致重新平衡的情况：

1. 新消费者加入：扩展、部署新实例。
2. 消费者主动离开组：关闭、滚动重启。
3. 消费者因未在session.timeout.ms内发送心跳而被视为死亡。默认值为45秒。
4. 消费者未在max.poll.interval.ms内调用poll()。默认值为5分钟。
5. 主题的分区数发生变化，或订阅pattern匹配新主题。

第3和第4项中的两个超时机制不同，经常被混淆：

配置	默认值	检查内容
heartbeat.interval.ms	3000	发送心跳的间隔（来自后台线程）
session.timeout.ms	45000	进程是否仍活着
max.poll.interval.ms	300000	处理线程是否卡住

心跳从单独的后台线程发送。因此，当处理线程卡住时，进程仍然“活着”。相反，max.poll.interval.ms可以正确检测到您的消息处理代码过慢的情况。

4. 消费者为什么被停止

4.1. 急切重新平衡：停止整个组

使用传统协议和急切类型的分配器（RangeAssignor、RoundRobinAssignor、StickyAssignor），每次重新平衡都会发生以下情况：

1. 每个消费者撤销其持有的所有分区。
2. 每个消费者重新发送JoinGroup请求并相互等待。
3. 组的领导者计算新的分配方案，然后协调器将结果发回。
4. 消费者接收新分区并继续处理。

在步骤1-3进行期间，组中没有人处理消息。仅启动一个新实例也会导致整个组停止。这称为“停止世界”现象。

4.2. 处理过慢：超过max.poll.interval.ms

实际中最常见的原因是处理代码过慢。以下示例对外部服务的调用每条消息花费1秒：

@KafkaListener(topics = "orders", groupId = "order-service")
public void onOrder(ConsumerRecord<String, String> record) throws InterruptedException {
    paymentService.charge(record.value()); // 假设花费约1秒
}

默认情况下max.poll.records = 500。一次poll()可以返回500条记录，即大约500秒的处理时间。这个数字超过了max.poll.interval.ms的300秒。此时消费者自动离开组，您将看到如下日志：

[Consumer clientId=consumer-order-service-1, groupId=order-service] consumer poll timeout has expired.
This means the time between subsequent calls to poll() was longer than the configured
max.poll.interval.ms, which typically implies that the poll loop is spending too much time
processing messages. You can address this either by increasing max.poll.interval.ms or by
reducing the maximum size of batches returned in poll() with max.poll.records.

然后offset提交命令失败：

org.apache.kafka.clients.consumer.CommitFailedException: Offset commit cannot be completed
since the consumer is not part of an active group for auto partition assignment; it is likely
that the consumer was kicked out of the group.

这导致了一系列问题：

• 此消费者的分区被分配给其他人。
• 由于offset未提交，接收分区的人重新处理这些消息。
• 旧消费者重新加入，导致另一次重新平衡。
• 如果下一个批次仍然很慢，循环会重复。滞后持续增加。

5. 解决方法

5.1. 调整批处理大小和轮询时间

要记住的规则：max.poll.records × 每条记录的处理时间必须远小于max.poll.interval.ms。

spring:
  kafka:
    consumer:
      group-id: order-service
      max-poll-records: 50
      properties:
        max.poll.interval.ms: 120000

使用50条记录和每条1秒，一个批次花费约50秒。这个数字在120秒的限制内是安全的。

**注意：**不要将max.poll.interval.ms增加到几小时来“救急”。此值也是Kafka检测真正卡住的消费者的最大时间。此外，重新平衡必须等待最慢的消费者再次调用poll()，因此较大的值会使重新平衡延长。

如果每条消息本身处理缓慢，例如调用缓慢的API，您应该为外部调用设置超时。另一种方法是将消息推送到内部队列，而不是让poll线程等待太长时间。

5.2. 协作重新平衡

CooperativeStickyAssignor使用增量重新平衡。消费者只撤销真正需要转移的分区，其他分区继续正常处理。

spring:
  kafka:
    consumer:
      properties:
        partition.assignment.strategy: org.apache.kafka.clients.consumer.CooperativeStickyAssignor

当将order-service-4添加到第2部分中的3实例组时，只有少数分区转移到新实例：

order-service-1  -> orders-0, orders-1        (保持不变，不停止)
order-service-2  -> orders-2                  (失去orders-3)
order-service-3  -> orders-4                  (失去orders-5)
order-service-4  -> orders-3, orders-5

**注意：**不能在正在运行的组上直接从eager切换到cooperative。您需要滚动部署两次。第一次，配置两者：CooperativeStickyAssignor,RangeAssignor。下一次，删除RangeAssignor。如果成员没有共同的分配器，组将报告错误。

5.3. 静态成员身份

每次滚动重启，一个实例离开组然后回归。结果是每个pod两次重新平衡。静态成员身份解决了这个问题。您通过group.instance.id为每个实例分配一个固定标识符：

spring:
  kafka:
    consumer:
      properties:
        group.instance.id: ${HOSTNAME}
        session.timeout.ms: 60000

具有group.instance.id的消费者关闭时不会发送LeaveGroup。如果它在session.timeout.ms内以相同id返回，协调器将返回确切的旧分区，无需重新平衡。

注意：HOSTNAME必须在重启时保持稳定，例如Kubernetes上StatefulSet的pod名称。使用普通的Deployment，pod名称每次都会改变，所以静态成员身份无效。两个同时运行且具有相同id的实例将彼此踢出组（FencedInstanceIdException）。当concurrency > 1时，Spring Kafka会自动为每个子消费者在id中添加后缀-0、-1等。

这种方法的权衡：当实例真正宕机时，其分区将无法处理，直到session.timeout.ms过期。

5.4. 新消费者组协议

从Kafka 4.0开始，新的重新平衡协议(KIP-848)已准备好用于生产环境。您可以通过group.protocol=consumer启用它。分区分配完全转移到broker，始终以增量方式进行。消费者不再需要通过共同的JoinGroup“障碍”相互等待。

使用此协议，客户端的partition.assignment.strategy将不再使用。相反，broker选择分配器（客户端配置是group.remote.assignor）。如果您的集群和客户端足够新，这是一个值得考虑的选择。在启用之前，您需要检查Spring Boot正在拉取的kafka-clients版本。

6. 分区被撤销时的处理

无论优化到什么程度，重新平衡仍然会发生。重要的是在失去分区之前提交已处理的offset，以减少重复处理。

Spring Kafka在分区被撤销之前自动提交待处理的offset。如果您直接使用KafkaConsumer，必须通过ConsumerRebalanceListener自己做这件事：

public class CommitOnRevokeListener implements ConsumerRebalanceListener {

    private final Consumer<String, String> consumer;
    private final Map<TopicPartition, OffsetAndMetadata> pending = new ConcurrentHashMap<>();

    public CommitOnRevokeListener(Consumer<String, String> consumer) {
        this.consumer = consumer;
    }

    public void markProcessed(ConsumerRecord<String, String> record) {
        pending.put(new TopicPartition(record.topic(), record.partition()),
                new OffsetAndMetadata(record.offset() + 1));
    }

    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        Map<TopicPartition, OffsetAndMetadata> toCommit = new HashMap<>();
        for (TopicPartition tp : partitions) {
            OffsetAndMetadata offset = pending.remove(tp);
            if (offset != null) {
                toCommit.put(tp, offset);
            }
        }
        if (!toCommit.isEmpty()) {
            consumer.commitSync(toCommit);
        }
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        // 可以在这里为新分区加载本地状态
    }
}

逐部分解释：

• markProcessed()记录下一个需要读取的offset，即offset + 1，在记录处理完成后。
• onPartitionsRevoked()在poll线程上运行，就在分区被取走之前。它只提交被撤销的分区的offset。这对于协作重新平衡很重要，因为那时只有部分分区被撤销。
• commitSync()阻塞直到broker确认，确保在其他消费者接收分区之前offset已保存。

在订阅时注册监听器：consumer.subscribe(List.of("orders"), listener)。

使用kafka-clients中提供的MockConsumer检查此行为：

class CommitOnRevokeListenerTest {

    @Test
    void shouldCommitOnlyRevokedPartitions() {
        MockConsumer<String, String> consumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST);
        TopicPartition p0 = new TopicPartition("orders", 0);
        TopicPartition p1 = new TopicPartition("orders", 1);
        consumer.assign(List.of(p0, p1));

        CommitOnRevokeListener listener = new CommitOnRevokeListener(consumer);
        listener.markProcessed(new ConsumerRecord<>("orders", 0, 41L, "order-1", "{}"));
        listener.markProcessed(new ConsumerRecord<>("orders", 1, 7L, "order-2", "{}"));

        listener.onPartitionsRevoked(List.of(p0));

        Map<TopicPartition, OffsetAndMetadata> committed = consumer.committed(Set.of(p0, p1));
        assertThat(committed.get(p0).offset()).isEqualTo(42L);
        assertThat(committed.get(p1)).isNull();
    }
}

测试模拟只有orders-0被撤销的情况。预期结果是为orders-0提交offset42。而orders-1仍然被保持，所以没有offset被提交。

注意：onPartitionsRevoked()中的代码必须快速执行。此回调在重新平衡期间运行。如果它很慢，整个组必须等待。

7. 常见错误

• 异步处理但仍然按poll提交。 您将记录推送到ExecutorService，然后容器在poll()之后立即提交offset。如果发生重新平衡，正在处理的消息可能会丢失。只提交已完成处理的记录的offset。
• 混淆session.timeout.ms和max.poll.interval.ms。 当原因是处理缓慢时，增加会话超时没有帮助。
• 设置concurrency大于分区数。 多余的消费者没有工作。它们只会使组更拥挤，重新平衡花费更长时间。
• 长时间的GC暂停或容器上的CPU限流。 心跳线程无法及时运行，消费者即使代码不慢也会被踢出组。检查JVM指标和CPU限制。
• 没有设计幂等处理。 重新平衡总是可能导致某些消息的重复处理。不要假设“恰好一次”，除非您使用了事务或有去重机制，例如检查orderId是否已处理。

8. 结论

重新平衡是使消费者组能够自动扩展和容错的必要机制。代价是消费者必须暂停。导致消费者持续“被停止”的最常见原因是poll()循环超过max.poll.interval.ms。此时消费者被踢出组，提交失败，重新平衡不断重复。

要最小化，您可以：

• 通过max.poll.records保持每个批次较小。
• 切换到CooperativeStickyAssignor或新消费者组协议。
• 为具有稳定标识符的部署使用静态成员身份。
• 在onPartitionsRevoked()中及时提交offset。

最后，始终设计幂等消息处理。重新平衡无法完全消除，只能使其发生频率降低并伤害更少。