ngibanel opened a new issue, #40984:
URL: https://github.com/apache/beam/issues/40984
### What happened?
What happened?
### Summary
`SolaceIO.Write` can fail in `finishBundle()` with `Timed out waiting for
ACKs` even when the corresponding ACK callbacks have already been received.
Concurrent writer instances for the same write transform can share a pooled
Solace session and its publish-results queue. Each bundle tracks its own
outstanding message IDs, but drains the shared queue. One bundle can therefore
consume and emit another bundle's ACK, leaving the owning bundle waiting for a
result that has already been removed.
This was observed in a streaming pipeline on Google Cloud Dataflow and
reproduced deterministically.
### Environment and configuration
- Affected SDK: SolaceIO development sources for Beam **2.78.0** (initially
used through a privately built artifact as the version is not released yet)
- Runner where the error was observed: Google Cloud Dataflow, streaming.
- The relevant write options were:
```java
SolaceIO.write()
.to(SolaceIO.topicFromName("test/ack-routing"))
.withSessionServiceFactory(sessionServiceFactory)
.withDeliveryMode(DeliveryMode.PERSISTENT)
.withWriterType(SolaceIO.WriterType.BATCHED)
.withSubmissionMode(SolaceIO.SubmissionMode.HIGHER_THROUGHPUT)
.withNumberOfClientsPerWorker(6)
.withNumShards(40);
```
### Actual behavior
The Dataflow job reported timeouts for varying numbers of outstanding
messages. A representative worker stack trace was:
```
java.lang.RuntimeException: SolaceIO.Write: Timed out waiting for ACKs of 1
messages.
Outstanding message IDs: [f9284e67-5db6-49ab-9015-08c0b0f41b2d]
at
org.apache.beam.sdk.io.solace.write.UnboundedSolaceWriter.waitForAcks(UnboundedSolaceWriter.java:261)
at
org.apache.beam.sdk.io.solace.write.UnboundedBatchedSolaceWriter.finishBundle(UnboundedBatchedSolaceWriter.java:142)
```
The local reproduction proves that this exception can occur after all ACK
callbacks have arrived. It does not assume that every ACK timeout has this
cause.
### Expected behavior
Each publication result should reach the bundle that owns that publication.
An ACK consumed by another bundle must not leave the owner waiting until
timeout. Concurrent bundles should be able to share pooled sessions without
mixing their ACK/NACK ownership.
### Steps to reproduce deterministically
1. Use the SolaceIO sources at the baseline above (main branch or 2.78.0
branch)
2. Add the reproductible test
https://gist.github.com/ngibanel/dd957dca79bd830010fa18b9961031da in
`sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/write/SolaceAckTimeoutReproTest.java`
3. Run the test from the Beam repository root `./gradlew
:sdks:java:io:solace:test --tests
org.apache.beam.sdk.io.solace.write.SolaceAckTimeoutReproTest`
4. On the affected implementation, observe that writer A emits both
publication results, then writer B times out after the approximately 10-second
ACK deadline.
The test controls the following interleaving:
1. Create writer A and serialize/deserialize it to create writer B. Both
represent the same Beam write transform and retain the same transform UUID.
2. Start A's bundle, advance the producer index with five additional cloned
writer instances, then start B's bundle. With six client slots, A and B select
the same pooled session. The test asserts that they share that session.
3. Publish message `B` and deliver its ACK into the session's result
handler. Pause B before its publishing call returns.
4. Publish and ACK message `A`. Allow A to run `finishBundle()`, which
drains the shared result queue.
5. Resume B. Its ACK has already been consumed by A, so B's outstanding-ID
set still contains `B` and `waitForAcks()` times out.
The messages have different IDs. No dropped ACK, broker outage, network
delay, high message volume or Dataflow-specific scheduling is needed to trigger
the defect.
### Issue Priority
Priority: 2 (default / most bugs should be filed as P2)
### Issue Components
- [ ] Component: Python SDK
- [x] Component: Java SDK
- [ ] Component: Go SDK
- [ ] Component: Typescript SDK
- [x] Component: IO connector
- [ ] Component: Beam YAML
- [ ] Component: Beam examples
- [ ] Component: Beam playground
- [ ] Component: Beam katas
- [ ] Component: Website
- [ ] Component: Infrastructure
- [ ] Component: Spark Runner
- [ ] Component: Flink Runner
- [ ] Component: Prism Runner
- [ ] Component: Hazelcast Jet Runner
- [ ] Component: Google Cloud Dataflow Runner
--
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]
