You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by li...@apache.org on 2025/11/13 09:52:29 UTC
(doris) branch master updated: [fix](job) fix transaction timeout do not match routine load task adaptive timeout (#57967)
This is an automated email from the ASF dual-hosted git repository.
liaoxin 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 4ce3e6fc1ea [fix](job) fix transaction timeout do not match routine load task adaptive timeout (#57967)
4ce3e6fc1ea is described below
commit 4ce3e6fc1ead1d9410ced8725f73e98010efe578
Author: hui lai <la...@selectdb.com>
AuthorDate: Thu Nov 13 17:52:21 2025 +0800
[fix](job) fix transaction timeout do not match routine load task adaptive timeout (#57967)
Fix transaction timeout do not match routine load task adaptive timeout,
introduced by https://github.com/apache/doris/pull/56930
---
.../org/apache/doris/load/routineload/KafkaTaskInfo.java | 7 +++++--
.../apache/doris/nereids/load/NereidsLoadTaskInfo.java | 3 +++
.../doris/nereids/load/NereidsRoutineLoadTaskInfo.java | 15 +++++++++++++--
3 files changed, 21 insertions(+), 4 deletions(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaTaskInfo.java b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaTaskInfo.java
index 32ccf96fc2f..fafb34d960d 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaTaskInfo.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaTaskInfo.java
@@ -24,6 +24,7 @@ import org.apache.doris.catalog.Table;
import org.apache.doris.common.Config;
import org.apache.doris.common.UserException;
import org.apache.doris.common.util.DebugUtil;
+import org.apache.doris.nereids.load.NereidsLoadTaskInfo;
import org.apache.doris.nereids.load.NereidsStreamLoadPlanner;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.thrift.TFileFormatType;
@@ -100,6 +101,7 @@ public class KafkaTaskInfo extends RoutineLoadTaskInfo {
tRoutineLoadTask.setKafkaLoadInfo(tKafkaLoadInfo);
tRoutineLoadTask.setType(TLoadSourceType.KAFKA);
tRoutineLoadTask.setIsMultiTable(isMultiTable);
+ adaptiveBatchParam(tRoutineLoadTask, routineLoadJob);
if (!isMultiTable) {
Table tbl = database.getTableOrMetaException(routineLoadJob.getTableId());
tRoutineLoadTask.setTbl(tbl.getName());
@@ -107,7 +109,6 @@ public class KafkaTaskInfo extends RoutineLoadTaskInfo {
} else {
Env.getCurrentEnv().getRoutineLoadManager().addMultiLoadTaskTxnIdToRoutineLoadJobId(txnId, jobId);
}
- adaptiveBatchParam(tRoutineLoadTask, routineLoadJob);
if (!routineLoadJob.getFormat().isEmpty() && routineLoadJob.getFormat().equalsIgnoreCase("json")) {
tRoutineLoadTask.setFormat(TFileFormatType.FORMAT_JSON);
} else {
@@ -152,9 +153,11 @@ public class KafkaTaskInfo extends RoutineLoadTaskInfo {
TUniqueId loadId = new TUniqueId(id.getMostSignificantBits(), id.getLeastSignificantBits());
// plan for each task, in case table has change(rollup or schema change)
Database db = Env.getCurrentInternalCatalog().getDbOrMetaException(routineLoadJob.getDbId());
+ NereidsLoadTaskInfo taskInfo = routineLoadJob.toNereidsRoutineLoadTaskInfo();
+ taskInfo.setTimeout((int) (this.timeoutMs / 1000));
NereidsStreamLoadPlanner planner = new NereidsStreamLoadPlanner(db,
(OlapTable) db.getTableOrMetaException(routineLoadJob.getTableId(),
- Table.TableType.OLAP), routineLoadJob.toNereidsRoutineLoadTaskInfo());
+ Table.TableType.OLAP), taskInfo);
TPipelineFragmentParams tExecPlanFragmentParams = routineLoadJob.plan(planner, loadId, txnId);
TPlanFragment tPlanFragment = tExecPlanFragmentParams.getFragment();
tPlanFragment.getOutputSink().getOlapTableSink().setTxnId(txnId);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsLoadTaskInfo.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsLoadTaskInfo.java
index c5f99bc60c1..ec5195dfd60 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsLoadTaskInfo.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsLoadTaskInfo.java
@@ -42,6 +42,9 @@ public interface NereidsLoadTaskInfo {
int getTimeout();
+ default void setTimeout(int timeout) {
+ }
+
long getMemLimit();
String getTimezone();
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsRoutineLoadTaskInfo.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsRoutineLoadTaskInfo.java
index e5c7fae7a5d..3cf39955ab8 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsRoutineLoadTaskInfo.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsRoutineLoadTaskInfo.java
@@ -67,6 +67,7 @@ public class NereidsRoutineLoadTaskInfo implements NereidsLoadTaskInfo {
protected boolean loadToSingleTablet;
protected boolean isPartialUpdate;
protected boolean memtableOnSinkNode;
+ protected int timeoutSec;
/**
* NereidsRoutineLoadTaskInfo
@@ -96,6 +97,7 @@ public class NereidsRoutineLoadTaskInfo implements NereidsLoadTaskInfo {
this.loadToSingleTablet = loadToSingleTablet;
this.isPartialUpdate = isPartialUpdate;
this.memtableOnSinkNode = memtableOnSinkNode;
+ this.timeoutSec = calTimeoutSec();
}
@Override
@@ -108,14 +110,23 @@ public class NereidsRoutineLoadTaskInfo implements NereidsLoadTaskInfo {
return -1L;
}
- @Override
- public int getTimeout() {
+ public int calTimeoutSec() {
int timeoutSec = (int) maxBatchIntervalS * Config.routine_load_task_timeout_multiplier;
int realTimeoutSec = timeoutSec < Config.routine_load_task_min_timeout_sec
? Config.routine_load_task_min_timeout_sec : timeoutSec;
return realTimeoutSec;
}
+ @Override
+ public int getTimeout() {
+ return this.timeoutSec;
+ }
+
+ @Override
+ public void setTimeout(int timeout) {
+ this.timeoutSec = timeout;
+ }
+
@Override
public long getMemLimit() {
return execMemLimit;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org