schizophrenicmaniac opened a new pull request, #40550: URL: https://github.com/apache/beam/pull/40550
BeamStatefulProcessor wrote beamWatermark on every visit and never removed it, so each key that ever went through a streaming stateful ParDo kept a row in the state store even after all its windows expired. beamTimers was already cleared by persistTimers once no timers remain. The watermark was the only thing left behind. storeWatermark, called from handleInputRows and handleExpiredTimer, now clears beamWatermark when the key has no timers and beamState is empty, for example after the GC timer of its last window fired. The change stays inside those two code paths, so it should rebase cleanly onto #40516 and the GroupByKey base-class split. On late data: after the cleanup, the key's next visit is a first visit and filters late data against TIMESTAMP_MIN_VALUE, so a late element for an already expired window of that key is processed instead of dropped. I went with accepting this as part of #40446. A brand-new key already behaves this way. Keeping any marker per key means keeping the same Long the issue wants gone. A TTLConfig is processing-time based, so it doesn't line up with allowed lateness. I updated the contract comment in timerInternals to cover cleared keys. The new test, testExpiredKeyLeavesNoState, lets a key's only window expire, then sends a late element back into that window. It checks the element is handled as a first visit, then reads the checkpoint through the statestore data source and asserts that beamState, beamTimers and beamWatermark have no rows left. Without the change the late element is dropped and one beamWatermark row remains. Fixes #40517 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
