使用 apache Storm 从 kafka 读取标头

如何解决使用 apache Storm 从 kafka 读取标头

我们有一个用例，我们需要使用 apache Storm 从 kafka 读取消息头并将其传递给下游 Bolt。在 Storm 文档中，没有提到如何将 kafka 标头传递给 Storm 拓扑。有没有人想出如何做到这一点？

解决方法

一种方法是在构建 spout 时使用 setRecordTranslator。 示例：

return KafkaSpoutConfig.builder(
        "<bootstrap-servers>","topic-to-consume")
        .setRecordTranslator((r) -> new Values(r.topic(),r.partition(),r.offset(),r.key(),r.value(),r.headers()),new Fields("topic","partition","offset","key","value","headers"))
        .build();

现在，在 Bolt 的 execute 中，您将能够访问上述所有字段，包括标题，如下所示：

public class MyNewBolt extends BaseRichBolt {

   @Override
   public void execute(Tuple tuple) {

      RecordHeaders messageHeaders = (RecordHeaders) tuple.getValueByField("headers");
      Iterator headerIterater = messageHeaders.iterator();
      while (headerIterater.hasNext()) {
         Header header = (Header) headerIterater.next();
         String headerKey = header.key();
         byte[] headerBytes = header.value();
         String headerValue = new String(headerBytes);
         System.out.println("Header Key: " + headerKey);
         System.out.println("Header Value: " + headerValue);

      }
}

}