使用 Cloud Stream Binder 处理 Springboot Kafka Producer 错误

Question

我创建了一个 Springboot 应用程序来将消息推送到 Kafka 主题。该应用程序运行良好。我正在尝试的是在将消息发送到 Kafka 主题时出现故障时处理异常。我在发送消息时使用错误通道来跟踪错误。但实际问题是，我能够看到错误消息，但我无法看到错误消息中失败的实际有效负载。实际上，我想记录该有效负载。

我尝试发送的 JSON 消息：{"key1":"value1"}

服务类：

@AllArgsConstructor
@EnableBinding(Source.class)
public class SendMessageToKafka {

    private final Source source;

    public void sendMessage(String sampleMessage) {
        source.output.send(MessageBuilder.withPayLoad(sampleMessage).build());
    }

    @ServiceActivator(inputChannel = "errorChannel")
    public void errorHandler(ErrorMessage em) {
        System.out.println(em);
    }
}

application.yml:

spring:
 cloud:
  stream:
   bindings:
    output:
     producer:
      error-channel-enabled: true

通过以上配置，当Kafka服务器宕机时，控制权来到errorHandler方法并打印消息。但我无法从错误消息中看到 {"key1":"value1"} 的实际有效负载。如何从错误消息中检索它？

Answer 1

您可以根据类型过滤负载，例如KafkaSendFailureException（https://docs.spring.io/spring-kafka/docs/2.2.0.RELEASE/reference/html/_spring_integration.html 表示ErrorMessage payload 就是这种类型。） 之后，在我的案例中起作用的是将其转换为原始发送的消息，如下所示（通过断点分析对象以确定适当的值类型，例如 byte[]）：

@ServiceActivator(inputChannel = "errorChannel")
public void errorHandler(ErrorMessage em) {
    log.debug("got error message over errorChannel: {}", em);
    if (null != em.getPayload() && em.getPayload() instanceof KafkaSendFailureException) {
        KafkaSendFailureException kafkaSendFailureException = (KafkaSendFailureException) em.getPayload();
        if (kafkaSendFailureException.getRecord() != null && kafkaSendFailureException.getRecord().value() != null
                && kafkaSendFailureException.getRecord().value() instanceof byte[]) {
            log.warn("error channel message. Payload {}", new String((byte[])(kafkaSendFailureException.getRecord().value())));
        }
    }
}