From 05b2a7e012a21a0a9308dd7d78773a2fb854bfc4 Mon Sep 17 00:00:00 2001 From: Radek Stankiewicz Date: Mon, 11 May 2026 18:46:37 +0000 Subject: [PATCH 1/3] add drain docs --- CHANGES.md | 1 + .../site/content/en/blog/looping-timers.md | 11 ++- .../en/documentation/programming-guide.md | 85 +++++++++++++++++++ website/www/site/data/capability_matrix.yaml | 2 +- 4 files changed, 96 insertions(+), 3 deletions(-) diff --git a/CHANGES.md b/CHANGES.md index 2cf7f983fa35..bd9e272fa459 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -69,6 +69,7 @@ ## New Features / Improvements +* Capability introduces an indicator for aggregations and timers firing during a pipeline drain, allowing users and sinks to recognize and appropriately handle potentially incomplete or partial data ([#36884](https://github.com/apache/beam/issues/36884)). * Added support for setting disk provisioned IOPS and throughput in Dataflow runner via `--diskProvisionedIops` and `--diskProvisionedThroughputMibps` pipeline options (Java/Go/Python) ([#38349](https://github.com/apache/beam/issues/38349)). * TriggerStateMachineRunner changes from BitSetCoder to SentinelBitSetCoder to encode finished bitset. SentinelBitSetCoder and BitSetCoder are state diff --git a/website/www/site/content/en/blog/looping-timers.md b/website/www/site/content/en/blog/looping-timers.md index 2690dc967ecf..ad13aae75444 100644 --- a/website/www/site/content/en/blog/looping-timers.md +++ b/website/www/site/content/en/blog/looping-timers.md @@ -221,11 +221,18 @@ public static class LoopingStatefulTimer extends DoFn, KV key, - @TimerId("loopingTimer") Timer loopingTimer) { + @TimerId("loopingTimer") Timer loopingTimer, + CausedByDrain drain) { LOG.info("Timer @ {} fired", c.timestamp()); c.output(KV.of(key.read(), 0)); + // Check if drain is in progress and avoid resetting the timer + if (drain == CausedByDrain.CAUSED_BY_DRAIN) { + LOG.info("Drain in progress, stopping looping timer."); + return; + } + // If we do not put in a “time to live” value, then the timer would loop forever Instant nextTimer = c.timestamp().plus(Duration.standardMinutes(1)); if (nextTimer.isBefore(stopTimerTime)) { @@ -347,4 +354,4 @@ support for dealing with this use case in production. Runner specific notes: -Google Cloud Dataflow Runners Drain feature does not support looping timers (Link to matrix) +Support for cancelling looping timers on drain is currently limited to Dataflow and is being implemented (see [Issue #36884](https://github.com/apache/beam/issues/36884)). diff --git a/website/www/site/content/en/documentation/programming-guide.md b/website/www/site/content/en/documentation/programming-guide.md index 343fb128b3ef..956d7b997f16 100644 --- a/website/www/site/content/en/documentation/programming-guide.md +++ b/website/www/site/content/en/documentation/programming-guide.md @@ -7501,6 +7501,91 @@ class BufferDoFn(DoFn): {{< code_sample "sdks/go/examples/snippets/04transforms.go" batching_dofn_example >}} {{< /highlight >}} +#### 11.5.3. Looping timers {#looping-timers} + +Looping timers are a pattern where a timer sets another timer for a future time, creating a loop. This is useful for producing periodic outputs or heartbeats in the absence of data for a specific key. + +When draining a pipeline, it is important to terminate these loops to allow the pipeline to finish. In the Java SDK, you can use the `CausedByDrain` parameter in the `@OnTimer` method to check if the timer firing was induced by a drain operation. **Note:** `CausedByDrain` will be set only in certain runners. Check the [capability matrix](/documentation/runners/capability-matrix/) for more details. + +{{< highlight java >}} +public static class LoopingStatefulTimer extends DoFn, KV> { + @StateId("key") private final StateSpec> key = StateSpecs.value(); + @TimerId("loopingTimer") private final TimerSpec loopingTimer = TimerSpecs.timer(TimeDomain.EVENT_TIME); + + @ProcessElement + public void process( + @Element KV element, + @StateId("key") ValueState keyState, + @TimerId("loopingTimer") Timer timer) { + + keyState.write(element.getKey()); + // Set initial timer + timer.offset(Duration.standardMinutes(1)).setRelative(); + output.output(element); + } + + @OnTimer("loopingTimer") + public void onTimer( + @StateId("key") ValueState keyState, + @TimerId("loopingTimer") Timer timer, + OutputReceiver> output, + CausedByDrain drain) { + + output.output(KV.of(keyState.read(), 0)); + + // Cancel looping timer if drain is in progress + if (drain == CausedByDrain.CAUSED_BY_DRAIN) { + return; + } + + // Set next timer + timer.offset(Duration.standardMinutes(1)).setRelative(); + } +} +{{< /highlight >}} + +{{< highlight py >}} +# Python does not currently support detecting drain in OnTimer. +# The following example demonstrates a looping timer without drain support. + +class LoopingTimerDoFn(DoFn): + KEY_STATE = ValueStateSpec('key', coders.StrUtf8Coder()) + TIMER = TimerSpec('timer', TimeDomain.WATERMARK) + + def process(self, element, key_state=DoFn.StateParam(KEY_STATE), timer=DoFn.TimerParam(TIMER)): + key_state.write(element[0]) + timer.set(Timestamp.now() + Duration(seconds=60)) + yield element + + @on_timer(TIMER) + def on_timer(self, key_state=DoFn.StateParam(KEY_STATE), timer=DoFn.TimerParam(TIMER)): + yield (key_state.read(), 0) + # Loops forever, cannot handle drain safely if it never stops. + timer.set(Timestamp.now() + Duration(seconds=60)) +{{< /highlight >}} + +{{< highlight go >}} +// Go does not currently support detecting drain in OnTimer. +// The following example demonstrates a looping timer without drain support. + +type LoopingTimerFn struct { + KeyState state.Value[string] + Timer timers.EventTime +} + +func (fn *LoopingTimerFn) ProcessElement(sp state.Provider, tp timers.Provider, key string, value int, emit func(string, int)) { + fn.KeyState.Write(sp, key) + fn.Timer.Set(tp, time.Now().Add(60*time.Second)) + emit(key, value) +} + +func (fn *LoopingTimerFn) OnTimer(sp state.Provider, tp timers.Provider, key string, emit func(string, int)) { + emit(key, 0) + // Loops forever, cannot handle drain safely if it never stops. + fn.Timer.Set(tp, time.Now().Add(60*time.Second)) +} +{{< /highlight >}} + ## 12. Splittable `DoFns` {#splittable-dofns} diff --git a/website/www/site/data/capability_matrix.yaml b/website/www/site/data/capability_matrix.yaml index b7c236865ef3..a1afdc6f8abc 100644 --- a/website/www/site/data/capability_matrix.yaml +++ b/website/www/site/data/capability_matrix.yaml @@ -1475,7 +1475,7 @@ capability-matrix: - class: dataflow l1: "Partially" l2: - l3: Dataflow has a native drain operation, but it does not work in the presence of event time timer loops. Final implemention pending model support. + l3: Dataflow has a native drain operation, support for event time timer loops drain is limited to Non-portable runner. - class: prism l1: "No" l2: From 9ee4d629420c08dbde2ba81825e8322d1e588bc1 Mon Sep 17 00:00:00 2001 From: Radek Stankiewicz Date: Mon, 11 May 2026 19:09:42 +0000 Subject: [PATCH 2/3] simplify code --- .../en/documentation/programming-guide.md | 42 +++++++++---------- 1 file changed, 20 insertions(+), 22 deletions(-) diff --git a/website/www/site/content/en/documentation/programming-guide.md b/website/www/site/content/en/documentation/programming-guide.md index 956d7b997f16..76375a462c46 100644 --- a/website/www/site/content/en/documentation/programming-guide.md +++ b/website/www/site/content/en/documentation/programming-guide.md @@ -7509,16 +7509,14 @@ When draining a pipeline, it is important to terminate these loops to allow the {{< highlight java >}} public static class LoopingStatefulTimer extends DoFn, KV> { - @StateId("key") private final StateSpec> key = StateSpecs.value(); @TimerId("loopingTimer") private final TimerSpec loopingTimer = TimerSpecs.timer(TimeDomain.EVENT_TIME); @ProcessElement public void process( @Element KV element, - @StateId("key") ValueState keyState, - @TimerId("loopingTimer") Timer timer) { + @TimerId("loopingTimer") Timer timer, + OutputReceiver> output) { - keyState.write(element.getKey()); // Set initial timer timer.offset(Duration.standardMinutes(1)).setRelative(); output.output(element); @@ -7526,12 +7524,12 @@ public static class LoopingStatefulTimer extends DoFn, KV keyState, + @Key String key, @TimerId("loopingTimer") Timer timer, OutputReceiver> output, CausedByDrain drain) { - output.output(KV.of(keyState.read(), 0)); + output.output(KV.of(key, 0)); // Cancel looping timer if drain is in progress if (drain == CausedByDrain.CAUSED_BY_DRAIN) { @@ -7546,43 +7544,43 @@ public static class LoopingStatefulTimer extends DoFn, KV}} # Python does not currently support detecting drain in OnTimer. -# The following example demonstrates a looping timer without drain support. +# The following example demonstrates a looping timer without drain support, +# using event time. class LoopingTimerDoFn(DoFn): - KEY_STATE = ValueStateSpec('key', coders.StrUtf8Coder()) TIMER = TimerSpec('timer', TimeDomain.WATERMARK) - def process(self, element, key_state=DoFn.StateParam(KEY_STATE), timer=DoFn.TimerParam(TIMER)): - key_state.write(element[0]) - timer.set(Timestamp.now() + Duration(seconds=60)) + def process(self, element, ts=DoFn.TimestampParam, timer=DoFn.TimerParam(TIMER)): + timer.set(ts + Duration(seconds=60)) yield element @on_timer(TIMER) - def on_timer(self, key_state=DoFn.StateParam(KEY_STATE), timer=DoFn.TimerParam(TIMER)): - yield (key_state.read(), 0) + def on_timer(self, key=DoFn.KeyParam, timestamp=DoFn.TimestampParam, timer=DoFn.TimerParam(TIMER)): + yield (key, 0) # Loops forever, cannot handle drain safely if it never stops. - timer.set(Timestamp.now() + Duration(seconds=60)) + timer.set(timestamp + Duration(seconds=60)) {{< /highlight >}} {{< highlight go >}} // Go does not currently support detecting drain in OnTimer. -// The following example demonstrates a looping timer without drain support. +// The following example demonstrates a looping timer without drain support, +// using event time. type LoopingTimerFn struct { - KeyState state.Value[string] - Timer timers.EventTime + Timer timers.EventTime } -func (fn *LoopingTimerFn) ProcessElement(sp state.Provider, tp timers.Provider, key string, value int, emit func(string, int)) { - fn.KeyState.Write(sp, key) - fn.Timer.Set(tp, time.Now().Add(60*time.Second)) +func (fn *LoopingTimerFn) ProcessElement(et beam.EventTime, sp state.Provider, tp timers.Provider, key string, value int, emit func(string, int)) { + nextTime := et.ToTime().Add(60 * time.Second) + fn.Timer.Set(tp, nextTime) emit(key, value) } -func (fn *LoopingTimerFn) OnTimer(sp state.Provider, tp timers.Provider, key string, emit func(string, int)) { +func (fn *LoopingTimerFn) OnTimer(et beam.EventTime, sp state.Provider, tp timers.Provider, key string, timer timers.Context, emit func(string, int)) { emit(key, 0) // Loops forever, cannot handle drain safely if it never stops. - fn.Timer.Set(tp, time.Now().Add(60*time.Second)) + nextTime := et.ToTime().Add(60 * time.Second) + fn.Timer.Set(tp, nextTime) } {{< /highlight >}} From 1b564b5c60676a004bdbe496a3effc6dd2e96ac7 Mon Sep 17 00:00:00 2001 From: Radek Stankiewicz Date: Mon, 11 May 2026 19:26:33 +0000 Subject: [PATCH 3/3] trailing space --- website/www/site/content/en/documentation/programming-guide.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/website/www/site/content/en/documentation/programming-guide.md b/website/www/site/content/en/documentation/programming-guide.md index 76375a462c46..3f37f45bace6 100644 --- a/website/www/site/content/en/documentation/programming-guide.md +++ b/website/www/site/content/en/documentation/programming-guide.md @@ -7516,7 +7516,7 @@ public static class LoopingStatefulTimer extends DoFn, KV element, @TimerId("loopingTimer") Timer timer, OutputReceiver> output) { - + // Set initial timer timer.offset(Duration.standardMinutes(1)).setRelative(); output.output(element);