SolaceIO - add ack deadline #37164

stankiewicz · 2025-12-22T10:30:15Z

Stacked on top of #37162

SolaceIO data loss:

remove message ack from close and advance as it may lead to data loss during work rebalancing or retry.
add async NACK with configurable deadline so any not finalized messages are rejected and retried.

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.

…e as it may lead to data loss during work rebalancing or retry. " This reverts commit f277b5ec59ba213af547167080fd8f5bd210e6a7.

gemini-code-assist · 2025-12-22T10:30:36Z

Summary of Changes

Hello @stankiewicz, I'm Gemini Code Assist¹! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!

This pull request significantly enhances the robustness of the SolaceIO connector by addressing potential data loss scenarios. It refactors the message acknowledgment process, moving away from immediate acknowledgments in certain lifecycle methods. Instead, it implements a more resilient system where messages that are not successfully processed and checkpointed within a defined ackDeadlineSeconds are automatically negatively acknowledged (NACKed) and thus retried. This change improves the connector's fault tolerance and ensures data integrity, especially in dynamic environments with work rebalancing.

Highlights

Data Loss Prevention: Removed immediate message acknowledgment from close() and advance() methods in SolaceIO to prevent data loss during work rebalancing or retries.
Asynchronous NACK with Deadline: Introduced an asynchronous negative acknowledgment (NACK) mechanism with a configurable deadline. Messages not finalized within this deadline will be explicitly rejected and retried.
Configurable Acknowledgment Deadline: Added a new withAckDeadlineSeconds option to SolaceIO.read() with a default of 60 seconds, allowing users to specify how long messages are held before being NACKed.
Internal Message Queue Update: Changed the internal SolaceCheckpointMark from using a List to a Queue for managing messages awaiting acknowledgment, better suiting the new NACK logic.
Test Case Adjustments: Updated existing test cases in SolaceIOReadTest to reflect the new ackDeadlineSeconds parameter and removed an outdated test related to late checkpoints.

🧠 New Feature in Public Preview: You can now enable Memory to help Gemini Code Assist learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. Click here to enable Memory in your admin console.

Using Gemini Code Assist

The full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips.

Invoking Gemini

You can request assistance from Gemini at any point by creating a comment using either /gemini <command> or @gemini-code-assist <command>. Below is a summary of the supported commands on the current page.

Feature	Command	Description
Code Review	`/gemini review`	Performs a code review for the current pull request in its current state.
Pull Request Summary	`/gemini summary`	Provides a summary of the current pull request in its current state.
Comment	@gemini-code-assist	Responds in comments when explicitly tagged, both in pull request comments and review comments.
Help	`/gemini help`	Displays a list of available commands.

Customization

To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a .gemini/ folder in the base of the repository. Detailed instructions can be found here.

Limitations & Feedback

Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here.

You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension.

Review the Privacy Notices, Generative AI Prohibited Use Policy, Terms of Service, and learn how to configure Gemini Code Assist in GitHub here. Gemini can make mistakes, so double check it and use code with caution. ↩

stankiewicz · 2025-12-22T10:31:09Z

R: @scwhittle

stankiewicz · 2025-12-22T10:31:22Z

R: @sjvanrossum

github-actions · 2025-12-22T10:32:13Z

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment assign set of reviewers

github-actions · 2025-12-22T10:32:23Z

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment assign set of reviewers

sjvanrossum · 2025-12-22T13:32:40Z

sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/read/UnboundedSolaceReader.java

+    Queue<BytesXMLMessage> safeToAckMessages = new ConcurrentLinkedQueue<>();
+    safeToAckMessages.addAll(receivedMessages);


We're not modifying safeToAckMessages outside the scope of this function, right? This doesn't have to be a concurrent container if that's the case.

I can't recall if receivedMessages can safely be replaced by the time this function is called. If so, maybe this can be cleaned up a bit more by reassigning receivedMessages to a new instance and passing the old instance off to nackMessages.

safeToAckMessages is used async by cleanUpThread, see below

I see, nvm. It seems like we're doing the same work that's expected of the bundle finalizer though.
I'm trying to think if we could capture the outstanding checkpoint finalization calls instead with an ExecutorService for example. When the reader is closed you'd then call ExecutorService#shutdownNow() which returns a collection of all Runnable items that will not be executed and those items could be executed then instead.

sjvanrossum · 2025-12-22T13:51:10Z

sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/read/UnboundedSolaceReader.java

+    try {
+      if (nackCallback != null) {
+        // wait only for last one to finish, it will mean all the previous one are also done.
+        nackCallback.get(ackDeadlineSeconds * 2, TimeUnit.SECONDS);
+      }
+    } catch (InterruptedException | ExecutionException | TimeoutException e) {
+      LOG.error("SolaceIO.Read: Failed to wait till nack background thread is finished");
+    }


Seems like this attempts to await the same conditions as when callingExecutorService#shutdown() and ExecutorService#awaitTermination(long, TimeUnit)?

not really, there are other background tasks that are scheduled on executor service that I don't want to wait for.

stankiewicz · 2025-12-22T15:26:51Z

sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/read/UnboundedSolaceReader.java

-    return new SolaceCheckpointMark(bytesXMLMessages);
+    nackCallback =
+        cleanUpThread.schedule(
+            () -> nackMessages(safeToAckMessages), ackDeadlineSeconds, TimeUnit.SECONDS);


see here @sjvanrossum

Ah yes, and then there's another possible use of the container in SolaceCheckpointMark#finalizeCheckpoint(), nvm.

- remove message ack from close and advance as it may lead to data loss during work rebalancing or retry. - add async NACK with configurable deadline so any not finalized messages are rejected and retried.

Revert "SolaceIO data loss - remove message ack from close and advanc…

f4cbb28

…e as it may lead to data loss during work rebalancing or retry. " This reverts commit f277b5ec59ba213af547167080fd8f5bd210e6a7.

github-actions bot added java io labels Dec 22, 2025

stankiewicz force-pushed the solace_io_ack_deadline branch 2 times, most recently from 55edab3 to dc3b007 Compare December 22, 2025 10:54

stankiewicz changed the title ~~Solace io ack deadline~~ SolaceIO - add ack deadline Dec 22, 2025

stankiewicz force-pushed the solace_io_ack_deadline branch from dc3b007 to 201026c Compare December 22, 2025 13:23

sjvanrossum suggested changes Dec 22, 2025

View reviewed changes

stankiewicz commented Dec 22, 2025

View reviewed changes

stankiewicz force-pushed the solace_io_ack_deadline branch from 201026c to cd60511 Compare December 23, 2025 12:17

SolaceIO data loss

3149f96

- remove message ack from close and advance as it may lead to data loss during work rebalancing or retry. - add async NACK with configurable deadline so any not finalized messages are rejected and retried.

stankiewicz force-pushed the solace_io_ack_deadline branch from cd60511 to 3149f96 Compare December 23, 2025 12:32

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

SolaceIO - add ack deadline #37164

SolaceIO - add ack deadline #37164

stankiewicz commented Dec 22, 2025

Uh oh!

gemini-code-assist bot commented Dec 22, 2025

Uh oh!

stankiewicz commented Dec 22, 2025

Uh oh!

stankiewicz commented Dec 22, 2025

Uh oh!

github-actions bot commented Dec 22, 2025

Uh oh!

github-actions bot commented Dec 22, 2025

Uh oh!

sjvanrossum Dec 22, 2025

Uh oh!

stankiewicz Dec 22, 2025

Uh oh!

sjvanrossum Dec 22, 2025

Uh oh!

sjvanrossum Dec 22, 2025

Uh oh!

stankiewicz Dec 22, 2025

Uh oh!

stankiewicz Dec 22, 2025

Uh oh!

sjvanrossum Dec 22, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

		Queue<BytesXMLMessage> safeToAckMessages = new ConcurrentLinkedQueue<>();
		safeToAckMessages.addAll(receivedMessages);

SolaceIO - add ack deadline #37164

Are you sure you want to change the base?

SolaceIO - add ack deadline #37164

Conversation

stankiewicz commented Dec 22, 2025

GitHub Actions Tests Status (on master branch)

Uh oh!

gemini-code-assist bot commented Dec 22, 2025

Summary of Changes

Highlights

Footnotes

Uh oh!

stankiewicz commented Dec 22, 2025

Uh oh!

stankiewicz commented Dec 22, 2025

Uh oh!

github-actions bot commented Dec 22, 2025

Uh oh!

github-actions bot commented Dec 22, 2025

Uh oh!

sjvanrossum Dec 22, 2025

Choose a reason for hiding this comment

Uh oh!

stankiewicz Dec 22, 2025

Choose a reason for hiding this comment

Uh oh!

sjvanrossum Dec 22, 2025

Choose a reason for hiding this comment

Uh oh!

sjvanrossum Dec 22, 2025

Choose a reason for hiding this comment

Uh oh!

stankiewicz Dec 22, 2025

Choose a reason for hiding this comment

Uh oh!

stankiewicz Dec 22, 2025

Choose a reason for hiding this comment

Uh oh!

sjvanrossum Dec 22, 2025

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants