MapR [SPARK-966] Streaming application with the latest offset read 1 …

…message from mapr stream which was produced before application start (apache#924)

external/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/DirectKafkaInputDStream.scala

@@ -59,6 +59,8 @@ private[spark] class DirectKafkaInputDStream[K, V](
     ppc: PerPartitionConfig
   ) extends InputDStream[ConsumerRecord[K, V]](_ssc) with Logging with CanCommitOffsets {
 
+  var isFirstCompute = true
+
   val executorKafkaParams = {
     val ekp = new ju.HashMap[String, Object](consumerStrategy.executorKafkaParams)
     KafkaUtils.fixKafkaParams(ekp)
@@ -267,7 +269,12 @@ private[spark] class DirectKafkaInputDStream[K, V](
   override def compute(validTime: Time): Option[KafkaRDD[K, V]] = {
     val untilOffsets = clamp(latestOffsets())
     val offsetRanges = untilOffsets.map { case (tp, uo) =>
-      val fo = currentOffsets(tp)
+      var fo = currentOffsets(tp)
+      if (isFirstCompute && KafkaUtils.isStreams(untilOffsets) && uo.-(fo) == 1
+        && consumerStrategy.executorKafkaParams.get("auto.offset.reset").toString.equals("latest")) {
+        fo = uo
+        isFirstCompute = false
+      }
       OffsetRange(tp.topic, tp.partition, fo, uo)
     }
     val useConsumerCache = context.conf.get(CONSUMER_CACHE_ENABLED)