Spark structured streaming: Commit source offsets to Kafka on QueryProgress

App.scala

    val groupId = "demo-consumer"
    val bootstrapServers = "localhost:9092"

    val spark = SparkSession
      .builder()
      .master("local[*]")
      .appName("demo")
      //.config("spark.sql.streaming.stateStore.maintenanceInterval", "10s")
      //.config("spark.sql.streaming.metricsEnabled", true)
      .config("spark.sql.streaming.minBatchesToRetain", 2)
      .config("spark.sql.streaming.stateStore.minDeltasForSnapshot", 2)
      .getOrCreate()

    import spark.implicits._

    val props = new Properties()
    props.put("group.id", groupId)
    props.put("bootstrap.servers", bootstrapServers)
    props.put("key.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer")
    props.put("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer")
    props.put("enable.auto.commit", "false")
    val kafkaConsumer = new KafkaConsumer[Array[Byte], Array[Byte]](props)
    val listener = new CommitOffsetsOnProgressQueryListener(kafkaConsumer)
    spark.streams.addListener(listener)

CommitOffsetsOnProgressQueryListener.scala

  case class CommitOffsetsOnProgressQueryListener(kafkaConsumer: KafkaConsumer[_, _]) extends StreamingQueryListener {

    override def onQueryStarted(event: StreamingQueryListener.QueryStartedEvent): Unit = { }

    override def onQueryProgress(event: QueryProgressEvent): Unit = {

      val offsets = event
        .getSourceOffsets()
        .mapValues(new OffsetAndMetadata(_))

      import scala.collection.JavaConversions._
      kafkaConsumer.commitSync(offsets)
    }

    override def onQueryTerminated(event: StreamingQueryListener.QueryTerminatedEvent): Unit = { }
  }

QueryProgressEventExt.scala

 implicit class QueryProgressEventExt(event: QueryProgressEvent) {

    def getSourceOffsets(): Map[TopicPartition, Long] = event.progress
      .sources // assumption: sources are all kafka, all on the same cluster
      .map(_.endOffset)
      .flatMap(JsonUtilsWrapper.jsonToOffsets) // extract all offsets
      .groupBy(_._1) // take the smallest offset per topic partition
      .mapValues(_.sortWith((a, b) => a._2.compareTo(b._2) < 0).head._2)
  }

Hi,
Is it working for you?

Thanks

It once worked for me using spark 2.3 ..

The clue was in making sure the "separate" (different group-id) kafka consumer instance was subscribed to the topic(s).. otherwise the broker would not accept the commits..

Hi , i want this JsonUtilsWrapper code

https://stackoverflow.com/questions/46153105/how-to-get-kafka-offsets-for-structured-query-for-manual-and-reliable-offset-man