This is an automated email from the ASF dual-hosted git repository.
wenchen pushed a commit to branch branch-2.4
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-2.4 by this push:
new 705507f [SPARK-27494][SS] Null values don't work in Kafka source v2
705507f is described below
commit 705507facda11060f1a0beb04d1dd19bda5fc4f3
Author: uncleGen <husty...@gmail.com>
AuthorDate: Fri Apr 26 14:25:31 2019 +0800
[SPARK-27494][SS] Null values don't work in Kafka source v2
## What changes were proposed in this pull request?
Right now Kafka source v2 doesn't support null values. The issue is in
org.apache.spark.sql.kafka010.KafkaRecordToUnsafeRowConverter.toUnsafeRow which
doesn't handle null values.
## How was this patch tested?
add new unit tests
Closes #24441 from uncleGen/SPARK-27494.
Authored-by: uncleGen <husty...@gmail.com>
Signed-off-by: Wenchen Fan <wenc...@databricks.com>
(cherry picked from commit d2656aaecd4a7b5562d8d2065aaa66fdc72d253d)
Signed-off-by: Wenchen Fan <wenc...@databricks.com>
---
.../kafka010/KafkaRecordToUnsafeRowConverter.scala | 7 ++-
.../sql/kafka010/KafkaContinuousSourceSuite.scala | 4 ++
.../sql/kafka010/KafkaMicroBatchSourceSuite.scala | 58 ++++++++++++++++++++++
3 files changed, 68 insertions(+), 1 deletion(-)
diff --git
a/external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaRecordToUnsafeRowConverter.scala
b/external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaRecordToUnsafeRowConverter.scala
index f35a143..306ef10 100644
---
a/external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaRecordToUnsafeRowConverter.scala
+++
b/external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaRecordToUnsafeRowConverter.scala
@@ -30,13 +30,18 @@ private[kafka010] class KafkaRecordToUnsafeRowConverter {
def toUnsafeRow(record: ConsumerRecord[Array[Byte], Array[Byte]]): UnsafeRow
= {
rowWriter.reset()
+ rowWriter.zeroOutNullBytes()
if (record.key == null) {
rowWriter.setNullAt(0)
} else {
rowWriter.write(0, record.key)
}
- rowWriter.write(1, record.value)
+ if (record.value == null) {
+ rowWriter.setNullAt(1)
+ } else {
+ rowWriter.write(1, record.value)
+ }
rowWriter.write(2, UTF8String.fromString(record.topic))
rowWriter.write(3, record.partition)
rowWriter.write(4, record.offset)
diff --git
a/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaContinuousSourceSuite.scala
b/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaContinuousSourceSuite.scala
index a0e5818..649cb72 100644
---
a/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaContinuousSourceSuite.scala
+++
b/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaContinuousSourceSuite.scala
@@ -169,6 +169,10 @@ class KafkaContinuousSourceSuite extends
KafkaSourceSuiteBase with KafkaContinuo
}
}
}
+
+ test("SPARK-27494: read kafka record containing null key/values.") {
+ testNullableKeyValue(ContinuousTrigger(100))
+ }
}
class KafkaContinuousSourceTopicDeletionSuite extends KafkaContinuousTest {
diff --git
a/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchSourceSuite.scala
b/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchSourceSuite.scala
index 34cf335..da92019 100644
---
a/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchSourceSuite.scala
+++
b/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchSourceSuite.scala
@@ -988,6 +988,10 @@ abstract class KafkaMicroBatchSourceSuiteBase extends
KafkaSourceSuiteBase {
q.stop()
}
}
+
+ test("SPARK-27494: read kafka record containing null key/values.") {
+ testNullableKeyValue(Trigger.ProcessingTime(100))
+ }
}
@@ -1461,6 +1465,60 @@ abstract class KafkaSourceSuiteBase extends
KafkaSourceTest {
CheckAnswer(2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17)
)
}
+
+ protected def testNullableKeyValue(trigger: Trigger): Unit = {
+ val table = "kafka_null_key_value_source_test"
+ withTable(table) {
+ val topic = newTopic()
+ testUtils.createTopic(topic)
+ testUtils.withTranscationalProducer { producer =>
+ val df = spark
+ .readStream
+ .format("kafka")
+ .option("kafka.bootstrap.servers", testUtils.brokerAddress)
+ .option("kafka.isolation.level", "read_committed")
+ .option("startingOffsets", "earliest")
+ .option("subscribe", topic)
+ .load()
+ .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
+ .as[(String, String)]
+
+ val q = df
+ .writeStream
+ .format("memory")
+ .queryName(table)
+ .trigger(trigger)
+ .start()
+ try {
+ var idx = 0
+ producer.beginTransaction()
+ val expected1 = Seq.tabulate(5) { _ =>
+ producer.send(new ProducerRecord[String, String](topic, null,
null)).get()
+ (null, null)
+ }.asInstanceOf[Seq[(String, String)]]
+
+ val expected2 = Seq.tabulate(5) { _ =>
+ idx += 1
+ producer.send(new ProducerRecord[String, String](topic,
idx.toString, null)).get()
+ (idx.toString, null)
+ }.asInstanceOf[Seq[(String, String)]]
+
+ val expected3 = Seq.tabulate(5) { _ =>
+ idx += 1
+ producer.send(new ProducerRecord[String, String](topic, null,
idx.toString)).get()
+ (null, idx.toString)
+ }.asInstanceOf[Seq[(String, String)]]
+
+ producer.commitTransaction()
+ eventually(timeout(streamingTimeout)) {
+ checkAnswer(spark.table(table), (expected1 ++ expected2 ++
expected3).toDF())
+ }
+ } finally {
+ q.stop()
+ }
+ }
+ }
+ }
}
object KafkaSourceSuite {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
