Spring Cloud Stream 发送到 Kafka 错误控制处理

Question

如何捕获写入 Kafka 主题的错误？

import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.stereotype.Component;

import javax.xml.bind.JAXBException;

@Component
@EnableBinding(Processor.class)
public class Parser {

    @StreamListener(Processor.INPUT)
    @SendTo(Processor.OUTPUT)
    public String process(Message<?> input) {

        // do smth

        // how to catch an error if sending message to 'output' topic fails
    }

}

将生产者切换到同步模式。

spring.cloud.stream.kafka.bindings.output.producer.sync=true

接下来呢？有什么例子吗？扩展一些绑定实现，添加一些 AOP 魔法？

Answer 1

我认为您需要注册一个 KafkaProducerListener 来处理您的案例中的错误场景，并使其在您的应用程序中作为 bean 可用。

更多关于 KafkaProducerListener 的信息是here

另外，请注意，失败的消息通常会出现在errorChannel 上，频道名称为error。为错误通道配置 destination 名称后，所有错误消息都会发布：spring.cloud.stream.bindings.error.destination。