fix comments

Author: uncleGen

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

@@ -30,8 +30,7 @@ private[kafka010] class KafkaRecordToUnsafeRowConverter {
 
   def toUnsafeRow(record: ConsumerRecord[Array[Byte], Array[Byte]]): UnsafeRow = {
     rowWriter.reset()
-    rowWriter.unsetNullAt(0)
-    rowWriter.unsetNullAt(1)
+    rowWriter.zeroOutNullBytes()
 
     if (record.key == null) {
       rowWriter.setNullAt(0)

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

@@ -170,52 +170,8 @@ 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()
-        }
-      }
-    }
+  test(s"SPARK-27494: read kafka record containing null key/values in continuous mode") {
+    testNullableKeyValue(ContinuousTrigger(100))
   }
 }
 

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

@@ -1040,6 +1040,10 @@ abstract class KafkaMicroBatchSourceSuiteBase extends KafkaSourceSuiteBase {
       q.stop()
     }
   }
+
+  test(s"SPARK-27494: read kafka record containing null key/values in micro-batch mode") {
+    testNullableKeyValue(Trigger.ProcessingTime(100))
+  }
 }
 
 
@@ -1511,6 +1515,54 @@ 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.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(trigger)
+          .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()
+        }
+      }
+    }
+  }
 }
 
 object KafkaSourceSuite {

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

@@ -111,10 +111,6 @@ 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);