[Bug]: HBaseIO skips connection cleanup when an earlier close() throws, permanently corrupting the shared-connection refcount

[PDGGK]

What happened?

HBaseIO releases its resources as consecutive, unguarded statements in three places. If an earlier close() throws, everything after it is skipped. All line numbers are against master today.

1. HBaseIO.java:550-560 — HBaseReader.close()

if (scanner != null) { scanner.close(); scanner = null; }
if (connection != null) { connection.close(); connection = null; }

A throwing scanner.close() leaks the Connection.

2. HBaseIO.java:766-776 — HBaseWriterFn.tearDown()

if (mutator != null) { mutator.close(); mutator = null; }
if (connection != null) { connection.close(); connection = null; }

This is the sharpest of the three, because mutator.close() is expected to be able to throw. BufferedMutator.close() is documented as:

Performs a flush() and releases any resources held.
@throws IOException if a remote or network exception occurs.

So any bundle whose final buffered flush fails — a down region server, a network blip — skips connection.close() and leaks the whole Connection. The connection here is created per-@Setup via ConnectionFactory.createConnection (:744), so this is a real per-worker leak, not a shared handle.

3. HBaseIO.java:931-940 — HBaseRowMutationWriterFn.tearDown() — worst consequence

if (table != null) { table.close(); table = null; }
HBaseSharedConnection.close(configuration);

Here the skipped call is not an ordinary close(), it is a reference-count decrement. HBaseSharedConnection keeps a static HashMap<String, Pair<Connection, Integer>> connectionPool (HBaseSharedConnection.java:46); getOrCreate increments the count (:80) and close(Configuration) decrements it, closing the underlying Connection only once it reaches zero (:119, :128-129).

So a throwing table.close() means the count is never decremented. Because the map is static, that entry — and its ZooKeeper session and RPC threads — then survives for the lifetime of the JVM, and every later getOrCreate hands back the same permanently unreleasable connection. A leak here is not bounded by the bundle or even the worker's use of the sink.

What I'd propose

Make each teardown run every step unconditionally while preserving the first failure, attaching later ones as suppressed, so the caller still sees the error that actually broke the job rather than a teardown symptom. A plain nested try/finally would guarantee the calls happen but would silently swap which exception propagates, so it is only half a fix.

I'm happy to submit the PR if this looks right.

Issue Priority

Priority: 2 (default / most bugs should be filed as P2)

Issue Components

• Component: Java SDK
• Component: IO connector

[Dheeraj-1103]

I’d like to work on this if it’s still open for contribution. I’ve gone through all three call sites and here’s the fix I’d propose:

Approach: Replace the consecutive unguarded cleanup statements with a pattern that attempts every cleanup step even if an earlier one throws, while preserving the first exception as the primary failure and attaching subsequent failures via Throwable.addSuppressed(). A plain nested try/finally would guarantee cleanup, but could change which exception is ultimately propagated.

1. HBaseReader.close() (HBaseIO.java)

• Ensure scanner.close() is attempted.
• If it throws, still attempt connection.close().
• Preserve the scanner exception as the primary failure and suppress any connection-close exception.
• Clear the resource references regardless of whether close() succeeds.

2. HBaseWriterFn.tearDown()

• Apply the same pattern to mutator.close() followed by connection.close().
• This is particularly important because BufferedMutator.close() can legitimately throw when its final flush fails.

3. HBaseRowMutationWriterFn.tearDown()

• Ensure HBaseSharedConnection.close(configuration) executes even if table.close() throws.
• This is the most important case because skipping the shared-connection release can leave the static connection-pool reference count permanently elevated.

Tests I’d add:

• First cleanup step throws → second cleanup step is still executed.
• Both cleanup steps throw → first exception is propagated and the second appears in getSuppressed().
• Normal successful cleanup remains unchanged.
• For HBaseRowMutationWriterFn, verify that the shared connection is released/refcount decremented even when table.close() fails.

I noticed that #39711 and #39712 are already linked to this issue. Could a maintainer confirm their current status before I start? If either PR already covers the implementation and additional tests/review would be more useful, I’m happy to contribute there instead of duplicating the work.

[PDGGK]

@Dheeraj-1103 thanks for the interest, and sorry — this one is already taken twice over. Two PRs are open against it: #39711 by @waterWang (first, 02:20Z) and #39712 (mine, 02:30Z). Your outline matches what both of them already do, so anything you build now would be a third duplicate.

If you're looking for something in this area that genuinely is unclaimed, the same bug class — consecutive unguarded close() calls in a @Teardown or close() — tends to recur across the sdks/java/io/* connectors. Grepping for teardown methods that call close() on two or more resources in sequence is a decent way to find one, and the fix shape is the same: run every step, keep the first failure, attach the rest with addSuppressed. Worth filing an issue with the specific file and line before writing code, so this doesn't happen again.