title: Why Kafka Streams reset offsets to zero after Kafka Server upgrade 2.4?
body: <h2>Problem Description:</h2>
  
  <p>I have a kafka topic with 12 partitions which consumer application is polling from. After the Kafka broker upgrade to 2.4 from version 2.2.1, I noticed that the consumer application lag turns to a higher number and reprocessing the records again from offset zero.</p>
  
  <h2>Topic configuration:</h2>
  
  <pre><code>cleanup.policy=compact
  delete.retention.ms=100
  segment.ms=100
  min.cleanable.dirty.ratio=0.01 
  </code></pre>
  
  <h2>Streaming app libraries:</h2>
  
  <pre><code>kafka-streams (version 2.0.1)
  spring-cloud-stream-binder-kafka-streams (version 2.1.2.RELEASE)
  Spring-kafka (version 2.2.6.RELEASE)
  </code></pre>
  
  <h2>Application Logs:</h2>
  
  <pre><code>[Consumer clientId=web-streaming-application-8023d9d2-e761-41c5-9b90-d586a3bbe095-StreamThread-1-consumer, groupId=web-consumer-application] Resetting offset for partition web-streaming-application_employees-4 to offset 0.
  [Consumer clientId=web-streaming-application-8023d9d2-e761-41c5-9b90-d586a3bbe095-StreamThread-1-consumer, groupId=web-streaming-application] Fetch offset 3378632 is out of range for partition web-streaming-application_employees-4, resetting offset
  </code></pre>
  
  <p>The Kafka brokers logs and metrics seems to be fine and no replication issues are there. I don't have an evidence that the upgrade is causing a problem. So I'm seeking your help to understand what is going on.</p>
  
  <p>Any thoughts are appreciated.</p>
  
  <h2>Application Config:</h2>
  
  <pre><code>spring:
    application:
      name: web-streaming-application
    kafka:
      properties:
        security.protocol: SASL_PLAINTEXT
        sasl.mechanism: GSSAPI
        sasl.kerberos.service.name: kafka
        bootstrap.servers: ${kafka.bootstrap-servers}
        value.serializer: org.apache.kafka.connect.json.JsonSerializer
    cloud.stream:
      kafka.streams:
        binder:
          brokers: ${kafka.bootstrap-servers}
          configuration:
            security.protocol: SASL_PLAINTEXT
            sasl.mechanism: GSSAPI
            sasl.kerberos.service.name: kafka
            default.key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
          serdeError: logAndContinue
        bindings:
          employeeInput:
            consumer:
              value-serde: business.serde.EmployeeJsonSerde
              application.id: ${spring.application.name}
      bindings:
        employeeInput:
          destination: ${kafka.topics.employee}
          group: ${spring.application.name}
  </code></pre>