waterWang opened a new pull request, #39664:
URL: https://github.com/apache/beam/pull/39664
## Description
Fixes #39589
**Problem**: In batch pipelines, the SolaceIO write transform loses publish
results because the pipeline finishes before asynchronous Solace ACK callbacks
arrive. The `@FinishBundle` method calls `publishResults()` which drains the
`publishedResultsQueue`, but the ACK callbacks that populate the queue haven't
fired yet, so the queue is empty and output is lost.
**Root cause**: Both `UnboundedBatchedSolaceWriter` and
`UnboundedStreamingSolaceWriter` call `publishResults()` in `@FinishBundle`
without waiting for in-flight publish ACKs. The batched writer also relies on a
`@OnTimer("bundle_flusher")` to "finalize" outputting publish results, but
processing-time timers don't fire in batch pipelines before the pipeline ends.
**Fix**: Track the number of in-flight publish operations using an
`AtomicInteger` counter. The counter is incremented when a publish is sent and
decremented when the `PublishResultHandler` receives the ACK callback. In
`@FinishBundle`, the writer waits (with a 30-second timeout) for the counter to
reach 0 before emitting results.
### Changes
1. **`PublishResultHandler.java`**: Accept an `AtomicInteger
pendingPublishCount` in the constructor. Decrement it in `processKey()` when a
publish result callback arrives.
2. **`SessionService.java`**: Add abstract method `getPendingPublishCount()`
returning `AtomicInteger`.
3. **`JcsmpSessionService.java`**: Add `AtomicInteger pendingPublishCount`
field, implement `getPendingPublishCount()`, pass it to `PublishResultHandler`.
4. **`UnboundedSolaceWriter.java`**: Add `incrementPendingPublishes(int
count)` and `waitForPendingPublishes()` helper methods. The wait polls the
counter with 50ms intervals up to a 30-second timeout, logging a warning if the
timeout is exceeded.
5. **`UnboundedBatchedSolaceWriter.java`**: In `finishBundle()`, increment
the pending count by the number of entries published in each batch, then call
`waitForPendingPublishes()` before `publishResults()`. Also add the wait in
`flushBundle()` (the timer handler) for consistency.
6. **`UnboundedStreamingSolaceWriter.java`**: In `processElement()`,
increment the pending count after each successful `publishSingleMessage()`. In
`finishBundle()`, call `waitForPendingPublishes()` before `publishResults()`.
### Testing
The existing unit tests cover the normal streaming and batched writer paths.
The fix is exercised in batch pipeline mode where the `@FinishBundle` wait
ensures ACKs are processed before results are emitted. All existing tests
should continue to pass since the wait is a no-op when there are no pending
publishes.
### Wallet
Solana: `fj4WqyCCw3C5ShR1RfB7MoBPTpkRrBFYP1uT35g3MvT`
--
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]
