SpringXD：从 Kafka 读取消息时，消息不包含标头答案

【问题标题】：SpringXD: messages don't include headers when a message is read from KafkaSpringXD：从 Kafka 读取消息时，消息不包含标头
【发布时间】：2017-01-17 23:06:10
【问题描述】：

我有以下结构：

一个卡夫卡话题
我的 kafka 接收器/源的模块
kafka 源码之后的一个模块。基本上是阅读 kafka 源发送的内容。

问题是，当kafka sink向kafka Queue发送事件时，kafka source已经收到了来自kafka topic的消息，然后，下一个模块尝试读取headers，因为没有找到headers而失败.

我提出以下解决方案：将标头包装在消息中，因此，内部有效负载将包含原始有效负载 + 标头。

我还有什么选择？

【问题讨论】：

标签： apache-kafka spring-integration spring-xd

【解决方案1】：

由于 Kafka 本身不支持标头，因此 Kafka 消息总线支持您描述的确切机制 - 只是您必须选择要通过总线传输的标头名称，以避免传输不支持的标头需要。

见Application Configuration。

向下滚动到 Kafka 消息总线属性。

messagebus:
 kafka:
  # connection properties
  brokers:                                localhost:9092  (1)
  zkAddress:                              localhost:2181  (2)
  socketBufferSize:                       2097152         (3)
  # operating mode
  mode:                                   embeddedHeaders (4)
  offsetManagement:                       kafkaTopic      
  headers:
  # comma-delimited list of additional header names to transport (6)
  ...

要由总线传输的自定义标头列表。

编辑

对于rabbit，默认传输所有header：

    replyHeaderPatterns:       STANDARD_REPLY_HEADERS,*   
    ...
    requestHeaderPatterns:     STANDARD_REQUEST_HEADERS,*

编辑

如果您的意思是希望使用相同的技术使用接收器/源从 Kafka 发送/接收，那么您可以使用 XD 的支持类 - 请参阅 EmbeddedHeadersMessageConverter。标头值编码为 JSON。

【讨论】：

这意味着我应该改变我们自去年以来一直使用的数据传输（我们使用的是 rabbit）。这听起来不太划算，我的意思是如果我将数据传输更改为 Kafka。如果我不想更改数据传输，应该是我描述更合适的解决方案的选项吗？
如果你使用 Rabbit 作为传输，所有的 headers 都会被传输（映射到 rabbitmq headers）；仅对于 kafka 传输，您必须配置传输哪些标头。兔子传输可以配置为限制标头，但默认情况下，全部发送。
如果您的意思是希望使用相同的技术使用接收器/源从 Kafka 发送/接收，那么您可以使用 XD 的支持类 - 请参阅 EmbeddedHeadersMessageConverter。标头值使用 JSON 编码。