ppawel opened a new issue, #39588:
URL: https://github.com/apache/beam/issues/39588
### What happened?
Beam 2.75.0, local direct runner in tests, but also same in GCP Dataflow
Runner.
Any downstream transform attached to the output of `SolaceIO.Write`
transform does not receive publish results for all messages, only some of them.
This seems to depend on how many producers are used to publish messages and
also whether batched or streaming writer are used. With the streaming writer,
the impact is smaller but it still loses results.
One issue could be that `UnboundedSolaceWriter` base class keeps track of
producer "index" when publishing messages but then uses the same index in
`publishResults` method. Only messages for that producer are sent to the
output, queues of other producers are not polled.
I added a workaround that flushes all queues in `publishResults`, something
like this:
```
public void publishResults(BeamContextWrapper context) {
for (int producerIndex = 0; producerIndex < producersMapCardinality;
producerIndex++) {
SessionService session =
SolaceWriteSessionsHandler.getSessionServiceWithProducer(
producerIndex, sessionServiceFactory, writerTransformUuid);
publishResultsForQueue(context, session.getPublishedResultsQueue());
}
}
```
`publishResultsForQueue` is just the original version of the
`publishResults` method.
This seems to work a bit better but then there is another issue - the
timestamps of the published output elements is most likely wrong as this code
does not keep track of the windows of input elements, it just has the latest
bundle timestamp so that's how it publishes everything. That causes issues with
windows and also with batch mode pipelines but that is another bug I will
create shortly...
CC @stankiewicz @iht
### Issue Priority
Priority: 2 (default / most bugs should be filed as P2)
### Issue Components
- [ ] Component: Python SDK
- [ ] 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: Twister2 Runner
- [ ] Component: Hazelcast Jet Runner
- [x] 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]
