Github user HeartSaVioR commented on a diff in the pull request:
https://github.com/apache/spark/pull/22282#discussion_r214629265
--- 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(
--- End diff --
We could remove `.toArray` here and also `.toIterable` in
`headers.toIterable.asJava` unless there's performance difference.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
