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
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.Writetransform 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
UnboundedSolaceWriterbase class keeps track of producer "index" when publishing messages but then uses the same index inpublishResultsmethod. 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:publishResultsForQueueis just the original version of thepublishResultsmethod.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