fix leader board trigger

[aIbrahiim]

LeaderBoardIT was still failing after the longer waits and the job read all 500 messages, but the Python pipeline only wrote a couple of partial score rows, so the check for total_score 5000 never passed

This switches the Python leader board to processing time triggers, like the Java example, and actually uses the allowed lateness value. User scores fire every 10 minutes and team scores fire 5 minutes early and 10 minutes late

---

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

• Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
• Update CHANGES.md with noteworthy changes.
• If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.

[github-actions]

Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment assign set of reviewers

[github-actions]

Assigning reviewers:

R: @claudevdm for label python.

This pull request likely touches a core component ("core" label). Please review with scrutiny.

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[Abacn]

The fix touches core codes of trigger, tagging people to review @kennknowles (Beam semantics)

Also this test exists for long. While it started failing recently? And now requiring changing core code to fix it?

[aIbrahiim]

The fix touches core codes of trigger, tagging people to review @kennknowles (Beam semantics)

Also this test exists for long. While it started failing recently? And now requiring changing core code to fix it?

The example has used AfterCount since 2017, and the passing run and the failing run were the same code so I think this did not break recently and it just fails when that trigger does not emit the row the test was looking for

Dataflow already runs AfterProcessingTime and the change that fixes the IT is leader_board.py

execution.py and trigger.py are only for the unit tests as after the example change, FnApi crashed because the trigger driver has no clock, and it fired the processing time timers immediately

[kennknowles]

There are multiple changes here that need to be made separately IMO

• changing the leader_board.py to set allowed lateness (I don't believe changing the trigger matters)
• improving plumbing for a clock in trigger execution (this does look useful and good!)

I believe just using allowed lateness with existing trigger should have the same impact on the integration test, FWIW. But I do think your other changes looks useful.

[aIbrahiim]

There are multiple changes here that need to be made separately IMO

• changing the leader_board.py to set allowed lateness (I don't believe changing the trigger matters)
• improving plumbing for a clock in trigger execution (this does look useful and good!)

I believe just using allowed lateness with existing trigger should have the same impact on the integration test, FWIW. But I do think your other changes looks useful.

Thanks @kennknowles, I can split these

but allowed lateness is needed for the team windows. The test uses 1 minute windows and the event time is from before the job starts, so with lateness 0 those elements get dropped. The code already calculated 120 minutes, but it never passed that value to the window

I don't think that fixes the user query though. leader_board_users is a global window, so those rows are not dropped for lateness. On the failing job all 500 messages were read, but the user score step only wrote 2 rows, and neither one was total_score 5000 so AfterCount(10) should have kept firing until that row showed up

The processing time trigger is the same one the Java example uses as it fire after the data is already there so that pane is 5000 The clock changes are only for the unit tests, so I can move them to another PR

[aIbrahiim]

There are multiple changes here that need to be made separately IMO

• changing the leader_board.py to set allowed lateness (I don't believe changing the trigger matters)
• improving plumbing for a clock in trigger execution (this does look useful and good!)

I believe just using allowed lateness with existing trigger should have the same impact on the integration test, FWIW. But I do think your other changes looks useful.

I split the clock change out of this PR.
The FnApi clock plumbing is in a separate PR #40429

[kennknowles]

There are multiple changes here that need to be made separately IMO

• changing the leader_board.py to set allowed lateness (I don't believe changing the trigger matters)
• improving plumbing for a clock in trigger execution (this does look useful and good!)

I believe just using allowed lateness with existing trigger should have the same impact on the integration test, FWIW. But I do think your other changes looks useful.

Thanks @kennknowles, I can split these

but allowed lateness is needed for the team windows. The test uses 1 minute windows and the event time is from before the job starts, so with lateness 0 those elements get dropped. The code already calculated 120 minutes, but it never passed that value to the window

I don't think that fixes the user query though. leader_board_users is a global window, so those rows are not dropped for lateness. On the failing job all 500 messages were read, but the user score step only wrote 2 rows, and neither one was total_score 5000 so AfterCount(10) should have kept firing until that row showed up

The processing time trigger is the same one the Java example uses as it fire after the data is already there so that pane is 5000 The clock changes are only for the unit tests, so I can move them to another PR

Yes, I think we agree. I think the addition of allowed lateness is the thing that fixes the example.

The change of trigger is neutral. I actually don't care too much either way about what trigger is used for the code demonstration. I was just thinking to keep the change to the minimum needed to fix it.

[kennknowles]

I think a time based trigger is more sensible for updating a leaderboard anyhow

[kennknowles]

Sorry, I had missed that the delay was super long. How about 5 seconds before the watermark and 30 seconds after it or something.

[codecov]

Codecov Report

❌ Patch coverage is 81.81818% with 2 lines in your changes missing coverage. Please review.
✅ Project coverage is 56.16%. Comparing base (78c9e10) to head (76f6e6d).
⚠️ Report is 84 commits behind head on master.

Files with missing lines	Patch %	Lines
sdks/python/apache_beam/transforms/trigger.py	77.77%	2 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #40370      +/-   ##
============================================
+ Coverage     56.15%   56.16%   +0.01%     
  Complexity     2288     2288              
============================================
  Files          1121     1124       +3     
  Lines        177193   178336    +1143     
  Branches       1489     1489              
============================================
+ Hits          99504   100170     +666     
- Misses        75168    75645     +477     
  Partials       2521     2521              

Flag	Coverage Δ
python	79.55% <81.81%> (-0.31%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[aIbrahiim]

@kennknowles @Abacn could you please take a look?