title: Kafka uncommitted message not getting consumed again
body: <p>I am processing kafka messages and inserting into kudu table using spark streaming with manual offset commit here is my code.</p>
  
  <pre><code>val topicsSet = topics.split(",").toSet
  val kafkaParams = Map[String, Object](
    ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -&gt; brokers,
    ConsumerConfig.GROUP_ID_CONFIG -&gt; groupId,
    ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -&gt; classOf[StringDeserializer],
    ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -&gt; classOf[StringDeserializer],
    ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG -&gt; (false: java.lang.Boolean),
    ConsumerConfig.AUTO_OFFSET_RESET_CONFIG -&gt; "earliest" //"latest" //"earliest"
   )
  val stream = KafkaUtils.createDirectStream[String, String](
                          ssc,
                          PreferConsistent,
                          Subscribe[String, String](topicsSet, kafkaParams)
                         )
  stream.foreachRDD { rdd =&gt;
  var offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
  //offsetRanges.foreach(println)
  var msgOffsetsRdd = rdd.map(msg =&gt;{
  val msgOffset = OffsetRange(msg.topic(), msg.partition(),  msg.offset(), msg.offset()+1)
          println(msg)
          msgOffset 
        }
      )
     val msgOffsets = msgOffsetsRdd.collect() //here idea was to get only processed messages offsets for commit
     stream.asInstanceOf[CanCommitOffsets].commitAsync(msgOffsets)
  }
  </code></pre>
  
  <p>Let us table this example  While inserting data into kudu  I got the error I need to process those messages again, if I stop the job and start it again I am able to get uncommitted message can't we get all uncommitted messages in the streaming?</p>