[SPARK-9780] [STREAMING] [KAFKA] prevent NPE if KafkaRDD instantiation …

…fails Author: cody koeninger <[email protected]> Closes #8133 from koeninger/SPARK-9780 and squashes the following commits: 406259d [cody koeninger] [SPARK-9780][Streaming][Kafka] prevent NPE if KafkaRDD instantiation fails
kiszk · Aug 13, 2015 · 8ce6096 · 8ce6096
1 parent 660e6dc
commit 8ce6096
Showing 1 changed file with 5 additions and 1 deletion.
diff --git a/external/kafka/src/main/scala/org/apache/spark/streaming/kafka/KafkaRDD.scala b/external/kafka/src/main/scala/org/apache/spark/streaming/kafka/KafkaRDD.scala
@@ -197,7 +197,11 @@ class KafkaRDD[
         .dropWhile(_.offset < requestOffset)
     }
 
-    override def close(): Unit = consumer.close()
+    override def close(): Unit = {
+      if (consumer != null) {
+        consumer.close()
+      }
+    }
 
     override def getNext(): R = {
       if (iter == null || !iter.hasNext) {