This is an automated email from the ASF dual-hosted git repository.
RocMarshal pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-jdbc.git
The following commit(s) were added to refs/heads/main by this push:
new f8f84a49 [FLINK-39975][jdbc] Recover JdbcSourceSplitReader from a
connection closed during cancellation (#200)
f8f84a49 is described below
commit f8f84a4948e167611e11eaa09c0cc57b25b8cbf4
Author: laughingman7743 <[email protected]>
AuthorDate: Wed Jun 24 00:11:56 2026 +0900
[FLINK-39975][jdbc] Recover JdbcSourceSplitReader from a connection closed
during cancellation (#200)
---
.../source/reader/JdbcSourceSplitReader.java | 62 +++++++++++-
.../source/reader/JdbcSourceSplitReaderTest.java | 109 +++++++++++++++++++++
2 files changed, 170 insertions(+), 1 deletion(-)
diff --git
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/reader/JdbcSourceSplitReader.java
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/reader/JdbcSourceSplitReader.java
index 2ffca4bc..58928c26 100644
---
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/reader/JdbcSourceSplitReader.java
+++
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/reader/JdbcSourceSplitReader.java
@@ -67,6 +67,13 @@ public class JdbcSourceSplitReader<T>
private static final Logger LOG =
LoggerFactory.getLogger(JdbcSourceSplitReader.class);
+ /**
+ * Maximum number of times a split is re-opened after the connection was
found to be closed
+ * (e.g. torn down concurrently while the source is being cancelled).
Mirrors the sink default
+ * {@code JdbcExecutionOptions.DEFAULT_MAX_RETRY_TIMES}.
+ */
+ private static final int MAX_CONNECTION_RETRIES = 3;
+
private final Configuration config;
@Nullable private JdbcSourceSplit currentSplit;
private final Queue<JdbcSourceSplit> splits;
@@ -259,7 +266,7 @@ public class JdbcSourceSplitReader<T>
final JdbcSourceSplit nextSplit = splits.poll();
if (nextSplit != null) {
currentSplit = nextSplit;
- openResultSetForSplit(currentSplit);
+ openResultSetForSplitWithReconnect(currentSplit);
return true;
}
return false;
@@ -268,6 +275,59 @@ public class JdbcSourceSplitReader<T>
}
}
+ /**
+ * Opens the result set for the split, re-establishing the connection and
retrying if the
+ * connection was closed underneath us.
+ *
+ * <p>When a bounded query (e.g. {@code LIMIT}) produces enough rows, the
job finishes and the
+ * source is cancelled while this reader may still be opening the next
split. The connection
+ * that the provider validated can then be torn down by the time we
prepare/execute the
+ * statement, surfacing as a closed-connection error (e.g. Derby {@code
08003 "No current
+ * connection"}). That should not fail the whole job, so we re-establish
the connection and
+ * retry, mirroring the retry/reconnect handling in {@code
JdbcOutputFormat#flush()}. We only
+ * retry when the connection is no longer valid; a genuine query error on
a healthy connection
+ * is rethrown immediately, and a truly unreachable database makes
re-establishing fail so real
+ * errors still propagate once the retry budget is exhausted.
+ */
+ private void openResultSetForSplitWithReconnect(JdbcSourceSplit split)
+ throws SQLException, ClassNotFoundException {
+ for (int attempt = 0; attempt <= MAX_CONNECTION_RETRIES; attempt++) {
+ try {
+ openResultSetForSplit(split);
+ return;
+ } catch (SQLException e) {
+ // Only retry when the connection itself is no longer valid
(e.g. torn down while
+ // the source is being cancelled). A genuine query error on a
healthy connection is
+ // rethrown immediately, and a truly unreachable database
keeps failing the re-open,
+ // so real errors still propagate once the retry budget is
exhausted.
+ if (attempt >= MAX_CONNECTION_RETRIES ||
connectionStillValid()) {
+ throw e;
+ }
+ LOG.warn(
+ "Connection was closed while opening split {} (attempt
{}/{}); "
+ + "re-establishing the connection and
retrying.",
+ split.splitId(),
+ attempt + 1,
+ MAX_CONNECTION_RETRIES,
+ e);
+ // Drop the dead connection together with the statement/result
set left on it so
+ // the retried open re-establishes cleanly, without running
the close helpers
+ // (which rethrow as unchecked exceptions) against the closed
connection.
+ resultSet = null;
+ statement = null;
+ connectionProvider.closeConnection();
+ }
+ }
+ }
+
+ private boolean connectionStillValid() {
+ try {
+ return connectionProvider.isConnectionValid();
+ } catch (SQLException ignored) {
+ return false;
+ }
+ }
+
private void getOrEstablishConnection() throws SQLException,
ClassNotFoundException {
connection = connectionProvider.getOrEstablishConnection();
if (autoCommit == null) {
diff --git
a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/source/reader/JdbcSourceSplitReaderTest.java
b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/source/reader/JdbcSourceSplitReaderTest.java
index b28238f9..a81630f2 100644
---
a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/source/reader/JdbcSourceSplitReaderTest.java
+++
b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/source/reader/JdbcSourceSplitReaderTest.java
@@ -34,6 +34,8 @@ import
org.apache.flink.connector.testutils.source.reader.TestingReaderContext;
import org.junit.jupiter.api.Test;
+import java.sql.Connection;
+import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
@@ -144,4 +146,111 @@ class JdbcSourceSplitReaderTest extends JdbcDataTestBase {
assertThat(fetchedRecordsWithSplitIds.nextSplit()).isNull();
splitReader.close();
}
+
+ @Test
+ void testFetchReconnectsWhenConnectionClosedWhileOpeningSplit() throws
Exception {
+ // The provider hands back an already-closed connection the first
time, so the reader's
+ // first use of it fails (mimicking the connection being torn down,
e.g. on source
+ // cancellation, before the reader finishes opening the split). The
reader must
+ // re-establish the connection and read the whole split.
+ CountingConnectionProvider provider = new
CountingConnectionProvider(connectionProvider, 1);
+ try (JdbcSourceSplitReader<TestEntry> splitReader =
newReader(provider, split)) {
+ RecordsWithSplitIds<RecordAndOffset<TestEntry>> fetched =
splitReader.fetch();
+ assertThat(fetched.nextSplit()).isEqualTo("1");
+ List<TestEntry> records = new ArrayList<>();
+ RecordAndOffset<TestEntry> recordAndOffset =
fetched.nextRecordFromSplit();
+ while (recordAndOffset != null) {
+ records.add(recordAndOffset.record);
+ recordAndOffset = fetched.nextRecordFromSplit();
+ }
+ assertThat(records).hasSize(TEST_DATA.length);
+ // The first (closed) connection plus exactly one re-established
one: a reconnect ran.
+ assertThat(provider.establishCount).isEqualTo(2);
+ }
+ }
+
+ @Test
+ void testFetchRethrowsImmediatelyWhenConnectionStaysOpen() throws
Exception {
+ // A query error on a healthy (open) connection must be rethrown
immediately, with no
+ // reconnect attempt.
+ CountingConnectionProvider provider = new
CountingConnectionProvider(connectionProvider, 0);
+ JdbcSourceSplit invalidSplit =
+ new JdbcSourceSplit("1", "select * from NON_EXISTENT_TABLE",
null, null);
+ try (JdbcSourceSplitReader<TestEntry> splitReader =
newReader(provider, invalidSplit)) {
+
assertThatThrownBy(splitReader::fetch).isInstanceOf(RuntimeException.class);
+ assertThat(provider.establishCount).isEqualTo(1);
+ }
+ }
+
+ @Test
+ void testFetchFailsAfterExhaustingReconnectRetries() throws Exception {
+ // If the connection keeps being closed, the reader gives up after the
retry budget
+ // instead of looping forever.
+ CountingConnectionProvider provider =
+ new CountingConnectionProvider(connectionProvider,
Integer.MAX_VALUE);
+ try (JdbcSourceSplitReader<TestEntry> splitReader =
newReader(provider, split)) {
+
assertThatThrownBy(splitReader::fetch).isInstanceOf(RuntimeException.class);
+ // The initial attempt plus MAX_CONNECTION_RETRIES (3) reconnect
attempts.
+ assertThat(provider.establishCount).isEqualTo(4);
+ }
+ }
+
+ private JdbcSourceSplitReader<TestEntry> newReader(
+ JdbcConnectionProvider provider, JdbcSourceSplit sourceSplit) {
+ JdbcSourceSplitReader<TestEntry> reader =
+ new JdbcSourceSplitReader<>(
+ new TestingReaderContext(),
+ new Configuration(),
+ TypeInformation.of(TestEntry.class),
+ provider,
+ DeliveryGuarantee.NONE,
+ extractor);
+ reader.handleSplitsChanges(new
SplitsAddition<>(Collections.singletonList(sourceSplit)));
+ return reader;
+ }
+
+ /**
+ * A {@link JdbcConnectionProvider} that closes the first {@code
closeFirstN} connections it
+ * hands out (returning them already closed) to deterministically
reproduce a connection torn
+ * down underneath the reader, while counting how many connections were
established.
+ */
+ private static class CountingConnectionProvider implements
JdbcConnectionProvider {
+ private final JdbcConnectionProvider delegate;
+ private final int closeFirstN;
+ private int establishCount = 0;
+
+ CountingConnectionProvider(JdbcConnectionProvider delegate, int
closeFirstN) {
+ this.delegate = delegate;
+ this.closeFirstN = closeFirstN;
+ }
+
+ @Override
+ public Connection getOrEstablishConnection() throws SQLException,
ClassNotFoundException {
+ Connection connection = delegate.getOrEstablishConnection();
+ if (++establishCount <= closeFirstN) {
+ connection.close();
+ }
+ return connection;
+ }
+
+ @Override
+ public Connection getConnection() {
+ return delegate.getConnection();
+ }
+
+ @Override
+ public boolean isConnectionValid() throws SQLException {
+ return delegate.isConnectionValid();
+ }
+
+ @Override
+ public void closeConnection() {
+ delegate.closeConnection();
+ }
+
+ @Override
+ public Connection reestablishConnection() throws SQLException,
ClassNotFoundException {
+ return delegate.reestablishConnection();
+ }
+ }
}
