apache / apache/beam

[Bug]: SolaceIO write transform does not output all publish results

Open
#39,588 0 comments 0 reactions 0 assignees View on GitHub
awaiting triage bug dataflow io P2
Dominant language
Java
Stars
8.7k
Forks
4.7k
Avg merge
1d 20h
Merged PRs (30d)
196

Description

### 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

Contributor guide

Open the contributing guide

Research direction

Reproduce the missing results with the local Direct Runner tests, varying producer counts and batched versus streaming writers. Inspect UnboundedSolaceWriter.publishResults and SolaceWriteSessionsHandler, including the producer and published-results queue handling. Done means downstream transforms receive publish results for every message across those configurations; timestamp behavior is identified separately as noted in the issue.

Written by the indexing model from the issue text.

Assessment

Tech stack
google-cloud, java
Domain
distributed-systems
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Quiet
Clarity
Mostly clear
Newbie friendliness
48/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.