Copilot commented on code in PR #68432:
URL: https://github.com/apache/doris/pull/68432#discussion_r4080254727
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/JdbcIncrementalSourceReader.java:
##########
@@ -322,6 +305,17 @@ private synchronized SplitReadResult prepareSnapshotSplits(
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;
+ });
Review Comment:
Same issue as in `MySqlSourceReader`: `newFixedThreadPool(0)` will fail when
`splits.size() == 0`. Add an explicit empty-splits branch (preferred) or clamp
the thread count to at least 1 and ensure no work is submitted for the empty
case.
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingMultiTblTask.java:
##########
@@ -99,6 +100,7 @@ public StreamingMultiTblTask(Long jobId,
UserIdentity userIdentity,
String cloudCluster) {
super(jobId, taskId, userIdentity);
+ this.noRetry = true;
Review Comment:
This changes task retry behavior globally (previously `noRetry` was set only
on timeout/exception in `sendWriteRequest`). The PR description focuses on
snapshot parallelism adaptation; please either (a) justify this behavior change
in the PR description/release note, or (b) scope `noRetry` changes to the
original failure paths to avoid reducing resiliency for retryable failures.
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReader.java:
##########
@@ -382,6 +366,17 @@ private synchronized SplitReadResult prepareSnapshotSplits(
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;
+ });
Review Comment:
`Executors.newFixedThreadPool(splits.size())` will throw
`IllegalArgumentException` when `splits.size() == 0`. Since
`prepareSnapshotSplits` can be invoked with an empty split list (e.g., no
snapshot work), guard this by skipping executor creation when `splits` is empty
or using `Math.max(1, splits.size())` and ensuring no tasks are submitted in
the empty case.
##########
fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProvider.java:
##########
@@ -166,7 +166,9 @@ public Offset getNextOffset(StreamingJobProperties
jobProps, Map<String, String>
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)));
Review Comment:
`properties` is now dereferenced unconditionally; previously this method did
not require `properties` to be non-null. If any caller passes `null` for
`properties`, this will throw a `NullPointerException`. Consider treating
`null` as an empty map (or falling back to `snapshotParallelism`) before
calling `getOrDefault`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
