chamikaramj commented on code in PR #39729:
URL: https://github.com/apache/beam/pull/39729#discussion_r3787311530
##########
sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/write/UnboundedSolaceWriter.java:
##########
@@ -143,57 +143,61 @@ public void publishResults(BeamContextWrapper context) {
long minFailed = Long.MAX_VALUE;
long maxFailed = 0;
- Queue<PublishResult> publishResultsQueue =
- solaceSessionServiceWithProducer().getPublishedResultsQueue();
- Solace.PublishResult result = publishResultsQueue.poll();
-
- if (result != null) {
- if (getCurrentBundleTimestamp() == null) {
- setCurrentBundleTimestamp(Instant.now());
+ for (int producerIndex = 0; producerIndex < producersMapCardinality;
producerIndex++) {
+ SessionService session =
+ SolaceWriteSessionsHandler.getSessionServiceWithProducer(
+ producerIndex, sessionServiceFactory, writerTransformUuid);
+ Queue<PublishResult> publishResultsQueue =
session.getPublishedResultsQueue();
+ Solace.PublishResult result = publishResultsQueue.poll();
+
+ if (result != null) {
Review Comment:
I'm wondering if this can result in data duplication. What if two competing
bundles get access to the same queue with data and push them. Does the
underlying library guarantee that data only get pushed once ?
##########
sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/write/UnboundedSolaceWriter.java:
##########
@@ -143,57 +143,61 @@ public void publishResults(BeamContextWrapper context) {
long minFailed = Long.MAX_VALUE;
long maxFailed = 0;
- Queue<PublishResult> publishResultsQueue =
- solaceSessionServiceWithProducer().getPublishedResultsQueue();
- Solace.PublishResult result = publishResultsQueue.poll();
-
- if (result != null) {
- if (getCurrentBundleTimestamp() == null) {
- setCurrentBundleTimestamp(Instant.now());
+ for (int producerIndex = 0; producerIndex < producersMapCardinality;
producerIndex++) {
Review Comment:
Could you please add a unit test ?
##########
sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/write/UnboundedSolaceWriter.java:
##########
@@ -143,57 +143,61 @@ public void publishResults(BeamContextWrapper context) {
long minFailed = Long.MAX_VALUE;
long maxFailed = 0;
- Queue<PublishResult> publishResultsQueue =
- solaceSessionServiceWithProducer().getPublishedResultsQueue();
- Solace.PublishResult result = publishResultsQueue.poll();
-
- if (result != null) {
- if (getCurrentBundleTimestamp() == null) {
- setCurrentBundleTimestamp(Instant.now());
+ for (int producerIndex = 0; producerIndex < producersMapCardinality;
producerIndex++) {
Review Comment:
Have you considered potential perf implications due to all bundles going
through all queues ? I would still merge this to fix the correctness issue. But
I would at least file a bug to improve perf if this is a concern.
--
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]
