This is an automated email from the ASF dual-hosted git repository.
dataroaring pushed a commit to branch branch-2.0
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-2.0 by this push:
new 5b23ab70d20 [improve](routine load) delay schedule EOF tasks to avoid
too many small transactions (#39975) (#40498)
5b23ab70d20 is described below
commit 5b23ab70d20bc657bca183b1e6a76e9981cab377
Author: hui lai <[email protected]>
AuthorDate: Sun Sep 8 10:28:08 2024 +0800
[improve](routine load) delay schedule EOF tasks to avoid too many small
transactions (#39975) (#40498)
pick (#39975)
We encountered a scenario where a large number of small transactions
were generated, resulting in an impact on query performance: Kafka's
data comes in batches of very small data every very short time, which
leads to tasks being frequently scheduled and ending very quickly,
resulting in a large number of small transactions.
To solve this problem, we delay the scheduling of tasks that perceive
EOF, which would not delay data consumption, for perceiving EOF
indicates that the consumption speed is greater than the production
speed.
---
.../doris/load/routineload/KafkaTaskInfo.java | 1 +
.../doris/load/routineload/RoutineLoadJob.java | 2 +-
.../load/routineload/RoutineLoadTaskInfo.java | 22 +++++++++++++++++++++-
.../load/routineload/RoutineLoadTaskScheduler.java | 15 ++++++++++-----
4 files changed, 33 insertions(+), 7 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 de1cf5096d2..dbddc695a68 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
@@ -58,6 +58,7 @@ public class KafkaTaskInfo extends RoutineLoadTaskInfo {
kafkaTaskInfo.getTimeoutMs(),
kafkaTaskInfo.getTimeoutBackOffCount(),
kafkaTaskInfo.getBeId(), isMultiTable);
this.partitionIdToOffset = partitionIdToOffset;
+ this.isEof = kafkaTaskInfo.getIsEof();
}
public List<Integer> getPartitions() {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java
index 57aa84e7731..3a1c35bd10f 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java
@@ -1247,7 +1247,7 @@ public abstract class RoutineLoadJob extends
AbstractTxnStateChangeCallback impl
} else if (checkCommitInfo(rlTaskTxnCommitAttachment, txnState,
txnStatusChangeReason)) {
// step2: update job progress
updateProgress(rlTaskTxnCommitAttachment);
- routineLoadTaskInfo.selfAdaptTimeout(rlTaskTxnCommitAttachment);
+
routineLoadTaskInfo.handleTaskByTxnCommitAttachment(rlTaskTxnCommitAttachment);
}
if (rlTaskTxnCommitAttachment != null &&
!Strings.isNullOrEmpty(rlTaskTxnCommitAttachment.getErrorLogUrl())) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadTaskInfo.java
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadTaskInfo.java
index ae2570224c4..69c07507487 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadTaskInfo.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadTaskInfo.java
@@ -77,6 +77,8 @@ public abstract class RoutineLoadTaskInfo {
protected static final int MAX_TIMEOUT_BACK_OFF_COUNT = 3;
protected int timeoutBackOffCount = 0;
+ protected boolean isEof = false;
+
// this status will be set when corresponding transaction's status is
changed.
// so that user or other logic can know the status of the corresponding
txn.
protected TransactionStatus txnStatus = TransactionStatus.UNKNOWN;
@@ -167,6 +169,10 @@ public abstract class RoutineLoadTaskInfo {
return timeoutBackOffCount;
}
+ public boolean getIsEof() {
+ return isEof;
+ }
+
public boolean isTimeout() {
if (txnStatus == TransactionStatus.COMMITTED || txnStatus ==
TransactionStatus.VISIBLE) {
// the corresponding txn is already finished, this task can not be
treated as timeout.
@@ -181,7 +187,12 @@ public abstract class RoutineLoadTaskInfo {
return false;
}
- public void selfAdaptTimeout(RLTaskTxnCommitAttachment
rlTaskTxnCommitAttachment) {
+ public void handleTaskByTxnCommitAttachment(RLTaskTxnCommitAttachment
rlTaskTxnCommitAttachment) {
+ selfAdaptTimeout(rlTaskTxnCommitAttachment);
+ judgeEof(rlTaskTxnCommitAttachment);
+ }
+
+ private void selfAdaptTimeout(RLTaskTxnCommitAttachment
rlTaskTxnCommitAttachment) {
long taskExecutionTime =
rlTaskTxnCommitAttachment.getTaskExecutionTimeMs();
long timeoutMs = this.timeoutMs;
@@ -196,6 +207,15 @@ public abstract class RoutineLoadTaskInfo {
this.timeoutMs = timeoutMs;
}
+ private void judgeEof(RLTaskTxnCommitAttachment rlTaskTxnCommitAttachment)
{
+ RoutineLoadJob routineLoadJob = routineLoadManager.getJob(jobId);
+ if (rlTaskTxnCommitAttachment.getTotalRows() <
routineLoadJob.getMaxBatchRows()
+ && rlTaskTxnCommitAttachment.getReceivedBytes() <
routineLoadJob.getMaxBatchSizeBytes()
+ && rlTaskTxnCommitAttachment.getTaskExecutionTimeMs() <
this.timeoutMs) {
+ this.isEof = true;
+ }
+ }
+
abstract TRoutineLoadTask createRoutineLoadTask() throws UserException;
// begin the txn of this task
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadTaskScheduler.java
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadTaskScheduler.java
index d42b600fb1d..f76cf6205fe 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadTaskScheduler.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadTaskScheduler.java
@@ -101,11 +101,15 @@ public class RoutineLoadTaskScheduler extends
MasterDaemon {
try {
// This step will be blocked when queue is empty
RoutineLoadTaskInfo routineLoadTaskInfo =
needScheduleTasksQueue.take();
- if (System.currentTimeMillis() -
routineLoadTaskInfo.getLastScheduledTime()
- < routineLoadTaskInfo.getTimeoutMs()) {
- // try to delay scheduling this task for 'timeout', to void
too many failure
- needScheduleTasksQueue.addLast(routineLoadTaskInfo);
- return;
+ // try to delay scheduling tasks that are perceived as Eof to
MaxBatchInterval
+ // to avoid to much small transaction
+ if (routineLoadTaskInfo.getIsEof()) {
+ RoutineLoadJob routineLoadJob =
routineLoadManager.getJob(routineLoadTaskInfo.getJobId());
+ if (System.currentTimeMillis() -
routineLoadTaskInfo.getLastScheduledTime()
+ < routineLoadJob.getMaxBatchIntervalS()) {
+ needScheduleTasksQueue.addLast(routineLoadTaskInfo);
+ return;
+ }
}
scheduleOneTask(routineLoadTaskInfo);
} catch (Exception e) {
@@ -114,6 +118,7 @@ public class RoutineLoadTaskScheduler extends MasterDaemon {
}
private void scheduleOneTask(RoutineLoadTaskInfo routineLoadTaskInfo)
throws Exception {
+ routineLoadTaskInfo.setLastScheduledTime(System.currentTimeMillis());
if (LOG.isDebugEnabled()) {
LOG.debug("schedule routine load task info {} for job {}",
routineLoadTaskInfo.id, routineLoadTaskInfo.getJobId());
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
