使用 Spring Cloud Stream Kafka Binder 时 Kafka Producer 中的错误处理答案

【问题标题】：Error Handling in Kafka Producer while using Spring Cloud Stream Kafka Binder使用 Spring Cloud Stream Kafka Binder 时 Kafka Producer 中的错误处理
【发布时间】：2021-05-27 13:32:23
【问题描述】：

我有一个休息后端点，它使用 Spring Cloud Stream Kafka Binder 消耗数据并写入 Kafka。现在我们没有任何错误处理。但我们希望通过在数据未写入 Kafka 时添加额外检查来使此端点容错。我们打算在数据未写入 Kafka 时发送异常信息对象。我正在尝试以这种方式使用全局错误来实现这一目标

@ServiceActivator(inputChannel = "errorChannel")
public void handle(final ErrorMessage em) {
logger.error("encountered exception" + em.getOriginalMessage().toString());
throw new RuntimeException(em.getOriginalMessage().toString);
}

我的疑惑有两个：

当我们写入 Kafka 的数据失败时，这是处理异常的正确方法吗？
是否在数据写入失败时调用此句柄方法，并且此更改是否传播到系统级错误处理。

如果有其他流程请提出建议。我们目前正在探索应用程序级错误处理和全局级错误处理。系统级错误处理暂时不在讨论范围内。提前致谢。

【问题讨论】：

标签： java spring apache-kafka spring-cloud-stream spring-cloud-stream-binder-kafka

【解决方案1】：

你必须选择异步错误处理，使用errorChannelEnabledhttps://docs.spring.io/spring-cloud-stream/docs/3.1.3/reference/html/spring-cloud-stream.html#_producer_properties

您可以通过将 kafka 生产者属性 sync 设置为 true https://docs.spring.io/spring-cloud-stream-binder-kafka/docs/3.1.3/reference/html/spring-cloud-stream-binder-kafka.html#kafka-producer-properties 来获取发送线程上的异常

【讨论】：

这很有帮助。我还有一个问题。通过启用同步属性，我可以将用于发送数据的代码 sn-p 包含在 try catch 块中，并像我们在正常情况下一样传播异常。
是的；那是对的;使用同步，通道适配器等待发送未来完成（通常，或者有异常，可以被捕获）。
我们这里没有使用期货，我们使用消息通道发送来发送数据。那么我们可以将它包装在 try catch 块中并处理它。抱歉这个愚蠢的问题，请不要介意。
未来（由KafkaTemplate返回）在通道适配器下游使用；当同步为真时，适配器等待它完成（在调用线程上），因此如果发送失败，您将收到异常。
谢谢加里，这真的很有帮助。如果我想异步执行此操作，我可以使用我提到的代码 sn-p 来执行此操作，并且我需要正确启用 errorChannel。这样，每当 errorchannel 收到消息时，它都会正确抛出异常。