This is an automated email from the ASF dual-hosted git repository.
JNSimba pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 49c5c843487 [improvement](streaming-job) Adapt CDC snapshot
parallelism to backend CPU (#68432)
49c5c843487 is described below
commit 49c5c843487595a972d5c24cd1b9476250f3cb4c
Author: wudi <[email protected]>
AuthorDate: Tue Oct 13 17:41:04 2026 +0800
[improvement](streaming-job) Adapt CDC snapshot parallelism to backend CPU
(#68432)
### What problem does this PR solve?
Problem Summary:
FROM TO CDC snapshot tasks used a fixed snapshot parallelism without
considering the CPU capacity of the selected backend. This could create
excessive snapshot readers and CDC client threads on smaller backends.
This change derives the effective snapshot parallelism from half of the
selected backend CPU cores, caps it with the mutable FE configuration
`streaming_cdc_max_snapshot_parallelism` (default 16), and preserves a
lower user-configured value. MySQL and OceanBase `server_id` ranges
continue to cap the effective parallelism when the user does not
explicitly configure snapshot parallelism. The offset provider sends at
most that many pending splits, and the CDC client creates one worker per
split in the actual batch.
---
.../main/java/org/apache/doris/common/Config.java | 4 +
.../insert/streaming/StreamingMultiTblTask.java | 33 +++++--
.../doris/job/offset/SourceOffsetProvider.java | 5 +
.../job/offset/jdbc/JdbcSourceOffsetProvider.java | 11 ++-
.../streaming/DataSourceConfigValidatorTest.java | 14 +++
.../StreamingMultiTblTaskParallelismTest.java | 109 +++++++++++++++++++++
.../JdbcSourceOffsetProviderAsyncSplitTest.java | 14 +++
.../jdbc/JdbcSourceOffsetProviderOffsetTest.java | 31 ++++++
.../source/reader/JdbcIncrementalSourceReader.java | 30 +++---
.../source/reader/mysql/MySqlSourceReader.java | 30 +++---
10 files changed, 239 insertions(+), 42 deletions(-)
diff --git a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
index 82064f1ab07..c0e29b7314d 100644
--- a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
+++ b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
@@ -1228,6 +1228,10 @@ public class Config extends ConfigBase {
@ConfField(mutable = true, masterOnly = true)
public static int streaming_cdc_fetch_splits_batch_size = 16;
+ @ConfField(mutable = true, masterOnly = true,
+ description = "Maximum snapshot parallelism per FROM TO task;
values below 1 are treated as 1")
+ public static int streaming_cdc_max_snapshot_parallelism = 16;
+
/**
* the max timeout of get kafka meta.
*/
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingMultiTblTask.java
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingMultiTblTask.java
index f141762bdf8..8589d93a2f3 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingMultiTblTask.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingMultiTblTask.java
@@ -85,6 +85,7 @@ public class StreamingMultiTblTask extends
AbstractStreamingTask {
private long filteredRows = 0L;
private long loadedRows = 0L;
private volatile long runningBackendId;
+ private Backend runningBackend;
long lastScannedRows = -1;
long lastProgressMs = 0;
@@ -99,6 +100,7 @@ public class StreamingMultiTblTask extends
AbstractStreamingTask {
UserIdentity userIdentity,
String cloudCluster) {
super(jobId, taskId, userIdentity);
+ this.noRetry = true;
this.dataSourceType = dataSourceType;
this.offsetProvider = offsetProvider;
this.sourceProperties = sourceProperties;
@@ -117,7 +119,27 @@ public class StreamingMultiTblTask extends
AbstractStreamingTask {
this.startTimeMs = System.currentTimeMillis();
this.lastProgressMs = this.startTimeMs;
this.status = TaskStatus.RUNNING;
- this.runningOffset = offsetProvider.getNextOffset(null,
sourceProperties);
+ boolean snapshotPhase = offsetProvider.isSnapshotPhase();
+ this.runningBackend = resolveBackend(snapshotPhase);
+ this.runningBackendId = runningBackend.getId();
+ if (snapshotPhase) {
+ int maxParallelism = Math.max(1,
Config.streaming_cdc_max_snapshot_parallelism);
+ int autoParallelism = Math.max(1, runningBackend.getCputCores() /
2);
+ int parallelism = Math.min(maxParallelism,
Integer.parseInt(sourceProperties.getOrDefault(
+ DataSourceConfigKeys.SNAPSHOT_PARALLELISM,
String.valueOf(autoParallelism))));
+ if
(!sourceProperties.containsKey(DataSourceConfigKeys.SNAPSHOT_PARALLELISM)
+ && (dataSourceType == DataSourceType.MYSQL ||
dataSourceType == DataSourceType.OCEANBASE)
+ &&
sourceProperties.containsKey(DataSourceConfigKeys.SERVER_ID)) {
+ int[] serverIdRange =
DataSourceConfigValidator.parseServerIdRange(
+ sourceProperties.get(DataSourceConfigKeys.SERVER_ID));
+ parallelism = Math.min(parallelism, serverIdRange[1] -
serverIdRange[0] + 1);
+ }
+ this.sourceProperties = new HashMap<>(sourceProperties);
+
this.sourceProperties.put(DataSourceConfigKeys.SNAPSHOT_PARALLELISM,
String.valueOf(parallelism));
+ log.info("snapshot task {} on backend {} with {} CPU cores uses
parallelism {}",
+ taskId, runningBackendId, runningBackend.getCputCores(),
parallelism);
+ }
+ this.runningOffset = offsetProvider.getNextOffset(null,
this.sourceProperties);
log.info("streaming multi task {} get running offset: {}", taskId,
runningOffset.toString());
}
@@ -131,10 +153,9 @@ public class StreamingMultiTblTask extends
AbstractStreamingTask {
}
private void sendWriteRequest() throws JobException {
- Backend backend = resolveBackend();
+ Backend backend = runningBackend;
log.info("start to run streaming multi task {} in backend {}/{},
offset is {}",
taskId, backend.getId(), backend.getHost(),
runningOffset.toString());
- this.runningBackendId = backend.getId();
WriteRecordRequest params = buildRequestParams();
InternalService.PRequestCdcClientRequest request =
InternalService.PRequestCdcClientRequest.newBuilder()
.setApi("/api/writeRecords")
@@ -175,7 +196,6 @@ public class StreamingMultiTblTask extends
AbstractStreamingTask {
taskId, getJobId(), backend.getHost(),
backend.getBrpcPort(),
Config.streaming_cdc_heavy_rpc_timeout_sec);
// the request may have been dispatched and still running remotely
- noRetry = true;
throw new JobException("cdc_client RPC timeout: /api/writeRecords
jobId="
+ getJobId() + " taskId=" + taskId);
} catch (ExecutionException | InterruptedException ex) {
@@ -183,14 +203,13 @@ public class StreamingMultiTblTask extends
AbstractStreamingTask {
Thread.currentThread().interrupt();
}
log.error("Send write request failed: ", ex);
- noRetry = true;
throw new JobException(ex);
}
}
- private Backend resolveBackend() throws JobException {
+ private Backend resolveBackend(boolean snapshotPhase) throws JobException {
// Snapshot phase keeps per-round selection; binlog phase binds to a
fixed BE for reuse.
- if (((JdbcOffset) runningOffset).snapshotSplit()) {
+ if (snapshotPhase) {
return StreamingJobUtils.selectBackend(cloudCluster);
}
return getStreamingJob().resolveBoundBackend();
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/offset/SourceOffsetProvider.java
b/fe/fe-core/src/main/java/org/apache/doris/job/offset/SourceOffsetProvider.java
index e2c70b228ae..0fbd5755c03 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/job/offset/SourceOffsetProvider.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/job/offset/SourceOffsetProvider.java
@@ -214,6 +214,11 @@ public interface SourceOffsetProvider {
return 0;
}
+ /** Snapshot phase lasts until splitting and consumption of produced
splits both finish. */
+ default boolean isSnapshotPhase() {
+ return !noMoreSplits() || pendingSplitCount() > 0;
+ }
+
/** Get the latest successfully observed source-log lag in bytes, or -1
before any observation. */
default long getLagBytes() {
return -1;
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProvider.java
b/fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProvider.java
index 16e4d109159..0eb327af571 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProvider.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProvider.java
@@ -167,7 +167,9 @@ public class JdbcSourceOffsetProvider implements
SourceOffsetProvider {
synchronized (splitsLock) {
JdbcOffset nextOffset = new JdbcOffset();
if (!remainingSplits.isEmpty()) {
- int splitsNum = Math.min(remainingSplits.size(),
snapshotParallelism);
+ int taskParallelism = Integer.parseInt(properties.getOrDefault(
+ DataSourceConfigKeys.SNAPSHOT_PARALLELISM,
String.valueOf(snapshotParallelism)));
+ int splitsNum = Math.min(remainingSplits.size(),
taskParallelism);
List<SnapshotSplit> snapshotSplits = new
ArrayList<>(remainingSplits.subList(0, splitsNum));
nextOffset.setSplits(snapshotSplits);
return nextOffset;
@@ -745,6 +747,13 @@ public class JdbcSourceOffsetProvider implements
SourceOffsetProvider {
}
}
+ @Override
+ public boolean isSnapshotPhase() {
+ synchronized (splitsLock) {
+ return !remainingSplits.isEmpty() || !noMoreSplits();
+ }
+ }
+
@Override
public boolean noMoreSplits() {
if (!checkNeedSplitChunks(sourceProperties)) {
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/DataSourceConfigValidatorTest.java
b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/DataSourceConfigValidatorTest.java
index 231c7210e7d..7f1aaa9f7ff 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/DataSourceConfigValidatorTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/DataSourceConfigValidatorTest.java
@@ -417,6 +417,20 @@ public class DataSourceConfigValidatorTest {
DataSourceConfigValidator.validateSource(props,
DataSourceType.OCEANBASE.name());
}
+ @Test
+ public void testOceanBaseServerIdRangeKeepsOriginalValidation() {
+ Map<String, String> props = new HashMap<>();
+ props.put(DataSourceConfigKeys.SERVER_ID, "5400");
+ DataSourceConfigValidator.validateSource(props,
DataSourceType.OCEANBASE.name());
+
+ props.put(DataSourceConfigKeys.SNAPSHOT_PARALLELISM, "2");
+ Assertions.assertThrows(IllegalArgumentException.class,
+ () -> DataSourceConfigValidator.validateSource(props,
DataSourceType.OCEANBASE.name()));
+
+ props.put(DataSourceConfigKeys.SNAPSHOT_PARALLELISM, "1");
+ DataSourceConfigValidator.validateSource(props,
DataSourceType.OCEANBASE.name());
+ }
+
@Test
public void testOceanBaseRejectsNonMysqlJdbcUrl() {
Map<String, String> props = new HashMap<>();
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingMultiTblTaskParallelismTest.java
b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingMultiTblTaskParallelismTest.java
new file mode 100644
index 00000000000..59cfec6e848
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/job/extensions/insert/streaming/StreamingMultiTblTaskParallelismTest.java
@@ -0,0 +1,109 @@
+// 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.doris.job.extensions.insert.streaming;
+
+import org.apache.doris.common.Config;
+import org.apache.doris.job.cdc.DataSourceConfigKeys;
+import org.apache.doris.job.common.DataSourceType;
+import org.apache.doris.job.offset.SourceOffsetProvider;
+import org.apache.doris.job.offset.jdbc.JdbcOffset;
+import org.apache.doris.job.util.StreamingJobUtils;
+import org.apache.doris.system.Backend;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.util.HashMap;
+import java.util.Map;
+
+class StreamingMultiTblTaskParallelismTest {
+ @Test
+ void taskCapsAutomaticAndExplicitParallelismWithoutChangingJobProperties()
throws Exception {
+ int oldMax = Config.streaming_cdc_max_snapshot_parallelism;
+ try {
+ Config.streaming_cdc_max_snapshot_parallelism = 16;
+ assertParallelism(0, null, 1);
+ assertParallelism(1, null, 1);
+ assertParallelism(4, null, 2);
+ assertParallelism(8, null, 4);
+ assertParallelism(16, null, 8);
+ assertParallelism(32, null, 16);
+ assertParallelism(8, "1", 1);
+ assertParallelism(8, "4", 4);
+ assertParallelism(8, "8", 8);
+ assertParallelism(8, "10000", 16);
+ assertParallelism(32, "10000", 16);
+ assertParallelism(8, null, 4, DataSourceType.MYSQL, null);
+ assertParallelism(8, null, 1, DataSourceType.MYSQL, "5400");
+ assertParallelism(8, null, 4, DataSourceType.MYSQL, "5400-5403");
+ assertParallelism(8, null, 1, DataSourceType.OCEANBASE, "5400");
+ Config.streaming_cdc_max_snapshot_parallelism = 4;
+ assertParallelism(8, null, 4);
+ assertParallelism(32, null, 4);
+ assertParallelism(8, "8", 4);
+ Config.streaming_cdc_max_snapshot_parallelism = 32;
+ assertParallelism(8, null, 4);
+ assertParallelism(8, "32", 32, DataSourceType.MYSQL, null);
+ Config.streaming_cdc_max_snapshot_parallelism = 1;
+ assertParallelism(8, null, 1);
+ assertParallelism(8, "10000", 1);
+ Config.streaming_cdc_max_snapshot_parallelism = 0;
+ assertParallelism(8, null, 1);
+ assertParallelism(8, "8", 1);
+ } finally {
+ Config.streaming_cdc_max_snapshot_parallelism = oldMax;
+ }
+ }
+
+ private void assertParallelism(int cpuCores, String requested, int
expected) throws Exception {
+ assertParallelism(cpuCores, requested, expected, DataSourceType.MYSQL,
"5400-5415");
+ }
+
+ private void assertParallelism(int cpuCores, String requested, int
expected,
+ DataSourceType sourceType, String serverId) throws Exception {
+ Backend backend = Mockito.mock(Backend.class);
+ Mockito.when(backend.getId()).thenReturn(7L);
+ Mockito.when(backend.getCputCores()).thenReturn(cpuCores);
+ SourceOffsetProvider provider =
Mockito.mock(SourceOffsetProvider.class);
+ Mockito.when(provider.isSnapshotPhase()).thenReturn(true);
+ Mockito.when(provider.getNextOffset(Mockito.isNull(),
Mockito.anyMap()))
+ .thenReturn(Mockito.mock(JdbcOffset.class));
+ Map<String, String> properties = new HashMap<>();
+ if (serverId != null) {
+ properties.put(DataSourceConfigKeys.SERVER_ID, serverId);
+ }
+ if (requested != null) {
+ properties.put(DataSourceConfigKeys.SNAPSHOT_PARALLELISM,
requested);
+ }
+ Map<String, String> original = new HashMap<>(properties);
+ StreamingMultiTblTask task = new StreamingMultiTblTask(
+ 1L, 2L, sourceType, provider, properties, "db", null, null,
null, null);
+ try (MockedStatic<StreamingJobUtils> utils =
Mockito.mockStatic(StreamingJobUtils.class)) {
+ utils.when(() ->
StreamingJobUtils.selectBackend(null)).thenReturn(backend);
+ task.before();
+ utils.verify(() -> StreamingJobUtils.selectBackend(null));
+ }
+ Assertions.assertEquals(7L, task.getRunningBackendId());
+ Assertions.assertEquals(String.valueOf(expected),
+
task.getSourceProperties().get(DataSourceConfigKeys.SNAPSHOT_PARALLELISM));
+ Assertions.assertEquals(original, properties);
+ Mockito.verify(provider).getNextOffset(Mockito.isNull(),
Mockito.same(task.getSourceProperties()));
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProviderAsyncSplitTest.java
b/fe/fe-core/src/test/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProviderAsyncSplitTest.java
index 033f5c8f50d..2a21a5ec52c 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProviderAsyncSplitTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProviderAsyncSplitTest.java
@@ -125,10 +125,21 @@ public class JdbcSourceOffsetProviderAsyncSplitTest {
// ===== initOnCreate / noMoreSplits =====
+ @Test
+ public void testSnapshotPhaseSkipsCompletionScanWithPendingSplits() throws
JobException {
+ provider.initOnCreate(Collections.singletonList("db.tbl_a"));
+ provider.remainingSplits.add(split("db.tbl_a", 0, null, null));
+ JdbcSourceOffsetProvider spy = Mockito.spy(provider);
+
+ Assertions.assertTrue(spy.isSnapshotPhase());
+ Mockito.verify(spy, Mockito.never()).noMoreSplits();
+ }
+
@Test
public void testInitWithEmptySyncTablesIsAllDone() throws JobException {
provider.initOnCreate(Collections.emptyList());
Assertions.assertTrue(provider.noMoreSplits());
+ Assertions.assertFalse(provider.isSnapshotPhase());
}
@Test
@@ -138,6 +149,7 @@ public class JdbcSourceOffsetProviderAsyncSplitTest {
Assertions.assertNotNull(provider.cdcSplitProgress);
Assertions.assertNull(provider.cdcSplitProgress.getCurrentSplittingTable());
Assertions.assertFalse(provider.noMoreSplits());
+ Assertions.assertTrue(provider.isSnapshotPhase());
}
@Test
@@ -149,6 +161,7 @@ public class JdbcSourceOffsetProviderAsyncSplitTest {
provider.initOnCreate(Arrays.asList("db.tbl_a", "db.tbl_b"));
provider.setCurrentOffset(new JdbcOffset(Collections.singletonList(new
BinlogSplit())));
Assertions.assertTrue(provider.noMoreSplits());
+ Assertions.assertFalse(provider.isSnapshotPhase());
}
@Test
@@ -230,6 +243,7 @@ public class JdbcSourceOffsetProviderAsyncSplitTest {
Assertions.assertTrue(provider.noMoreSplits());
Assertions.assertEquals(1, provider.remainingSplits.size());
+ Assertions.assertTrue(provider.isSnapshotPhase());
}
@Test
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProviderOffsetTest.java
b/fe/fe-core/src/test/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProviderOffsetTest.java
index 9a6a601d8b7..ed80084ef5f 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProviderOffsetTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProviderOffsetTest.java
@@ -19,6 +19,7 @@ package org.apache.doris.job.offset.jdbc;
import org.apache.doris.common.Config;
import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.job.cdc.DataSourceConfigKeys;
import org.apache.doris.job.cdc.split.BinlogSplit;
import org.apache.doris.job.cdc.split.SnapshotSplit;
import org.apache.doris.job.extensions.insert.streaming.StreamingInsertJob;
@@ -26,12 +27,42 @@ import
org.apache.doris.job.extensions.insert.streaming.StreamingInsertJob;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
public class JdbcSourceOffsetProviderOffsetTest {
+ @Test
+ public void testTaskParallelismKeepsUncommittedSplitsForNextTask() {
+ JdbcSourceOffsetProvider provider = new JdbcSourceOffsetProvider();
+ provider.snapshotParallelism = 10000;
+ provider.remainingSplits.addAll(Arrays.asList(
+ snapshotSplit("source_table:0"),
snapshotSplit("source_table:1"), snapshotSplit("source_table:2")));
+ Map<String, String> taskProperties =
Collections.singletonMap(DataSourceConfigKeys.SNAPSHOT_PARALLELISM, "2");
+
+ JdbcOffset first = (JdbcOffset) provider.getNextOffset(null,
taskProperties);
+ Assertions.assertEquals(2, first.getSplits().size());
+ Assertions.assertEquals(3, provider.pendingSplitCount());
+ Assertions.assertEquals(10000, provider.snapshotParallelism);
+ // The next task must receive the same uncommitted splits.
+ JdbcOffset nextTask = (JdbcOffset) provider.getNextOffset(null,
taskProperties);
+ Assertions.assertEquals(first.getSplits(), nextTask.getSplits());
+
+ provider.updateOffset(first);
+ JdbcOffset second = (JdbcOffset) provider.getNextOffset(null,
taskProperties);
+ Assertions.assertEquals(1, second.getSplits().size());
+ Assertions.assertEquals("source_table:2",
second.getSplits().get(0).getSplitId());
+ Assertions.assertEquals(1, provider.pendingSplitCount());
+
+ provider.updateOffset(second);
+ Assertions.assertEquals(0, provider.pendingSplitCount());
+ JdbcOffset binlog = (JdbcOffset) provider.getNextOffset(null,
taskProperties);
+ Assertions.assertFalse(binlog.snapshotSplit());
+ Assertions.assertEquals(3, ((BinlogSplit)
binlog.getSplits().get(0)).getFinishedSplits().size());
+ }
+
@Test
public void testSnapshotOffsetUsesConfiguredPersistInterval() {
int oldInterval =
Config.streaming_job_snapshot_offset_persist_interval_sec;
diff --git
a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/JdbcIncrementalSourceReader.java
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/JdbcIncrementalSourceReader.java
index b4c7dcada31..0592d83c54b 100644
---
a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/JdbcIncrementalSourceReader.java
+++
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/JdbcIncrementalSourceReader.java
@@ -116,23 +116,6 @@ public abstract class JdbcIncrementalSourceReader extends
AbstractCdcSourceReade
@Override
public void initialize(String jobId, DataSource dataSource, Map<String,
String> config) {
this.serializer.init(config);
-
- // Initialize thread pool for parallel polling
- int parallelism =
- Integer.parseInt(
- config.getOrDefault(
- DataSourceConfigKeys.SNAPSHOT_PARALLELISM,
-
DataSourceConfigKeys.SNAPSHOT_PARALLELISM_DEFAULT));
- this.snapshotPollExecutor =
- Executors.newFixedThreadPool(
- parallelism,
- r -> {
- Thread t = new Thread(r);
- t.setName("snapshot-reader-" + jobId + "-" +
t.getId());
- t.setDaemon(true);
- return t;
- });
- LOG.info("Initialized poll executor with parallelism: {}",
parallelism);
}
/**
@@ -322,6 +305,17 @@ public abstract class JdbcIncrementalSourceReader extends
AbstractCdcSourceReade
this.snapshotReaderContexts.clear();
this.completedSplitIds.clear();
+ shutdownSnapshotPollExecutor();
+ this.snapshotPollExecutor =
+ Executors.newFixedThreadPool(
+ splits.size(),
+ r -> {
+ Thread t = new Thread(r);
+ t.setName("snapshot-reader-" + baseReq.getJobId()
+ "-" + t.getId());
+ t.setDaemon(true);
+ return t;
+ });
+
List<CompletableFuture<Void>> futures = new ArrayList<>();
// Create reader for each split and submit
for (int i = 0; i < splits.size(); i++) {
@@ -946,6 +940,7 @@ public abstract class JdbcIncrementalSourceReader extends
AbstractCdcSourceReade
@Override
public synchronized void finishSplitRecords() {
+ shutdownSnapshotPollExecutor();
// Cancel any active poll operations
if (activePollFutures != null) {
activePollFutures.forEach(f -> f.cancel(true));
@@ -1012,6 +1007,7 @@ public abstract class JdbcIncrementalSourceReader extends
AbstractCdcSourceReade
protected void shutdownSnapshotPollExecutor() {
if (snapshotPollExecutor != null) {
snapshotPollExecutor.shutdownNow();
+ snapshotPollExecutor = null;
}
}
diff --git
a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReader.java
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReader.java
index 19380dfe421..1e920969875 100644
---
a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReader.java
+++
b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReader.java
@@ -152,22 +152,6 @@ public class MySqlSourceReader extends
AbstractCdcSourceReader {
@Override
public void initialize(String jobId, DataSource dataSource, Map<String,
String> config) {
this.serializer.init(config);
-
- int parallelism =
- Integer.parseInt(
- config.getOrDefault(
- DataSourceConfigKeys.SNAPSHOT_PARALLELISM,
-
DataSourceConfigKeys.SNAPSHOT_PARALLELISM_DEFAULT));
- this.snapshotPollExecutor =
- Executors.newFixedThreadPool(
- parallelism,
- r -> {
- Thread t = new Thread(r);
- t.setName("snapshot-reader-" + jobId + "-" +
t.getId());
- t.setDaemon(true);
- return t;
- });
- LOG.info("Initialized poll executor with parallelism: {}",
parallelism);
}
/**
@@ -382,6 +366,17 @@ public class MySqlSourceReader extends
AbstractCdcSourceReader {
this.snapshotReaderContexts.clear();
this.completedSplitIds.clear();
+ shutdownSnapshotPollExecutor();
+ this.snapshotPollExecutor =
+ Executors.newFixedThreadPool(
+ splits.size(),
+ r -> {
+ Thread t = new Thread(r);
+ t.setName("snapshot-reader-" + baseReq.getJobId()
+ "-" + t.getId());
+ t.setDaemon(true);
+ return t;
+ });
+
List<CompletableFuture<Void>> futures = new ArrayList<>();
// Create reader for each split and submit
@@ -1143,7 +1138,7 @@ public class MySqlSourceReader extends
AbstractCdcSourceReader {
@Override
public synchronized void finishSplitRecords() {
-
+ shutdownSnapshotPollExecutor();
// Cancel any active poll operations
if (activePollFutures != null) {
activePollFutures.forEach(f -> f.cancel(true));
@@ -1308,6 +1303,7 @@ public class MySqlSourceReader extends
AbstractCdcSourceReader {
protected void shutdownSnapshotPollExecutor() {
if (snapshotPollExecutor != null) {
snapshotPollExecutor.shutdownNow();
+ snapshotPollExecutor = null;
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
