This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a commit to branch v3.4
in repository https://gitbox.apache.org/repos/asf/flink-connector-jdbc.git
The following commit(s) were added to refs/heads/v3.4 by this push:
new 93debc91 [FLINK-39206][jdbc] Give each split reader its own connection
provider
93debc91 is described below
commit 93debc91b1d6471dfd5c9022e518ef81729ab3e1
Author: Martijn Visser <[email protected]>
AuthorDate: Mon Sep 21 09:18:30 2026 +0200
[FLINK-39206][jdbc] Give each split reader its own connection provider
JdbcSource handed the same JdbcConnectionProvider to every split reader it
created, while flink-connector-base builds a split reader per split fetcher
and
closes the reader of a finished split while the reader of the next split is
already reading. A provider holds a single connection and is not thread
safe,
so the finished reader closed the running reader's result set and the job
failed instead of finishing.
Copy the provider per split reader, the way other connectors give each split
reader its own client. A reader now owns its copy, so it closes the result
set,
the connection and the provider together, and the interface says that it is
copied.
The deprecated org.apache.flink.connector.jdbc.source.JdbcSource builds the
same
split readers, so it copies the provider too.
Generated-by: Claude Code (Opus 5)
---
.../jdbc/core/datastream/source/JdbcSource.java | 35 +++++--
.../source/reader/JdbcSourceSplitReader.java | 14 ++-
.../connections/JdbcConnectionProvider.java | 9 +-
.../core/datastream/source/JdbcSourceTest.java | 111 +++++++++++++++++++++
.../flink/connector/jdbc/source/JdbcSource.java | 5 +-
5 files changed, 157 insertions(+), 17 deletions(-)
diff --git
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/JdbcSource.java
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/JdbcSource.java
index 8efb732e..0822ecc6 100644
---
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/JdbcSource.java
+++
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/core/datastream/source/JdbcSource.java
@@ -30,6 +30,7 @@ import
org.apache.flink.api.connector.source.SplitEnumeratorContext;
import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.base.DeliveryGuarantee;
+import org.apache.flink.connector.base.source.reader.splitreader.SplitReader;
import
org.apache.flink.connector.jdbc.core.datastream.source.config.ContinuousUnBoundingSettings;
import
org.apache.flink.connector.jdbc.core.datastream.source.enumerator.JdbcSourceEnumStateSerializer;
import
org.apache.flink.connector.jdbc.core.datastream.source.enumerator.JdbcSourceEnumerator;
@@ -37,11 +38,13 @@ import
org.apache.flink.connector.jdbc.core.datastream.source.enumerator.JdbcSou
import
org.apache.flink.connector.jdbc.core.datastream.source.enumerator.JdbcSqlSplitEnumeratorBase;
import
org.apache.flink.connector.jdbc.core.datastream.source.reader.JdbcSourceReader;
import
org.apache.flink.connector.jdbc.core.datastream.source.reader.JdbcSourceSplitReader;
+import
org.apache.flink.connector.jdbc.core.datastream.source.reader.RecordAndOffset;
import
org.apache.flink.connector.jdbc.core.datastream.source.reader.extractor.ResultExtractor;
import
org.apache.flink.connector.jdbc.core.datastream.source.split.JdbcSourceSplit;
import
org.apache.flink.connector.jdbc.core.datastream.source.split.JdbcSourceSplitSerializer;
import
org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider;
import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.util.InstantiationUtil;
import org.apache.flink.util.Preconditions;
import javax.annotation.Nullable;
@@ -49,6 +52,7 @@ import javax.annotation.Nullable;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Objects;
+import java.util.function.Supplier;
/** JDBC source. */
@PublicEvolving
@@ -98,16 +102,27 @@ public class JdbcSource<OUT>
public SourceReader<OUT, JdbcSourceSplit> createReader(SourceReaderContext
readerContext)
throws Exception {
return new JdbcSourceReader<>(
- () ->
- new JdbcSourceSplitReader<>(
- readerContext,
- configuration,
- typeInformation,
- connectionProvider,
- deliveryGuarantee,
- resultExtractor),
- configuration,
- readerContext);
+ splitReaderSupplier(readerContext), configuration,
readerContext);
+ }
+
+ /**
+ * Supplies the split readers of one {@link JdbcSourceReader}. Every
reader gets its own
+ * connection provider: a provider holds a single connection and is not
thread safe, while
+ * flink-connector-base builds a split reader per split fetcher and closes
the reader of a
+ * finished split while the reader of the next split is already reading.
Sharing one provider
+ * lets the first close the connection out from under the second.
+ */
+ @VisibleForTesting
+ Supplier<SplitReader<RecordAndOffset<OUT>, JdbcSourceSplit>>
splitReaderSupplier(
+ SourceReaderContext readerContext) {
+ return () ->
+ new JdbcSourceSplitReader<>(
+ readerContext,
+ configuration,
+ typeInformation,
+ InstantiationUtil.cloneUnchecked(connectionProvider),
+ deliveryGuarantee,
+ resultExtractor);
}
@Override
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 e9b1ca5d..2fa7733a 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
@@ -32,6 +32,7 @@ import
org.apache.flink.connector.base.source.reader.splitreader.SplitsChange;
import
org.apache.flink.connector.jdbc.core.datastream.source.reader.extractor.ResultExtractor;
import
org.apache.flink.connector.jdbc.core.datastream.source.split.JdbcSourceSplit;
import
org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider;
+import org.apache.flink.util.IOUtils;
import org.apache.flink.util.Preconditions;
import org.slf4j.Logger;
@@ -221,12 +222,15 @@ public class JdbcSourceSplitReader<T>
@Override
public void close() throws Exception {
- closeResultSetAndStatement();
- if (connection != null && !connection.isClosed()) {
- connection.close();
+ try {
+ // The provider is this reader's own copy, so it is closed here
too: that runs the
+ // teardown of a user-supplied provider, and covers the connection
if the provider
+ // hands out one it does not track itself.
+ IOUtils.closeAll(this::closeResultSetAndStatement, connection,
connectionProvider);
+ } finally {
+ connection = null;
+ currentSplit = null;
}
- connection = null;
- currentSplit = null;
}
@Override
diff --git
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/datasource/connections/JdbcConnectionProvider.java
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/datasource/connections/JdbcConnectionProvider.java
index 81c6bccb..b43b9e3e 100644
---
a/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/datasource/connections/JdbcConnectionProvider.java
+++
b/flink-connector-jdbc-core/src/main/java/org/apache/flink/connector/jdbc/datasource/connections/JdbcConnectionProvider.java
@@ -27,7 +27,14 @@ import java.sql.Connection;
import java.sql.SQLException;
import java.util.Properties;
-/** JDBC connection provider. */
+/**
+ * Provides the connection a JDBC source or sink works on.
+ *
+ * <p>A source copies the provider once per split reader, by Java
serialization, so an
+ * implementation that wraps a pool or caches credentials gets one independent
copy per reader
+ * rather than one per subtask. Keep anything that must not be duplicated out
of the serialized
+ * state.
+ */
@PublicEvolving
public interface JdbcConnectionProvider extends Serializable, AutoCloseable {
/**
diff --git
a/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/source/JdbcSourceTest.java
b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/source/JdbcSourceTest.java
new file mode 100644
index 00000000..d33faa4d
--- /dev/null
+++
b/flink-connector-jdbc-core/src/test/java/org/apache/flink/connector/jdbc/core/datastream/source/JdbcSourceTest.java
@@ -0,0 +1,111 @@
+/*
+ * 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.flink.connector.jdbc.core.datastream.source;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
+import org.apache.flink.connector.base.source.reader.splitreader.SplitReader;
+import
org.apache.flink.connector.base.source.reader.splitreader.SplitsAddition;
+import org.apache.flink.connector.jdbc.JdbcDataTestBase;
+import
org.apache.flink.connector.jdbc.core.datastream.source.reader.JdbcSourceSplitReader;
+import
org.apache.flink.connector.jdbc.core.datastream.source.reader.RecordAndOffset;
+import
org.apache.flink.connector.jdbc.core.datastream.source.split.JdbcSourceSplit;
+import org.apache.flink.connector.testutils.source.reader.TestingReaderContext;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.function.Supplier;
+
+import static org.apache.flink.connector.jdbc.JdbcTestFixture.INPUT_TABLE;
+import static org.apache.flink.connector.jdbc.JdbcTestFixture.TEST_DATA;
+import static org.apache.flink.connector.jdbc.JdbcTestFixture.TestEntry;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link JdbcSource}. */
+class JdbcSourceTest extends JdbcDataTestBase {
+
+ private static final String SQL = "SELECT id, title, author, price, qty
FROM " + INPUT_TABLE;
+
+ @Test
+ void testSplitReadersDoNotShareAConnection() throws Exception {
+ // flink-connector-base builds a split reader per split fetcher and
closes the reader of a
+ // finished split while the reader of the next split is already
reading. A connection
+ // provider holds a single connection and is not thread safe, so every
split reader needs
+ // its own: sharing one lets the finished reader close the running
reader's result set.
+ JdbcSource<TestEntry> source =
+ JdbcSource.<TestEntry>builder()
+
.setTypeInformation(TypeInformation.of(TestEntry.class))
+ .setSql(SQL)
+ .setDBUrl(getMetadata().getJdbcUrl())
+ .setDriverName(getMetadata().getDriverClass())
+ .setResultExtractor(extractor)
+ // One record per fetch, so the running reader still
has an open result set
+ // when the finished one is closed.
+ .setSplitReaderFetchBatchSize(1)
+ .build();
+
+ Supplier<SplitReader<RecordAndOffset<TestEntry>, JdbcSourceSplit>>
supplier =
+ source.splitReaderSupplier(new TestingReaderContext());
+
+ JdbcSourceSplitReader<TestEntry> finishedSplitReader =
newSplitReader(supplier, "1");
+ JdbcSourceSplitReader<TestEntry> runningSplitReader =
newSplitReader(supplier, "2");
+
+ // The first fetch on each opens its result set, which is what
establishes its connection.
+ drain(finishedSplitReader.fetch());
+ List<TestEntry> records = new
ArrayList<>(drain(runningSplitReader.fetch()));
+
+ finishedSplitReader.close();
+
+ // The finished reader is gone; the running one must still read its
split to the end.
+ // One record per fetch, so the remaining count is exactly the number
of fetches left.
+ for (int i = records.size(); i < TEST_DATA.length; i++) {
+ records.addAll(drain(runningSplitReader.fetch()));
+ }
+ runningSplitReader.close();
+ assertThat(records).containsExactlyInAnyOrder(TEST_DATA);
+ }
+
+ @SuppressWarnings("unchecked")
+ private static JdbcSourceSplitReader<TestEntry> newSplitReader(
+ Supplier<SplitReader<RecordAndOffset<TestEntry>, JdbcSourceSplit>>
supplier,
+ String splitId) {
+ JdbcSourceSplitReader<TestEntry> splitReader =
+ (JdbcSourceSplitReader<TestEntry>) supplier.get();
+ splitReader.handleSplitsChanges(
+ new SplitsAddition<>(
+ Collections.singletonList(new JdbcSourceSplit(splitId,
SQL, null, null))));
+ return splitReader;
+ }
+
+ private static List<TestEntry> drain(
+ RecordsWithSplitIds<RecordAndOffset<TestEntry>> fetchedRecords) {
+ List<TestEntry> records = new ArrayList<>();
+ while (fetchedRecords.nextSplit() != null) {
+ RecordAndOffset<TestEntry> recordAndOffset =
fetchedRecords.nextRecordFromSplit();
+ while (recordAndOffset != null) {
+ records.add(recordAndOffset.getRecord());
+ recordAndOffset = fetchedRecords.nextRecordFromSplit();
+ }
+ }
+ return records;
+ }
+}
diff --git
a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/source/JdbcSource.java
b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/source/JdbcSource.java
index cee58e75..6cb936a8 100644
---
a/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/source/JdbcSource.java
+++
b/flink-connector-jdbc/src/main/java/org/apache/flink/connector/jdbc/source/JdbcSource.java
@@ -42,6 +42,7 @@ import
org.apache.flink.connector.jdbc.core.datastream.source.split.JdbcSourceSp
import
org.apache.flink.connector.jdbc.datasource.connections.JdbcConnectionProvider;
import org.apache.flink.connector.jdbc.utils.ContinuousUnBoundingSettings;
import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.util.InstantiationUtil;
import org.apache.flink.util.Preconditions;
import javax.annotation.Nullable;
@@ -108,7 +109,9 @@ public class JdbcSource<OUT>
readerContext,
configuration,
typeInformation,
- connectionProvider,
+ // A provider holds a single connection and is
not thread safe, so
+ // every split reader gets its own copy, as in
the core JdbcSource.
+
InstantiationUtil.cloneUnchecked(connectionProvider),
deliveryGuarantee,
resultExtractor),
configuration,
