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

Question

我有一个rest postendpoint，它使用Spring Cloud Stream Kafka绑定器来使用数据和写到Kafka。现在我们没有任何错误处理到位。但是我们希望通过在没有将数据写入Kafka时添加一个额外的检查来使这个endpoint容错。当数据没有写入Kafka时，我们打算发送一个异常信息对象。我试图用这种方法使用全局错误来实现这一点

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

我的怀疑有两个方面：

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

如果有另一个过程，请建议。我们目前正在探索应用程序级错误处理和全局级错误处理。系统级错误处理目前是不可能的。提前谢了。

Answer 1

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

通过将kafka producer属性sync设置为truehttps://docs.spring.io/spring-cloud-stream-binder-kafka/docs/3.1.3/reference/html/spring-cloud-stream-binder-kafka.html#kafka-producer-properties