<p>We are facing an issue with kafka state store which has a standby replica.</p>
<p>We have two topologies(streams). One without any internal state store and one with an internal state store.</p>
<p>We have problem with the topology which has the state store. So here I am giving info about that topology only:</p>
<p>The active and standby tasks of that stream are running in two seperate kubernetes pods in two nodes.</p>
<p>The active tasks consumes some kafka messages , processes them and write them in the state store.
The pod that holds the active task is killed.
The stand by task is assigned to the changelog topics.
But the consumer is seeking to EARLIEST offset. Since this changelog topic has thousands of messages it takes very long time complete.</p>
<p>I was expecting that the latest offset from the previous active task should be available to the standby (now active) task so that it would not need to pump up all the events.</p>
<p>Please correct me as I ran out of options and dont know where to check furthur.</p>
<p>I am using kafka-streams version <em><strong>2.5.0</strong></em></p>
<p><strong>Configuration:</strong></p>
<p>These are the configured values for the topology:</p>
<pre><code>    Properties properties = new Properties();
    properties.put(StreamsConfig.APPLICATION_ID_CONFIG, &quot;myService&quot;);
    properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServer);
    properties.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, ONE_THOUSAND);
    properties.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 10 * 1024 * 1024L);
    
    properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, &quot;earliest&quot;);
    properties.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, &quot;exactly_once&quot;);
    properties.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, &quot;read_committed&quot;);
    properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
    
    properties.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 1);
    properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 6000);
    properties.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, THREE); 
</code></pre>
<p><strong>Topology:</strong></p>
<pre><code> topology.addSource(AGENT_ADMIN_SOURCE, Serdes.String().deserializer(), serde.deserializer(), inboundAdminTopic)
            .addSource(MS_SOURCE_MATCH, Serdes.String().deserializer(), kafkaAvroDeserializer(), inboundMatchTopic)
            .addProcessor(MsProcessor.PROCESSOR_NAME, 
                    () -&gt; new MsProcessor(msActionManager(), stateStoreDAO()), 
                    AGENT_ADMIN_SOURCE, MS_SOURCE_MATCH)
            .addStateStore(agentAdminStateStore(), MsProcessor.PROCESSOR_NAME)   //agent-state-store
            .addStateStore(resourceMatchingStateStore(), MsProcessor.PROCESSOR_NAME)   //service-matches-state-store
            .addSink(...)
            .addSink(...);
        
</code></pre>
<p>//when the service installed this pod gets the stand by task</p>
<pre><code>{&quot;@timestamp&quot;:&quot;2020-08-31T10:50:17.253+00:00&quot;,&quot;@version&quot;:&quot;1&quot;,&quot;message&quot;:&quot;stream-thread [myService-418e5529-d591-46b3-b1f4-1da74e3926f4-StreamThread-1] partition assignment took 10 ms.\n\tcurrently assigned active tasks: []\n\tcurrently assigned standby tasks: [0_0]\n\trevoked active tasks: []\n\trevoked standby tasks: []\n&quot;,&quot;logger_name&quot;:&quot;org.apache.kafka.streams.processor.internals.StreamThread&quot;,&quot;thread_name&quot;:&quot;myService-418e5529-d591-46b3-b1f4-1da74e3926f4-StreamThread-1&quot;,&quot;level&quot;:&quot;INFO&quot;,&quot;level_value&quot;:20000}
</code></pre>
<p>The other pod is killed. The stand by task is expected to read from the latest offset in the change log topic. But it goes to earliest.</p>
<pre><code>{&quot;@timestamp&quot;:&quot;2020-08-31T10:50:17.302+00:00&quot;,&quot;@version&quot;:&quot;1&quot;,&quot;message&quot;:&quot;[Consumer clientId=myService-418e5529-d591-46b3-b1f4-1da74e3926f4-StreamThread-1-restore-consumer, groupId=null] Subscribed to partition(s): myService-agent-state-store-changelog-0,myService-service-matches-state-store-changelog-0 &quot;, &quot;logger_name&quot;:&quot;org.apache.kafka.clients.consumer.KafkaConsumer&quot;,&quot;thread_name&quot;:&quot;myService-418e5529-d591-46b3-b1f4-1da74e3926f4-StreamThread-1&quot;,&quot;level&quot;:&quot;INFO&quot;,&quot;level_value&quot;:20000}

{&quot;@timestamp&quot;:&quot;2020-08-31T10:50:17.306+00:00&quot;,&quot;@version&quot;:&quot;1&quot;,&quot;message&quot;:&quot;[Consumer clientId=myService-418e5529-d591-46b3-b1f4-1da74e3926f4-StreamThread-1-restore-consumer, groupId=null] Seeking to EARLIEST offset of partition myService-agent-state-store-changelog-0&quot;,&quot;logger_name&quot;:&quot;org.apache.kafka.clients.consumer.internals.SubscriptionState&quot;,&quot;thread_name&quot;:&quot;myService-418e5529-d591-46b3-b1f4-1da74e3926f4-StreamThread-1&quot;,&quot;level&quot;:&quot;INFO&quot;,&quot;level_value&quot;:20000}

{&quot;@timestamp&quot;:&quot;2020-08-31T10:50:17.306+00:00&quot;,&quot;@version&quot;:&quot;1&quot;,&quot;message&quot;:&quot;[Consumer clientId=myService-418e5529-d591-46b3-b1f4-1da74e3926f4-StreamThread-1-restore-consumer, groupId=null] Seeking to EARLIEST offset of partition myService-service-matches-state-store-changelog-0&quot;,&quot;logger_name&quot;:&quot;org.apache.kafka.clients.consumer.internals.SubscriptionState&quot;,&quot;thread_name&quot;:&quot;myService-418e5529-d591-46b3-b1f4-1da74e3926f4-StreamThread-1&quot;,&quot;level&quot;:&quot;INFO&quot;,&quot;level_value&quot;:20000}
</code></pre>
<p><em><strong>Could you please tell why does this seek to earliest instead of the latest known offset.</strong></em></p>
<p>Also why I am getting <em><strong>hundreds of these messages</strong></em></p>
<pre><code>{&quot;@timestamp&quot;:&quot;2020-08-31T11:50:39.849+00:00&quot;,&quot;@version&quot;:&quot;1&quot;,&quot;message&quot;:&quot;[Consumer clientId=myService-418e5529-d591-46b3-b1f4-1da74e3926f4-StreamThread-1-restore-consumer, groupId=null] Subscribed to partition(s): myService-agent-state-store-changelog-0, myService-service-matches-state-store-changelog-0 &quot;, &quot;logger_name&quot;:&quot;org.apache.kafka.clients.consumer.KafkaConsumer&quot;,&quot;thread_name&quot;:&quot;myService-418e5529-d591-46b3-b1f4-1da74e3926f4-StreamThread-1&quot;,&quot;level&quot;:&quot;INFO&quot;,&quot;level_value&quot;:20000}
</code></pre>

<p>Your expectation is correct, and what you observe is a bug that was fixed in 2.6.0 release: <a href="https://issues.apache.org/jira/browse/KAFKA-9501" rel="nofollow noreferrer">https://issues.apache.org/jira/browse/KAFKA-9501</a></p>
<p>The default behavior of Kafka Streams in a rebalancs is to close all tasks, and reopen them later. However, for StandbyTasks with in-memory store it implies that the state would be lost when a standby is closed in order to re-open it as an active task. For persistent stores, there is no such problem, because, well, the state is persistent to disk.</p>
<p>Since 2.6.0, the logic was changes and standby tasks are converted to active task without closing them, and thus the state is not lost any longer for in-memory stores.</p>
