Как реализовать «Ровно один раз» потребителя кафки с использованием Apache Kafka? - PullRequest
Новая
       84

Как реализовать «Ровно один раз» потребителя кафки с использованием Apache Kafka?

0 голосов
/ 05 мая 2020

У нас есть приложение, которое потребляет сообщения от Kafka topi c (3 раздела), обогащает данные и сохраняет запись в БД (Spring JPA), а затем публикует сообщения в другой kafka topi c (на тот же брокер), и все это организовано с помощью Camel 2.4.1 и Spring Boot 2.1.7.RELEASE.

Мы хотели бы реализовать семантику «ровно один раз» для комбинации потребитель-производитель kafka.

Пользовательские настройки: 

   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 Wiring: 

   @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;
    }

Camel Route: 

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 зафиксировать сообщения, когда есть сценарий с обработкой исключений. Что мне не хватает, и каков правильный подход?

1 Ответ

0 голосов
/ 05 мая 2020

Кажется, вы используете Spring Kafka, а их KafkaTransactionManager не является настоящим менеджером транзакций с поддержкой XA (ограничения см. В их документации), поэтому вы не можете использовать его для отката как для Kafka, так и для базы данных JDB C et c .

И camel-kafka не поддерживает транзакции Kafka (пока). Я создал заявку: https://issues.apache.org/jira/browse/CAMEL-15016

...