Spring Cloud SteamKafka误差通道

Question

我正在尝试建立一个绑定，将Kafka消息从Spring Integration errorChannel转发到一个自定义通道（用于集中错误处理）。

错误消息被发送到配置的通道，但它们是以genericmessage的形式到达的，负载为byte[]（包含异常详细信息和失败消息）。

我的配置：

spring:
  cloud:
    stream:
      kafka:
        bindings:
          accountOut.producer:
            sync: true
        binder:
          autoCreateTopics: false
          headers:
               - spanId
               - spanTraceId
               - spanSampled
               - spanParentSpanId
               - spanName
               - spanFlags
               - eventType
               - Authorization
      bindings:
        error:
          destination: test-error
        accountOut:
          producer.partitionKeyExpression: payload.key
          content-type: application/json
          destination: account
  kafka:
    producer.keySerializer: org.apache.kafka.common.serialization.StringSerializer
    consumer.valueDeserializer: org.apache.kafka.common.serialization.StringDeserializer

阅读文档时，我希望消息以errormessage的形式到达。我有什么办法可以做到这一点吗？还是将有效负载配置为作为对象到达？

我正在使用的版本：

• Spring Boot 1.5.8
• Spring Cloud Edgware
• Kafka11客户
• Spring Integration core 4.3.12

@Bean
@ServiceActivator(inputChannel = "errorChannel")
public MessageHandler handler() throws Exception {
    KafkaProducerMessageHandler<String, String> handler =
            new KafkaProducerMessageHandler<>(kafkaTemplate());
    handler.setTopicExpression(new LiteralExpression("someTopic"));
    handler.setMessageKeyExpression(new LiteralExpression("someKey"));
    return handler;
}

但是是否可以在属性yaml中配置这个流，而不是在代码中配置这个流呢？这就是所有其他Kafka配置的地方，所以在代码中配置Kafka模板并不理想。

另一个选择是显式侦听ErrorMessage，并在代码中发送到kafka输出通道：

@ServiceActivator(inputChannel = "errorChannel")
public void handle(ErrorMessage em) {
    outputChannel.kieranError().send(...)
}

Answer 1

你到底是在哪里消费这样的信息呢？

当enabledlq对于使用者为true时，您描述的消息听起来像是发送到DLQ主题的消息；你没有显示消费者配置，所以我很难猜。

发送到目标特定错误通道（并桥接到全局errorchannel的errormessage可以使用

@ServiceActivator(inputChannel = "errorChannel")
public void handle(ErrorMessage em) {
    ...
}

error:
  destination: