RabbitMqIO doesn't handle closed channels (#35471)

liferoad · web-flow · commit d0036c39198e · 2025-07-02T13:52:15.000-04:00
diff --git a/sdks/java/io/rabbitmq/src/main/java/org/apache/beam/sdk/io/rabbitmq/RabbitMqIO.java b/sdks/java/io/rabbitmq/src/main/java/org/apache/beam/sdk/io/rabbitmq/RabbitMqIO.java
@@ -442,7 +442,9 @@ public void finalizeCheckpoint() throws IOException {
         channel.basicAck(sessionId, false);
         sessionId = sessionIds.poll();
       }
-      channel.txCommit();
+      if (channel.isOpen()) {
+        channel.txCommit();
+      }
       latestTimestamp = Instant.now();
     }
   }

Original file line number	Diff line number	Diff line change
`@@ -442,7 +442,9 @@ public void finalizeCheckpoint() throws IOException {`
`442`	`442`	`channel.basicAck(sessionId, false);`
`443`	`443`	`sessionId = sessionIds.poll();`
`444`	`444`	`}`
`445`		`- channel.txCommit();`
	`445`	`+ if (channel.isOpen()) {`
	`446`	`+ channel.txCommit();`
	`447`	`+ }`
`446`	`448`	`latestTimestamp = Instant.now();`
`447`	`449`	`}`
`448`	`450`	`}`