Skip to content

[improve](routine load) delay schedule EOF tasks to avoid too many small transactions (#39975) #40498

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
dataroaring merged 1 commit into from
Sep 8, 2024
Merged

[improve](routine load) delay schedule EOF tasks to avoid too many small transactions (#39975) #40498

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaTaskInfo.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,7 @@ public KafkaTaskInfo(KafkaTaskInfo kafkaTaskInfo, Map<Integer, Long> partitionId
kafkaTaskInfo.getTimeoutMs(), kafkaTaskInfo.getTimeoutBackOffCount(),
kafkaTaskInfo.getBeId(), isMultiTable);
this.partitionIdToOffset = partitionIdToOffset;
this.isEof = kafkaTaskInfo.getIsEof();
}

public List<Integer> getPartitions() {
Expand Down
2 changes: 1 addition & 1 deletion fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1247,7 +1247,7 @@ private void executeTaskOnTxnStatusChanged(RoutineLoadTaskInfo routineLoadTaskIn
} 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())) {
Expand Down
22 changes: 21 additions & 1 deletion fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadTaskInfo.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -167,6 +169,10 @@ public int getTimeoutBackOffCount() {
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.
Expand All @@ -181,7 +187,12 @@ public boolean isTimeout() {
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;

Expand All @@ -196,6 +207,15 @@ public void selfAdaptTimeout(RLTaskTxnCommitAttachment rlTaskTxnCommitAttachment
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
Expand Down
15 changes: 10 additions & 5 deletions fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadTaskScheduler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -101,11 +101,15 @@ private void process() throws UserException, InterruptedException {
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) {
Expand All @@ -114,6 +118,7 @@ private void process() throws UserException, InterruptedException {
}

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());
Expand Down