this could be due to uneven distribution of keys (doc ids) across parallel tasks

task manager failure and restart seem to coincide with start of the issue maybe task left in inconsistent state …

Did you perform a key distribution analysis? you might need to implement a custom partitioner to ensure more even distribution across tasks

You also might look at rocksdb tuning. ie buffer sizes cache sizes etc to make sure they are optimized

also review your shutdown strategy a bit

its crucial to check that the system clock on the machine running the affected subtask remains accurate through the window processing

a drift or inconsistency in processing time can disrupt the window boundaries and result in misalignment

These are somewhat rare but could cause this

Another thing you could look at is resource starvation. if task manger failure and restar left the subtask in a state where it could access sufficient resources to catch up effectively. I think in a way this is more likely and you might look at resource settings.

I think its either that or something related to custom timer firings due to something like incorrect timer setup, missed signals or interference with a custom trigger.

Other than that all I can think of is that its trying to checkpoint while the operator is overwhelmed. Need to maybe look closer at how flink checkpointing interacts with the stateful operator. If system tries to checkout while operator is overwhelmed this could cause something like this.

Let me take a look at the logs …

Ok so you faced a Taskmanager Disconnection. Looks like it crashed or was terminated. This in tern caused a failure of the sink task. Checkpoint was completed before failure.

The sub-task might resume processing from the last successful checkpoint but fails to recognize or properly process the accumulated windows that were supposed to be handled between the last checkpoint and the failure. This could stem from incomplete state restoration, misalignment of timers, or issues in the state backend. Given the scale of the problem, it’s possible that during recovery, the system did not properly manage the resumption of processing time timers or the cleanup of outdated state, leading to the backlog of unprocessed windows.

let me check logs for the recovery …

also check that task manager and job managers have their clocks synced

clock skew can cause this by making tasks misinterpret when windows should close

what is allowed lateness on the windows? does it seem reasonable or quite long?

what is allowed lateness on the windows? does it seem reasonable or quite long?

where i could check it?

look for any changes in input data patterns, config updates etc

check for late events, out of order events or surge in events

the allowed lateness should be in the window definitions somewhere for example:

stream
    .keyBy(<key selector>)
    .window(TumblingEventTimeWindows.of(Time.seconds(30))) // window definition
    .allowedLateness(Time.minutes(5)) // allowing events to be late by 5 minutes
    .process(new YourWindowFunction());

You might need to try reduce the allowed lateness as too long can cause accumulation

search code and config files for allowedLateness

Another thing you can do is try using Flink’s State Processor API to compare the clean save state with the first one that has an issue.

if its available for your version of Flink

Another little known tool to examine state which is a bit of a black box is Queryable State which has been deprecated but in testing or dev you could use it. Sometimes its needed to debug issues like the one you are describing to understand what is going on.

Its not well documented but it can be used.

In summary, I think we are looking at either a clock/timer configuration issue or skew, or there is an unusual network condition or data volume that periodically occurs for which the window settings/timeouts etc are not adjusted for. I think you need to examine all events and event flows around the timeframe between successful clean save and the incidence to look for patterns.