Spring：带有 Jpa 和 Kafka 的 ChainedKafkaTransactionManager 不是原子的？答案

【问题标题】：Spring: ChainedKafkaTransactionManager with Jpa and Kafka is not atomic?Spring：带有 Jpa 和 Kafka 的 ChainedKafkaTransactionManager 不是原子的？
【发布时间】：2021-12-13 16:41:34
【问题描述】：

我们的基础架构必须管理 2 个 Spring Boot 应用程序之间的信息交换。一些信息基于 Oracle DB。我们使用 Kafka 来通知哪些信息必须由接收者管理。

（注意：我必须概括代码）

我们有 Kafka 生产者，配置如下：

@Configuration
@Lazy(false)
public class KafkaConfig {

    //...

    @Bean(name="trManagerJpaKafka")
    public ChainedKafkaTransactionManager<Object, Object> chainedTm(KafkaTransactionManager<String, String> ktm,
            JpaTransactionManager jpaTransactionManager) {

        return new ChainedKafkaTransactionManager<>(jpaTransactionManager,ktm);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(KafkaOperations<?, ?> template,
            ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
            ConsumerFactory<Object, Object> kafkaConsumerFactory, ChainedKafkaTransactionManager<Object, Object> chainedTm) {

        ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
        
        configurer.configure(factory, kafkaConsumerFactory);
        
        //...

        factory.setErrorHandler(new SeekToCurrentErrorHandler(recoverer,new FixedBackOff(0L, 2)));
        factory.getContainerProperties().setTransactionManager(chainedTm);
        return factory;
    }

}

然后我们有一些服务产生 Kafka 消息，像这样

public class SomeService implements ISomeService {
    
    @Override
    @Transactional(transactionManager = "trManagerJpaKafka" , rollbackFor = { Exception.class })
    public boolean createMessage(Long id, CreateRDM createRDM) {

        ...
        
        Entity entity = new Entity();
        entity.setSomething(something);
        entity.setSomethingElse(somethingElse);
        
        entity = entityRepository.save(entity);

        JSONObject message = new JSONObject();
        
        try {
            message.put("id", entity.getId());
            kafkaProducerService.sendMessage(message.toString());
            
        ...
    }
    ...
}

public class SomeGenericKafkaService implements ISomeGenericKafkaService {

    ...

    public void sendMessage(String data) {
        Map<String, Object> headers = new HashMap<String, Object>();
        headers.put(KafkaHeaders.TOPIC, someTopic);
        headers.put(KafkaHeaders.MESSAGE_KEY, UUID.randomUUID().toString());
        
        KafkaProducerCallback callback = new KafkaProducerCallback();
        
        kafkaTemplate.send(new GenericMessage<String>(data, headers)).addCallback(callback);
        
        if (callback.isError()) {
            throw new KafkaException(callback.getThrowable());
        }
    }   

    ...

}

在消费者 Spring Boot 应用程序中，我们有时会遇到这个问题：它使用来自 Kafka 的消息，但有时 DB 上的记录尚不存在，因此当我们尝试访问 DB 记录时出现异常...... 生产者中的事务对于 Jpa 和 Kafka 都不是原子的？ Kafka 消息在 Jpa 插入之前提交？

同时，作为紧急解决方案，为了管理从 Oracle 检索数据的异常，我们在接收器中放置了一些重试，类似于他的：

    ...
    Optional<Entity> optionalEntity = entityRepository.findById(id);
    
    if (!optionalEntity.isPresent()) {
        for (int i = 0; i < kafkaDbMaxRetry; i++) {
            Thread.sleep(kafkaMessageDelay);
            optionalEntity = entityRepository.findById(id);
            if (optionalEntity.isPresent()) break;
        }
    }
    
    entity = optionalEntity.get();
    ...

如何调整生产者，避免消费者出现这种糟糕的代码？

【问题讨论】：

标签： java spring spring-boot spring-data-jpa spring-kafka

【解决方案1】：

Kafka 仅按顺序处理来自一个分区的事件，但是如果事件最终位于两个不同的分区中，则它们可能会被乱序处理。因此，您可以实现发送消息逻辑，例如将数据库记录的主键包含为分区键。

Kafka 使用 key 来指定目标分区。默认策略是根据 key 的 hash 选择分区，如果 key 为 null，则使用循环算法。

您可以实现自定义org.apache.kafka.clients.producer.Partitioner 将消息映射到您需要的分区。您的类的名称必须设置为生产者的 partitioner.class 属性。

【讨论】：

【解决方案2】：

链式事务管理器只提供“Best Effort 1 Phase Commit”——Kafka不能参与JTA/XA事务；用这些不同的技术提供原子更新是不可能的。

您必须使您的侦听器具有幂等性 - 一种常见的模式是将主题/分区/偏移量与数据一起存储，以便您可以检查它是否已被处理。

【讨论】：