Github user HeartSaVioR commented on a diff in the pull request:
https://github.com/apache/spark/pull/22282#discussion_r214632570
--- 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 --
Looks like RecordHeader is in `org.apache.kafka.common.header.internals`
package. Is there any alternative public methods/classes to create Header? Or
it is just a missing spot in Kafka?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
