1.實現(xiàn)BatchErrorHandler

@Component
public  class KafkaBatchExceptionHandler implements BatchErrorHandler {

    @Override
    public void handle(Exception e, ConsumerRecords<?, ?> consumerRecords) {
    }

    @Override
    public void handle(Exception thrownException, ConsumerRecords<?, ?> consumerRecords, Consumer<?, ?> consumer) {
        Map<TopicPartition, Long> offsetsToReset = new HashMap<>();
        consumerRecords.forEach((record) -> {
            offsetsToReset.compute(new TopicPartition(record.topic(), record.partition()),
                    (k, v) -> v == null ? record.offset() : Math.min(v, record.offset()));

        });
        offsetsToReset.forEach((k, v) -> consumer.seek(k, v));

    }


}

2.注冊進KafkaListenerContainerFactory中

   @Bean
    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, GenericRecord>> kafkaListenerContainerFactory(@Autowired KafkaBatchExceptionHandler batchErrorHandler) {
        ConcurrentKafkaListenerContainerFactory<String, GenericRecord> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(null);
        factory.setBatchListener(true);
        factory.getContainerProperties()
                .setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        factory.getContainerProperties().setPollTimeout(30000);
        factory.setBatchErrorHandler(batchErrorHandler);
        return factory;
    }