如何判断Kafka中的记录已提交？集成测试无休眠断言方案问询

嘿，我来帮你把这两个Kafka相关的问题理清楚，一步步来：

这个得分生产者端和消费者端两个视角来看：

• 生产者视角：当你调用producer.send(record).get()成功返回RecordMetadata时，就代表这条记录已经被Kafka集群的ISR（同步副本集合）确认持久化了——这是最严格的提交标准，对应生产者配置acks=all（生产环境首选）。如果是acks=1，只要leader副本写入成功就算提交；acks=0的话生产者完全不等待确认，根本没法判断提交状态。
• 消费者视角：消费者处理完记录后，会把这条记录的偏移量（offset）提交到Kafka的__consumer_offsets主题里。所以对消费者来说，“记录已提交”指的是：消费者不仅处理完了这条记录，还调用了commitSync()或commitAsync()完成了偏移量的提交，这样就算消费者重启，也能从正确的位置继续消费。

用Thread.sleep()确实不靠谱，要么等太久浪费时间，要么等不够导致断言失败。这里给你两种可行的方案，都是基于轮询检查的思路：

方案一：手动实现轮询检查 #

核心逻辑是：生产者发送记录并确认提交后，用一个测试专用的消费者轮询目标分区，直到找到目标记录（或者确认业务消费者已经提交了对应偏移量），再执行断言。

完整代码示例：

import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.TopicPartition;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public void sendRecordAndWaitUntilProcessed(ProducerRecord<String, String> record) throws ExecutionException, InterruptedException {
    // 第一步：发送记录并确认生产者端提交成功
    RecordMetadata recordMetadata = producer.send(record).get();
    TopicPartition targetPartition = new TopicPartition(recordMetadata.topic(), recordMetadata.partition());
    long targetOffset = recordMetadata.offset();

    // 初始化测试消费者（用唯一的消费者组ID，避免和业务消费者的位移冲突）
    Properties consumerProps = new Properties();
    consumerProps.put("bootstrap.servers", "your-kafka-broker-address");
    consumerProps.put("group.id", "test-group-" + System.currentTimeMillis());
    consumerProps.put("auto.offset.reset", "earliest");
    consumerProps.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    consumerProps.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

    try (KafkaConsumer<String, String> testConsumer = new KafkaConsumer<>(consumerProps)) {
        // 订阅目标分区
        testConsumer.assign(Collections.singleton(targetPartition));
        
        boolean isProcessed = false;
        int maxWaitAttempts = 30; // 最多尝试30次，每次100ms，总计3秒，可按需调整
        int attemptCount = 0;

        while (!isProcessed && attemptCount < maxWaitAttempts) {
            ConsumerRecords<String, String> records = testConsumer.poll(Duration.ofMillis(100));
            if (!records.isEmpty()) {
                // 遍历拿到的记录，检查是否包含我们发送的目标记录
                for (var consumerRecord : records) {
                    if (consumerRecord.offset() == targetOffset && consumerRecord.value().equals(record.value())) {
                        // 如果需要确认业务消费者已经提交了位移，可以加下面的检查
                        // OffsetAndMetadata committedOffset = testConsumer.committed(targetPartition);
                        // isProcessed = committedOffset != null && committedOffset.offset() > targetOffset;
                        // 如果只需要确认记录被消费到，直接标记完成
                        isProcessed = true;
                        break;
                    }
                }
            }
            attemptCount++;
        }

        if (!isProcessed) {
            throw new AssertionError("超时了！没等到记录被处理：" + record);
        }
    }

    // 到这里就可以执行你的断言操作了
    // 比如 assert someService.getLatestRecord().equals(record.value());
}

方案二：用Awaitility框架简化等待逻辑 #

Awaitility是专门用来处理异步操作等待的测试框架，能让代码更简洁易读。先引入依赖后，代码可以写成这样：

import org.awaitility.Awaitility;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.TopicPartition;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public void sendRecordAndWaitUntilProcessedWithAwaitility(ProducerRecord<String, String> record) throws ExecutionException, InterruptedException {
    RecordMetadata recordMetadata = producer.send(record).get();
    TopicPartition targetPartition = new TopicPartition(recordMetadata.topic(), recordMetadata.partition());
    long targetOffset = recordMetadata.offset();

    Properties consumerProps = new Properties();
    // 同上面的消费者配置...
    try (KafkaConsumer<String, String> testConsumer = new KafkaConsumer<>(consumerProps)) {
        testConsumer.assign(Collections.singleton(targetPartition));

        // 用Awaitility等待条件满足，最多等3秒，每100ms检查一次
        Awaitility.await()
                .atMost(Duration.ofSeconds(3))
                .pollInterval(Duration.ofMillis(100))
                .until(() -> {
                    var records = testConsumer.poll(Duration.ofMillis(50));
                    return records.records(targetPartition).stream()
                            .anyMatch(r -> r.offset() == targetOffset && r.value().equals(record.value()));
                });
    }

    // 执行断言
    // assert ...
}

额外提醒 #

• 如果你的业务消费者是自动提交位移（enable.auto.commit=true），那只要测试消费者能拿到这条记录，基本可以确认业务消费者已经处理并提交了；如果要更严谨，可以检查业务消费者组的已提交偏移量。
• 如果是手动提交位移，一定要确保业务逻辑在处理完记录后调用了commitSync()/commitAsync()，不然测试代码里的位移检查会失效。

内容的提问来源于stack exchange，提问作者BnJ