Skip to content

Changes cleanup thread to forceExecute. #38073

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
acrites wants to merge 6 commits into
base: master
Choose a base branch
from
Open

Changes cleanup thread to forceExecute. #38073

Changes from 1 commit
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
46f8236
Changes cleanup thread to forceExecute. Otherwise, we can deadlock if…
acrites Apr 4, 2026
19b704f
Changes cacheCommitFinalizers loop to only acquire the lock once and …
acrites Apr 6, 2026
f823ffa
Changes bundle finalizers to run in a dedicated ScheduledExecutorServ…
acrites Apr 7, 2026
19e3225
Changes thread pool to a single thread and moves scheduling cleanup u…
acrites Apr 7, 2026
a47758a
Merge branch 'apache:master' into master
acrites Apr 8, 2026
ba3533e
Uses other applied_finalize_ids after merging in https://github.com/a…
acrites Apr 8, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion ...pache/beam/runners/dataflow/worker/windmill/work/processing/StreamingCommitFinalizer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -138,7 +138,7 @@ public void cacheCommitFinalizers(Map<Long, Pair<Instant, Runnable>> callbacks)
// Start the cleanup thread lazily for pipelines that don't use finalization callbacks
// and some tests.
cleanUpThreadStarted = true;
finalizationExecutor.execute(this::cleanupThreadBody, 0);
finalizationExecutor.forceExecute(this::cleanupThreadBody, 0);
Copy link
Copy Markdown
Contributor

@arunpandianp arunpandianp Apr 6, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can we move the forceExecute to after releasing the lock lock.unlock?

Copy link
Copy Markdown
Contributor

@arunpandianp arunpandianp Apr 6, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Similar to how it is done in finalizeCommits below.

}
cleanUpQueue.add(info);
@SuppressWarnings("ReferenceEquality")
Expand Down
Loading