Repository navigation
Conversation
|
Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment |
| Queue<PublishResult> publishResultsQueue = session.getPublishedResultsQueue(); | ||
| Solace.PublishResult result = publishResultsQueue.poll(); | ||
|
|
||
| if (result != null) { |
There was a problem hiding this comment.
I'm wondering if this can result in data duplication. What if two competing bundles get access to the same queue with data and push them. Does the underlying library guarantee that data only get pushed once ?
| if (result != null) { | ||
| if (getCurrentBundleTimestamp() == null) { | ||
| setCurrentBundleTimestamp(Instant.now()); | ||
| for (int producerIndex = 0; producerIndex < producersMapCardinality; producerIndex++) { |
There was a problem hiding this comment.
Have you considered potential perf implications due to all bundles going through all queues ? I would still merge this to fix the correctness issue. But I would at least file a bug to improve perf if this is a concern.
| if (result != null) { | ||
| if (getCurrentBundleTimestamp() == null) { | ||
| setCurrentBundleTimestamp(Instant.now()); | ||
| for (int producerIndex = 0; producerIndex < producersMapCardinality; producerIndex++) { |
There was a problem hiding this comment.
Could you please add a unit test ?
|
hi, #39612 is being tested to fix #39589 @waterWang |
|
This pull request has been marked as stale due to 60 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@beam.apache.org list. Thank you for your contributions. |
Description
The
UnboundedSolaceWriter.publishResults()method only polls the publish results queue for the current bundle's producer. When multiple producers are configured (producersMapCardinality > 1), messages published by other producers have their publish results left in their respective queues, never drained or output. This causes downstream transforms attached to theSolaceIO.Writeoutput to receive only a subset of publish results.Root Cause
In
publishResults():solaceSessionServiceWithProducer()returns the session forcurrentBundleProducerIndex(set at@StartBundletime). Other producers' queues are never polled.Fix
Iterate over all producers (0 to
producersMapCardinality - 1) and poll each one's publish results queue. This ensures all publish results are drained and output, regardless of which producer published the message.Related Issues
Fixes #39588
Also related to #39589 (batch pipeline output loss in SolaceIO writer).
CC @stankiewicz @iht