Github user HeartSaVioR commented on a diff in the pull request:
https://github.com/apache/spark/pull/22282#discussion_r215092933
--- Diff:
external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaWriteTask.scala
---
@@ -88,7 +92,30 @@ private[kafka010] abstract class KafkaRowWriter(
throw new NullPointerException(s"null topic present in the data. Use
the " +
s"${KafkaSourceProvider.TOPIC_OPTION_KEY} option for setting a
default topic.")
}
- val record = new ProducerRecord[Array[Byte],
Array[Byte]](topic.toString, key, value)
+ val record = if (projectedRow.isNullAt(3)) {
+ new ProducerRecord[Array[Byte], Array[Byte]](
+ topic.toString,
+ null,
+ key,
+ value
+ )
+ } else {
+ val headerMap = projectedRow.getMap(3)
+ val headers = (0 until headerMap.numElements()).toArray.map(
+ i =>
+ new RecordHeader(
--- End diff --
Yeah then I also think it is missing spot for Kafka. Just asked it to Kafka
dev mailing list.
https://lists.apache.org/thread.html/2ec3e7e2345e64ac559d98aaa28e0980f07a9778db447168e19d41d2@%3Cdev.kafka.apache.org%3E
If Kafka community says it's missing spot, either of us can go ahead fixing
that. You can take it forward if you're happy to do it.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
