📚 课程目录第4章消息消费者集成测试

4、消息消费者集成测试：自动提交与手动提交Offset、消费组重平衡测试

消费者 Offset 提交策略与消费组重平衡（Rebalance）是 Kafka 集成测试中最容易引发隐蔽缺陷的两个领域。自动提交可能导致消息丢失或重复消费；手动提交则要求精确控制提交时机；而重平衡期间的 Offset 管理不当会直接破坏消费语义。本章通过集成测试用例，系统验证这三种场景下的消费者行为。

4.1 自动提交 Offset 的集成测试

自动提交模式下，消费者定期将当前消费位置写入 Kafka 内部主题 __consumer_offsets。测试需要验证：提交间隔、消费进度与故障恢复后的重复消费范围。

测试维度	验证点	预期行为
提交间隔	`enable.auto.commit=true` + `auto.commit.interval.ms=5000`	每 5 秒自动提交一次 Offset
消费进度	消费 10 条消息后模拟消费者崩溃	重启后从最近一次提交点开始消费，可能重复消费 0~5 条
重复范围	崩溃前最后提交的 Offset 与当前消费位置之差	重复消息数 ≤ 提交间隔内消费的消息数

测试要点：自动提交的重复消费窗口由提交间隔决定。间隔越短，重复窗口越小，但提交频率增加会加重 Broker 负载。集成测试应覆盖极端间隔值（如 100ms 和 30s）。

// 自动提交消费者集成测试示例（Java + JUnit 5 + Testcontainers）
@Test
void testAutoCommitOffset() throws Exception {
    try (KafkaContainer kafka = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:7.5.0"))) {
        kafka.start();
        String topic = "auto-commit-test";
        createTopic(kafka, topic, 1, (short) 1);

        // 生产者发送 20 条消息
        try (Producer<String, String> producer = createProducer(kafka.getBootstrapServers())) {
            for (int i = 0; i < 20; i++) {
                producer.send(new ProducerRecord<>(topic, "key-" + i, "value-" + i)).get();
            }
        }

        // 消费者：自动提交，间隔 2 秒，消费 10 条后模拟崩溃
        Properties consumerProps = new Properties();
        consumerProps.setProperty("bootstrap.servers", kafka.getBootstrapServers());
        consumerProps.setProperty("group.id", "auto-group");
        consumerProps.setProperty("enable.auto.commit", "true");
        consumerProps.setProperty("auto.commit.interval.ms", "2000");
        consumerProps.setProperty("key.deserializer", StringDeserializer.class.getName());
        consumerProps.setProperty("value.deserializer", StringDeserializer.class.getName());
        consumerProps.setProperty("auto.offset.reset", "earliest");

        try (Consumer<String, String> consumer = new KafkaConsumer<>(consumerProps)) {
            consumer.subscribe(Collections.singletonList(topic));
            int count = 0;
            while (count < 10) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                for (ConsumerRecord<String, String> record : records) {
                    count++;
                }
            }
            // 不关闭 consumer，模拟崩溃（直接退出）
        }

        // 新消费者加入，验证重复消费范围
        try (Consumer<String, String> consumer2 = new KafkaConsumer<>(consumerProps)) {
            consumer2.subscribe(Collections.singletonList(topic));
            List<String> reConsumed = new ArrayList<>();
            int pollCount = 0;
            while (pollCount < 3) {
                ConsumerRecords<String, String> records = consumer2.poll(Duration.ofMillis(1000));
                records.forEach(r -> reConsumed.add(r.value()));
                pollCount++;
            }
            // 断言：重复消费的消息数 ≤ 10（提交间隔内最多消费 10 条）
            assertTrue(reConsumed.size() >= 10 && reConsumed.size() <= 20);
        }
    }
}

4.2 手动提交 Offset 的集成测试

手动提交模式下，消费者在业务逻辑完成后显式调用 commitSync() 或 commitAsync()。测试需要验证：同步提交的阻塞行为、异步提交的回调处理、以及提交失败后的重试策略。

提交方式	阻塞行为	失败处理	适用场景
`commitSync()`	阻塞直到提交成功或超时	抛出异常，需业务层重试	精确一次语义，低吞吐场景
`commitAsync()`	非阻塞，立即返回	通过回调处理失败，可重试	高吞吐，允许少量重复

最佳实践：生产环境通常组合使用：异步提交提升吞吐，在消费者关闭前同步提交一次确保 Offset 落盘。集成测试应覆盖“异步提交失败后同步重试”的混合模式。

// 手动提交 Offset 集成测试：同步提交 + 异步提交混合模式
@Test
void testManualCommitOffset() throws Exception {
    try (KafkaContainer kafka = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:7.5.0"))) {
        kafka.start();
        String topic = "manual-commit-test";
        createTopic(kafka, topic, 2, (short) 1);

        // 生产者发送 50 条消息
        try (Producer<String, String> producer = createProducer(kafka.getBootstrapServers())) {
            for (int i = 0; i < 50; i++) {
                producer.send(new ProducerRecord<>(topic, 0, "key-" + i, "value-" + i)).get();
            }
        }

        Properties consumerProps = new Properties();
        consumerProps.setProperty("bootstrap.servers", kafka.getBootstrapServers());
        consumerProps.setProperty("group.id", "manual-group");
        consumerProps.setProperty("enable.auto.commit", "false");
        consumerProps.setProperty("key.deserializer", StringDeserializer.class.getName());
        consumerProps.setProperty("value.deserializer", StringDeserializer.class.getName());
        consumerProps.setProperty("auto.offset.reset", "earliest");

        // 第一轮消费：手动提交
        try (Consumer<String, String> consumer = new KafkaConsumer<>(consumerProps)) {
            consumer.subscribe(Collections.singletonList(topic));
            int totalConsumed = 0;
            while (totalConsumed < 30) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                for (ConsumerRecord<String, String> record : records) {
                    // 模拟业务处理
                    processRecord(record);
                    totalConsumed++;
                }
                // 异步提交，带回调
                consumer.commitAsync((offsets, exception) -> {
                    if (exception != null) {
                        log.warn("异步提交失败，准备同步重试", exception);
                        consumer.commitSync(); // 同步重试
                    }
                });
            }
        }

        // 第二轮消费：验证 Offset 已提交，不会重复消费已提交部分
        try (Consumer<String, String> consumer2 = new KafkaConsumer<>(consumerProps)) {
            consumer2.subscribe(Collections.singletonList(topic));
            List<String> secondRound = new ArrayList<>();
            int pollCount = 0;
            while (pollCount < 5) {
                ConsumerRecords<String, String> records = consumer2.poll(Duration.ofMillis(1000));
                records.forEach(r -> secondRound.add(r.value()));
                pollCount++;
            }
            // 第二轮消费的消息数应 ≤ 20（第一轮消费了 30 条，剩余 20 条）
            assertTrue(secondRound.size() <= 20);
        }
    }
}

4.3 消费组重平衡测试

消费组重平衡发生在消费者加入、离开或分区数变更时。重平衡期间，所有消费者暂停消费，重新分配分区。测试需要验证：重平衡触发条件、分区分配结果、以及重平衡期间的消费中断时间。

4.4 重平衡期间的 Offset 管理测试

重平衡发生时，消费者需要提交当前已消费但未提交的 Offset，否则新消费者可能重复消费。测试需要验证：ConsumerRebalanceListener 中 onPartitionsRevoked 和 onPartitionsAssigned 的正确实现。

// 重平衡 Offset 管理集成测试
@Test
void testRebalanceOffsetManagement() throws Exception {
    try (KafkaContainer kafka = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:7.5.0"))) {
        kafka.start();
        String topic = "rebalance-offset-test";
        createTopic(kafka, topic, 3, (short) 1);

        // 生产者发送 100 条消息
        try (Producer<String, String> producer = createProducer(kafka.getBootstrapServers())) {
            for (int i = 0; i < 100; i++) {
                producer.send(new ProducerRecord<>(topic, "key-" + i, "value-" + i)).get();
            }
        }

        Properties baseProps = new Properties();
        baseProps.setProperty("bootstrap.servers", kafka.getBootstrapServers());
        baseProps.setProperty("group.id", "rebalance-group");
        baseProps.setProperty("enable.auto.commit", "false");
        baseProps.setProperty("key.deserializer", StringDeserializer.class.getName());
        baseProps.setProperty("value.deserializer", StringDeserializer.class.getName());
        baseProps.setProperty("auto.offset.reset", "earliest");

        // 消费者 A：带 RebalanceListener
        CountDownLatch revokedLatch = new CountDownLatch(1);
        CountDownLatch assignedLatch = new CountDownLatch(1);

        try (Consumer<String, String> consumerA = new KafkaConsumer<>(baseProps)) {
            consumerA.subscribe(Collections.singletonList(topic), new ConsumerRebalanceListener() {
                @Override
                public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                    // 在分区被回收前提交 Offset
                    consumerA.commitSync();
                    revokedLatch.countDown();
                }

                @Override
                public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                    assignedLatch.countDown();
                }
            });

            // 消费 30 条消息
            int consumed = 0;
            while (consumed < 30) {
                ConsumerRecords<String, String> records = consumerA.poll(Duration.ofMillis(1000));
                for (ConsumerRecord<String, String> record : records) {
                    consumed++;
                }
                consumerA.commitAsync();
            }

            // 启动消费者 B，触发重平衡
            Properties propsB = new Properties();
            propsB.putAll(baseProps);
            propsB.setProperty("group.instance.id", "consumer-B");
            try (Consumer<String, String> consumerB = new KafkaConsumer<>(propsB)) {
                consumerB.subscribe(Collections.singletonList(topic));
                // 等待重平衡完成
                assertTrue(assignedLatch.await(10, TimeUnit.SECONDS));
                assertTrue(revokedLatch.await(10, TimeUnit.SECONDS));

                // 验证消费者 B 不会重复消费已提交的 Offset
                List<String> messagesB = new ArrayList<>();
                int pollCount = 0;
                while (pollCount < 5) {
                    ConsumerRecords<String, String> records = consumerB.poll(Duration.ofMillis(1000));
                    records.forEach(r -> messagesB.add(r.value()));
                    pollCount++;
                }
                // 消费者 B 消费的消息数应 ≤ 70（剩余 70 条）
                assertTrue(messagesB.size() <= 70);
            }
        }
    }
}

4.5 测试结果对比与故障排查清单

测试场景	常见故障	排查方向	验证方法
自动提交	重复消费过多	检查 `auto.commit.interval.ms` 是否过大	缩短间隔后重新测试重复窗口
手动提交	消息丢失	检查是否在业务处理前提交了 Offset	在提交点前后添加日志，对比消费进度
手动提交	提交超时	检查网络延迟或 Broker 负载	增加 `max.poll.interval.ms` 或优化业务处理时间
重平衡	消费中断时间过长	检查 `session.timeout.ms` 和 `heartbeat.interval.ms`	缩短超时时间，观察重平衡耗时
重平衡	重复消费	检查 `onPartitionsRevoked` 中是否提交了 Offset	在 Listener 中添加日志，验证提交行为
重平衡	分区分配不均	检查分区分配策略（Range / RoundRobin / Sticky）	打印分配结果，验证是否符合预期策略

关键提醒：重平衡测试必须在多消费者环境下进行，且需要模拟消费者崩溃、网络分区等异常场景。单消费者测试无法触发重平衡逻辑，容易遗漏 Offset 管理缺陷。建议使用 Testcontainers 启动多 Broker 集群，配合线程控制模拟消费者加入/离开。

4.6 测试覆盖率检查清单

自动提交：覆盖不同提交间隔（100ms、1s、5s、30s）
手动提交：覆盖同步提交、异步提交、异步+同步重试
重平衡触发：消费者加入、消费者离开、分区数变更
重平衡监听器：验证 onPartitionsRevoked 和 onPartitionsAssigned 的调用时机
Offset 提交时机：在业务处理前、中、后分别提交，验证语义正确性
异常场景：Broker 宕机、网络超时、消费者进程 Kill -9
多分区多消费者：至少 3 个分区、3 个消费者，验证分区分配均匀性

通过上述集成测试用例，可以系统性地验证 Kafka 消费者在自动提交、手动提交和重平衡场景下的行为正确性。这些测试应作为 CI/CD 流水线的一部分持续运行，确保每次代码变更不会引入消费语义缺陷。