[SolaceIO] refactor SolaceIO's writers finishBundle by stankiewicz · Pull Request #39612 · apache/beam

stankiewicz · 2026-08-04T12:37:49Z

Refactor SolaceIO's writers to block in @FinishBundle until all published persistent messages have received either an acknowledgment (ACK) or a negative acknowledgment (NACK/error) from the Solace broker, up to a timeout.

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.

…shed persistent messages have received either an acknowledgment (ACK) or a negative acknowledgment (NACK/error) from the Solace broker, up to a timeout.

gemini-code-assist · 2026-08-04T12:38:56Z

Caution

The consumer version of Gemini Code Assist on GitHub has been sunset. All code review activity has officially ceased.

github-actions · 2026-08-04T14:18:31Z

Assigning reviewers:

R: @kennknowles for label java.

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).

ppawel · 2026-08-06T07:24:08Z

+      publishResults(context, messageIdsToAck);
+      if (!messageIdsToAck.isEmpty()) {
+        try {
+          Thread.sleep(10);


This will increase my project's test suite run by 10x :) Maybe this could be done without sleep, e.g. the waiting thread is woken up as soon as Solace callback with the ack arrives, and then checks the ack queue.

codecov · 2026-08-06T11:39:15Z

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 58.13%. Comparing base (4fd1744) to head (a1173bb).
⚠️ Report is 294 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #39612      +/-   ##
============================================
+ Coverage     58.12%   58.13%   +0.01%     
- Complexity    13086    13089       +3     
============================================
  Files          2521     2522       +1     
  Lines        264308   264352      +44     
  Branches      10786    10788       +2     
============================================
+ Hits         153616   153670      +54     
+ Misses       104922   104919       -3     
+ Partials       5770     5763       -7

Flag	Coverage Δ
java	`64.26% <ø> (+0.02%)`	⬆️

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.

stankiewicz · 2026-08-17T07:35:31Z

fixes #39589

stankiewicz · 2026-08-17T07:35:48Z

status: under testing by @ppawel .

chamikaramj

Thanks!

chamikaramj · 2026-08-17T18:00:02Z


-    publishResults(BeamContextWrapper.of(context));
+    if (getDeliveryMode() == DeliveryMode.PERSISTENT && messageIdsToAck != null) {
+      waitForAcks(BeamContextWrapper.of(context), messageIdsToAck);


Don't we need to invoke publishResults in this path as well ?

waitForAcks is waiting for Acks and publishing results in it.

chamikaramj · 2026-08-17T18:08:16Z

  @OnTimer("bundle_flusher")
  public void flushBundle(OnTimerContext context) throws IOException {
-    publishResults(BeamContextWrapper.of(context));
+    publishResults(BeamContextWrapper.of(context), null);


What would happen if the same results get published via this timer and the finishBundle above ? If this is safe, probably add a comment to clarify.

that is good point. publishResults is removing data from queue so it won't happen. In future or even this pr I also want to remove timer as it is not needed anymore. It was something also that shouldn't work correctly in batch.

@chamikaramj I've removed timer.

chamikaramj · 2026-08-17T18:10:53Z

+  public void waitForAcks(BeamContextWrapper context, Set<String> messageIdsToAck) {
+    BlockingQueue<PublishResult> queue =
+        solaceSessionServiceWithProducer().getPublishedResultsQueue();
+    long timeoutMs = System.currentTimeMillis() + ACKS_FLUSHING_INTERVAL_SECS * 1000;


Should we use exponential backoff here and set a larger overall timeout (O (minutes)) to support potentially large bundles and/or API rate limit issues ?

so the tiny cost here will be cpu of worker on checking the queue. there is no api rate limit issue as different thread is publishing to the queue.

chamikaramj · 2026-08-17T18:11:22Z

  public void finishBundle(FinishBundleContext context) {
-    publishResults(BeamContextWrapper.of(context));
+    if (getDeliveryMode() == DeliveryMode.PERSISTENT) {
+      waitForAcks(BeamContextWrapper.of(context), messageIdsToAck);


ppawel · 2026-08-18T14:07:54Z

From my testing it looks OK so far, all input elements have an output element (Solace publishing result).

I tested both in streaming pipelines and batch pipelines.

I am only testing using the default writer type (BATCHED). Does it make sense to also test it with the STREAMING writer type? I think I will do that as in fact we might consider switching writer types at some point... will report back.

stankiewicz · 2026-08-28T09:34:39Z

From my testing it looks OK so far, all input elements have an output element (Solace publishing result).

I tested both in streaming pipelines and batch pipelines.

I am only testing using the default writer type (BATCHED). Does it make sense to also test it with the STREAMING writer type? I think I will do that as in fact we might consider switching writer types at some point... will report back.

Paweł, committed new version, it doesn't change too much, it's still stateful, but I've removed timer which wasn't needed as all ACKs are received and it was invalid for batch pipelines.

Regarding STREAMING, it is waiting for ACK on finishBundle, similarly to BATCHED mode, you should see same improvement there.

refactor SolaceIO's writers to block in @FinishBundle until all publi…

2aa47a1

…shed persistent messages have received either an acknowledgment (ACK) or a negative acknowledgment (NACK/error) from the Solace broker, up to a timeout.

github-actions Bot added java io labels Aug 4, 2026

stankiewicz changed the title ~~[SolaceIO] refactor SolaceIO's writers finishBunlde~~ [SolaceIO] refactor SolaceIO's writers finishBundle Aug 4, 2026

github-actions Bot added the Next Action: Reviewers label Aug 4, 2026

ppawel reviewed Aug 6, 2026

View reviewed changes

Transition to blocking queue

a1173bb

stankiewicz mentioned this pull request Aug 17, 2026

fix: poll publish results from all producers in SolaceIO writer #39729

Open

chamikaramj requested changes Aug 17, 2026

View reviewed changes

remove timer, introduce state variable to still force checkpoint

085cf35

stankiewicz requested review from chamikaramj and ppawel August 28, 2026 09:27

Uh oh!

Conversation

stankiewicz commented Aug 4, 2026

GitHub Actions Tests Status (on master branch)

Uh oh!

gemini-code-assist Bot commented Aug 4, 2026

Uh oh!

github-actions Bot commented Aug 4, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

codecov Bot commented Aug 6, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

stankiewicz commented Aug 17, 2026

Uh oh!

stankiewicz commented Aug 17, 2026

Uh oh!

chamikaramj left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ppawel commented Aug 18, 2026

Uh oh!

stankiewicz commented Aug 28, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

codecov Bot commented Aug 6, 2026 •

edited

Loading