title: Spark Streaming from Kafka topic throws offset out of range with no option to restart the stream
body: <p>I have a streaming job running on Spark 2.1.1, polling Kafka 0.10.  I am using the Spark <a href="https://github.com/apache/spark/blob/master/external/kafka-0-10/src/main/scala/org/apache/spark/streaming/kafka010/KafkaUtils.scala" rel="nofollow noreferrer">KafkaUtils</a> class to create a DStream, and everything is working fine until I have data that ages out of the topic because of the retention policy.  My problem comes when I stop my job to make some changes if any data has aged out of the topic I get an error saying that my offsets are out of range.  I have done a lot of research including looking at the spark source code, and I see lots of comments like the comments in this issue:  <a href="https://issues.apache.org/jira/browse/SPARK-19680" rel="nofollow noreferrer">SPARK-19680</a> - basically saying that data should not be lost silently - so auto.offset.reset is ignored by spark.  My big question, though, is what can I do now?  My topic will not poll in spark - it dies on startup with the offsets exception.  I don't know how to reset the offsets so my job will just get started again.  I have not enabled checkpoints since I read that those are unreliable for this use.  I used to have a lot of code to manage offsets, but it appears that spark ignores requested offsets if there are any committed, so I am currently managing offsets like this:</p>
  
  <pre><code>val stream = KafkaUtils.createDirectStream[String, T](
      ssc,
      PreferConsistent,
      Subscribe[String, T](topics, kafkaParams))
  
  stream.foreachRDD { (rdd, batchTime) =&gt;
      val offsets = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
      Log.debug("processing new batch...")
  
      val values = rdd.map(x =&gt; x.value())
      val incomingFrame: Dataset[T] = SparkUtils.sparkSession.createDataset(values)(consumer.encoder()).persist
  
      consumer.processDataset(incomingFrame, batchTime)
      stream.asInstanceOf[CanCommitOffsets].commitAsync(offsets)
  }
  ssc.start()
  ssc.awaitTermination()
  </code></pre>
  
  <p>As a workaround I have been changing my group ids but that is really lame.  I know this is expected behavior and should not happen, I just need to know how to get the stream running again.  Any help would be appreciated.</p>