...

• On start-up, we will construct and initialize state stores for every corresponding Task store directory we find on-disk, and call committedOffset to determine their stored offset(s). We will then cache these offsets in-memory and close() these stores. This cache will be shared among all StreamThreads, and will therefore be thread-safe.
• On rebalance, we will consult this cache to compute our Task offset lags.
• After state restore completes for an assigned Task, we will update the offset cache to Task.LATEST_OFFSET, to indicate that the Task now has the currently latest offset.
• When closing a StateStore, we will update the offset cache with the current changelog offset for the store.
  • This will ensure that when a Task is reassigned to another instance, the Task lag for the local state matches what's on-disk, instead of using the sentinel value Task.LATEST_OFFSET, which is only valid for Tasks currently assigned to a thread on the local instance.

...

• • .

Compatibility, Deprecation, and Migration Plan

...