...

• How frequently do we fetch the changelog offsets today? Is there a config? Need to introduce a config to control how frequently we fetch the tail record from changelog.
• lagInfoForStore(String storeName) is not quite right. We need to be able to use StreamsMetadata object and find the right StreamsLagInfo object from the list. 

Compatibility, Deprecation, and Migration Plan

...