[Task]: Clear per key watermark and timer state of the Spark 4 streaming stateful processor #40517

Activity

1. github-actions

   added

   spark

   P2

   on Oct 12, 2026
2. schizophrenicmaniac commented on Oct 12, 2026

   

   schizophrenicmaniac

   on Oct 12, 2026

   Contributor

   More actions

   .take-issue

   Hi, I'd like to work on this. In BeamStatefulProcessor, storeWatermark always writes beamWatermark, so I'd clear beamWatermark and beamTimers after handleInputRows/handleExpiredTimer once no timers remain and beamState is empty, with a streaming test checking the key's state is gone after window expiry.
3. github-actions

   assigned

   schizophrenicmaniac

   on Oct 12, 2026
4. tkaymak commented on Oct 12, 2026

   

   tkaymak

   on Oct 12, 2026

   ContributorAuthor

   More actions

   Thanks for taking this. Two things to line up:

   1. Clearing beamWatermark turns the next visit of that key into a first visit, which filters late data against TIMESTAMP_MIN_VALUE (contract comment in timerInternals, [Task]: Tighten the late data bound of the Spark 4 streaming stateful ParDo to the previous batch watermark #40446). A late element for a cleared key would then be accepted instead of dropped. Please cover that case in the test and say in the PR how you handle it, for example by keeping a much smaller marker or by accepting it as part of [Task]: Tighten the late data bound of the Spark 4 streaming stateful ParDo to the previous batch watermark #40446.
   2. BeamStatefulProcessor is about to change shape: [Spark 4] Group the streaming stateful ParDo by encoded key bytes to support null keys #40516 (open) switches the Spark grouping key to encoded bytes, and the next PR for windowed GroupByKey turns it into an abstract base with a ParDo and a GroupByKey subclass. The cleanup belongs in the base, so building on top of [Spark 4] Group the streaming stateful ParDo by encoded key bytes to support null keys #40516 and keeping the change inside handleInputRows and handleExpiredTimer keeps the conflict small. I will link the GroupByKey PR here when it is open.