Spark streaming skipped message with zero offset from Kafka 0.9

Author: rsotn-mapr

external/kafka-0-9/src/main/scala/org/apache/spark/streaming/kafka09/CachedKafkaConsumer.scala

@@ -16,17 +16,14 @@
  */
 
 package org.apache.spark.streaming.kafka09
-
 import java.{ util => ju }
 
 import collection.JavaConverters._
-
 import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord, KafkaConsumer}
 import org.apache.kafka.common.{KafkaException, TopicPartition}
 
 import org.apache.spark.internal.Logging
 
-
 /**
  * Consumer of single topicpartition, intended for cached reuse.
  * Underlying consumer is not threadsafe, so neither is this,
@@ -104,14 +101,8 @@ class CachedKafkaConsumer[K, V] private(
     val p = consumer.poll(timeout)
     val r = p.records(topicPartition)
 
-    val preparedRecords = if (r.size > 1) {
-      r.asScala.filter(_.offset() > 0).asJava
-    } else {
-      r
-    }
-
-    logDebug(s"Polled ${p.partitions()}  ${preparedRecords.size}")
-    buffer = preparedRecords.iterator()
+    logDebug(s"Polled ${p.partitions()}  ${r.size}")
+    buffer = r.iterator()
   }
 }
 

external/kafka-0-9/src/main/scala/org/apache/spark/streaming/kafka09/KafkaRDD.scala

@@ -227,16 +227,18 @@ private[spark] class KafkaRDD[K, V](
     //    override def hasNext(): Boolean = requestOffset < part.untilOffset
     override def getNext(): ConsumerRecord[K, V] = {
 
+      val isStreams = consumer.topic.startsWith("/") && consumer.topic.contains(":")
+
       @tailrec
       def skipGapsAndGetNext: ConsumerRecord[K, V] = {
         if (requestOffset < part.untilOffset) {
           val r = consumer.get(requestOffset, pollTimeout)
 
-          requestOffset = if (r.offset() == 0) {part.untilOffset} else {r.offset() + 1}
-
-          if (null == r && r.offset() == 0) {
+          if (isStreams && r.offset() == 0) {
+            requestOffset = requestOffset + 1
             skipGapsAndGetNext
           } else {
+            requestOffset = r.offset() + 1
             r
           }
         } else {