如何使用ApacheKafka实现“恰好一次”kafka消费者？

Question

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

我们想为 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制作人提交消息。我缺少什么，正确的方法是什么？

Answer 1

您似乎正在使用Spring Kafka，而他们的KafkaTransactionManager并不是一个真正支持XA的事务管理器（有关限制，请参阅他们的文档），因此您不能将其用于Kafka和JDBC数据库等的回滚。

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