This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a commit to branch v4.1
in repository https://gitbox.apache.org/repos/asf/flink-connector-jdbc.git
The following commit(s) were added to refs/heads/v4.1 by this push:
new f08f23cf [FLINK-39206][jdbc] Give each split reader its own connection
provider
f08f23cf is described below
commit f08f23cf1a223068720f1486409c7efb72b6b848
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.
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 +++++++++++++++++++++
4 files changed, 153 insertions(+), 16 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 427c9669..fafd5b37 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;
@@ -39,6 +40,7 @@ import
org.apache.flink.connector.jdbc.core.datastream.source.enumerator.splitte
import
org.apache.flink.connector.jdbc.core.datastream.source.enumerator.splitter.SplitterEnumerator;
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;
@@ -49,6 +51,7 @@ import org.apache.flink.core.io.SimpleVersionedSerializer;
import org.apache.flink.streaming.api.lineage.LineageDataset;
import org.apache.flink.streaming.api.lineage.LineageVertex;
import org.apache.flink.streaming.api.lineage.LineageVertexProvider;
+import org.apache.flink.util.InstantiationUtil;
import org.apache.flink.util.Preconditions;
import javax.annotation.Nullable;
@@ -59,6 +62,7 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.Objects;
import java.util.Optional;
+import java.util.function.Supplier;
/** JDBC source. */
@PublicEvolving
@@ -120,16 +124,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 58928c26..ac6218a0 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;
+ }
+}
