[Bug]: SolaceIO write transform does not output all publish results #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
• 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
• Component: Google Cloud Dataflow Runner