Status

Current state: Under DiscussionAccepted

Discussion thread: Thread

JIRA:  

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

...

• 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.

We will conduct performance testing for this strategy, and if we find that initializing and closing many on-disk stores on startup is prohibitively expensive, we will parallelize this process by forking a new thread for each Task directory.

Compatibility, Deprecation, and Compatibility, Deprecation, and Migration Plan

Kafka Streams will automatically migrate offsets found in an existing .checkpoint file, and/or an existing .position file, to store those offsets directly in the StateStore, if managesOffsets returns true. Users of the in-built store types will not need to make any changes. See Upgrading.

Because checkpointed offsets are now stored inside the state stores, the state stores must be opened during KafkaStreams#start() to read the offsets, what implies that #start() might fail with new types of errors. This change is not a backward incompatible change because #start() already declares that StreamsException might be thrown, and thus existing code which is prepared for error handling would not be impacted.

flush deprecation

All internal usage of the StateStore#flush method has been removed/replaced. Therefore, the main concern is with end-users calling StateStore#flush directly from custom Processors. Obviously, users cannot safely call StateStore#commit, because they do not know the changelog offsets that correspond to the locally written state. Forcibly flushing/fsyncing recently written records to disk may also violate any transactional guarantees that StateStores are providing. Therefore, the default implementation of flush has been updated to a no-op. Users are now advised (via JavaDoc) that should instead request an early commit via. ProcessingContext#commit().

...