[damccorm]

Why This Is Needed

In #40205, we updated RestrictionTrackers.RestrictionTrackerObserverWithProgress.getProgress() to use lock.tryLock(60, TimeUnit.SECONDS) and return the cached lastProgress on lock timeout so that a long-running tryClaim() (such as a filtered scan in BoundedSourceAsSDFWrapperFn) does not block progress reporting indefinitely.

However, in runner harnesses (such as Dataflow's worker progress/split update loop), getProgress() and trySplit() execute sequentially within the same control-plane progress update cycle and share the same 10-minute lease renewal deadline:

1. getProgress() is called to gather bundle progress. With Return last evaluated progress on lock timeout in RestrictionTrackers.getProgress #40205, if tryClaim() is holding RestrictionTrackers.lock for an extended period (e.g. >10 minutes), getProgress() times out after 60 seconds and returns lastProgress.
2. That unchanged lastProgress is sent to the runner backend. Because the work item's reported progress is not advancing while other shards complete, dynamic work rebalancing (liquid sharding) flags the shard as a straggler and replies with a split request (ProcessBundleSplitRequest).
3. The worker control loop then immediately invokes RestrictionTrackerObserver.trySplit() as part of handling the update response—which calls lock.lock() on the exact same ReentrantLock still held by the long-running tryClaim().
4. trySplit() blocks indefinitely on lock.lock(), stalling the worker's progress/lease-renewal loop until the 10-minute deadline expires and fails the work item lease.

Solution & Why It Is Safe

This change updates RestrictionTrackerObserver.trySplit() to use lock.tryLock(300, TimeUnit.SECONDS) (5 minutes) instead of an unbounded lock.lock(), returning null if the lock cannot be acquired within the timeout:

• Follows the SDF trySplit contract: In Apache Beam's Splittable DoFn specification, RestrictionTracker.trySplit() is always advisory and is explicitly permitted to return null whenever a restriction cannot be split at the current moment.
• No state mutation on timeout: When lock.tryLock times out, delegate.trySplit() is never invoked, so the active restriction, ClaimObserver state, and cached progress remain untouched.
• Automatic retry on subsequent update cycles: Returning null causes FnApiDoFnRunner to return an empty ProcessBundleSplitResponse, allowing the runner's progress/update loop to complete normally and keep renewing the work item lease. On subsequent progress update cycles, the runner can continue requesting splits, which will succeed as soon as the slow tryClaim() finishes and releases lock.
• 5-minute timeout: Using a longer 5-minute timeout (300s) ensures splits still wait out moderately slow claims while guaranteeing that getProgress() (60s) + trySplit() (300s) always complete well within a 10-minute runner lease renewal window.

[Abacn]

Would this change break previously working slow split, in particular the initial split?

After #36750, initial trySplit costs >10 min are tolerated. In fact, it was a response to scenarios of initial split cost >10 min worked for Dataflow v1 but not v2. I'm afraid this change will cause those pipeline failing again (initial split return null)

[damccorm]

After #36750, initial trySplit costs >10 min are tolerated. In fact, it was a response to scenarios of initial split cost >10 min worked for Dataflow v1 but not v2. I'm afraid this change will cause those pipeline failing again (initial split return null)

It will only fail if we hit a 5 minute timeout trying to acquire the lock. Once the lock is acquired, we can take just as long to perform the split.

[codecov]

Codecov Report

❌ Patch coverage is 77.77778% with 2 lines in your changes missing coverage. Please review.
✅ Project coverage is 58.56%. Comparing base (d1ebefb) to head (973e0ca).
⚠️ Report is 4 commits behind head on master.

Files with missing lines	Patch %	Lines
...eam/sdk/fn/splittabledofn/RestrictionTrackers.java	77.77%	2 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #40251      +/-   ##
============================================
+ Coverage     57.29%   58.56%   +1.27%     
- Complexity     3696    13700   +10004     
============================================
  Files          1201     2579    +1378     
  Lines        194397   269892   +75495     
  Branches       3889    11138    +7249     
============================================
+ Hits         111373   158055   +46682     
- Misses        79416   105845   +26429     
- Partials       3608     5992    +2384     

Flag	Coverage Δ
java	64.86% <77.77%> (-7.22%)	⬇️

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.

[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

[scwhittle]

trySplit has this comment:

• @return a {@link SplitResult} if a split was possible, otherwise returns {@code null}. If the

• {@code fractionOfRemainder == 0}, a {@code null} result MUST imply that the restriction
  

• tracker is done and there is no more work left to do.
  

So it seems like data loss to return null if fractionOfRemainder == 0.

[damccorm]

That's a good catch. I think in that case, we have 2 options:

1. Raise a runtime exception
2. Time out (keep the current behavior).

I lean towards keeping the current behavior for 2 reasons:

1. It is less invasive
2. It allows jobs to succeed if acquiring the split lock takes e.g. 8 minutes (so no potential regression)

I updated to do this, let me know what you think. It does mean we could fail a work item on a runner-initiated checkpoint, but at least the odds of doing that 4 times in a row should be low. I guess its also more motivation to move forward with your fix.

Btw, I think that the null protection is mostly for this code path -

beam/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java

Line 676 in e3b3e7c

trySplitForElementAndRestriction(0, continuation.resumeDelay(), false);

- which would actually be ok since the lock would no longer be held after processElement returned. But I do think we should honor this to protect against data loss here nonetheless.

[github-actions]

Assigning reviewers:

R: @ahmedabu98 for label java.

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]

@damccorm shall we cherry-pick this PR and #40205 into release-2.77 ?