Quarkus中Kafka消费Postgres Upsert事务异常求助

问题根源 #

核心矛盾是阻塞式JTA事务与SmallRye Reactive Messaging的异步线程模型不兼容：你使用的quarkus-hibernate-orm-panache是阻塞式ORM，在返回CompletionStage的异步方法上标注@Transactional时，事务上下文无法正确跨线程传递，导致多线程共享事务连接，最终触发XA异常和连接无活跃事务的错误。

解决方案1：切换到Reactive技术栈（推荐） #

Quarkus的Reactive生态天然适配异步消息场景，能从根本上避免线程上下文问题：

1. 替换依赖：移除阻塞式ORM，改用Reactive版本

<!-- 移除原阻塞式依赖 -->
<!-- <dependency>
    <groupId>io.quarkus</groupId>
    <artifactId>quarkus-hibernate-orm-panache</artifactId>
</dependency> -->

<!-- 添加Reactive Hibernate与Postgres驱动 -->
<dependency>
    <groupId>io.quarkus</groupId>
    <artifactId>quarkus-hibernate-reactive-panache</artifactId>
</dependency>
<dependency>
    <groupId>io.quarkus</groupId>
    <artifactId>quarkus-reactive-pg-client</artifactId>
</dependency>

2. 调整代码为Reactive风格：用@ReactiveTransactional替代@Transactional，确保事务上下文在Reactive流中传递

@Incoming("sink")
@ReactiveTransactional
public Uni<Void> sink(KafkaRecordBatch<String, String> messages) {
    // 确保事务提交后再确认消息
    return persistReactive(messages)
            .flatMap(ignored -> messages.ack());
}

// Reactive版数据持久化方法
private Uni<Integer> persistReactive(KafkaRecordBatch<String, String> messages) {
    return Panache.getEntityManager()
            .createNativeQuery(sql)
            .executeUpdate(); // 返回Uni<Integer>表示影响行数
}

解决方案2：强制阻塞式处理（适合不愿切换技术栈的场景） #

如果必须保留阻塞式ORM，需确保整个消息处理流程在单线程中执行：

1. 修改方法为同步阻塞式：去掉异步返回值，同步等待消息确认

@Incoming("sink")
@Transactional
public void sink(KafkaRecordBatch<String, String> messages) {
    persist(messages);
    // 同步等待ack完成，避免事务未提交就确认消息
    messages.ack().toCompletableFuture().join();
}

2. 配置专属阻塞线程池：在application.properties中指定该消费者使用阻塞线程池

mp.messaging.incoming.sink.connector=smallrye-kafka
mp.messaging.incoming.sink.thread-pool=blocking

关键注意事项 #

• 禁止在Reactive异步流中混用阻塞式JDBC/Hibernate操作，线程切换会直接导致事务连接失效。
• 使用@ReactiveTransactional时，所有数据库操作必须是Reactive类型（返回Uni/Multi），不能包含任何阻塞调用。
• 消息确认必须在事务提交完成后执行，否则会出现「事务回滚但消息已被确认」的重复消费问题。

内容的提问来源于stack exchange，提问作者Ike