如何使用 Apache Kafka 实现“Exactly once” kafka 消费者？答案

【问题标题】：How to implement "Exactly once" kafka consumer using Apache Kafka?如何使用 Apache Kafka 实现“Exactly once” kafka 消费者？
【发布时间】：2020-08-20 14:55:58
【问题描述】：

我们有一个应用程序，它使用来自 Kafka 主题（3 个分区）的消息并丰富数据并将记录保存在 DB（Spring JPA）中，然后将消息发布到另一个 kafka 主题（在同一个代理上）和所有这是通过使用 Camel 2.4.1 和 Spring Boot 2.1.7.RELEASE 编排的。

我们希望为 kafka 消费者-生产者组合实现“exactly-once”语义。

消费者设置：

   autoOffsetReset: earliest
   autoCommitEnable: false
   allowManualCommit: true
   breakOnFirstError: true
   group.id : CONSUMER.GROUP.ID
   count: 3
   max.poll.records = 1  # rollback when message processing fails.

生产者设置：

   idempotence: true
   transactionIdPrefix: txn-prefix-id

豆接线：

   @Bean
    SpringTransactionPolicy springTransactionPolicy() throws Exception {
        SpringTransactionPolicy txRequired = new SpringTransactionPolicy();
        txRequired.setTransactionManager(transactionManager());
        txRequired.setPropagationBehaviorName("PROPAGATION_REQUIRED");
        return txRequired;
    }

    @Bean
    public DefaultKafkaProducerFactory<byte[], byte[]> producerFactory() {
        DefaultKafkaProducerFactory<byte[], byte[]> defaultKafkaProducerFactory = new DefaultKafkaProducerFactory<byte[], byte[]>(
                kafkaConfigs());
        // enable transaction manager
        defaultKafkaProducerFactory.setTransactionIdPrefix(transactionIdPrefix);
        return defaultKafkaProducerFactory;
    }


    @Bean
    @Primary
    public ChainedKafkaTransactionManager<byte[], byte[]> transactionManager() throws Exception {
        return new ChainedKafkaTransactionManager<>(kafkaTransactionManager(),jpaTransactionManager());
    }

    @Bean
    public PlatformTransactionManager kafkaTransactionManager() {
        KafkaTransactionManager<byte[], byte[]> kafkaTransactionManager = new KafkaTransactionManager<>(producerFactory);
        kafkaTransactionManager.setTransactionSynchronization(AbstractPlatformTransactionManager.SYNCHRONIZATION_ON_ACTUAL_TRANSACTION);
        kafkaTransactionManager.setRollbackOnCommitFailure(true);
        return kafkaTransactionManager;
    }

    @Bean
    JpaTransactionManager jpaTransactionManager() {
        JpaTransactionManager transactionManager = new JpaTransactionManager();
        transactionManager.setRollbackOnCommitFailure(true);
        return transactionManager;
    }

骆驼路线：

public RoutesBuilder inboundRoute() {
        return new RouteBuilder() {

            @Override
            public void configure() throws Exception {
                //Common error handler
                onException(UnsupportedMessageTypeException.class).
                   maximumRedeliveries(redeliveryCount).
                   handled(true).
                   bean(ExceptionPropagatorProcessor.class, "process").
                   bean(manualCommitProcessor).
                end();

                onException(AppRuntimeException.class).
                   maximumRedeliveries(redeliveryCount).                
                   bean(ExceptionPropagatorProcessor.class, "process")
                end();  

                onException(RetryExhaustedException.class).
                   maximumRedeliveries(0).// No retry for this exception
                   handled(true).
                   bean(ExceptionPropagatorProcessor.class, "process").
                   bean(kafkaManualCommitProcessor).
                end(); 

                from("kafka:inboundTopic").
                    routeId("consume-msg").
                    transacted("springTransactionPolicy").
                               bean(transactionBeginProcessor).
                               //check if this is a retry scenario, the max retry count reached then throw RetryExhaustedException.
                               bean(retryEvaluationProcessor). 
                               bean(enrichProcessor). // publish kafka messages
                               bean(persistenceProcessor).
                               bean(transactionEndProcessor). // publish kafka messages
                    bean(manualCommitProcessor);

但是当出现异常处理场景时，我们无法让 kafka 生产者提交消息。我缺少什么，正确的方法是什么？

【问题讨论】：

标签： java spring apache-kafka apache-camel spring-camel

【解决方案1】：

您似乎使用的是 Spring Kafka，而他们的 KafkaTransactionManager 不是真正的 XA 事务管理器（请参阅他们的文档了解限制），因此您不能将它用于 Kafka 和 JDBC 数据库等的回滚。

而且 camel-kafka 不支持 Kafka 事务（目前）。我已经创建了一张票：https://issues.apache.org/jira/browse/CAMEL-15016

【讨论】：