Python DirectRunner does not emit data at GC time

[damccorm]

The following should succeed but does not:

test_options = PipelineOptions(flags=['--allow_unsafe_triggers'])
 with TestPipeline(options=test_options)
as pipeline:
  pcoll = (
    pipeline
    | beam.Create([(1, 1), (1, 2), (1, 3), (1, 4)])
    |
WindowInto(
      window.GlobalWindows(),
      trigger=trigger.AfterCount(5),
      accumulation_mode=trigger.AccumulationMode.ACCUMULATING)
 
  | beam.GroupByKey())
  assert_that(pcoll, equal_to([(1, [1, 2, 3, 4])]))

However, it  currently fails, because  pcoll will be empty. It appears that the Direct Runner drops data if the trigger never fired.

Imported from Jira BEAM-13078. Original Jira may contain additional context.
Reported by: zhoufek.

[kennknowles]

@zhoufek are you aware of any follow-up to this? If not, this is certainly an issue. It would be a direct runner issue, not a Python SDK issue, I think. Dataflow has one implementation of triggers while Java-based runners have another, and neither is shared with the local Python runner.