如何通过由REST api触发的Spring云流优雅地处理Kafka发布失败

Question

我有一个场景，必须通过服务层中RESTAPI的触发器发布多条消息。

此服务还具有db操作（插入、更新）。这里的要求是，如果存在导致publisher无法发布的问题的网络问题，或者当kafka broker不可用时，我们希望回滚包括数据库在内的所有内容，并通过API响应通知用户它失败。

问题1：是否可以通过启用Kafkahtml" target="_blank">事务性来实现需求？

据我所知，默认情况下，同一事务中的数据库将首先提交，然后由Kafka提交，但我不确定的是，如果Kafka没有提交，我们将完全丢失消息，数据库也不会回滚。

我使用以下设置对一个简单的spring boot项目进行了一些测试：

1. Spring启动版本：2.5.2
2. Spring云版本：2020.0.3
3. 使用org.springframework.cloud.stream.function.StreamBridge

我无法再现丢失消息的场景。我所尝试的是在事务方法的末尾放置一个延迟，在事务开始提交和关闭kafka broker之前放置一个暂停（在本地环境中），我可以看到发布者将尝试联系kafka，最终会失败，超时导致整个事务（包括db）回滚。

    @Transactional
    public PurchaseOrder createOrder(PurchaseOrder purchaseOrder) throws InterruptedException {
        log.info("Starting create order");

        var result = purchaseOrderRepository.save(purchaseOrder);
        log.info("purchase order is created: {}", result.getId());

        var kafkaResult = streamBridge.send("toStream-out-0", purchaseOrder);
        log.info("Kafka send result {}", kafkaResult);

        Thread.sleep(10000);
        return result;
    }

这是发布服务器端的超时<代码>组织。阿帕奇。Kafka。常见的错误。TimeoutException：等待EndTxn（true）时，超时在60000毫秒后过期

问题2：如果kafka transactional无法实现此要求，是否有更好的方法？

Answer 1

您可以在生产者端使用initTransactions：

https://www.baeldung.com/kafka-exactly-once