[Bug]: Python event-time timers clearing behavior not consistent on different bundle sizes

[shunping]

What happened?

Assuming the first event's event time is T, the following code sets an event-time timer to T+1 after receiving the first event. Then on the third event, it clears the timer.

The code behaves differently on different bundle sizes of different runners.

• On prism, the bundle size seems to be small. So the timer setting operation is on a different bundle than the timer clearing operation. As a result, timer is triggered before the timer clearing operation.
• On dataflow, the bundle size seems to be bigger. Both setting and clearing are in the same bundle. In the current python implementation, the timer will not be fired because it is cleared before the bundle ends.

class AnotherTimerDoFn(beam.DoFn):
  COUNT = ReadModifyWriteStateSpec('count', coders.VarInt32Coder())
  TIMER = TimerSpec('timer', TimeDomain.WATERMARK)

  def __init__(self):
    pass

  def start_bundle(self):
    logging.info("start bundle")
    pass

  def finish_bundle(self):
    logging.info("finish bundle")
    pass

  def process(self,
              element_pair,
              t=beam.DoFn.TimestampParam,
              count=beam.DoFn.StateParam(COUNT),
              timer=beam.DoFn.TimerParam(TIMER)):
    local_count = count.read() or 0
    local_count += 1

    logging.info(f"get element {element_pair[1]}, count={local_count}")
    if local_count == 1:
      logging.info(f"set timer to {t+1}")
      timer.set(t+1)

    if local_count == 3: # change this to 5 so the clear is on a different bundle
      logging.info("clear timer")
      timer.clear()

    count.write(local_count)

  @on_timer(TIMER)
  def timer_callback(self, t=beam.DoFn.TimestampParam):
    logging.error("Timer should not fire here")
    logging.info(f"timer callback start (timestamp={t})")
    yield (">>>timer fired")


with beam.Pipeline(options=options) as p:
  _ = (
      p | PeriodicImpulse(
          start_timestamp=Timestamp.now(),
          stop_timestamp=Timestamp.now() + 14,
          fire_interval=1)
      | beam.WithKeys(0)
      | beam.ParDo(AnotherTimerDoFn())
      | beam.Map(print))

A workaround of this is to add a state to keep track of the last set timestamp, and only clear the timer when the event time when clearing is triggered is smaller than or equal to the set timestamp.

class AnotherTimerDoFn(beam.DoFn):
  COUNT = ReadModifyWriteStateSpec('count', coders.VarInt32Coder())
  LAST_TIMESTAMP = ReadModifyWriteStateSpec('ts', coders.TimestampCoder())
  TIMER = TimerSpec('timer', TimeDomain.WATERMARK)

  def __init__(self):
    pass

  def start_bundle(self):
    logging.info("start bundle")
    pass

  def finish_bundle(self):
    logging.info("finish bundle")
    pass

  def process(self,
              element_pair,
              t=beam.DoFn.TimestampParam,
              count=beam.DoFn.StateParam(COUNT),
              last_ts=beam.DoFn.StateParam(LAST_TIMESTAMP),
              timer=beam.DoFn.TimerParam(TIMER)):
    local_count = count.read() or 0
    local_count += 1

    logging.info(f"get element {element_pair[1]}, count={local_count}")
    if local_count == 1:
      logging.info(f"set timer to {t+1}")
      timer.set(t+1)
      last_ts.write(t+1)

    if local_count == 3:
      logging.info("clear timer")
      ts = last_ts.read() or 0
      if t <= ts:
        timer.clear()

    count.write(local_count)

  @on_timer(TIMER)
  def timer_callback(self, t=beam.DoFn.TimestampParam):
    logging.error("Timer should not fire here")
    logging.info(f"timer callback start (timestamp={t})")
    yield (">>>timer fired")

Issue Priority

Priority: 2 (default / most bugs should be filed as P2)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam YAML
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Infrastructure
• Component: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner

[shunping]

In summary, the timing of clear and set matters here.

• If we set the event time timer to trigger at T + 10, and we choose to clear the timer when processing event at T + 3 event time, then the timer should be cleared successfully.

• If we set the event time timer to trigger at T + 1, and we choose to clear the timer when processing event at T + 3 event time, I think we need to define a consistent behavior of whether the timer should be triggered regardless of bundle size. In this case, I feel it makes sense to trigger the timer because the triggering should happen before the clear.

WDYT? @lostluck

[lostluck]

• yes, this seems correct.
• This is only defined if the runner promises event time ordering of elements per key, which isn't guaranteed across all transforms in the model.

As a separate point, is it reasonable for a data element to clear a timer if the timer Callback code is assuming it runs?

Timers self clear when they run, so why manually clear it? It's situational.

Clear is a convenience action for when the user code knows the timers shouldn't run based on state they've stored. That state is independent of whether the timer executed or not.

We can't always assume we know better than the user code being executed. Recall that the user code can't know or assume anything about the bundle size either. To do so is a programing error.

[shunping]

Clear is a convenience action for when the user code knows the timers shouldn't run based on state they've stored. That state is independent of whether the timer executed or not.

I agree. That's also what I have in my first example: the clear action is called when the count (a state) reaches 3.

Anyway, the workaround I proposed (by adding the LAST_TIMESTAMP state) seems to work on different bundle sizes: the clear action is only called when it is going to clear a timer in the future ET. Maybe this is too specific to my use case though.

With all that said, I still think that if a timer behavior is relying on bundle size, it is a problem. It will be great if we can clearly say this is a "yes" or "no", rather than a "maybe"...

[github-actions]

This issue has been marked as stale due to 150 days of inactivity. It will be closed in 30 days if no further activity occurs. If you think that’s incorrect or this issue still needs to be addressed, please simply write any comment. If closed, you can reopen the issue at any time. Thank you for your contributions.

[github-actions]

This issue has been closed due to lack of activity. If you think that is incorrect, you can reopen the issue at any time.