将消息写入代理后，Spark kafka 连接器抛出空指针异常答案

【问题标题】：Spark kafka connector throwing nullpointerexception after writing the messages into the broker将消息写入代理后，Spark kafka 连接器抛出空指针异常
【发布时间】：2021-08-18 10:43:00
【问题描述】：

我正在使用 spark kafka 驱动程序将数据集写入 kafka 代理。下面是示例代码：

dataset
   .write()
   .format("kafka")
   .option("kafka.bootstrap.servers", kafkaProducer.getKafkaBrokerUrl())
   .option("topic", kafkaProducer.getTopic())
   .option(KEY_SERIALIZER, kafkaProducer.getKafkaConfig().getKeySerializer())
   .option(VALUE_SERIALIZER, kafkaProducer.getKafkaConfig().getValueSerializer())
   .save();

虽然这是将数据集的内容写入代理，但对于数据集中的每条记录，我都会收到以下错误。

ERROR ProducerBatch:207: Error executing user-provided callback on message for topic-partition 'Kafka-topic-0'
java.lang.NullPointerException
    at org.apache.kafka.clients.producer.KafkaProducer$InterceptorCallback.onCompletion(KafkaProducer.java:1422)
    at org.apache.kafka.clients.producer.internals.ProducerBatch.completeFutureAndFireCallbacks(ProducerBatch.java:201)
    at org.apache.kafka.clients.producer.internals.ProducerBatch.done(ProducerBatch.java:187)
    at org.apache.kafka.clients.producer.internals.Sender.completeBatch(Sender.java:591)
    at org.apache.kafka.clients.producer.internals.Sender.completeBatch(Sender.java:567)
    at org.apache.kafka.clients.producer.internals.Sender.handleProduceResponse(Sender.java:481)
    at org.apache.kafka.clients.producer.internals.Sender.access$100(Sender.java:74)
    at org.apache.kafka.clients.producer.internals.Sender$1.onComplete(Sender.java:692)
    at org.apache.kafka.clients.ClientResponse.onComplete(ClientResponse.java:101)
    at org.apache.kafka.clients.NetworkClient.completeResponses(NetworkClient.java:482)
    at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:474)
    at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:239)
    at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:163)
    at java.lang.Thread.run(Thread.java:748)

我查看了spark structured streaming documentation 并没有提及回调。我正在使用以下库依赖项：

<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-sql-kafka-0-10_2.12</artifactId>
  <version>${spark.kafka.client.version}</version>
</dependency>

谁能帮我解决这个问题？

【问题讨论】：

kafka/spark 集群是否在同一台主机上？
@Chandan：它在不同的集群中。我能够将数据写入此主机。但是在写入数据后，我得到了这个错误。

标签： java apache-spark apache-kafka spark-structured-streaming

【解决方案1】：

我看不出代码有什么问题。我唯一的怀疑是使用的依赖项（不确定该特定版本中是否存在错误）。这是我没有遇到任何问题的确切版本。请用这些版本测试同样的东西，看看是否有帮助。

如果这没有帮助，您可以使用完整的 pom.xml 更新问题，以便我们进一步调查

    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_2.11</artifactId>
        <version>2.4.2</version>
    </dependency>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-sql_2.11</artifactId>
        <version>2.4.2</version>
    </dependency>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-sql-kafka-0-10_2.11</artifactId>
        <version>2.4.2</version>
    </dependency>
    <dependency>
        <groupId>org.scala-lang</groupId>
        <artifactId>scala-library</artifactId>
        <version>2.11.8</version>
    </dependency>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-streaming_2.11</artifactId>
        <version>2.4.2</version>
    </dependency>

【讨论】：

让我试试这个。