设置auto.offset.reset=none时，如何为新消费者组手动指定初始偏移量？

你已经将auto.offset.reset设为none以感知OutOfRangeException，要处理新消费者组无偏移量的问题，无需修改该配置，可通过以下两种方式实现：

方法一：通过ConsumerAwareRebalanceListener提前设置初始偏移量 #

新消费者组首次分配分区时会触发重平衡，我们可以监听这个事件，检查每个分区是否存在已提交的偏移量，若不存在则直接指定初始位置（比如末尾），从根源避免NoOffsetForPartitionException。

示例代码：

@Component
public class CustomRebalanceListener implements ConsumerAwareRebalanceListener {

    @Override
    public void onPartitionsAssigned(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
        for (TopicPartition partition : partitions) {
            OffsetAndMetadata committedOffset = consumer.committed(partition);
            // 无已提交偏移量，判定为新消费者组，手动定位到分区末尾
            if (committedOffset == null) {
                consumer.seekToEnd(Collections.singleton(partition));
            }
        }
        // 执行父类默认逻辑
        ConsumerAwareRebalanceListener.super.onPartitionsAssigned(consumer, partitions);
    }
}

将监听器绑定到你的MessageListenerContainer：

@Bean
public MessageListenerContainer messageListenerContainer(ConsumerFactory<String, String> consumerFactory,
                                                        CustomRebalanceListener rebalanceListener) {
    ContainerProperties containerProps = new ContainerProperties("your-target-topic");
    containerProps.setMessageListener((MessageListener<String, String>) message -> {
        // 你的消息处理逻辑
    });
    // 添加自定义重平衡监听器
    containerProps.getConsumerRebalanceListeners().add(rebalanceListener);
    
    KafkaMessageListenerContainer<String, String> container = new KafkaMessageListenerContainer<>(consumerFactory, containerProps);
    // 其他容器配置（如并发数、批量消费等）
    return container;
}

方法二：扩展自定义ErrorHandler捕获异常后处理 #

如果你希望在异常抛出后再处理，可以扩展已有的自定义ErrorHandler，捕获NoOffsetForPartitionException并执行seek操作：

public class CustomKafkaErrorHandler extends SeekToCurrentErrorHandler {

    public CustomKafkaErrorHandler() {
        super((record, exception) -> {
            // 原有的异常处理逻辑
        });
    }

    @Override
    public void handle(Exception thrownException, List<ConsumerRecord<?, ?>> records, Consumer<?, ?> consumer, MessageListenerContainer container) {
        if (thrownException instanceof NoOffsetForPartitionException) {
            NoOffsetForPartitionException noOffsetEx = (NoOffsetForPartitionException) thrownException;
            // 对异常涉及的分区执行seek到末尾
            noOffsetEx.partitions().forEach(partition -> 
                consumer.seekToEnd(Collections.singleton(partition))
            );
        } else {
            // 处理其他类型异常
            super.handle(thrownException, records, consumer, container);
        }
    }
}

将该ErrorHandler配置到容器：

containerProps.setErrorHandler(new CustomKafkaErrorHandler());

补充说明 #

• 两种方式都能保留auto.offset.reset=none的配置，不影响你感知OutOfRangeException的需求。
• 若需要将初始偏移量设置到分区开头，只需把seekToEnd替换为seekToBeginning即可。

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