如何使用ConsumerSeekAware接口从指定Kafka偏移量拉取消息及代码改造

一、实现ConsumerSeekAware接口 #

写个自定义消息监听器类，同时实现ConsumerSeekAware和MessageListener接口，通过重写接口方法控制消息偏移量。核心是在分区分配时指定拉取的起始位置，也支持消费过程中动态调整偏移量。

示例代码：

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.listener.ConsumerSeekAware;
import org.springframework.kafka.listener.MessageListener;
import java.util.Map;

public class CustomSeekAwareListener implements MessageListener<String, String>, ConsumerSeekAware {

    private ConsumerSeekCallback seekCallback;

    // 保存偏移量控制回调，用于后续动态调整
    @Override
    public void registerSeekCallback(ConsumerSeekCallback callback) {
        this.seekCallback = callback;
    }

    // 分区分配给消费者时，设置初始偏移量
    @Override
    public void onPartitionsAssigned(Map<org.apache.kafka.common.TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
        // 遍历所有分配到的分区，设置指定偏移量
        assignments.forEach((topicPartition, currentOffset) -> {
            // 方式1：指定具体偏移量（比如从第100条开始拉取）
            callback.seek(topicPartition.topic(), topicPartition.partition(), 100L);
            // 方式2：从头开始消费
            // callback.seekToBeginning(topicPartition.topic(), topicPartition.partition());
            // 方式3：从最新消息开始消费
            // callback.seekToEnd(topicPartition.topic(), topicPartition.partition());
        });
    }

    @Override
    public void onIdleContainer(Map<org.apache.kafka.common.TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
        // 容器空闲时的处理逻辑，可选实现
    }

    // 核心消息消费逻辑
    @Override
    public void onMessage(ConsumerRecord<String, String> message) {
        System.out.println("Message Received : " + message.value());
        // 消费过程中如需动态调整偏移量，可直接调用seekCallback
        // 示例：跳过下一条消息，直接拉取当前偏移量+2的位置
        // seekCallback.seek("mytopic", message.partition(), message.offset() + 2);
    }
}

二、修改现有代码注册使用 #

把原来的普通消息监听器替换为自定义的CustomSeekAwareListener，注册到容器中即可生效。修改后的kafkaListenerContainerFactory方法如下：

@EventListener(ApplicationReadyEvent.class)
public void kafkaListenerContainerFactory() {
    ConcurrentMessageListenerContainer<String, String> factory = 
        new ConcurrentMessageListenerContainer<>(consumerFactory(), new ContainerProperties("mytopic"));
    
    factory.setConcurrency(3);
    factory.setAutoStartup(true);
    // 替换为自定义的SeekAware监听器
    factory.setupMessageListener(new CustomSeekAwareListener());
    factory.start();
}

三、关键注意事项 #

• onPartitionsAssigned是设置初始偏移量的最佳时机，会在消费者分配到分区时自动触发。
• 动态调整偏移量时，直接调用保存的seekCallback即可，无需手动操作Kafka消费者实例。
• 并发数setConcurrency(3)会创建多个消费者实例，每个实例处理不同分区，偏移量设置会针对各自分配的分区生效。
• 手动指定偏移量会覆盖Kafka的自动偏移量提交逻辑，需注意消费者组group-id的配置一致性。

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