如何防止主主题重新使用 DLQ 处理的消息？

Question

我有一个 kafka 消费者类，它有一个主主题侦听器和一个 DLQ 侦听器。当主主题监听器无法处理消费者记录时，根据我的 bean 工厂，记录被推送到 DLQ 主题中。因此，DLQ 成功处理了该消息。但是，当我重新启动使用者应用程序时，我看到 DLQ 处理的消息再次被主主题侦听器使用，尽管它已成功处理。有人可以帮助我如何防止主要主题重新使用DLQ处理的消息吗？提前感谢您！

Kafka·Consumer.java

public class KafkaConsumer {
//MAIN TOPIC LISTENER
@KafkaListener(id = "main-topic", topics = "main-topic", groupId = "main", containerFactory = "kafkaListenerContainerFactory", clientIdPrefix = "main-topic")
    public void mainListener(ConsumerRecord<String, String> consumerRecord, Acknowledgement ack) {
     //The below line of code converts the consumerRecord value into an Object class and save the converted object to DB.   
     dbService.saveTodb(consumerRecord.value(), new ObjectMapper());
        ack.acknowledge();
    }

//DLQ LISTENER
@KafkaListener(id = "DLQ-topic", topics = "DLQ-topic", groupId = "main", clientIdPrefix = "DLQ", autostartup= "false")
    public void mainListener(ConsumerRecord<String, String> consumerRecord, Acknowledgement ack) {
        dbService.saveTodb(consumerRecord.value(), new ObjectMapper());
        ack.acknowledge();
    }
}

KafkaBeanFactory.java

public class KafkaBeanFactory{
@Bean(name = "kafkaListenerContainerFactory")
    public ConcurrentKafkaListenerContainerFactory<Object, Object> kafkaListenerContainerFactory(
            ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
            ConsumerFactory<Object, Object> kafkaConsumerFactory, KafkaTemplate<Object, Object> template) {
        ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
        configurer.configure(factory, kafkaConsumerFactory);
        var recoverer = new DeadLetterPublishingRecoverer(template,
                (record, ex) -> new TopicPartition("DLQ-topic", record.partition()));
        var errorHandler = new DefaultErrorHandler(recoverer, new FixedBackOff(3, 20000));
        errorHandler.addRetryableExceptions(JsonProcessingException.class, DBException.class);
        errorHandler.setAckAfterHandle(true);
        factory.setCommonErrorHandler(errorHandler);
        return factory;
}
}

应用程序.yaml

 kafka:
    bootstrap-servers: localhost:9092 # sample value 
    client-id: main&DLQ
    properties:
      security:
        protocol: SASL_SSL
      sasl:
        mechanism: PLAIN
        jaas:
          config: org.apache.kafka.common.security.plain.PlainLoginModule required username="$ConnectionString" password="<string>";
          security:
            protocol: SASL_SSL
    consumer:
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      groupId: main
      enable-auto-commit: false
      auto.offset.reset: earliest
    listener:
      ack-mode: MANUAL_IMMEDIATE

Answer 1

您需要在默认错误处理程序上设置< code>commitRecovered属性。

    /**
     * Set to true to commit the offset for a recovered record.
     * The container must be configured with
     * {@link org.springframework.kafka.listener.ContainerProperties.AckMode#MANUAL_IMMEDIATE}.
     * Whether or not the commit is sync or async depends on the container's syncCommits
     * property.
     * @param commitRecovered true to commit.
     */
    @Override
    public void setCommitRecovered(boolean commitRecovered) { // NOSONAR enhanced javadoc