[FLINK-38992] Expire associated state together for non-time over windows #27487
Conversation
- Non-Time over windows uses multiple states to manage data - These states define ttl independently which causes out of sync expiry for all associated states for the same key thus leading to NullPointerException - To fix this, register a cleanup timer state and use that to expire all associated state for a key when the timer fires. This guarantees expiration of all associated keys across states.
bvarghese1
force-pushed
the
fix_over_window_state_ttl
branch
from
f3d9505 to
ade3f38
Compare
|
|
||
| // Initialize state to maintain id counter | ||
| idStateDescriptor = new ValueStateDescriptor<Long>("idState", Long.class); | ||
| if (ttlConfig.isEnabled()) { |
There was a problem hiding this comment.
In addition to the fix could we have a validation check when defining a state ttl on NonTime Over Windows, and error in this case, so the user is not led to think that ttl might be valid in this case.
fhueske
left a comment
fhueske
left a comment
There was a problem hiding this comment.
Thanks for the fix @bvarghese1!
Looks mostly good, left a few minor comments.
Thanks, Fabian
| cleanupState(idState, valueMapState, accMapState, sortedListState); | ||
| resetAndCleanupAggFuncs(); |
There was a problem hiding this comment.
I think we should check that the firing timer is actually a clean up timer, even though this operator does not register any other timers. This makes it a bit safer and the intent more clear.
KeyedProcessFunctionWithCleanupState.isProcessingTimeTimer()KeyedProcessFunctionWithCleanupState.needToCleanupState()
| generatedSortKeyComparator.newInstance( | ||
| getRuntimeContext().getUserCodeClassLoader()); | ||
|
|
||
| StateTtlConfig ttlConfig = createTtlConfig(stateRetentionTime); |
There was a problem hiding this comment.
Can this be removed (including import of class and static function)?
| } | ||
|
|
||
| @Override | ||
| public void onTimer(long timestamp, OnTimerContext ctx, Collector<RowData> out) |
There was a problem hiding this comment.
add some comments explaining that we are cleaning up state due to State TTL?
4 participants
What is the purpose of the change
Brief change log
Verifying this change
This change added tests and can be verified as follows:
Does this pull request potentially affect one of the following parts:
@Public(Evolving): (no)Documentation