[SPARK-27494][SS] Null keys/values don't work in Kafka source v2

external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaRecordToUnsafeRowConverter.scala

@@ -30,13 +30,19 @@ private[kafka010] class KafkaRecordToUnsafeRowConverter {
 
   def toUnsafeRow(record: ConsumerRecord[Array[Byte], Array[Byte]]): UnsafeRow = {
     rowWriter.reset()
+    rowWriter.unsetNullAt(0)
+    rowWriter.unsetNullAt(1)
 
     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)

external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaContinuousSourceSuite.scala

@@ -169,6 +169,54 @@ class KafkaContinuousSourceSuite extends KafkaSourceSuiteBase with KafkaContinuo
       }
     }
   }
+
+  test("read kafka record containing null key/values") {
+    val table = "kafka_null_key_value_source_test"
+     withTable(table) {
+      val topic = newTopic()
+      testUtils.createTopic(topic)
+      testUtils.withProducer { producer =>
+        val df = spark
+          .readStream
+          .format("kafka")
+          .option("kafka.bootstrap.servers", testUtils.brokerAddress)
+          .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(ContinuousTrigger(100))
+          .start()
+        try {
+          val expected1 = (1 to 5).map { _ =>
+            producer.send(new ProducerRecord[String, String](topic, null, null)).get()
+            (null, null)
+          }.asInstanceOf[Seq[(String, String)]]
+
+          val expected2 = (6 to 10).map { i =>
+            producer.send(new ProducerRecord[String, String](topic, i.toString, null)).get()
+            (i.toString, null)
+          }.asInstanceOf[Seq[(String, String)]]
+
+          val expected3 = (11 to 15).map { i =>
+            producer.send(new ProducerRecord[String, String](topic, null, i.toString)).get()
+            (null, i.toString)
+          }.asInstanceOf[Seq[(String, String)]]
+
+          eventually(timeout(streamingTimeout)) {
+            checkAnswer(spark.table(table), (expected1 ++ expected2 ++ expected3).toDF())
+          }
+        } finally {
+          q.stop()
+        }
+      }
+    }
+  }
 }
 
 class KafkaContinuousSourceTopicDeletionSuite extends KafkaContinuousTest {

external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala

@@ -350,6 +350,17 @@ class KafkaTestUtils(withBrokerProps: Map[String, Object] = Map.empty) extends L
     props
   }
 
+  /** Call `f` with a `KafkaProducer`. */
+  def withProducer(f: KafkaProducer[String, String] => Unit): Unit = {
+    val props = producerConfiguration
+    val producer = new KafkaProducer[String, String](props)
+    try {
+      f(producer)
+    } finally {
+      producer.close()
+    }
+  }
+
   /** Call `f` with a `KafkaProducer` that has initialized transactions. */
   def withTranscationalProducer(f: KafkaProducer[String, String] => Unit): Unit = {
     val props = producerConfiguration

sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/codegen/UnsafeRowWriter.java

@@ -111,6 +111,10 @@ public void setNullAt(int ordinal) {
     write(ordinal, 0L);
   }
 
+  public void unsetNullAt(int ordinal) {
+    BitSetMethods.unset(getBuffer(), startingOffset, ordinal);
+  }
+
   @Override
   public void setNull1Bytes(int ordinal) {
     setNullAt(ordinal);