title: Strimzi Kafka + Structured Streaming - Offsets not available (is this because of segment.bytes)?
body: <p>i've a long-running Apache Spark structured streaming job running in GCP Dataproc, which reads data from Kafka every 10 mins, and does some processing.
  Kafka topic has 3 partitions, and a retention period of 3 days.</p>
  <p>The issue i'm facing is that after few hours, the program stops reading data from Kafka.
  If i delete the gcp bucket (which is the checkpoint directory), and then restart the streaming job - it starts consuming the data again.</p>
  <p>here is the code, where I'm reading data from Kafka &amp; using foreachbatch to call a function where the processing happens</p>
  <pre><code>
  df_stream = spark.readStream.format('kafka') \
          .option(&quot;kafka.security.protocol&quot;, &quot;SSL&quot;) \
          .option(&quot;kafka.ssl.truststore.location&quot;, ssl_truststore_location) \
          .option(&quot;kafka.ssl.truststore.password&quot;, ssl_truststore_password) \
          .option(&quot;kafka.ssl.keystore.location&quot;, ssl_keystore_location) \
          .option(&quot;kafka.ssl.keystore.password&quot;, ssl_keystore_password) \
          .option(&quot;kafka.bootstrap.servers&quot;, kafkaBrokers) \
          .option(&quot;subscribe&quot;, topic) \
          .option(&quot;startingOffsets&quot;, &quot;latest&quot;) \
          .option(&quot;failOnDataLoss&quot;, &quot;false&quot;) \
          .option(&quot;kafka.metadata.max.age.ms&quot;, &quot;1000&quot;) \
          .option(&quot;kafka.ssl.keystore.type&quot;, &quot;PKCS12&quot;) \
          .option(&quot;kafka.ssl.truststore.type&quot;, &quot;PKCS12&quot;) \
          .option(&quot;maxOffsetsPerTrigger&quot;, 100000) \
          .option(&quot;max.poll.records&quot;, 500) \
          .option(&quot;max.poll.interval.ms&quot;, 1000000) \
          .load()
  
  
  
  query = df_stream.selectExpr(&quot;CAST(value AS STRING)&quot;, &quot;timestamp&quot;, &quot;topic&quot;).writeStream \
          .outputMode(&quot;append&quot;) \
          .trigger(processingTime='3 minutes') \
          .option(&quot;truncate&quot;, &quot;false&quot;) \
          .option(&quot;checkpointLocation&quot;, checkpoint) \
          .foreachBatch(convertToDictForEachBatch) \
          .start()
  
  </code></pre>
  <p>Here is snippet of the log, where the</p>
  <pre><code> total time take, convertToDict :  0:00:00.006695
  22/08/26 17:45:03 INFO org.apache.kafka.clients.consumer.internals.AbstractCoordinator: [Consumer clientId=consumer-spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0-1, groupId=spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0] Member consumer-spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0-1-26755c4c-93d6-4ab6-8799-411439e310bc sending LeaveGroup request to coordinator 35.185.24.226:9094 (id: 2147483646 rack: null) due to consumer poll timeout has expired. This means the time between subsequent calls to poll() was longer than the configured max.poll.interval.ms, which typically implies that the poll loop is spending too much time processing messages. You can address this either by increasing max.poll.interval.ms or by reducing the maximum size of batches returned in poll() with max.poll.records.
  22/08/26 17:50:00 INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator: [Consumer clientId=consumer-spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0-1, groupId=spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0] Giving away all assigned partitions as lost since generation has been reset,indicating that consumer is no longer part of the group
  22/08/26 17:50:00 INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator: [Consumer clientId=consumer-spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0-1, groupId=spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0] Lost previously assigned partitions syslog.ueba-us4.v1.versa.demo3-0, syslog.ueba-us4.v1.versa.demo3-1, syslog.ueba-us4.v1.versa.demo3-2
  22/08/26 17:50:00 INFO org.apache.kafka.clients.consumer.internals.AbstractCoordinator: [Consumer clientId=consumer-spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0-1, groupId=spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0] (Re-)joining group
  22/08/26 17:50:00 INFO org.apache.kafka.clients.consumer.internals.AbstractCoordinator: [Consumer clientId=consumer-spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0-1, groupId=spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0] Join group failed with org.apache.kafka.common.errors.MemberIdRequiredException: The group member needs to have a valid member id before actually entering a consumer group.
  22/08/26 17:50:00 INFO org.apache.kafka.clients.consumer.internals.AbstractCoordinator: [Consumer clientId=consumer-spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0-1, groupId=spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0] (Re-)joining group
  22/08/26 17:50:03 INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator: [Consumer clientId=consumer-spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0-1, groupId=spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0] Finished assignment for group at generation 1: {consumer-spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0-1-307570e0-ca20-42d5-b4d1-255da4fca485=Assignment(partitions=[syslog.ueba-us4.v1.versa.demo3-0, syslog.ueba-us4.v1.versa.demo3-1, syslog.ueba-us4.v1.versa.demo3-2])}
  22/08/26 17:50:03 INFO org.apache.kafka.clients.consumer.internals.AbstractCoordinator: [Consumer clientId=consumer-spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0-1, groupId=spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0] Successfully joined group with generation 1
  22/08/26 17:50:03 INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator: [Consumer clientId=consumer-spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0-1, groupId=spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0] Notifying assignor about the new Assignment(partitions=[syslog.ueba-us4.v1.versa.demo3-0, syslog.ueba-us4.v1.versa.demo3-1, syslog.ueba-us4.v1.versa.demo3-2])
  22/08/26 17:50:03 INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator: [Consumer clientId=consumer-spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0-1, groupId=spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0] Adding newly assigned partitions: syslog.ueba-us4.v1.versa.demo3-0, syslog.ueba-us4.v1.versa.demo3-1, syslog.ueba-us4.v1.versa.demo3-2
  22/08/26 17:50:03 INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator: [Consumer clientId=consumer-spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0-1, groupId=spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0] Found no committed offset for partition syslog.ueba-us4.v1.versa.demo3-0
  22/08/26 17:50:03 INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator: [Consumer clientId=consumer-spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0-1, groupId=spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0] Found no committed offset for partition syslog.ueba-us4.v1.versa.demo3-1
  22/08/26 17:50:03 INFO org.apache.kafka.clients.consumer.internals.ConsumerCoordinator: [Consumer clientId=consumer-spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0-1, groupId=spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0] Found no committed offset for partition syslog.ueba-us4.v1.versa.demo3-2
  22/08/26 17:50:03 INFO org.apache.kafka.clients.consumer.internals.SubscriptionState: [Consumer clientId=consumer-spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0-1, groupId=spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0] Resetting offset for partition syslog.ueba-us4.v1.versa.demo3-1 to offset 247969068.
  22/08/26 17:50:03 INFO org.apache.kafka.clients.consumer.internals.SubscriptionState: [Consumer clientId=consumer-spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0-1, groupId=spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0] Resetting offset for partition syslog.ueba-us4.v1.versa.demo3-2 to offset 246383018.
  22/08/26 17:50:03 INFO org.apache.kafka.clients.consumer.internals.SubscriptionState: [Consumer clientId=consumer-spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0-1, groupId=spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0] Resetting offset for partition syslog.ueba-us4.v1.versa.demo3-0 to offset 248913006.
  22/08/26 17:50:03 INFO org.apache.kafka.clients.consumer.internals.SubscriptionState: [Consumer clientId=consumer-spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0-1, groupId=spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0] Seeking to LATEST offset of partition syslog.ueba-us4.v1.versa.demo3-0
  22/08/26 17:50:03 INFO org.apache.kafka.clients.consumer.internals.SubscriptionState: [Consumer clientId=consumer-spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0-1, groupId=spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0] Seeking to LATEST offset of partition syslog.ueba-us4.v1.versa.demo3-1
  22/08/26 17:50:03 INFO org.apache.kafka.clients.consumer.internals.SubscriptionState: [Consumer clientId=consumer-spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0-1, groupId=spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0] Seeking to LATEST offset of partition syslog.ueba-us4.v1.versa.demo3-2
  22/08/26 17:50:03 INFO org.apache.kafka.clients.consumer.internals.SubscriptionState: [Consumer clientId=consumer-spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0-1, groupId=spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0] Resetting offset for partition syslog.ueba-us4.v1.versa.demo3-2 to offset 248534038.
  22/08/26 17:50:03 INFO org.apache.kafka.clients.consumer.internals.SubscriptionState: [Consumer clientId=consumer-spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0-1, groupId=spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0] Resetting offset for partition syslog.ueba-us4.v1.versa.demo3-1 to offset 248185455.
  22/08/26 17:50:03 INFO org.apache.kafka.clients.consumer.internals.SubscriptionState: [Consumer clientId=consumer-spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0-1, groupId=spark-kafka-source-a1532573-fad4-4127-ac20-ab9878913643-594190416-driver-0] Resetting offset for partition syslog.ueba-us4.v1.versa.demo3-0 to offset 248990456.
   IN CONVERT TO DICT  84  currentTime  2022-08-26 17:50:03.995993  df -&gt;  DataFrame[value: string, timestamp: timestamp, topic: string]
   before adding topic, count in batch  0
  +-----+---------+-----+
  |value|timestamp|topic|
  +-----+---------+-----+
  +-----+---------+-----+
  
  </code></pre>
  <p>Here is the oldest &amp; latest offset in the Kafka topic (from Prometheus)</p>
  <pre><code>Current Offset (metric : kafka_topic_partition_current_offset)
  partition 0 : 249185343
  partition 1 : 248380971
  partition 2 : 248728475
  
  Oldest Offset (metric : kafka_topic_partition_oldest_offset)
  partition 0 : 248913006
  partition 1 : 247969068
  partition 2 : 248541752
  
  </code></pre>
  <p>What i see in the checkpoint bucket:</p>
  <pre><code>(base) Karans-MacBook-Pro:prometheus-yamls karanalang$ gsutil cat gs://ss-checkpoint-10m-noconsumergrp/offsets/96 
  v1
  {&quot;batchWatermarkMs&quot;:0,&quot;batchTimestampMs&quot;:1661538780007,&quot;conf&quot;:{&quot;spark.sql.streaming.stateStore.providerClass&quot;:&quot;org.apache.spark.sql.execution.streaming.state.HDFSBackedStateStoreProvider&quot;,&quot;spark.sql.streaming.join.stateFormatVersion&quot;:&quot;2&quot;,&quot;spark.sql.streaming.stateStore.compression.codec&quot;:&quot;lz4&quot;,&quot;spark.sql.streaming.flatMapGroupsWithState.stateFormatVersion&quot;:&quot;2&quot;,&quot;spark.sql.streaming.multipleWatermarkPolicy&quot;:&quot;min&quot;,&quot;spark.sql.streaming.aggregation.stateFormatVersion&quot;:&quot;2&quot;,&quot;spark.sql.shuffle.partitions&quot;:&quot;200&quot;}}
  {&quot;syslog.ueba-us4.v1.versa.demo3&quot;:{&quot;2&quot;:246124174,&quot;1&quot;:245765547,&quot;0&quot;:246582707}}
  
  </code></pre>
  <p>The offsets in the checkpoint bucket are lower than the oldest offset in the topic, is that the reason why the data is not being read ?
  What needs to be done to resolve this issue ?</p>
  <p>tia!</p>
  <p>Also, here is the yaml used to create the Strimzi Kafka Topic, this</p>
  <pre><code>apiVersion: kafka.strimzi.io/v1beta2
  kind: KafkaTopic
  metadata:
    annotations:
      kubectl.kubernetes.io/last-applied-configuration: |
        {&quot;apiVersion&quot;:&quot;kafka.strimzi.io/v1beta2&quot;,&quot;kind&quot;:&quot;KafkaTopic&quot;,&quot;metadata&quot;:{&quot;annotations&quot;:{},&quot;labels&quot;:{&quot;strimzi.io/cluster&quot;:&quot;versa-kafka-gke&quot;},&quot;name&quot;:&quot;syslog.ueba-us4.v1.versa.demo3&quot;,&quot;namespace&quot;:&quot;kafka&quot;},&quot;spec&quot;:{&quot;config&quot;:{&quot;retention.ms&quot;:259200,&quot;segment.bytes&quot;:1073741824},&quot;partitions&quot;:3,&quot;replicas&quot;:3}}
    creationTimestamp: &quot;2022-07-24T18:11:32Z&quot;
    generation: 3
    labels:
      strimzi.io/cluster: versa-kafka-gke
    name: syslog.ueba-us4.v1.versa.demo3
    namespace: kafka
    resourceVersion: &quot;140557580&quot;
    uid: 26c32f6f-5e4f-48ae-8026-007ab59624ec
  spec:
    config:
      retention.ms: 259200
      segment.bytes: 1073741824
    partitions: 3
    replicas: 3
  status:
    conditions:
    - lastTransitionTime: &quot;2022-08-21T17:25:44.833287Z&quot;
      status: &quot;True&quot;
      type: Ready
    observedGeneration: 3
    topicName: syslog.ueba-us4.v1.versa.demo3
  </code></pre>
  <p>segment.bytes = 1073741824,(i.e. ~1 GB), is this causing the older data to be unavailable in kafka ?</p>