问题描述
我们有一个用例，我们需要使用 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);
}
}
}