在Nifi中创建自定义处理器并发送到PublishKafka

Question

    try {
         this.getLogger().info("[INFO - ListenMW] - Message Received: " +
                            data.getMsgID().toString() + " Size: " +
                            data.getMsgData().length);
         this.currentSession.adjustCounter("MW Counter", 1, true);

         // Setup the flowfile to transfer
         FlowFile flowfile = this.currentSession.create();
         flowfile = this.currentSession.putAttribute(flowfile, "key",data.getMsgID().toString());
         flowfile = this.currentSession.putAttribute(flowfile, "value", new String(data.getMsgData(),StandardCharsets.UTF_8));

         this.currentSession.transfer(flowfile, SUCCESS);


     }catch(Exception e) {
          this.getLogger().error("[INFO - ListenMW] - "+e.getMessage());
          this.currentSession.adjustCounter("MW Failure", 1, true);
     }

我无法确定要为msgID和msgData使用什么属性，所以我现在创建了自己的属性。我看到一篇文章，有人建议构建您自己的json结构，并将其作为有效负载发送，但同样，您将通过哪个属性发送该结构，以便将其正确映射到kafka消息？我对Kafka很陌生，到目前为止，我只尝试了一些基本的测试案例，所以请原谅我的无知，因为我有任何错误的假设。

谢谢您的指导！我使用的是Kafka2.0.1和PublishKafka2.0处理器。

Answer 1

根据您所分享的内容，看起来您没有将任何内容发布到Kafka中的主要原因是您实际上没有向flowfile内容写入任何内容。作为参考，这里有一个javadocs For NiFi的副本（这里还有处理器文档）。您应该这样做：

flowFile = session.write(flowFile, outStream -> {
  outStream.write("some string here".getBytes());
});

我使用publishkafkarecord，但是publishkafka处理器在概念上非常相似。您可以按照在那里的方式设置消息的键，但需要通过将其写入flowfile正文来设置值。

在不了解更广泛的用例的情况下，似乎可以使用executescript来完成需要做的事情。将此视为具有多个脚本语言引用的ExecuteScript的起点。