You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "tkaymak (via GitHub)" <gi...@apache.org> on 2026/10/12 11:34:14 UTC

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

tkaymak opened a new issue, #40517:
URL: https://github.com/apache/beam/issues/40517

   ### What needs to happen?
   
   `BeamStatefulProcessor` keeps a `beamWatermark` and a `beamTimers` `ValueState` per grouping key and never clears `beamWatermark`, so a stateful ParDo, and the upcoming windowed GroupByKey, over an unbounded key space grows the state store forever, even after all windows of a key expired and its `beamState` map is empty. Clear both when no timers remain and `beamState` is empty, or attach a `TTLConfig` derived from allowed lateness.
   
   Part of #36841, follows #40447.
   
   ### Issue Priority
   
   Priority: 2 (default / most feature requests should be filed as P2)
   
   ### Issue Components
   
   - [x] Component: Spark Runner
   


-- 
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: github-unsubscribe@beam.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [I] [Task]: Clear per key watermark and timer state of the Spark 4 streaming stateful processor [beam]

Posted by "schizophrenicmaniac (via GitHub)" <gi...@apache.org>.
schizophrenicmaniac commented on issue #40517:
URL: https://github.com/apache/beam/issues/40517#issuecomment-6125921417

   .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.


-- 
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: github-unsubscribe@beam.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [I] [Task]: Clear per key watermark and timer state of the Spark 4 streaming stateful processor [beam]

Posted by "tkaymak (via GitHub)" <gi...@apache.org>.
tkaymak commented on issue #40517:
URL: https://github.com/apache/beam/issues/40517#issuecomment-6316612835

   The GroupByKey PR is open as #40703. `BeamStatefulProcessor` is now an abstract base for the stateful ParDo and the GroupByKey processor, the cleanup from #40550 sits in the base unchanged and covers GroupByKey keys too, with a test for it. Thanks again for the fix.
   


-- 
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: github-unsubscribe@beam.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [I] [Task]: Clear per key watermark and timer state of the Spark 4 streaming stateful processor [beam]

Posted by "Abacn (via GitHub)" <gi...@apache.org>.
Abacn closed issue #40517: [Task]: Clear per key watermark and timer state of the Spark 4 streaming stateful processor
URL: https://github.com/apache/beam/issues/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: github-unsubscribe@beam.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [I] [Task]: Clear per key watermark and timer state of the Spark 4 streaming stateful processor [beam]

Posted by "tkaymak (via GitHub)" <gi...@apache.org>.
tkaymak commented on issue #40517:
URL: https://github.com/apache/beam/issues/40517#issuecomment-6130511840

   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`, #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 #40446.
   2. `BeamStatefulProcessor` is about to change shape: #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 #40516 and keeping the change inside `handleInputRows` and `handleExpiredTimer` keeps the conflict small. I will link the GroupByKey PR here when it is open.
   


-- 
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: github-unsubscribe@beam.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org