在事务内包装StreamBridge send和JPA save

Question

我使用的是Spring Boot 2.5.2和Spring Cloud 2020.0.3。我正在尝试包装一个rest服务调用，该调用使用JPA（CrudRepository.save）将记录保存到DB，然后使用StreamBridge使用spring cloud stream（Kafka binder）将消息发布到Kafka主题。我试过好几种方法，但似乎没有一种效果很好。我故意造成JPA问题（插入一行将违反唯一密钥约束），但Kafka消息似乎仍会传递给代理。

1. 我已经配置了一个KafkaTransactionManager（没有使用ChainedKafkaTransactionManager，因为现在不推荐使用它）。但是，它似乎被忽略了，因为当配置中存在事务id前缀时，StreamBridge似乎在内部创建自己的tx mgr
2. 如果没有事务id前缀，ProducerFactory就根本不是事务性的，这会导致KafkatTransactionManager实例化失败
3. 我试图完全避免创建自己的事务管理器，但这似乎也失败了，并继续发送Kafka消息

配置这种类型的流以使对db和代理的写入都是原子的，正确的方法是什么？

HTTP协议-

Answer 1

您不需要事务管理器，但需要事务管理器。生产商工厂的id。

如果发送是在JPA事务的范围内执行的（例如，使用JPA TM的Transactional方法），kafka模板将使kafka事务与现有事务同步，并根据主事务提交或回滚。

您是否知道，即使是回滚的记录，实际上也会写入日志？必须设置使用者属性隔离。将级别设置为read\u committed以不接收回滚记录；它默认为“read\u uncommitted”（读取未提交）。

编辑

将仅生产者事务与现有事务同步时出现错误；而是在本地事务中执行发送。

您可以使用交易模板（TransactionTemplate）启动Kafka交易，作为解决方案：

@SpringBootApplication
public class So68460690Application {

    public static void main(String[] args) {
        SpringApplication.run(So68460690Application.class, args);
    }

    @Bean
    public ApplicationRunner runner(StreamBridge bridge, Foo foo, KafkaTransactionManager<byte[], byte[]> ktm) {
        return args -> {
            new TransactionTemplate(ktm).executeWithoutResult(
                    status -> foo.doInTx(bridge)); // or execute() to return a result
        };
    }

    @Bean
    KafkaTransactionManager<byte[], byte[]> binderTM(BinderFactory bf) {
        return new KafkaTransactionManager<>(((KafkaMessageChannelBinder) bf.getBinder("kafka", MessageChannel.class))
                .getTransactionalProducerFactory());
    }

}

@Component
class Foo {

    @Transactional
    public void doInTx(StreamBridge bridge) {
        bridge.send("ouutput", "test");
        throw new RuntimeException("testEx");
    }

}

spring.cloud.stream.bindings.output.destination=so68460690

spring.cloud.stream.kafka.binder.transaction.transaction-id-prefix=tx.
spring.cloud.stream.kafka.binder.configuration.acks=all


logging.level.org.springframework.kafka=trace

2021-07-27 17:31:37.923 DEBUG 55933 --- [           main] o.s.k.core.DefaultKafkaProducerFactory   : CloseSafeProducer [delegate=org.apache.kafka.clients.producer.KafkaProducer@56c8e6f0] beginTransaction()
2021-07-27 17:31:37.924 DEBUG 55933 --- [           main] o.s.k.t.KafkaTransactionManager          : Created Kafka transaction on producer [CloseSafeProducer [delegate=org.apache.kafka.clients.producer.KafkaProducer@56c8e6f0]]
2021-07-27 17:31:37.927 DEBUG 55933 --- [           main] o.s.k.t.KafkaTransactionManager          : Initiating transaction rollback
2021-07-27 17:31:37.928 DEBUG 55933 --- [           main] o.s.k.core.DefaultKafkaProducerFactory   : CloseSafeProducer [delegate=org.apache.kafka.clients.producer.KafkaProducer@56c8e6f0] abortTransaction()