waterWang opened a new pull request, #39729:
URL: https://github.com/apache/beam/pull/39729
## 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 the
`SolaceIO.Write` output to receive only a subset of publish results.
### Root Cause
In `publishResults()`:
```java
Queue<PublishResult> publishResultsQueue =
solaceSessionServiceWithProducer().getPublishedResultsQueue();
```
`solaceSessionServiceWithProducer()` returns the session for
`currentBundleProducerIndex` (set at `@StartBundle` time). 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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
