From 1948b867fd130b00087cc9dcdb80e01dce4808e9 Mon Sep 17 00:00:00 2001
From: water <672684719@qq.com>
Date: Wed, 12 Aug 2026 07:00:34 +0800
Subject: [PATCH] fix: poll publish results from all producers in SolaceIO
 writer

---
 .../solace/write/UnboundedSolaceWriter.java   | 90 ++++++++++---------
 1 file changed, 47 insertions(+), 43 deletions(-)

diff --git a/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/write/UnboundedSolaceWriter.java b/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/write/UnboundedSolaceWriter.java
index 1c98113c2416..954b837fda19 100644
--- a/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/write/UnboundedSolaceWriter.java
+++ b/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) {
+        if (getCurrentBundleTimestamp() == null) {
+          setCurrentBundleTimestamp(Instant.now());
+        }
       }
-    }
 
-    while (result != null) {
-      Long latency = result.getLatencyNanos();
+      while (result != null) {
+        Long latency = result.getLatencyNanos();
 
-      if (latency == null && shouldPublishLatencyMetrics()) {
-        LOG.error(
-            "SolaceIO.Write: Latency is null but user asked for latency metrics."
-                + " This may be a bug.");
-      }
+        if (latency == null && shouldPublishLatencyMetrics()) {
+          LOG.error(
+              "SolaceIO.Write: Latency is null but user asked for latency metrics."
+                  + " This may be a bug.");
+        }
 
-      if (latency != null) {
+        if (latency != null) {
+          if (result.getPublished()) {
+            sumPublish += latency;
+            countPublish++;
+            minPublish = Math.min(minPublish, latency);
+            maxPublish = Math.max(maxPublish, latency);
+          } else {
+            sumFailed += latency;
+            countFailed++;
+            minFailed = Math.min(minFailed, latency);
+            maxFailed = Math.max(maxFailed, latency);
+          }
+        }
         if (result.getPublished()) {
-          sumPublish += latency;
-          countPublish++;
-          minPublish = Math.min(minPublish, latency);
-          maxPublish = Math.max(maxPublish, latency);
+          context.output(
+              SUCCESSFUL_PUBLISH_TAG, result, getCurrentBundleTimestamp(), GlobalWindow.INSTANCE);
         } else {
-          sumFailed += latency;
-          countFailed++;
-          minFailed = Math.min(minFailed, latency);
-          maxFailed = Math.max(maxFailed, latency);
+          try {
+            BadRecord b =
+                BadRecord.fromExceptionInformation(
+                    result,
+                    null,
+                    null,
+                    Optional.ofNullable(result.getError()).orElse("SolaceIO.Write: unknown error."));
+            context.output(FAILED_PUBLISH_TAG, b, getCurrentBundleTimestamp(), GlobalWindow.INSTANCE);
+          } catch (IOException e) {
+            // ignore, the exception is thrown when the exception argument in the
+            // `BadRecord.fromExceptionInformation` is not null.
+          }
         }
-      }
-      if (result.getPublished()) {
-        context.output(
-            SUCCESSFUL_PUBLISH_TAG, result, getCurrentBundleTimestamp(), GlobalWindow.INSTANCE);
-      } else {
-        try {
-          BadRecord b =
-              BadRecord.fromExceptionInformation(
-                  result,
-                  null,
-                  null,
-                  Optional.ofNullable(result.getError()).orElse("SolaceIO.Write: unknown error."));
-          context.output(FAILED_PUBLISH_TAG, b, getCurrentBundleTimestamp(), GlobalWindow.INSTANCE);
-        } catch (IOException e) {
-          // ignore, the exception is thrown when the exception argument in the
-          // `BadRecord.fromExceptionInformation` is not null.
-        }
-      }
 
-      result = publishResultsQueue.poll();
+        result = publishResultsQueue.poll();
+      }
     }
 
     if (shouldPublishLatencyMetrics()) {