title: Kafka Streams with state stores - Reprocessing of messages on app restart
body: <p>We have the following topology with two transformers, and each transformer uses persistent state store:</p>
  
  <pre><code>kStreamBuilder.stream(inboundTopicName)
      .transform(() -&gt; new FirstTransformer(FIRST_STATE_STORE), FIRST_STATE_STORE)
      .map((key, value) -&gt; ...)
      .transform(() -&gt; new SecondTransformer(SECOND_STATE_STORE), SECOND_STATE_STORE)
      .to(outboundTopicName);
  </code></pre>
  
  <p>and Kafka settings has <code>auto.offset.reset: latest</code>. After app was launched, I see two internal compacted topics were creates (and it's expected): <code>appId_inbound_firstStateStore-changelog</code> and <code>appId_inbound_secondStateStore-changelog</code></p>
  
  <p>Our app was down for two days, and after we started app again, messages were reprocessed from the beginning for specific partition (but we have multiple partitions). 
  I know that committed offsets are stored during ~ 1 day for kafka brokers prior to version 2, so our offsets should be cleaned up by retention. <strong>But why messages were reprocessed from beginning if we use <code>auto.offset.reset: latest</code>?</strong> Maybe it's somehow relate to stateful operations or changelog internal topics.</p>
  
  <p>I see the following logs (most of them are duplicated multiple times):</p>
  
  <pre><code>StoreChangelogReader Restoring task 0_55's state store firstStateStore from beginning of the changelog
  Fetcher [Consumer clientId=xxx-restore-consumer, groupId=] Resetting offset for partition xxx-55 to offset 0
  ConsumerCoordinator Setting newly assigned partitions
  ConsumerCoordinator Revoking previously assigned partitions
  StreamsPartitionAssignor Assigned tasks to clients
  AbstractCoordinator Successfully joined group with generation
  StreamThread partition revocation took xxx ms
  Unsubscribed all topics or patterns and assigned partitions
  AbstractCoordinator (Re-)joining group
  Attempt to heartbeat failed since group is rebalancing
  AbstractCoordinator Group coordinator xxx:9092 (id: xxx rack: null) is unavailable or invalid, will attempt rediscovery
  FetchSessionHandler - [Consumer clientId=xxx-restore-consumer, groupId=] Error sending fetch request (sessionId=INVALID, epoch=INITIAL) to node 2: org.apache.kafka.common.errors.DisconnectException
  </code></pre>
  
  <p>Kafka broker version <code>0.11.0.2</code>; Kafka Streams version <code>2.1.0</code></p>