apache / apache/beam

Timers with same family ids in same stage (but different transforms) are buffered together

Open
#21,995 0 comments 0 reactions 0 assignees View on GitHub
core harness P2 python sub-task
Dominant language
Java
Stars
8.7k
Forks
4.7k
Avg merge
1d 20h
Merged PRs (30d)
196

Description

The following test case does not work properly:

 
```

def test_dynamic_timer_clear_then_set_timer(self):

class EmitTwoEvents(DoFn):
EMIT_CLEAR_SET_TIMER
= TimerSpec('emitclear', TimeDomain.WATERMARK)

def process(self, element, emit=DoFn.TimerParam(EMIT_CLEAR_SET_TIMER)):

yield ('1', 'set')
emit.set(1)

@on_timer(EMIT_CLEAR_SET_TIMER)
def emit_clear(self):

yield ('1', 'clear')

class DynamicTimerDoFn(DoFn):
EMIT_TIMER_FAMILY = TimerSpec('emit',
TimeDomain.WATERMARK)

def process(self, element, emit=DoFn.TimerParam(EMIT_TIMER_FAMILY)):

if element[1] == 'set':
emit.set(10, dynamic_timer_tag='emit1')
emit.set(20,
dynamic_timer_tag='emit2')
if element[1] == 'clear':
emit.set(30, dynamic_timer_tag='emit3')

emit.clear(dynamic_timer_tag='emit3')
emit.set(40, dynamic_timer_tag='emit3')

return []

@on_timer(EMIT_TIMER_FAMILY)
def emit_callback(
self, ts=DoFn.TimestampParam,
tag=DoFn.DynamicTimerTagParam):
yield (tag, ts)

with TestPipeline() as p:
res = (

p
| beam.Create([('1', 'impulse')])
| beam.ParDo(EmitTwoEvents())

| beam.ParDo(DynamicTimerDoFn()))
assert_that(res, equal_to([('emit1', 10), ('emit2', 20), ('emit3',
40)])
```

Imported from Jira [BEAM-14127](https://issues.apache.org/jira/browse/BEAM-14127). Original Jira may contain additional context.
Reported by: pabloem.
Subtask of issue #21987

Contributor guide

Open the contributing guide

Research direction

Start by running the provided TestPipeline case with the Python Beam timer APIs and isolate the behavior around dynamic_timer_tag values in DynamicTimerDoFn. Done means the pipeline produces exactly ('emit1', 10), ('emit2', 20), and ('emit3', 40), including after the clear-then-set sequence.

Written by the indexing model from the issue text.

Assessment

Tech stack
python
Domain
stream-processing
Issue type
Bug
Difficulty
3/5
Estimated time
1-2 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
45/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.