Kafka Headers作为批处理模式的列表答案

【问题标题】：Kafka Headers as List in batch modeKafka Headers作为批处理模式的列表
【发布时间】：2021-05-24 12:02:15
【问题描述】：

我正在尝试访问一些标头，同时以批处理模式使用消息。如果我设置监听器来处理Message<?>，我可以手动提取标题

@KafkaListener(topics = "${kafka.topic}")
public void receive (List<Message<?> data, Acknowledgment ack) throws SQLException {
  for (int i = 0; i < data.size(); ++) {
    Object message = data.get(i).getPayload();
    MessageHeaders mh = data.get(i).getHeaders();
    Object value = mh.get("test");

我想为我完成其中的一些，但是当我尝试时

@KafkaListener(topics = "${kafka.topic}")
public voic receive (List<string> data,
  @Header (KafkaHeaders.OFFSET) List<Integer> offsets,
  @Header ("test") List<String> testHeaders,
  Acknowledgment ack) throws SQLException {

我得到MessageHandlingException: Missing header 'test' for method parameter type [interface.java.util.list] 但是，此方法适用于偏移标头。

这是因为有内置代码来处理标准标题，而这种方法不能用于自定义标题，还是我错过了一些可以使这种方法起作用的东西？

【问题讨论】：

你尝试过新鲜的话题吗？也许您发布了没有“测试”标头且您的消费者的 spring.kafka.consumer.auto-offset-reset 设置为 earliest 的早期消息？
我没有使用Message<?> 代码转储数据，并且每条记录都设置了标题。
顺便说一句，请记住，您正在尝试从标头字段中读取字符串列表，但这不是问题。可以分享一下你的制作人吗？
还有一个问题……我无权联系制片人。为了转储数据，我使用代码的第一位将其插入数据库，并将标头转换为字符串。我确实尝试在列表中使用不同的类型（Byte[]、Object、?）但仍然遇到同样的问题。
你真的成功打印出你从mh.get("test")得到的value。我问这个是因为它可能返回 null。

标签： apache-kafka kafka-consumer-api spring-kafka

【解决方案1】：

是的，框架只映射它知道的标头；它将所有其他映射的标头放入

/**
 * The header for a list of Maps of converted native Kafka headers. Used for batch
 * listeners; the map at a particular list position corresponds to the data in the
 * payload list position.
 */
public static final String BATCH_CONVERTED_HEADERS = PREFIX + "batchConvertedHeaders";

如果您想要该标头的离散映射，则需要创建自定义 BatchMessageConverter - 可能是 BatchMessagingMessageConverter 的子类。

可能最简单的方法是覆盖this method，调用super.toMessage()，然后添加您的标题。

return MessageBuilder.fromMessage(super.toMessage(...))
    .setHeader("test", ...)
    .build();

如果您使用的是 Spring Boot，只需将转换器添加为 bean，boot 就会将其连接。

否则将转换器添加到容器工厂。

编辑

如果消息转换器没有标头映射器；所有标题都放在标题KafkaHeaders.NATIVE_HEADERS 中，即List<Headers>。

【讨论】：