连接 Confluent Kafka 和 InfluxDB 时出现 NullPointerException答案

【问题标题】：NullPointerException when connecting Confluent Kafka and InfluxDB连接 Confluent Kafka 和 InfluxDB 时出现 NullPointerException
【发布时间】：2019-03-19 04:14:43
【问题描述】：

我正在尝试使用 Confluent InfluxDB Sink Connector 将来自 kafka 主题的数据获取到我的 InfluxDB 中。

首先，我使用 nifi 将数据从日志文件传输到 kafka 主题，效果很好。 kafka topic获取数据，如下：

  {
    "topic": "testDB5",
    "key": null,
    "value": {
      "timestamp": "2019-03-20 01:24:29,461",
      "measurement": "INFO",
      "thread": "NiFi Web Server-795",
      "class": "org.apache.nifi.web.filter.RequestLogger",
      "message": "Attempting request for (anonymous) 
    },
    "partition": 0,
    "offset": 0
  }

然后，我通过 Kafka Connect UI 创建 InfluxDB 接收器连接器，我得到以下异常：

org.apache.kafka.connect.errors.ConnectException: Exiting WorkerSinkTask due to unrecoverable exception.
    at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:587)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:323)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:226)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:194)
    at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:175)
    at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:219)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.NullPointerException
    at io.confluent.influxdb.InfluxDBSinkTask.put(InfluxDBSinkTask.java:140)
    at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:565)
    ... 10 more

但是如果我手动输入数据到另一个主题 testDB1 使用

./bin/kafka-avro-console-producer --broker-list localhost:9092 --topic testDB1 --property value.schema='{"type":"record","name":"myrecord","fields":[{"name":"measurement","type":"string"},{"name":"timestamp","type":"string"}]}'

有效，我的influxDB可以获取数据。

这里是连接配置：

connector.class=io.confluent.influxdb.InfluxDBSinkConnector
influxdb.url=http://myurl
tasks.max=1
topics=testDB5

连接主题testDB1的配置除了主题名称相同。

nifi 有什么问题吗？但它可以很好地将数据传输到主题。

【问题讨论】：

如何序列化写入 Kafka 主题的数据？您在 Kafka Connect 工作程序属性文件中使用了哪些序列化程序 (converter) 设置？
在etc/schema-registry/connect-avro-distributed.properties中，设置为：key.converter=io.confluent.connect.avro.AvroConverter key.converter.schema.registry.url=http://localhost:8081 value.converter=io.confluent.connect.avro.AvroConverter value.converter.schemas.enable=false value.converter.schema.registry.url=http://localhost:8081
如何序列化写入 Kafka 主题的数据？
在 nifi flow 中，我使用 GrokReader 读取日志文件，然后使用 AvroRecordSetWriter 写入 kafka 主题，您可以再次点击链接了解更多详情link

标签： apache-nifi apache-kafka-connect confluent-platform

【解决方案1】：

当您将 Avro 与 Kafka Connect 一起使用时，Avro 反序列化器预计数据已使用 Avro serialiser 进行序列化。这就是 kafak-avro-console-producer 使用的，这就是为什么你的管道在你使用它时工作的原因。

This article 为 Avro 和 Schema Registry 提供了很好的背景知识。另见Kafka Connect Deep Dive – Converters and Serialization Explained。

我不熟悉 Nifi，但查看文档似乎AvroRecordSetWriter 可以选择使用Confluent Schema Registry。猜测一下，您还需要将Schema Write Strategy 设置为Confluent Schema Registry Reference。

一旦您可以使用 kafka-avro-console-consumer 使用您的主题中的数据，那么您就知道它已正确序列化并且可以与您的 Kafka Connect 接收器一起使用。

【讨论】：

实际上，写入 Kafka 并不强制使用模式注册表，所以我建议他实际上使用 Embed Avro Schema 选项，这不需要他提供任何模式。
这不是强制性的，除非您想使用带有默认 Avro 反序列化器的 Kafka Connect 反序列化 Avro 数据。所以是的，您可以将任何您想要的内容放到 Kafka 主题上，但是如果您希望另一端的消费者可以读取它，您必须使用相同的序列化器/反序列化器。在这种情况下，Kafka Connect 被配置为使用 Schema Registry Avro 反序列化器，因此要处理的消息需要使用 Schema Registry Avro 序列化器放置在 Kafka 主题上。
感谢您的回复。刚才又试了，虽然数据是通过Nifi传到Kafka topic的，但是数据可以用kafka-avro-console-consumer消费！但是日志里还是有那个异常java.lang.NullPointerException，我的influxDB也拿不到数据。
是的，我已经将Schema Write Strategy 设置为Confluent Schema Registry Reference。
所以如果你写一个新的主题，使用新的配置，你可以使用kafka-avro-console-consumer消费来自该主题的每条消息吗？并且连接器仍然失败并出现同样的错误？如果是这样，反序列化后该主题的数据是什么样的？

【解决方案2】：

我找到了原因。这是因为在Nifi中，我使用PublishKafka_0_10将数据发布到Kafka主题，但它的版本太低了！

当我在 ksql 中进行查询时，它会说

Input record ConsumerRecord(..data..) has invalid (negative) timestamp.
Possibly because a pre-0.10 producer client was used to write this record to Kafka without embedding a timestamp, 
or because the input topic was created before upgrading the Kafka cluster to 0.10+. Use a different TimestampExtractor to process this data.

所以，我将其更改为 PublishKafka_1_0 ，然后重新开始，它可以工作！我的 influxDB 可以获取数据。我无语了。

感谢 Robin Moffatt 的回复，这对我很有帮助。

【讨论】：