如何从 spring 云流应用程序启动源生成的 kafka 消息中删除内容类型 header

Question

对于 twitter 源，如果我们使用 @StreamListener（接收器）以外的自定义 kafka 消费者，它会为每个 kafka 消息创建一个 30 位 header。 @StreamListener 将消息正确发送到带注释的方法（没有 header）。问题是每个来源的 header 大小都不同。那么，自从我使用自定义 kafka 消费者以来，我如何动态理解消息的 header 大小？或者如何删除每条消息的 header。？

更新：-

请在下面找到使用 TCP 源时在 kafka-console-consumer 中收到的消息。

�
  contentType"application/octet-stream"hello world

我只需要收到消息而已。在这种情况下，你好世界

Answer 1

您使用的是什么版本的 Kafka？我问的原因是从 0.11 Kafka 开始支持原生 headers 并且你不会再有这个问题了。

Answer 2

在源中设置 spring.cloud.stream.default.producer.headerMode=raw 将在读取消费者中的消息时删除 header。

Answer 3

也可以更改此设置以删除烦人的 header

spring.cloud.stream.default.producer.useNativeEncoding=true

如何从 spring 云流应用程序启动源生成的 kafka 消息中删除内容类型 header

How to remove the content type header from kafka messages produced by the spring cloud stream app starter sources

java

spring

spring-boot

spring-cloud-stream