当我使用ConcurrentMessageListenerContainer时，如何跳过Kafka中有错误的消息？

Question

我用的是spring-kafka-2.1.10。RELEASE尝试跳过错误数据，但失败，

它落在一个无限消耗的循环中

我有什么要说的吗？

如果你能帮忙，我将不胜感激

简单代码：

=========================

@Service
public class testConsumerFactoryService {
    @Autowired
    private PlatformTransactionManager tx;
    @Autowired
    private TestRepository testRepository;  // table only have 1 column , String
    @Autowired
    private ConsumerFactory<String, String> consumerFactory;

    @PostConstruct
    public void consumerFactoryTest() {
      ContainerProperties containerProps = new ContainerProperties("test_1");
        containerProperties.setTransactionManager(this.tx);
        containerProperties
                .setMessageListener(new ConsumerAwareMessageListener<String, String>() {
                    @Override
                    public void onMessage(final ConsumerRecord<String, String> record,
                            final Consumer<?, ?> consumer) {
                        final String rec = record.value();
                        //add the 'rec' string to TestRepository's entity , aka: tests
                        try {
                            this.testRepository.save(tests);  //  >> 1. try to save in DB, but failed
                        } catch (final Throwable e) {
                            System.out.println("error !!" + e);
                            System.out.println("##records:" + record);
                            consumer.commitSync();  // >> 2. try to commit offset, but seems not working ?
                            System.out.println("##after:");
                        }
                    }
                }
            );
        final ConcurrentMessageListenerContainer<String, String> container = new ConcurrentMessageListenerContainer<>(
                this.consumerFactory, containerProperties);
        container.start();

    }
}

====================================================配置：===============================

@Configuration
@EnableKafka
public class Config {
@Bean
ConcurrentKafkaListenerContainerFactory<String, String>
                    kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, String> factory =
                            new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    //use AckMode.RECORD
    factory.getContainerProperties().setAckMode(AckMode.RECORD);
    return factory;
}

@Bean
public ConsumerFactory<String, String> consumerFactory() {
    return new DefaultKafkaConsumerFactory<>(consumerConfigs());
}

@Bean
public Map<String, Object> consumerConfigs() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, embeddedKafka.getBrokersAsString());
    props.put(ConsumerConfig.GROUP_ID_CONFIG, 'Test1');
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
    return props;
}

Answer 1

在您的ConcurrentKafkaListenerContainerFactory中，只需关联一个SeekTocurInterrorHandler：

factory.setErrorHandler(
        new SeekToCurrentErrorHandler(
            (rec, ex) -> {
              log.error(
                  "An exception happened during a message processing: {}", ex);
              // good to put something like a dispatch to a DLQ etc
            }));

有了这个，你将继续到下一个偏移量，你的应用程序不会因为坏消息等而被阻止。