Spring Boot中如何向EmbeddedKafka发送指定offset的消息？

问题根因 #

Kafka 分区的消息 offset 是由 broker 端维护的自增序列，生产者无法通过消息头自定义发送消息的 offset。你设置的KafkaHeaders.OFFSET属于 Spring Kafka 提供的消费端只读属性，发送阶段配置会被直接忽略，这就是你消费时拿到的 offset 始终从 0 开始递增的根本原因。 #

解决方案 #

针对不同测试场景，可选以下三种适配方案：

方案1：手动 seek 到指定 offset（最推荐，符合原生 Kafka 逻辑）

如果你的核心需求是测试「按指定 offset 消费」的业务逻辑，不需要篡改实际消息的 offset 值，直接利用你已经实现的ConsumerSeekAware接口的定位能力即可：

1. 先往 EmbeddedKafka 目标分区发送足量测试消息，此时分区内消息的 offset 会自然从 0 递增到 N-1
2. 在消费分区分配回调中手动定位到你需要的起始 offset，示例代码补全如下：

private class MessagesWithOffsetsConsumer implements BatchMessageListener<String, String>, ConsumerSeekAware {
    private ConsumerSeekCallback seekCallback;
    // 填入你测试需要的起始offset值
    private final Long expectedStartOffset = 10L;

    @Override
    public void registerSeekCallback(ConsumerSeekCallback callback) {
        this.seekCallback = callback;
    }

    @Override
    public void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
        assignments.keySet().forEach(tp -> {
            // 直接seek到指定offset
            callback.seek(tp.topic(), tp.partition(), expectedStartOffset);
        });
    }

    @Override
    public void onMessage(List<ConsumerRecord<String, String>> records) {
        records.forEach(record -> {
            // 此处拿到的第一条消息offset就是你指定的expectedStartOffset
            LOGGER.info("Received record: {} offset: {}", record.key(), record.offset());
        });
    }
}

该方案完全符合 Kafka 原生运行逻辑，测试结果和生产环境行为完全一致，无需额外 hack。

方案2：自定义头传递业务 offset

如果你的业务逻辑需要读取自己定义的 offset 值做处理，和 Kafka 原生 offset 无关，可以自行新增自定义消息头传递：

• 发送端配置：

messageBuilder.setHeader("CUSTOM_BUSINESS_OFFSET", yourOffsetVal);

• 消费端从消息头读取即可：

record.headers().headers("CUSTOM_BUSINESS_OFFSET").forEach(header -> {
    Long customOffset = ByteBuffer.wrap(header.value()).getLong();
    // 业务逻辑处理
});

方案3：修改消费者提交偏移量（特殊场景使用）

如果需要让消费者启动后直接从你指定的 offset 开始消费，可以在发送完测试消息后手动提交偏移量：

Map<TopicPartition, OffsetAndMetadata> offsetMap = new HashMap<>();
offsetMap.put(new TopicPartition(topic, partition), new OffsetAndMetadata(targetOffset));
kafkaTemplate.execute(consumer -> {
    consumer.commitSync(offsetMap);
    return null;
});

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