KAFKA-20875: Headers window store adapters must expose the retention they wrap - #23037
Open
alanlau28 wants to merge 1 commit into
Open
KAFKA-20875: Headers window store adapters must expose the retention they wrap#23037alanlau28 wants to merge 1 commit into
alanlau28 wants to merge 1 commit into
Conversation
…they wrap PlainToHeadersWindowStoreAdapter and TimestampedToHeadersWindowStoreAdapter hold the store they wrap in a field rather than participating in the WrappedStateStore chain, so ProcessorStateManager.StateStoreMetadata.extractRetentionPeriod -- which finds a store's retention by unwrapping to the innermost layer -- terminates on the adapter and resolves -1. A -1 fails the windowed restore gate "retentionPeriod > 0 && retentionPeriod != Long.MAX_VALUE", so the restore seeks the beginning of the changelog rather than skipping data the store would discard anyway. On a changelog with delete retention that means starting at the offset retention is deleting from, with no margin against being lapped. Each adapter now implements WithRetentionPeriod and unwraps its own delegate, which follows the convention already in this code: TimestampedWindowStoreWithHeadersBuilder special-cases the adapter by type to walk through it rather than making it a WrappedStateStore. Making the adapters WrappedStateStore instead would let chain walks reach the inner store's format flags -- isTimestamped() would report true for a store whose observable format is post-conversion -- which is what these adapters exist to prevent. Also raises the log level when a windowed store reports no usable retention. Every branch of this path logged at debug, so a store that never got optimised looked identical to one that did.
github-actions
Bot
added
triage
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
issue: https://issues.apache.org/jira/browse/KAFKA-20875
PlainToHeadersWindowStoreAdapterandTimestampedToHeadersWindowStoreAdapterhold the store they wrap in afield rather than participating in the
WrappedStateStorechain, soProcessorStateManager.StateStoreMetadata.extractRetentionPeriod—which finds a store's retention by unwrapping to the innermost layer —
terminates on the adapter and resolves -1.
A -1 fails the windowed restore gate
retentionPeriod > 0 && retentionPeriod != Long.MAX_VALUE, so the restore seeks the beginningof the changelog rather than skipping data the store would discard
anyway. On a changelog with delete retention that means starting at the
offset retention is deleting from, with no margin against being lapped.
Each adapter now implements
WithRetentionPeriodand unwraps its owndelegate.