This is an automated email from the ASF dual-hosted git repository.
loserwang1024 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new f183bbdf9 [FLINK-40623][cdc-base] Release chunk splitter JDBC
connection once a table is split (#4534)
f183bbdf9 is described below
commit f183bbdf9732617a8f7cd75caa848e8c1cdc612f
Author: xucq07 <[email protected]>
AuthorDate: Mon Sep 21 10:43:29 2026 +0800
[FLINK-40623][cdc-base] Release chunk splitter JDBC connection once a table
is split (#4534)
---
.../assigner/splitter/JdbcSourceChunkSplitter.java | 43 ++++----
.../splitter/JdbcSourceChunkSplitterTest.java | 117 ++++++++++++++++++++-
2 files changed, 134 insertions(+), 26 deletions(-)
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/assigner/splitter/JdbcSourceChunkSplitter.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/assigner/splitter/JdbcSourceChunkSplitter.java
index 8bbc93588..56882e1c6 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/assigner/splitter/JdbcSourceChunkSplitter.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/assigner/splitter/JdbcSourceChunkSplitter.java
@@ -67,7 +67,6 @@ public abstract class JdbcSourceChunkSplitter implements
ChunkSplitter {
@Nullable private ChunkSplitterState.ChunkBound nextChunkStart;
@Nullable private Integer nextChunkId;
- private JdbcConnection jdbcConnection;
private Table currentSplittingTable;
private TableChanges.TableChange currentSchema;
private Column splitColumn;
@@ -101,16 +100,21 @@ public abstract class JdbcSourceChunkSplitter implements
ChunkSplitter {
}
@Override
- public void open() {
- this.jdbcConnection = dialect.openJdbcConnection(sourceConfig);
- }
+ public void open() {}
/** Generates all snapshot splits (chunks) for the give table path. */
@Override
public Collection<SnapshotSplit> generateSplits(TableId tableId) throws
Exception {
+ try (JdbcConnection jdbc = dialect.openJdbcConnection(sourceConfig)) {
+ return generateSplits(tableId, jdbc);
+ }
+ }
+
+ private Collection<SnapshotSplit> generateSplits(TableId tableId,
JdbcConnection jdbc)
+ throws Exception {
if (!hasNextChunk()) {
// split a new table.
- analyzeTable(tableId);
+ analyzeTable(tableId, jdbc);
Optional<List<SnapshotSplit>> evenlySplitChunks =
trySplitAllEvenlySizedChunks(tableId);
if (evenlySplitChunks.isPresent()) {
return evenlySplitChunks.get();
@@ -119,7 +123,7 @@ public abstract class JdbcSourceChunkSplitter implements
ChunkSplitter {
this.currentSplittingTableId = tableId;
this.nextChunkStart =
ChunkSplitterState.ChunkBound.START_BOUND;
this.nextChunkId = 0;
- return
Collections.singletonList(splitOneUnevenlySizedChunk(tableId));
+ return
Collections.singletonList(splitOneUnevenlySizedChunk(tableId, jdbc));
}
}
} else {
@@ -127,10 +131,10 @@ public abstract class JdbcSourceChunkSplitter implements
ChunkSplitter {
currentSplittingTableId.equals(tableId),
"Can not split a new table before the previous table
splitting finish.");
if (currentSplittingTable == null) {
- analyzeTable(currentSplittingTableId);
+ analyzeTable(currentSplittingTableId, jdbc);
}
synchronized (lock) {
- return
Collections.singletonList(splitOneUnevenlySizedChunk(tableId));
+ return
Collections.singletonList(splitOneUnevenlySizedChunk(tableId, jdbc));
}
}
}
@@ -154,11 +158,7 @@ public abstract class JdbcSourceChunkSplitter implements
ChunkSplitter {
}
@Override
- public void close() throws Exception {
- if (jdbcConnection != null) {
- jdbcConnection.close();
- }
- }
+ public void close() throws Exception {}
/**
* Query the maximum value of the next chunk, and the next chunk must be
greater than or equal
@@ -357,21 +357,22 @@ public abstract class JdbcSourceChunkSplitter implements
ChunkSplitter {
}
/** Analyze the meta information for given table. */
- private void analyzeTable(TableId tableId) {
+ private void analyzeTable(TableId tableId, JdbcConnection jdbc) {
try {
- currentSchema = dialect.queryTableSchema(jdbcConnection, tableId);
+ currentSchema = dialect.queryTableSchema(jdbc, tableId);
currentSplittingTable =
Objects.requireNonNull(currentSchema).getTable();
splitColumn = getSplitColumn(currentSplittingTable,
sourceConfig.getChunkKeyColumn());
splitType = getSplitType(splitColumn);
- minMaxOfSplitColumn = queryMinMax(jdbcConnection, tableId,
splitColumn);
- approximateRowCnt = queryApproximateRowCnt(jdbcConnection,
tableId);
+ minMaxOfSplitColumn = queryMinMax(jdbc, tableId, splitColumn);
+ approximateRowCnt = queryApproximateRowCnt(jdbc, tableId);
} catch (Exception e) {
throw new RuntimeException("Fail to analyze table in chunk
splitter.", e);
}
}
/** Generates one snapshot split (chunk) for the give table path. */
- private SnapshotSplit splitOneUnevenlySizedChunk(TableId tableId) throws
SQLException {
+ private SnapshotSplit splitOneUnevenlySizedChunk(TableId tableId,
JdbcConnection jdbc)
+ throws SQLException {
final int chunkSize = sourceConfig.getSplitSize();
final Object chunkStartVal = nextChunkStart.getValue();
LOG.info(
@@ -384,7 +385,7 @@ public abstract class JdbcSourceChunkSplitter implements
ChunkSplitter {
// we start from [null, min + chunk_size) and avoid [null, min)
Object chunkEnd =
nextChunkEnd(
- jdbcConnection,
+ jdbc,
nextChunkStart ==
ChunkSplitterState.ChunkBound.START_BOUND
? minMaxOfSplitColumn[0]
: chunkStartVal,
@@ -395,7 +396,7 @@ public abstract class JdbcSourceChunkSplitter implements
ChunkSplitter {
// may sleep a while to avoid DDOS on MySQL server
maySleep(nextChunkId, tableId);
if (chunkEnd != null
- && isChunkEndLeMax(jdbcConnection, chunkEnd,
minMaxOfSplitColumn[1], splitColumn)) {
+ && isChunkEndLeMax(jdbc, chunkEnd, minMaxOfSplitColumn[1],
splitColumn)) {
nextChunkStart = ChunkSplitterState.ChunkBound.middleOf(chunkEnd);
return createSnapshotSplit(tableId, nextChunkId++, splitType,
chunkStartVal, chunkEnd);
} else {
@@ -495,7 +496,7 @@ public abstract class JdbcSourceChunkSplitter implements
ChunkSplitter {
Object chunkStart = null;
Object chunkEnd = nextChunkEnd(jdbc, min, tableId, splitColumn, max,
chunkSize);
int count = 0;
- while (chunkEnd != null && isChunkEndLeMax(jdbcConnection, chunkEnd,
max, splitColumn)) {
+ while (chunkEnd != null && isChunkEndLeMax(jdbc, chunkEnd, max,
splitColumn)) {
// we start from [null, min + chunk_size) and avoid [null, min)
splits.add(ChunkRange.of(chunkStart, chunkEnd));
// may sleep a while to avoid DDOS on PostgreSQL server
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/test/java/org/apache/flink/cdc/connectors/base/source/assigner/splitter/JdbcSourceChunkSplitterTest.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/test/java/org/apache/flink/cdc/connectors/base/source/assigner/splitter/JdbcSourceChunkSplitterTest.java
index a52edc636..3ff46acb6 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/test/java/org/apache/flink/cdc/connectors/base/source/assigner/splitter/JdbcSourceChunkSplitterTest.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/test/java/org/apache/flink/cdc/connectors/base/source/assigner/splitter/JdbcSourceChunkSplitterTest.java
@@ -17,21 +17,31 @@
package org.apache.flink.cdc.connectors.base.source.assigner.splitter;
+import org.apache.flink.cdc.connectors.base.config.JdbcSourceConfig;
+import org.apache.flink.cdc.connectors.base.dialect.JdbcDataSourceDialect;
+import org.apache.flink.cdc.connectors.base.source.meta.split.SnapshotSplit;
import org.apache.flink.table.api.DataTypes;
import io.debezium.jdbc.JdbcConnection;
import io.debezium.relational.Column;
+import io.debezium.relational.Table;
import io.debezium.relational.TableId;
+import io.debezium.relational.history.TableChanges;
import org.assertj.core.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
import javax.annotation.Nullable;
import java.sql.SQLException;
+import java.sql.Types;
+import java.util.Collection;
/** Tests for {@link JdbcSourceChunkSplitter}. */
class JdbcSourceChunkSplitterTest {
+ private static final TableId TABLE_ID = TableId.parse("db.schema.table");
+
/**
* Guards against regressions where {@code queryNextChunkMax} may return
{@code null} (e.g. when
* the max row was removed after MIN/MAX was determined). {@code
nextChunkEnd} must handle this
@@ -57,16 +67,105 @@ class JdbcSourceChunkSplitterTest {
Assertions.assertThat(result).isNull();
}
+ /** The splitter must not hold a JDBC connection while idle, before or
after splitting. */
+ @Test
+ void testConnectionIsReleasedWhenEvenlySizedTableIsSplit() throws
Exception {
+ // 100 rows in [1, 100] is evenly distributed and fits in one chunk
+ Fixture fixture = new Fixture(new Object[] {1, 100}, 100L);
+
+ fixture.splitter.open();
+ Mockito.verify(fixture.dialect,
Mockito.never()).openJdbcConnection(Mockito.any());
+
+ Collection<SnapshotSplit> splits =
fixture.splitter.generateSplits(TABLE_ID);
+
+ Assertions.assertThat(splits).hasSize(1);
+ Assertions.assertThat(fixture.splitter.hasNextChunk()).isFalse();
+ Mockito.verify(fixture.dialect,
Mockito.times(1)).openJdbcConnection(Mockito.any());
+ Mockito.verify(fixture.connection, Mockito.times(1)).close();
+ }
+
+ /**
+ * An unevenly distributed table is split one chunk per {@code
generateSplits} call; each call
+ * must release its connection instead of holding it until the table is
fully split.
+ */
+ @Test
+ void testConnectionIsReleasedAfterEachUnevenlySizedChunk() throws
Exception {
+ // 10 rows in [1, 1000000] exceeds the upper distribution factor, so
chunks are queried
+ // one by one: the first query ends at 10, the second finds no more
rows
+ Fixture fixture = new Fixture(new Object[] {1, 1_000_000}, 10L, 10,
null);
+
+ Collection<SnapshotSplit> firstSplits =
fixture.splitter.generateSplits(TABLE_ID);
+
+ Assertions.assertThat(firstSplits).hasSize(1);
+ Assertions.assertThat(fixture.splitter.hasNextChunk()).isTrue();
+ Mockito.verify(fixture.dialect,
Mockito.times(1)).openJdbcConnection(Mockito.any());
+ Mockito.verify(fixture.connection, Mockito.times(1)).close();
+
+ Collection<SnapshotSplit> lastSplits =
fixture.splitter.generateSplits(TABLE_ID);
+
+ Assertions.assertThat(lastSplits).hasSize(1);
+ Assertions.assertThat(fixture.splitter.hasNextChunk()).isFalse();
+ Mockito.verify(fixture.dialect,
Mockito.times(2)).openJdbcConnection(Mockito.any());
+ Mockito.verify(fixture.connection, Mockito.times(2)).close();
+ }
+
+ /** A splitter over a single-column table whose dialect and connection are
Mockito mocks. */
+ private static class Fixture {
+ private final JdbcDataSourceDialect dialect =
Mockito.mock(JdbcDataSourceDialect.class);
+ private final JdbcConnection connection =
Mockito.mock(JdbcConnection.class);
+ private final JdbcSourceChunkSplitter splitter;
+
+ Fixture(Object[] minMax, long approximateRowCnt, Object...
nextChunkMaxResults) {
+ Column idColumn =
+
Column.editor().name("id").type("INT").jdbcType(Types.INTEGER).create();
+ Table table =
+ Table.editor()
+ .tableId(TABLE_ID)
+ .addColumn(idColumn)
+ .setPrimaryKeyNames("id")
+ .create();
+
Mockito.when(dialect.openJdbcConnection(Mockito.any())).thenReturn(connection);
+ Mockito.when(dialect.queryTableSchema(Mockito.any(),
Mockito.eq(TABLE_ID)))
+ .thenReturn(
+ new TableChanges.TableChange(
+ TableChanges.TableChangeType.CREATE,
table));
+
+ JdbcSourceConfig sourceConfig =
Mockito.mock(JdbcSourceConfig.class);
+ Mockito.when(sourceConfig.getSplitSize()).thenReturn(8096);
+
Mockito.when(sourceConfig.getDistributionFactorUpper()).thenReturn(1000.0d);
+
Mockito.when(sourceConfig.getDistributionFactorLower()).thenReturn(0.05d);
+
+ splitter =
+ new TestingJdbcSourceChunkSplitter(
+ sourceConfig, dialect, minMax, approximateRowCnt,
nextChunkMaxResults);
+ }
+ }
+
/** Minimal testing implementation that stubs out JDBC interactions. */
private static class TestingJdbcSourceChunkSplitter extends
JdbcSourceChunkSplitter {
- @Nullable private final Object nextChunkMaxResult;
+ private final Object[] minMax;
+ private final long approximateRowCnt;
+ private final Object[] nextChunkMaxResults;
+ private int nextChunkMaxCalls;
TestingJdbcSourceChunkSplitter(@Nullable Object nextChunkMaxResult) {
- super(null, null, null, null, null);
- this.nextChunkMaxResult = nextChunkMaxResult;
+ this(null, null, new Object[] {null, null}, 0L, new Object[]
{nextChunkMaxResult});
}
+ TestingJdbcSourceChunkSplitter(
+ JdbcSourceConfig sourceConfig,
+ JdbcDataSourceDialect dialect,
+ Object[] minMax,
+ long approximateRowCnt,
+ Object[] nextChunkMaxResults) {
+ super(sourceConfig, dialect, null, null, null);
+ this.minMax = minMax;
+ this.approximateRowCnt = approximateRowCnt;
+ this.nextChunkMaxResults = nextChunkMaxResults;
+ }
+
+ /** Returns the configured results in order, then {@code null} once
they are used up. */
@Override
protected Object queryNextChunkMax(
JdbcConnection jdbc,
@@ -75,13 +174,21 @@ class JdbcSourceChunkSplitterTest {
int chunkSize,
Object includedLowerBound)
throws SQLException {
- return nextChunkMaxResult;
+ return nextChunkMaxCalls < nextChunkMaxResults.length
+ ? nextChunkMaxResults[nextChunkMaxCalls++]
+ : null;
+ }
+
+ @Override
+ protected Object[] queryMinMax(JdbcConnection jdbc, TableId tableId,
Column splitColumn)
+ throws SQLException {
+ return minMax;
}
@Override
protected Long queryApproximateRowCnt(JdbcConnection jdbc, TableId
tableId)
throws SQLException {
- return 0L;
+ return approximateRowCnt;
}
@Override
