如何使用Reactive Panache在同一事务中链式执行两次更新？

1. 给KafkaState添加唯一约束 #

要实现重复消息的检测，首先需要在KafkaState实体上添加唯一约束，确保同一个topic+partition+offset的记录只能插入一次，以此作为重复消息的判断依据：

@Entity
@Table(uniqueConstraints = {
    @UniqueConstraint(columnNames = {"topic", "partition", "offsetN"})
})
public class KafkaState extends PanacheEntity {
    public String topic;
    public int partition;
    public long offsetN;
    // 其他必要字段
}

2. 修复响应式流的链式调用逻辑 #

你原代码的核心问题是state.persist()是异步操作，但没有将其加入到事务的响应式执行链中，导致这个操作不会被实际执行。同时需要捕获唯一约束异常，跳过重复消息的业务处理：

@Inject
Mutiny.Session session;

@ActivateRequestContext
public Uni<Void> persist(ConsumerRecord<Long, String> record) {
    return session.withTransaction(tx -> {
        KafkaState state = new KafkaState();
        state.topic = record.topic();
        state.partition = record.partition();
        state.offsetN = record.offset();

        // 先执行KafkaState的持久化，再链式触发业务逻辑
        return state.persist()
            .chain(() -> {
                Event event = new Event();
                event.key = record.key();
                event.message = record.value();
                return event.persistAndFlush();
            })
            .replaceWithVoid();
    })
    .onFailure(ConstraintViolationException.class)
    .recoverWithUni(t -> {
        // 捕获唯一约束异常，判定为重复消息，直接返回成功跳过业务处理
        System.out.println("重复消息，跳过处理: " + t.getMessage());
        return Uni.createFrom().voidItem();
    })
    .onTermination()
    .call(session::close);
}

关键细节说明 #

• 使用chain()方法串联两个异步操作，确保KafkaState持久化完成后才会执行业务数据的插入，符合事务的原子性要求。
• 通过onFailure(ConstraintViolationException.class)精准捕获重复插入的异常，此时无需执行业务逻辑，直接返回成功即可。
• 简化了Session的关闭逻辑，onTermination()会在操作无论成功还是失败时都触发关闭，避免冗余代码。

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