使用xk6-kafka扩展时，如何仅消费测试发布的Kafka消息？

要实现只消费测试过程中发布的新消息、避开Topic中已有旧数据的需求，可以通过手动记录偏移量+消费者Seek定位的方式解决，以下是具体方案和问题分析：

问题根源 #

直接使用LAST_OFFSET或START_OFFSET_LAST_OFFSET失效的常见原因：

• 若在发布消息前初始化消费者，LAST_OFFSET指向的是发布前Topic的最后一条消息偏移量，后续发布的测试消息在该偏移量之后，消费者不会自动向后拉取。
• 若复用旧的消费者组ID，Kafka会优先读取该组之前提交的偏移量，而非当前Topic的最新偏移量。

可行解决方案 #

核心思路 #

先记录发布测试消息前的Topic最新偏移量，再发布消息，最后让消费者从该偏移量位置开始消费，确保只读取测试期间产生的新消息。

完整代码示例 #

const kafka = require('k6/x/kafka');

// 1. 获取发布前的Topic各分区最新偏移量
const admin = kafka.admin({ brokers: ['localhost:9092'] });
admin.connect();
const targetTopic = 'your-test-topic';
const partitions = admin.listPartitions(targetTopic);
const latestOffsets = {};

for (const p of partitions) {
    latestOffsets[p] = admin.getLatestOffset(targetTopic, p);
}
admin.disconnect();

// 2. 发布测试消息
const producer = kafka.producer({ brokers: ['localhost:9092'] });
producer.connect();
const testMessages = [
    { key: kafka.string('test-key-1'), value: kafka.string('test-value-1') },
    { key: kafka.string('test-key-2'), value: kafka.string('test-value-2') }
];
producer.produce({ topic: targetTopic, messages: testMessages });
producer.disconnect();

// 3. 初始化消费者并手动定位到指定偏移量
const consumer = kafka.consumer({
    brokers: ['localhost:9092'],
    groupId: 'test-group-202405', // 使用全新消费者组ID，避免历史偏移量干扰
    autoOffsetReset: kafka.OFFSET_RESET_LATEST
});
consumer.subscribe(targetTopic);
consumer.connect();

// 为每个分区设置消费起始位置
for (const partition in latestOffsets) {
    consumer.seek(targetTopic, parseInt(partition), latestOffsets[partition]);
}

// 消费测试消息
const receivedMessages = consumer.consume({ limit: testMessages.length, timeout: 10000 });
console.log('Received test messages:', receivedMessages);

consumer.disconnect();

额外注意事项 #

• 必须使用全新的消费者组ID：复用旧组ID会导致Kafka读取历史提交的偏移量，无法避开旧消息。
• 调整超时时间：确保consume方法的timeout足够覆盖消息生产、同步的时间，避免因超时触发无消息返回的报错。
• 适配xk6-kafka版本：不同版本的参数名称可能存在差异，若遇到参数不生效，可检查官方文档确认当前版本的配置项。

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