This is an automated email from the ASF dual-hosted git repository.
kxiao 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 837d8ce900f [fix](routine-load) fix total task exec time is far more
than actual #31279 (#31293)
837d8ce900f is described below
commit 837d8ce900f8ca1b064cd011035a375db67ee06b
Author: HHoflittlefish777 <[email protected]>
AuthorDate: Thu Feb 22 23:09:24 2024 +0800
[fix](routine-load) fix total task exec time is far more than actual #31279
(#31293)
---
.../java/org/apache/doris/load/routineload/RoutineLoadJob.java | 9 ++++-----
.../org/apache/doris/load/routineload/RoutineLoadJobTest.java | 4 ++--
2 files changed, 6 insertions(+), 7 deletions(-)
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 0da102e2212..0d9ae516351 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
@@ -774,17 +774,16 @@ public abstract class RoutineLoadJob extends
AbstractTxnStateChangeCallback impl
// if rate of error data is more than max_filter_ratio, pause job
protected void updateProgress(RLTaskTxnCommitAttachment attachment) throws
UserException {
updateNumOfData(attachment.getTotalRows(),
attachment.getFilteredRows(), attachment.getUnselectedRows(),
- attachment.getReceivedBytes(),
attachment.getTaskExecutionTimeMs(),
- false /* not replay */);
+ attachment.getReceivedBytes(), false /* not replay */);
}
private void updateNumOfData(long numOfTotalRows, long numOfErrorRows,
long unselectedRows, long receivedBytes,
- long taskExecutionTime, boolean isReplay)
throws UserException {
+ boolean isReplay) throws UserException {
this.jobStatistic.totalRows += numOfTotalRows;
this.jobStatistic.errorRows += numOfErrorRows;
this.jobStatistic.unselectedRows += unselectedRows;
this.jobStatistic.receivedBytes += receivedBytes;
- this.jobStatistic.totalTaskExcutionTimeMs += taskExecutionTime;
+ this.jobStatistic.totalTaskExcutionTimeMs = System.currentTimeMillis()
- createTimestamp;
if (MetricRepo.isInit && !isReplay) {
MetricRepo.COUNTER_ROUTINE_LOAD_ROWS.increase(numOfTotalRows);
@@ -857,7 +856,7 @@ public abstract class RoutineLoadJob extends
AbstractTxnStateChangeCallback impl
protected void replayUpdateProgress(RLTaskTxnCommitAttachment attachment) {
try {
updateNumOfData(attachment.getTotalRows(),
attachment.getFilteredRows(), attachment.getUnselectedRows(),
- attachment.getReceivedBytes(),
attachment.getTaskExecutionTimeMs(), true /* is replay */);
+ attachment.getReceivedBytes(), true /* is replay */);
} catch (UserException e) {
LOG.error("should not happen", e);
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java
b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java
index b6f3a1c3a64..ba3ab2100be 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java
@@ -288,7 +288,7 @@ public class RoutineLoadJobTest {
RoutineLoadJob routineLoadJob = new KafkaRoutineLoadJob();
Deencapsulation.setField(routineLoadJob, "maxErrorNum", 0);
Deencapsulation.setField(routineLoadJob, "maxBatchRows", 0);
- Deencapsulation.invoke(routineLoadJob, "updateNumOfData", 1L, 1L, 0L,
1L, 1L, false);
+ Deencapsulation.invoke(routineLoadJob, "updateNumOfData", 1L, 1L, 0L,
1L, false);
Assert.assertEquals(RoutineLoadJob.JobState.PAUSED,
Deencapsulation.getField(routineLoadJob, "state"));
@@ -303,7 +303,7 @@ public class RoutineLoadJobTest {
RoutineLoadStatistic jobStatistic =
Deencapsulation.getField(routineLoadJob, "jobStatistic");
Deencapsulation.setField(jobStatistic, "currentErrorRows", 1);
Deencapsulation.setField(jobStatistic, "currentTotalRows", 99);
- Deencapsulation.invoke(routineLoadJob, "updateNumOfData", 2L, 0L, 0L,
1L, 1L, false);
+ Deencapsulation.invoke(routineLoadJob, "updateNumOfData", 2L, 0L, 0L,
1L, false);
Assert.assertEquals(RoutineLoadJob.JobState.RUNNING,
Deencapsulation.getField(routineLoadJob, "state"));
Assert.assertEquals(new Long(0),
Deencapsulation.getField(jobStatistic, "currentErrorRows"));
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
