如何通过从 REST api 触发的 Spring Cloud Stream 优雅地处理 Kafka 发布失败答案

【问题标题】：How to gracefully handle Kafka publication failure, via spring cloud stream, triggered from REST api如何通过从 REST api 触发的 Spring Cloud Stream 优雅地处理 Kafka 发布失败
【发布时间】：2021-07-25 05:26:17
【问题描述】：

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

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

Q1：是否可以通过启用Kafka transactional来实现需求？

据我了解，默认情况下，在同一个事务数据库中将首先提交，然后由 kafka 提交，但我不确定是否有任何情况下 kafka 无法提交，我们将完全丢失消息而数据库不会回滚。

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

Spring Boot 版本：2.5.2
春云版本：2020.0.3
使用了org.springframework.cloud.stream.function.StreamBridge

我无法重现丢失消息的场景。我尝试的是在事务方法结束时延迟以在事务开始提交和关闭 kafka 代理之前暂停（在本地环境中），我可以看到发布者将尝试联系 kafka 并最终会超时失败导致整个事务回滚，包括数据库。

    @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;
    }

这是发布者方面的超时。 org.apache.kafka.common.errors.TimeoutException: Timeout expired after 60000milliseconds while awaiting EndTxn(true)

Q2：如果 kafka transactional 无法达到这个要求，有没有更好的方法呢？

【问题讨论】：

标签： apache-kafka spring-integration spring-cloud-stream spring-transactions spring-cloud-stream-binder-kafka

【解决方案1】：

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

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

【讨论】：

感谢您的回答。我实际上是在示例中的生产者端使用启用了 kafka 事务的 Spring Cloud Stream。我的问题（Q1）更多的是关于这是否足够安全以防止消息在任何失败情况下丢失。以及（Q2）是否有其他更好的方法来解决这个问题。
@Peeranut 比你可以使用重试机制为blogs.perficient.com/2021/02/15/…
您好，您发送的链接更多是关于消费者发起错误处理的。在这种情况下，正如我所解释的，发布者的触发器来自 rest api。至于重试，例如，如果无法访问代理，它将失败。如果该事件发生，我的第一个想法是回滚所有内容。并且 Kafka 事务似乎正在完成这项工作，因为我在最初的帖子中提交事务时从 kafka 获得了超时并且它回滚了。问题是这是否是防弹的，或者在 Spring Cloud Stream 不会丢失消息的上下文中是否有任何其他更好的解决方案。