[GitHub] [spark] srowen commented on a change in pull request #22282: [SPARK-23539][SS] Add support for Kafka headers in Structured Streaming

[GitBox <gi...@apache.org>]

srowen commented on a change in pull request #22282: [SPARK-23539][SS] Add support for Kafka headers in Structured Streaming
URL: https://github.com/apache/spark/pull/22282#discussion_r310352440
 
 

 ##########
 File path: external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaRelation.scala
 ##########
 @@ -100,17 +98,16 @@ private[kafka010] class KafkaRelation(
     // Create an RDD that reads from Kafka and get the (key, value) pair as byte arrays.
     val executorKafkaParams =
       KafkaSourceProvider.kafkaParamsForExecutors(specifiedKafkaParams, uniqueGroupId)
-    val rdd = new KafkaSourceRDD(
-      sqlContext.sparkContext, executorKafkaParams, offsetRanges,
-      pollTimeoutMs, failOnDataLoss, reuseKafkaConsumer = false).map { cr =>
-      InternalRow(
-        cr.key,
-        cr.value,
-        UTF8String.fromString(cr.topic),
-        cr.partition,
-        cr.offset,
-        DateTimeUtils.fromJavaTimestamp(new java.sql.Timestamp(cr.timestamp)),
-        cr.timestampType.id)
+    val rdd = if (includeHeaders) {
+      new KafkaSourceRDD(
+        sqlContext.sparkContext, executorKafkaParams, offsetRanges,
+        pollTimeoutMs, failOnDataLoss, reuseKafkaConsumer = false)
+        .map(KafkaOffsetReader.toInternalRowWithHeaders(_))
 
 Review comment:
   Instead of duplicating this block, can you have one method that's called with an additional boolean param?

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org