SolaceIO - fix for data loss during scaling/rebalancing (#36991) - #38603
Conversation
Introduced a sequential pending checkpoints tracking mechanism using a TreeMap in the reader. Created a JVM-global ActiveReadersRegistry using weak references to resolve serialized checkpoint marks back to their originating active reader. This enables reliable sequential acknowledgments of checkpoints, ensuring we only ack committed data, while allowing subsequent finalizations to catch up and prevent message leaks (stuckness) if intermediate finalizations are lost. Also synchronized received messages access for thread safety with minimal lock duration (network I/O done outside locks). Fixed initialization order by registering the reader post-construction.
Summary of ChangesHello, I'm Gemini Code Assist1! 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 addresses data loss issues in the SolaceIO connector during Dataflow scaling and rebalancing events. By moving away from premature message acknowledgments and implementing a robust, sequential, and thread-safe finalization mechanism, the changes ensure that messages are only acknowledged after being successfully committed by the runner. The solution includes a global reader registry and a self-healing checkpoint finalization logic that prevents message leaks and pipeline stalls. Highlights
New Features🧠 You can now enable Memory (public preview) 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 AssistThe 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
Customization To customize the 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 Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counterproductive. 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. Footnotes
|
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request refactors the Solace checkpointing mechanism to support sequential acknowledgments and decouple the checkpoint mark from the reader's lifecycle using a global registry. Key changes include the introduction of the ActiveReadersRegistry to track active readers via UUIDs, updating SolaceCheckpointMark to store a reader UUID and checkpoint ID, and modifying UnboundedSolaceReader to manage pending checkpoints in a TreeMap. Review feedback identifies several improvement opportunities: using Guava's Cache with weak values for the registry to prevent potential memory leaks of keys, storing UUID objects directly in the checkpoint mark to avoid redundant string parsing, using private lock objects instead of synchronizing on this, reducing method visibility to package-private where appropriate, and restoring detailed message identifiers in error logs for better debugging.
- Use Guava Cache with weak values in ActiveReadersRegistry to prevent memory leaks. - Standardize on String for readerUuid in SolaceCheckpointMark and UnboundedSolaceReader to avoid Avro serialization issues with UUID on JDK 17+, while still eliminating UUID.fromString() overhead. - Use private lock object in UnboundedSolaceReader instead of synchronizing on 'this'. - Reduce visibility of UnboundedSolaceReader.finalizeCheckpoint to package-private. - Restore applicationMessageId and ackMessageId in failed ack logs.
|
Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment |
|
/gemini review |
There was a problem hiding this comment.
Code Review
This pull request refactors the SolaceIO acknowledgment logic by introducing a global ActiveReadersRegistry and a sequence-based checkpointing system. SolaceCheckpointMark now uses a reader UUID and a checkpoint ID to trigger acknowledgments via the originating reader, allowing for sequential processing and catch-up of missed checkpoints. Feedback was provided regarding an unnecessary synchronization block in the advance() method, as the thread model in Apache Beam ensures that advance() and getCheckpointMark() are called by the same thread, making the lock on receivedMessages redundant.
Removed unnecessary synchronization in advance() when adding to receivedMessages. Reduced the scope of the synchronized block in getCheckpointMark() to only cover pendingCheckpoints.put(). These changes are safe because advance() and getCheckpointMark() are executed sequentially by the same reader thread, so receivedMessages does not require synchronization. pendingCheckpoints still requires synchronization as it is shared with the asynchronous finalizeCheckpoint() thread. TAG=agy CONV=f94654e7-4a0a-4667-8a8b-d5bcf77e2609
| List<BytesXMLMessage> messagesToAck = new ArrayList<>(); | ||
|
|
||
| synchronized (lock) { | ||
| SortedMap<Long, List<BytesXMLMessage>> toAck = pendingCheckpoints.headMap(checkpointId, true); |
There was a problem hiding this comment.
@scwhittle this will allow reader to acknowledge any past checkpoint that wasn't finalized for some reason that was produced by this reader instance. is it correct?
There was a problem hiding this comment.
acking previous messages seems safe if Solace itself returns things in order. However if this is just stored in-memory in this process we may have cases where the worker crashes and we don't have the previous messages that were committed as processed to the backend in memory.
Or if ranges are reassigned so that this process no longer has the source assigned to it we could not get a more recent finalization. We may need some timeout on the cache so that we nack in that case (perhaps the existing reader cache could be sufficient).
There was a problem hiding this comment.
I will try to add a timeout mechanism to nack any pending checkpoint that has not been acked yet.
|
Assigning reviewers: R: @kennknowles for label java. Note: If you would like to opt out of this review, comment Available commands:
The PR bot will only process comments in the main thread (not review comments). |
|
|
||
| for (BytesXMLMessage msg : messagesToAck) { | ||
| try { | ||
| msg.ackMessage(); |
There was a problem hiding this comment.
It was raised that ackMessage may be slow, I'm curious if we could just add this and make ExecutorService ioExecutor = Executors.newFixedThreadPool(numberOfThreads); configurable with some good default.
// 1. Map messages to asynchronous tasks
CompletableFuture<?>[] futures = messagesToAck.stream()
.map(msg -> CompletableFuture.runAsync(() -> {
try {
msg.ackMessage();
} catch (IllegalStateException e) {
// irrelevant
}
}, ioExecutor))
.toArray(CompletableFuture[]::new);
// 2. Wait for all of them to complete
CompletableFuture.allOf(futures).join();
There was a problem hiding this comment.
Nice addition, I will submit a commit with this change.
| receivedMessages.clear(); | ||
| return new SolaceCheckpointMark(safeToAckMessages); | ||
| synchronized (lock) { | ||
| pendingCheckpoints.put(checkpointId, messages); |
There was a problem hiding this comment.
as this may grow and grow, can we track with Gauge how many messages are unacked per connection? UnboundedSolaceSource could have ID (integer) passed which could be part of metric name.
There was a problem hiding this comment.
I can add two counter metrics, for the total messages read and acked.
Uses an ExecutorService in UnboundedSolaceReader to acknowledge messages in parallel, preventing the finalizer thread from blocking on slow broker calls. TAG=agy CONV=63b71cca-a6fc-4840-b094-874e14c7b9e5
Adds received and acked counters to UnboundedSolaceReader to track the number of messages ingested and successfully acknowledged. These aggregate globally to monitor in-flight messages. TAG=agy CONV=63b71cca-a6fc-4840-b094-874e14c7b9e5
Introduces a configurable checkpoint timeout and nack mechanism in SolaceIO.Read. - Configurable Deadline: Adds withAckDeadline(Duration) to SolaceIO.Read, propagating it to UnboundedSolaceSource and UnboundedSolaceReader. Defaults to 30 seconds. - Timeout Detection: UnboundedSolaceReader checks for expired checkpoints during advance(). If a checkpoint is not finalized within the deadline, it is removed from memory to prevent leaks. - Asynchronous Nack: Expired checkpoints explicitly Nack their messages back to the Solace broker using JCSMP settle(Outcome.FAILED) asynchronously via the ackExecutor. - Unit Test: Adds UnboundedSolaceReaderTest to verify the timeout and async nack logic using a mock clock. - Integration Test: Adds test04ReadWithNackAndTimeout in SolaceIOIT using Testcontainers to verify end-to-end redelivery and successful reprocessing with a real Solace broker. TAG=agy CONV=63b71cca-a6fc-4840-b094-874e14c7b9e5
Removes test04ReadWithNackAndTimeout from SolaceIOIT because DirectRunner does not support bundle retries upon user DoFn exceptions. The timeout and Nack logic is already robustly covered in the unit test UnboundedSolaceReaderTest. TAG=agy CONV=63b71cca-a6fc-4840-b094-874e14c7b9e5
Fixes spotless formatting issues in SolaceIOIT.java by removing extra newline at the end of the file. TAG=agy CONV=9412686b-0a8b-4351-99e8-6284f708c8e2
|
Thx again for looking into this issue. I did some tests with our pipelines by applying these commits to v2.73 and noticed some things which I'm not sure are intended:
|
|
Reminder, please take a look at this pr: @kennknowles |
|
Thanks for that feedback @Robbllle, I am looking into those issues. |
|
Assigning new set of reviewers because Pr has gone too long without review. If you would like to opt out of this review, comment R: @chamikaramj for label java. Available commands:
|
|
Reminder, please take a look at this pr: @chamikaramj |
|
Reminder, please take a look at this pr: @chamikaramj |
|
Assigning new set of reviewers because Pr has gone too long without review. If you would like to opt out of this review, comment R: @kennknowles for label java. Available commands:
|
|
Reminder, please take a look at this pr: @kennknowles |
|
Assigning new set of reviewers because Pr has gone too long without review. If you would like to opt out of this review, comment R: @ahmedabu98 for label java. Available commands:
|
|
Reminder, please take a look at this pr: @ahmedabu98 |
|
Assigning new set of reviewers because Pr has gone too long without review. If you would like to opt out of this review, comment R: @chamikaramj for label java. Available commands:
|
hi @Robbllle , sorry it took so long. Can you validate this PR from the perspective of data loss/ missing finalization? |
|
@stankiewicz @iht In our tests with this PR on top of Beam 2.75.0 we see in multiple tests some unexpected behavior. Example logs: I will analyze this deeper and let you know if I manage to find some useful info. |
|
NACKs are supported on event brokers 10.2.1 and later. If an event broker does not support NACKs, an InvalidOperationException occurs during the flow bind request when an outcome is specified. |
We are using the It also doesn't happen in every test execution, and it seems to randomly affect different tests. Need to look deeper there to find out more. |
|
In Solace JCSMP code, this exception is thrown here: So it looks like you can only NACK from a redelivery flow? I guess that would be strange.. |
| } | ||
| } | ||
|
|
||
| for (PendingCheckpoint cp : expired) { |
There was a problem hiding this comment.
i think this step should be optional/configurable as NACKing is not always supported. By default we should just skip expired, as Solace will redeliver those.
…he latest Beam versions
|
I have merged with the latest master branch to be able to test on top of the latest Beam versions, as some of the comments refer to newer Beam versions than what was available when this PR was started. I am currently working on additional fixes to address the latest comments and findings. |
… by default - Disables explicit NACKing by default to prevent InvalidOperationException on standard flows - Evicts expired checkpoints from reader memory on timeout allowing broker redelivery - Adds withNackOnTimeout(boolean) option in SolaceIO.Read for flows supporting Outcome.FAILED - Handles settlement exceptions gracefully without failing the reader - Updates UnboundedSolaceReaderTest and SolaceIOReadTest to verify default eviction, optional NACK, and error handling
|
Hi @stankiewicz @ppawel, I have pushed commit 2dc0ddc to address the findings and review feedback regarding the Summary of Changes
|
OK, in our integration tests, it now works better, basically without any visible issues so far, the only difference to 2.75.0 is that Solace is redelivering certain messages in some tests - i.e. tests with "invalid" messages that cannot be parsed. This is in fact good/expected behavior I think which was missing before (such messages were "lost"). I plan to test still today inside Dataflow and with some real data streams plus forcing scaling up/down. (BTW, @Robbllle and I are in the same team :) ) |
From my testing in Dataflow, so far I was not able to reproduce message loss with your branch on top of Beam 2.75.0. The same message stream was going to a pipeline using stock 2.75.0 and that pipeline failed to complete 20% of transactions (meaning message loss occurred). We will do more tests in the next days/weeks to ensure we don't get any regressions, but so far it looks like data loss does not occur anymore. |
|
Reminder, please take a look at this pr: @chamikaramj |
5 participants
This PR may solve #36991.
Problem Statement
The SolaceIO connector was experiencing data loss during Dataflow scaling/rebalancing events.
finalizeCheckpoint) was lost or delayed, the associated messages remained unacknowledged indefinitely on the Solace broker, eventually hitting the max-delivered-unacked-msgs-per-flow limit and halting message delivery. That fix was subsequently reverted in Revert "SolaceIO data loss - remove message ack from close and advanc… #37162Proposed Solution
This PR introduces a robust, thread-safe sequential finalization catch-up mechanism that ensures we only acknowledge committed messages without risking leaks or stuckness if finalizations are lost.
1. Active Reader Tracking ( ActiveReadersRegistry )
We introduce a JVM-global ActiveReadersRegistry that tracks active UnboundedSolaceReader instances using WeakReferences.
2. Sequential Catch-Up Finalization
Instead of a shared queue, the reader now maintains a
TreeMap<Long, List<BytesXMLMessage>> pendingCheckpointsto track messages per individual checkpoint ID.3. Concurrency & Thread Safety
To avoid blocking the critical reader thread (calling advance and getCheckpointMark ) during network operations in the finalizer thread:
Changes Made
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.