title: Spark structured streaming query always starts with auto.offset.rest=earliest even though auto.offset.reset=latest is set
body: <p>I have a weird issue with trying to read data from Kafka using Spark structured streaming.
  My use case is to be able to read from a topic from the largest/latest offset available.</p>
  
  <p>My read configs:</p>
  
  <pre><code>val data = spark
       .readStream
       .format("kafka")
       .option("kafka.bootstrap.servers", "some xyz server")
       .option("subscribe", "sampletopic")
       .option("auto.offset.reset", "latest")
       .option("startingOffsets", "latest")
       .option("kafkaConsumer.pollTimeoutMs", 20000)
       .option("failOnDataLoss","false")
       .option("maxOffsetsPerTrigger",20000)
       .load()
  </code></pre>
  
  <p>My write configs:</p>
  
  <pre><code>data
      .writeStream
      .outputMode("append")
      .queryName("test") 
      .format("parquet")
      .option("checkpointLocation", "s3://somecheckpointdir")
      .start("s3://outpath").awaitTermination()
  </code></pre>
  
  <p>Versions used:</p>
  
  <ul>
  <li><code>spark-core_2.11 : 2.2.1</code></li>
  <li><code>spark-sql_2.11 : 2.2.1</code></li>
  <li><code>spark-sql-kafka-0-10_2.11 : 2.2.1</code></li>
  </ul>
  
  <p>I have done my research online and from [the Kafka documentation](<a href="https://kafka.apache.org/0100/documentation.html0/" rel="nofollow noreferrer">https://kafka.apache.org/0100/documentation.html0/</a></p>
  
  <p>I am using the new consumer apis and as the documentation suggests i just need to set auto.offset.reset to "latest" or startingOffsets to "latest"  to ensure that my Spark job starts consuming from the the latest offset available per partition in Kafka.</p>
  
  <p>I am also aware that the setting <code>auto.offset.reset</code> only kicks in when a new query is started for the first time and not on a restart of an application in which case it will continue to read from the last saved offset.</p>
  
  <p>I am using s3 for checkpointing my offsets. and I see them being generated under s3://somecheckpointdir.</p>
  
  <p>The issue I am facing is that the Spark job always read from earliest offset even though latest option is specified in the code during startup of application when it is started for the first time and I see this in the Spark logs.
  <code>auto.offset.reset = earliest</code> being used. I have not seen posts related to this particular issue.</p>
  
  <p>I would like to know if I am missing something here and if someone has seen this behavior before. Any help/direction will indeed be useful. Thank you.</p>