如果 Kafka 服务器不可用，则终止 Spring Kafka 事务答案

【问题标题】：Terminate Spring Kafka transaction if Kafka server is unavailable如果 Kafka 服务器不可用，则终止 Spring Kafka 事务
【发布时间】：2018-12-23 20:50:51
【问题描述】：

在我们的项目中，我们需要通过多个线程定期从 3rd 方获取数据，然后将这些数据推送到 Kafka。如果此时 Kafka 服务器不可用，则应终止流程，获取的数据应丢失并在下一次计划执行期间重新获取。此外，还需要使用事务管理，因为我们需要将这批消息发送到 Kafka 到不同的主题。如果没有发送一条消息，则应回滚所有其他消息。

如果 Kafka 服务器不可用，我们会遇到终止执行的问题。当事务管理被禁用时，一切正常，我们得到了

org.apache.kafka.common.errors.TimeoutException: Failed to update metadata after 60000 ms.

但是启用事务管理后，Kafka 生产者会尝试无限地访问服务器，并且所有启动将消息推送到 Kafka 的线程都会卡住。我们尝试了不同的设置以使其在一段时间不可用后失败，但没有帮助。设置spring.kafka.producer.retries: 0 或spring.kafka.producer.acks: 0 导致（不是完整的堆栈跟踪）：

Caused by: org.apache.kafka.common.config.ConfigException: Must set retries to non-zero when using the idempotent producer.

Caused by: org.apache.kafka.common.config.ConfigException: Must set acks to all in order to use the idempotent producer. Otherwise we cannot guarantee idempotence.

当 Kafka 服务器不可用且启用事务管理时，是否有一些设置组合使 Kafka Producer 在超时后失败？这可能吗？

【问题讨论】：

标签： java apache-kafka spring-kafka

【解决方案1】：

Spring-Kafka 使用 DefaultAfterRollbackProcessor 寻找失败的偏移量并重试 - 这将继续循环，直到偏移量被正确处理。这是默认行为。如果事务失败，您将回滚，这取决于@Transactional 上的rollbackFor 属性。

你有一个特殊情况，如果 Kafka 服务器不可用，它将回滚。您可以通过实现AfterRollbackProcessor 创建自己的处理器。由于连接超时，您需要区分常规回滚和回滚。

编辑： 您还可以在@Transactional 上定义属性noRollbackFor 以排除TimeoutException 并让此异常到达Container。您可以创建自定义 ExceptionHandler 并在容器上使用 setErrorHandler(..)。你可以阅读容器错误处理程序here

【讨论】：

感谢您的回答，但两种方式都不适用于我。我尝试调试并发现 DefaultAfterRollbackProcessor 在我的情况下从未执行过。另外，我注意到org.springframework.kafka.listener.DefaultAfterRollbackProcessor#process 仅在KafkaMessageListenerContainer 中被调用，但我的应用程序是消息生产者，而不是消费者。另外，我实际上需要抛出TimeoutException，并且当启用事务管理时，它永远不会被抛出，这是我的问题:)
我再次查看了您的问题。我能够重现您描述的问题。我查看了org.apache.kafka.clients.producer.internals.Sender - 我找不到任何属性来调整这种行为。我以前从未遇到过这种行为。你的 Kafka 经纪人从一开始就倒闭了吗？ issues.apache.org/jira/browse/KAFKA-5415 看起来与您的问题相似。
这是 kafka 客户端中的一个错误（请参阅this answer）。它将在 kafka-clients 2.0.0 中修复 - 请参阅 KAFKA-6446。
@GaryRussell 谢谢！目前，我们正在使用 spring kafka 2.1.6.RELEASE，它使用 kafka-clients 1.0.1。我现在会考虑一种解决方法，但如果您能告知 kafka-clients 2.0.0 的计划发布日期，我将不胜感激。
对不起，我不是 kafka 团队的一员（只有 spring-kafka，它使用 kafka-clients jar）。他们的release plan 说是 6 月 26 日。我已经针对他们的 2.0.0.rc1 分支测试了一个 spring-kafka 2.2 版本；所以我正在等待他们的发布，以便发布 2.2.0.M1 里程碑。