prodriguezdefino opened a new pull request, #41073:
URL: https://github.com/apache/beam/pull/41073
Problem:
In decodeTimerIter
(sdks/go/pkg/beam/runners/prism/internal/engine/timers.go), a timer with
clear=true decodes to an element with sequence -1, timestamp 0, and
holdTimestamp 0. statefulStageKind.addPending handles sequence < 0 for
event-time timers by deleting the pending timer. Processing-time timers go
through ElementManager.triageTimers (elementmanager.go) and
timerHandler.Persist (timers.go), and neither function checks sequence < 0:
timers := th.timers(timer)
key := timerKey{family: timer.family, tag: timer.tag, window:
timer.window}
newTimer := fireElement{firing: fire, timer: timer}
if oldTimer, ok := timers[key]; ok {
timers[key] = newTimer
th.replace(key, oldTimer, newTimer)
...
return 0
}
timers[key] = newTimer
th.add(key, newTimer)
holdChanges[newTimer.timer.holdTimestamp] += 1
return 1
When an SDK clears a processing-time timer (for example in process or inside
another timer's callback), triageTimers rebases timestamp 0 into the past and
schedules an immediate refresh. If a timer was pending, Persist replaces it
with the clear element (returning 0 instead of -1), so Prism fires the cleared
timer at once. If no timer was pending, Persist schedules a new entry
(returning 1) and fires a timer that was never set.
Fix:
Check elm.sequence < 0 in triageTimers and timerHandler.Persist. Skip
rebasing and stageRefreshTimes insertion for clear elements, unschedule any
active timer from nextFiring, toFire, and order, decrement its watermark hold,
and return -1 when an active timer was removed (or 0 when none existed). Add
unit tests in timers_test.go and an ElementManager test in
elementmanager_test.go.
Reproduction (Python SDK on PrismRunner):
```python
import time
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.testing.util import assert_that, equal_to
from apache_beam.transforms.timeutil import TimeDomain
from apache_beam.transforms.userstate import TimerSpec, on_timer
from apache_beam.transforms.window import TimestampedValue
class ClearProcessingTimerDoFn(beam.DoFn):
EVENT_TIMER = TimerSpec("et", TimeDomain.WATERMARK)
PROC_TIMER = TimerSpec("pt", TimeDomain.REAL_TIME)
def process(
self,
element,
et=beam.DoFn.TimerParam(EVENT_TIMER),
pt=beam.DoFn.TimerParam(PROC_TIMER),
):
et.set(10)
pt.set(int(time.time()) + 3600)
@on_timer(EVENT_TIMER)
def on_event(self, pt=beam.DoFn.TimerParam(PROC_TIMER)):
pt.clear()
yield "event_fired"
@on_timer(PROC_TIMER)
def on_proc(self):
yield "proc_fired_after_clear"
if __name__ == "__main__":
options = PipelineOptions(["--runner=PrismRunner",
"--environment_type=LOOPBACK"])
with beam.Pipeline(options=options) as p:
out = (
p
| beam.Create([TimestampedValue(("k", 1), 1)])
| beam.ParDo(ClearProcessingTimerDoFn())
)
assert_that(out, equal_to(["event_fired"]))
```
Before the fix: Prism outputs `['event_fired', 'proc_fired_after_clear']`,
and `assert_that` fails.
After the fix: Prism outputs `['event_fired']`.
<!--
Thank you for your contribution! Follow this checklist to help us
incorporate your contribution quickly and easily:
- [ ] If this contribution is large, please file an Apache [Individual
Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
- [ ] Replace these instructions with a description of your change and why
you have made it.
- [ ] Mention the appropriate issue in your description (for example:
`addresses #123` or `fixes #456`).
- [ ] Update `CHANGES.md` with noteworthy changes.
See [CONTRIBUTING.md]() for more details.
-->
--
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]
