simplify code

Author: stankiewicz

website/www/site/content/en/documentation/programming-guide.md

@@ -7509,29 +7509,27 @@ When draining a pipeline, it is important to terminate these loops to allow the
 
 {{< highlight java >}}
 public static class LoopingStatefulTimer extends DoFn<KV<String, Integer>, KV<String, Integer>> {
-    @StateId("key") private final StateSpec<ValueState<String>> key = StateSpecs.value();
     @TimerId("loopingTimer") private final TimerSpec loopingTimer = TimerSpecs.timer(TimeDomain.EVENT_TIME);
 
     @ProcessElement
     public void process(
         @Element KV<String, Integer> element,
-        @StateId("key") ValueState<String> keyState,
-        @TimerId("loopingTimer") Timer timer) {
+        @TimerId("loopingTimer") Timer timer,
+        OutputReceiver<KV<String, Integer>> output) {
       
-      keyState.write(element.getKey());
       // Set initial timer
       timer.offset(Duration.standardMinutes(1)).setRelative();
       output.output(element);
     }
 
     @OnTimer("loopingTimer")
     public void onTimer(
-        @StateId("key") ValueState<String> keyState,
+        @Key String key,
         @TimerId("loopingTimer") Timer timer,
         OutputReceiver<KV<String, Integer>> 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<String, Integer>, KV<St
 
 {{< highlight py >}}
 # 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 >}}