title: Right way to read stream from Kafka topic using checkpointLocation offsets
body: <p>I'm trying to develop a small Spark app (using Scala) to read messages from Kafka (Confluent) and write them (insert) into Hive table. Everything works as expected, except for one important feature - managing offsets when the application is restarted (submited). It confuses me.</p>
  <p><strong>Cut from my code</strong>:</p>
  <pre><code>  def main(args: Array[String]): Unit = {
  
      val sparkSess = SparkSession
        .builder
        .appName(&quot;Kafka_to_Hive&quot;)
        .config(&quot;spark.sql.warehouse.dir&quot;, &quot;/user/hive/warehouse/&quot;)
        .config(&quot;hive.metastore.uris&quot;, &quot;thrift://localhost:9083&quot;)
        .config(&quot;hive.exec.dynamic.partition&quot;, &quot;true&quot;)
        .config(&quot;hive.exec.dynamic.partition.mode&quot;, &quot;nonstrict&quot;)
        .enableHiveSupport()
        .getOrCreate()
  
      sparkSess.sparkContext.setLogLevel(&quot;ERROR&quot;)
  
      // don't consider this code block please, it's just a part of Confluent avro message deserializing adventures
      sparkSess.udf.register(&quot;deserialize&quot;, (bytes: Array[Byte]) =&gt;
        DeserializerWrapper.deserializer.deserialize(bytes)
      )
      
  
      val kafkaDataFrame = sparkSess
        .readStream
        .format(&quot;kafka&quot;)
        .option(&quot;kafka.bootstrap.servers&quot;, 'localhost:9092')
        .option(&quot;group.id&quot;, 'kafka-to-hive-1')
        // ------&gt;   which Kafka options do I need to set here for starting from last right offset to ensure completenes of data and &quot;exactly once&quot; writing?   &lt;--------
        .option(&quot;failOnDataLoss&quot;, (false: java.lang.Boolean))
        .option(&quot;subscribe&quot;, 'some_topic')
        .load()
  
      import org.apache.spark.sql.functions._
      
      // don't consider this code block please, it's just a part of Confluent avro message deserializing adventures
      val valueDataFrame = kafkaDataFrame.selectExpr(&quot;&quot;&quot;deserialize(value) AS message&quot;&quot;&quot;)
      val df = valueDataFrame.select(
        from_json(col(&quot;message&quot;), sparkSchema.dataType).alias(&quot;parsed_value&quot;))
        .select(&quot;parsed_value.*&quot;)
  
  
      df.writeStream
        .foreachBatch((batchDataFrame, batchId) =&gt; {
          batchDataFrame.createOrReplaceTempView(&quot;`some_view_name`&quot;)
          val sqlText = &quot;SELECT * FROM `some_view_name` a where some_field='some value'&quot;
          val batchDataFrame_view = batchDataFrame.sparkSession.sql(sqlText);
          batchDataFrame_view.write.insertInto(&quot;default.some_hive_table&quot;)
        })
        .option(&quot;checkpointLocation&quot;, &quot;/user/some_user/tmp/checkpointLocation&quot;)
        .start()
        .awaitTermination()
    }
  </code></pre>
  <p><strong>Questions</strong> (the questions are related to each other):</p>
  <ol>
  <li>Which Kafka options do I need to apply on <code>readStream.format(&quot;kafka&quot;)</code> for starting from last right offset on every submit of spark app?</li>
  <li>Do I need to manually read 3rd line of checkpointLocation/offsets/latest_batch file to find last offsets to read from Kafka? I mean something like that: <code>readStream.format(&quot;kafka&quot;).option(&quot;startingOffsets&quot;, &quot;&quot;&quot;{&quot;some_topic&quot;:{&quot;2&quot;:35079,&quot;5&quot;:34854,&quot;4&quot;:35537,&quot;1&quot;:35357,&quot;3&quot;:35436,&quot;0&quot;:35213}}&quot;&quot;&quot;)</code></li>
  <li>What is the right/convenient way to read stream from Kafka (Confluent) topic? (I'm not considering offsets storing engine of Kafka)</li>
  </ol>