fix: poll publish results from all producers in SolaceIO writer - #39729
fix: poll publish results from all producers in SolaceIO writer#39729waterWang wants to merge 1 commit into
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 |
3 participants
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