From 2953492438d4cd099061cf76f314ca766d491a8a Mon Sep 17 00:00:00 2001 From: ffccites <99155080+PDGGK@users.noreply.github.com> Date: Tue, 11 Aug 2026 11:29:54 +1000 Subject: [PATCH 1/2] [core] Restore the interrupt status in the two JDBC catalog paths that drop it The jdbc package turns an InterruptedException into an unchecked exception in nine places. Seven re-assert the flag before rethrowing; two do not, so a thread that gets cancelled inside them comes back out looking un-cancelled and every later blocking call on it behaves as if nothing happened. JdbcCatalog:131 the constructor's initializeCatalogTablesIfNeed() call JdbcUtils:677 insertTable Both now follow the shape already used next to them: the single-catch site mirrors JdbcCatalog:1018, and the multi-catch site mirrors JdbcCatalog:1047, which keeps its instanceof guard. No exception type or message changes. JdbcInterruptStatusTest covers both. The catalog constructor needs no mocking: ClientPoolImpl.run waits on LinkedBlockingDeque.pollFirst, whose lockInterruptibly() throws as soon as it sees a thread that already carries the flag, so setting it first drives the real code down its real interrupt path. Both cases fail on master at the interrupt-status assertion. --- .../org/apache/paimon/jdbc/JdbcCatalog.java | 1 + .../org/apache/paimon/jdbc/JdbcUtils.java | 3 + .../paimon/jdbc/JdbcInterruptStatusTest.java | 104 ++++++++++++++++++ 3 files changed, 108 insertions(+) create mode 100644 paimon-core/src/test/java/org/apache/paimon/jdbc/JdbcInterruptStatusTest.java diff --git a/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalog.java b/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalog.java index bdcd2453effa..484ac3803bca 100644 --- a/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalog.java +++ b/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcCatalog.java @@ -129,6 +129,7 @@ protected JdbcCatalog( } catch (SQLException e) { throw new RuntimeException("Cannot initialize JDBC catalog", e); } catch (InterruptedException e) { + Thread.currentThread().interrupt(); throw new RuntimeException("Interrupted in call to initialize", e); } } diff --git a/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcUtils.java b/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcUtils.java index 79ca0db1b248..17e273d05e59 100644 --- a/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcUtils.java +++ b/paimon-core/src/main/java/org/apache/paimon/jdbc/JdbcUtils.java @@ -675,6 +675,9 @@ public static boolean insertTable( }); return insertRecord == 1; } catch (SQLException | InterruptedException e) { + if (e instanceof InterruptedException) { + Thread.currentThread().interrupt(); + } throw new RuntimeException("Failed to insert table: " + tableName, e); } } diff --git a/paimon-core/src/test/java/org/apache/paimon/jdbc/JdbcInterruptStatusTest.java b/paimon-core/src/test/java/org/apache/paimon/jdbc/JdbcInterruptStatusTest.java new file mode 100644 index 000000000000..b4a08d131c48 --- /dev/null +++ b/paimon-core/src/test/java/org/apache/paimon/jdbc/JdbcInterruptStatusTest.java @@ -0,0 +1,104 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.jdbc; + +import org.apache.paimon.catalog.CatalogContext; +import org.apache.paimon.fs.local.LocalFileIO; +import org.apache.paimon.options.CatalogOptions; +import org.apache.paimon.options.Options; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.nio.file.Path; +import java.util.HashMap; +import java.util.Map; +import java.util.UUID; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** + * The JDBC catalog turns an {@link InterruptedException} into an unchecked exception in nine + * places. Seven of them re-assert the interrupt before rethrowing; these tests cover the two that + * did not, so the thread does not silently come back out of them looking un-cancelled. + * + *

No mocking is needed for the catalog constructor: {@code ClientPoolImpl.run} waits on {@code + * LinkedBlockingDeque.pollFirst(10, SECONDS)}, whose {@code lockInterruptibly()} throws immediately + * when the calling thread already carries the flag. Setting the flag first is therefore enough to + * drive the real code down its real interrupt path. + */ +class JdbcInterruptStatusTest { + + @TempDir Path tempDir; + + @AfterEach + void clearInterruptFlag() { + // These tests deliberately leave the flag set; clear it so it cannot leak into whatever + // JUnit runs next on this thread. + Thread.interrupted(); + } + + @Test + void catalogConstructorKeepsTheInterruptStatus() { + Map properties = new HashMap<>(); + properties.put( + CatalogOptions.URI.key(), + "jdbc:sqlite:file:" + + UUID.randomUUID().toString().replace("-", "") + + "?mode=memory&cache=shared"); + properties.put(JdbcCatalog.PROPERTY_PREFIX + "username", "user"); + properties.put(JdbcCatalog.PROPERTY_PREFIX + "password", "password"); + properties.put(CatalogOptions.WAREHOUSE.key(), tempDir.toString()); + CatalogContext context = CatalogContext.create(Options.fromMap(properties)); + + Thread.currentThread().interrupt(); + + assertThatThrownBy( + () -> + new JdbcCatalog( + LocalFileIO.create(), + "interrupt-test-catalog", + context, + tempDir.toString())) + .isInstanceOf(RuntimeException.class) + .hasMessageContaining("Interrupted in call to initialize"); + + assertThat(Thread.currentThread().isInterrupted()).isTrue(); + } + + @Test + void insertTableKeepsTheInterruptStatus() throws Exception { + JdbcClientPool connections = mock(JdbcClientPool.class); + when(connections.run(any())).thenThrow(new InterruptedException("interrupted")); + + assertThatThrownBy( + () -> + JdbcUtils.insertTable( + connections, "catalog-key", "some_db", "some_table")) + .isInstanceOf(RuntimeException.class) + .hasMessageContaining("Failed to insert table: some_table"); + + assertThat(Thread.currentThread().isInterrupted()).isTrue(); + } +} From 394346d07cd6a5595a26de7b543d494d68d8d811 Mon Sep 17 00:00:00 2001 From: ffccites <99155080+PDGGK@users.noreply.github.com> Date: Tue, 11 Aug 2026 13:05:03 +1000 Subject: [PATCH 2/2] [core] Make the constructor interrupt test independent of the environment CI caught this: catalogConstructorKeepsTheInterruptStatus passed locally but failed on Linux with "Expecting code to raise a throwable". The first version set the thread's interrupt flag and relied on the real pool reaching LinkedBlockingDeque.pollFirst, whose lockInterruptibly() throws when the flag is already set. That assumed nothing between the flag and the wait consumes it -- but the constructor opens a real JDBC connection first, and driver initialisation apparently swallows the interrupt on Linux. So no exception, so no assertion. Both cases now stub JdbcClientPool.run to throw InterruptedException outright. For the constructor that means seeding CachedJdbcClientPool's shared cache through its existing @VisibleForTesting clientPools() accessor, the same seam CachedJdbcClientPoolTest already uses, so no real connection is opened at all and there is nothing left to be environment-dependent about. Both still fail against the unfixed catch blocks. --- .../paimon/jdbc/JdbcInterruptStatusTest.java | 65 ++++++++++++------- 1 file changed, 41 insertions(+), 24 deletions(-) diff --git a/paimon-core/src/test/java/org/apache/paimon/jdbc/JdbcInterruptStatusTest.java b/paimon-core/src/test/java/org/apache/paimon/jdbc/JdbcInterruptStatusTest.java index b4a08d131c48..cccdd1d99e21 100644 --- a/paimon-core/src/test/java/org/apache/paimon/jdbc/JdbcInterruptStatusTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/jdbc/JdbcInterruptStatusTest.java @@ -30,7 +30,6 @@ import java.nio.file.Path; import java.util.HashMap; import java.util.Map; -import java.util.UUID; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; @@ -43,43 +42,34 @@ * places. Seven of them re-assert the interrupt before rethrowing; these tests cover the two that * did not, so the thread does not silently come back out of them looking un-cancelled. * - *

No mocking is needed for the catalog constructor: {@code ClientPoolImpl.run} waits on {@code - * LinkedBlockingDeque.pollFirst(10, SECONDS)}, whose {@code lockInterruptibly()} throws immediately - * when the calling thread already carries the flag. Setting the flag first is therefore enough to - * drive the real code down its real interrupt path. + *

Both cases drive the interrupt through a stubbed {@link JdbcClientPool} rather than through a + * real one. For the constructor that means seeding {@link CachedJdbcClientPool}'s shared cache, so + * no real connection is ever opened and the interrupt cannot be consumed by driver initialisation + * before the code under test runs. */ class JdbcInterruptStatusTest { @TempDir Path tempDir; @AfterEach - void clearInterruptFlag() { + void tearDown() { + CachedJdbcClientPool.resetCache(); // These tests deliberately leave the flag set; clear it so it cannot leak into whatever // JUnit runs next on this thread. Thread.interrupted(); } @Test - void catalogConstructorKeepsTheInterruptStatus() { - Map properties = new HashMap<>(); - properties.put( - CatalogOptions.URI.key(), - "jdbc:sqlite:file:" - + UUID.randomUUID().toString().replace("-", "") - + "?mode=memory&cache=shared"); - properties.put(JdbcCatalog.PROPERTY_PREFIX + "username", "user"); - properties.put(JdbcCatalog.PROPERTY_PREFIX + "password", "password"); - properties.put(CatalogOptions.WAREHOUSE.key(), tempDir.toString()); - CatalogContext context = CatalogContext.create(Options.fromMap(properties)); - - Thread.currentThread().interrupt(); + void catalogConstructorKeepsTheInterruptStatus() throws Exception { + Options options = catalogOptions(); + seedPoolCache(options, interruptingPool()); assertThatThrownBy( () -> new JdbcCatalog( LocalFileIO.create(), "interrupt-test-catalog", - context, + CatalogContext.create(options), tempDir.toString())) .isInstanceOf(RuntimeException.class) .hasMessageContaining("Interrupted in call to initialize"); @@ -89,16 +79,43 @@ void catalogConstructorKeepsTheInterruptStatus() { @Test void insertTableKeepsTheInterruptStatus() throws Exception { - JdbcClientPool connections = mock(JdbcClientPool.class); - when(connections.run(any())).thenThrow(new InterruptedException("interrupted")); - assertThatThrownBy( () -> JdbcUtils.insertTable( - connections, "catalog-key", "some_db", "some_table")) + interruptingPool(), "catalog-key", "some_db", "some_table")) .isInstanceOf(RuntimeException.class) .hasMessageContaining("Failed to insert table: some_table"); assertThat(Thread.currentThread().isInterrupted()).isTrue(); } + + private static JdbcClientPool interruptingPool() throws Exception { + JdbcClientPool connections = mock(JdbcClientPool.class); + when(connections.run(any())).thenThrow(new InterruptedException("interrupted")); + return connections; + } + + private Options catalogOptions() { + Map properties = new HashMap<>(); + properties.put(CatalogOptions.URI.key(), "jdbc:sqlite:file:interrupt-test?mode=memory"); + properties.put(JdbcCatalog.PROPERTY_PREFIX + "username", "user"); + properties.put(JdbcCatalog.PROPERTY_PREFIX + "password", "password"); + properties.put(CatalogOptions.WAREHOUSE.key(), tempDir.toString()); + return Options.fromMap(properties); + } + + /** + * Mirrors how {@link CachedJdbcClientPool} derives its key, so {@code get()} finds this pool. + */ + private static void seedPoolCache(Options options, JdbcClientPool pool) { + CachedJdbcClientPool.clientPools() + .put( + CachedJdbcClientPool.Key.of( + options.get(CatalogOptions.URI), + options.get(JdbcCatalogOptions.CATALOG_KEY), + options.get(CatalogOptions.CLIENT_POOL_SIZE), + JdbcUtils.extractJdbcConfiguration( + options.toMap(), JdbcCatalog.PROPERTY_PREFIX)), + pool); + } }