title: [IllegalStateException]: Spark Structured Streaming is termination Streaming Query with Error
body: <p>I have created a Spark Structured Streaming application. In the app, I am pulling data from Kafka topics. For recovery purpose, I am using checkpointing.</p>
  
  <p>The challenge I am facing is as follows:</p>
  
  <pre><code>ERROR StreamExecution: Query [id = cf9e0f0a-653a-4246-a767-079b38f03b2f, runId = c8b52e1f-a51a-42c2-9934-6927ba79e44c] terminated with error
  java.lang.IllegalStateException: Set(mytopic-1) are gone. Some data may have been missed.
  Some data may have been lost because they are not available in Kafka any more; either the
   data was aged out by Kafka or the topic may have been deleted before all the data in the
   topic was processed. If you don't want your streaming query to fail on such cases, set the
   source option "failOnDataLoss" to "false".
  
  at org.apache.spark.sql.kafka010.KafkaSource.org$apache$spark$sql$kafka010$KafkaSource$$reportDataLoss(KafkaSource.scala:329)
      at org.apache.spark.sql.kafka010.KafkaSource.getBatch(KafkaSource.scala:250)
      at org.apache.spark.sql.execution.streaming.StreamExecution$$anonfun$org$apache$spark$sql$execution$streaming$StreamExecution$$runBatch$2$$anonfun$apply$7.apply(StreamExecution.scala:607)
      at org.apache.spark.sql.execution.streaming.StreamExecution$$anonfun$org$apache$spark$sql$execution$streaming$StreamExecution$$runBatch$2$$anonfun$apply$7.apply(StreamExecution.scala:603)
      at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)
      at scala.collection.TraversableLike$$anonfun$flatMap$1.apply(TraversableLike.scala:241)
      at scala.collection.Iterator$class.foreach(Iterator.scala:893)
      at scala.collection.AbstractIterator.foreach(Iterator.scala:1336)
      at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
      at org.apache.spark.sql.execution.streaming.StreamProgress.foreach(StreamProgress.scala:25)
      at scala.collection.TraversableLike$class.flatMap(TraversableLike.scala:241)
      at org.apache.spark.sql.execution.streaming.StreamProgress.flatMap(StreamProgress.scala:25)
      at org.apache.spark.sql.execution.streaming.StreamExecution$$anonfun$org$apache$spark$sql$execution$streaming$StreamExecution$$runBatch$2.apply(StreamExecution.scala:603)
      at org.apache.spark.sql.execution.streaming.StreamExecution$$anonfun$org$apache$spark$sql$execution$streaming$StreamExecution$$runBatch$2.apply(StreamExecution.scala:603)
      at org.apache.spark.sql.execution.streaming.ProgressReporter$class.reportTimeTaken(ProgressReporter.scala:279)
      at org.apache.spark.sql.execution.streaming.StreamExecution.reportTimeTaken(StreamExecution.scala:58)
      at org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runBatch(StreamExecution.scala:602)
      at org.apache.spark.sql.execution.streaming.StreamExecution$$anonfun$org$apache$spark$sql$execution$streaming$StreamExecution$$runBatches$1$$anonfun$apply$mcZ$sp$1.apply$mcV$sp(StreamExecution.scala:306)
      at org.apache.spark.sql.execution.streaming.StreamExecution$$anonfun$org$apache$spark$sql$execution$streaming$StreamExecution$$runBatches$1$$anonfun$apply$mcZ$sp$1.apply(StreamExecution.scala:294)
      at org.apache.spark.sql.execution.streaming.StreamExecution$$anonfun$org$apache$spark$sql$execution$streaming$StreamExecution$$runBatches$1$$anonfun$apply$mcZ$sp$1.apply(StreamExecution.scala:294)
      at org.apache.spark.sql.execution.streaming.ProgressReporter$class.reportTimeTaken(ProgressReporter.scala:279)
      at org.apache.spark.sql.execution.streaming.StreamExecution.reportTimeTaken(StreamExecution.scala:58)
      at org.apache.spark.sql.execution.streaming.StreamExecution$$anonfun$org$apache$spark$sql$execution$streaming$StreamExecution$$runBatches$1.apply$mcZ$sp(StreamExecution.scala:294)
      at org.apache.spark.sql.execution.streaming.OneTimeExecutor.execute(TriggerExecutor.scala:40)
      at org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark$sql$execution$streaming$StreamExecution$$runBatches(StreamExecution.scala:290)
      at org.apache.spark.sql.execution.streaming.StreamExecution$$anon$1.run(StreamExecution.scala:206)
  </code></pre>
  
  <p>After starting the application, above error occurs. The issue that I am not able to understand is that I am not running any streaming query for topic <code>myTopic</code>. However, other streaming queries keeps on working fine.</p>
  
  <p>Before running the query for <code>myTopic2</code>, I was indeed running the same query for <code>myTopic</code>. So, only the metadata logs for <code>myTopic</code> were created. Might be that can be a reason for the bug. But, other than that I cannot see any other reason for this exception.</p>
  
  <p>Can anyone help me to understand this exception?</p>
  
  <p>I am using Spark (v2.2.0) and Kafka (v0.10.2.1).</p>
  
  <p><strong>Code</strong>:</p>
  
  <pre><code>val df = spark
    .readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "localhost:9092")
    .option("subscribe", "myTopic2")
    .load()
  
  val query = df
    .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
    .writeStream
    .format("parquet")
    .option("checkpointLocation", "/path/to/hdfs/checkpoint")
    .option("path", "/path/to/hdfs/data")
    .start()
  </code></pre>