From 8183bcdc8704b4699911a982712ed49ee8b4536f Mon Sep 17 00:00:00 2001
From: water <672684719@qq.com>
Date: Tue, 11 Aug 2026 10:20:02 +0800
Subject: [PATCH] fix: use try/finally to ensure HBaseIO resource cleanup even
 if an earlier close() throws

---
 .../org/apache/beam/sdk/io/hbase/HBaseIO.java | 87 ++++++++++++++++---
 1 file changed, 74 insertions(+), 13 deletions(-)

diff --git a/sdks/java/io/hbase/src/main/java/org/apache/beam/sdk/io/hbase/HBaseIO.java b/sdks/java/io/hbase/src/main/java/org/apache/beam/sdk/io/hbase/HBaseIO.java
index bc575b50af54..67462b10605c 100644
--- a/sdks/java/io/hbase/src/main/java/org/apache/beam/sdk/io/hbase/HBaseIO.java
+++ b/sdks/java/io/hbase/src/main/java/org/apache/beam/sdk/io/hbase/HBaseIO.java
@@ -549,16 +549,38 @@ public boolean advance() {
     @Override
     public void close() throws IOException {
       LOG.debug("Closing reader after reading {} records.", recordsReturned);
+      IOException thrown = null;
       if (scanner != null) {
-        scanner.close();
-        scanner = null;
+        try {
+          scanner.close();
+        } catch (IOException e) {
+          thrown = e;
+        } finally {
+          scanner = null;
+        }
       }
       if (connection != null) {
-        connection.close();
-        connection = null;
+        try {
+          connection.close();
+        } catch (IOException e) {
+          thrown = addSuppressed(thrown, e);
+        } finally {
+          connection = null;
+        }
+      }
+      if (thrown != null) {
+        throw thrown;
       }
     }
 
+    private static IOException addSuppressed(IOException first, IOException next) {
+      if (first != null) {
+        first.addSuppressed(next);
+        return first;
+      }
+      return next;
+    }
+
     @Override
     public synchronized HBaseSource getCurrentSource() {
       return source;
@@ -765,13 +787,34 @@ public void finishBundle() throws Exception {
 
       @Teardown
       public void tearDown() throws Exception {
+        Exception thrown = null;
         if (mutator != null) {
-          mutator.close();
-          mutator = null;
+          try {
+            mutator.close();
+          } catch (Exception e) {
+            thrown = e;
+          } finally {
+            mutator = null;
+          }
         }
         if (connection != null) {
-          connection.close();
-          connection = null;
+          try {
+            connection.close();
+          } catch (Exception e) {
+            if (thrown != null) {
+              thrown.addSuppressed(e);
+            } else {
+              thrown = e;
+            }
+          } finally {
+            connection = null;
+          }
+        }
+        if (thrown != null) {
+          if (thrown instanceof IOException) {
+            throw (IOException) thrown;
+          }
+          throw thrown;
         }
       }
 
@@ -930,13 +973,31 @@ public void finishBundle() throws Exception {
 
       @Teardown
       public void tearDown() throws Exception {
-
+        Exception thrown = null;
         if (table != null) {
-          table.close();
-          table = null;
+          try {
+            table.close();
+          } catch (Exception e) {
+            thrown = e;
+          } finally {
+            table = null;
+          }
+        }
+        try {
+          HBaseSharedConnection.close(configuration);
+        } catch (Exception e) {
+          if (thrown != null) {
+            thrown.addSuppressed(e);
+          } else {
+            thrown = e;
+          }
+        }
+        if (thrown != null) {
+          if (thrown instanceof IOException) {
+            throw (IOException) thrown;
+          }
+          throw thrown;
         }
-
-        HBaseSharedConnection.close(configuration);
       }
 
       @ProcessElement