Skip to content

Commit fa0fe2e

Browse files
Mrhs121Mrhs121
authored
[Fix-18692][Master] Fix ignored dependent task check interval (#18694)
1 parent 7cead6c commit fa0fe2e

3 files changed

Lines changed: 386 additions & 5 deletions

File tree

‎dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/executor/plugin/dependent/DependentLogicTask.java‎

Lines changed: 52 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -32,17 +32,29 @@
3232
import org.apache.dolphinscheduler.server.master.engine.workflow.execution.IWorkflowExecution;
3333
import org.apache.dolphinscheduler.server.master.exception.MasterTaskExecuteException;
3434

35+
import java.util.concurrent.TimeUnit;
36+
import java.util.function.LongSupplier;
37+
3538
import lombok.extern.slf4j.Slf4j;
3639

3740
import com.fasterxml.jackson.core.type.TypeReference;
41+
import com.google.common.annotations.VisibleForTesting;
3842

3943
@Slf4j
4044
public class DependentLogicTask extends AbstractLogicTask<DependentParameters> {
4145

46+
private static final int DEFAULT_CHECK_INTERVAL_SECONDS = 10;
47+
4248
private final TaskExecutionContext taskExecutionContext;
4349

4450
private final DependentTaskTracker dependentTaskTracker;
4551

52+
private final long checkIntervalNanos;
53+
54+
private final LongSupplier nanoTimeSupplier;
55+
56+
private Long lastCheckTimeNanos;
57+
4658
public DependentLogicTask(TaskExecutionContext taskExecutionContext,
4759
ProjectDao projectDao,
4860
WorkflowDefinitionDao workflowDefinitionDao,
@@ -51,8 +63,26 @@ public DependentLogicTask(TaskExecutionContext taskExecutionContext,
5163
WorkflowInstanceDao workflowInstanceDao,
5264
IWorkflowExecution workflowExecution,
5365
TaskInstanceContextDao taskInstanceContextDao) {
66+
this(taskExecutionContext, projectDao, workflowDefinitionDao, taskDefinitionDao, taskInstanceDao,
67+
workflowInstanceDao, workflowExecution, taskInstanceContextDao, System::nanoTime);
68+
}
69+
70+
@VisibleForTesting
71+
DependentLogicTask(TaskExecutionContext taskExecutionContext,
72+
ProjectDao projectDao,
73+
WorkflowDefinitionDao workflowDefinitionDao,
74+
TaskDefinitionDao taskDefinitionDao,
75+
TaskInstanceDao taskInstanceDao,
76+
WorkflowInstanceDao workflowInstanceDao,
77+
IWorkflowExecution workflowExecution,
78+
TaskInstanceContextDao taskInstanceContextDao,
79+
LongSupplier nanoTimeSupplier) {
5480
super(taskExecutionContext);
5581
this.taskExecutionContext = taskExecutionContext;
82+
this.nanoTimeSupplier = nanoTimeSupplier;
83+
final Integer checkInterval = taskParameters.getDependence().getCheckInterval();
84+
this.checkIntervalNanos = TimeUnit.SECONDS.toNanos(
85+
checkInterval == null ? DEFAULT_CHECK_INTERVAL_SECONDS : checkInterval);
5686
this.dependentTaskTracker = new DependentTaskTracker(
5787
taskExecutionContext,
5888
taskParameters,
@@ -72,22 +102,39 @@ public void start() throws MasterTaskExecuteException {
72102

73103
@Override
74104
public TaskExecutionStatus getTaskExecutionState() {
75-
if (isRunning()) {
76-
taskExecutionStatus = dependentTaskTracker.getDependentTaskStatus();
105+
synchronized (this) {
106+
if (!isRunning()) {
107+
return taskExecutionStatus;
108+
}
109+
final long now = nanoTimeSupplier.getAsLong();
110+
if (lastCheckTimeNanos != null && now - lastCheckTimeNanos < checkIntervalNanos) {
111+
return taskExecutionStatus;
112+
}
113+
lastCheckTimeNanos = now;
114+
}
115+
final TaskExecutionStatus dependentState = dependentTaskTracker.getDependentTaskStatus();
116+
synchronized (this) {
117+
// A pause or kill may have completed while the dependency query was running.
118+
if (isRunning()) {
119+
taskExecutionStatus = dependentState;
120+
}
77121
return taskExecutionStatus;
78122
}
79-
return taskExecutionStatus;
80123
}
81124

82125
@Override
83126
public void pause() throws MasterTaskExecuteException {
84-
onTaskPaused();
127+
synchronized (this) {
128+
onTaskPaused();
129+
}
85130
log.info("Pause task : {} success", taskExecutionContext.getTaskName());
86131
}
87132

88133
@Override
89134
public void kill() throws MasterTaskExecuteException {
90-
onTaskKilled();
135+
synchronized (this) {
136+
onTaskKilled();
137+
}
91138
log.info("Kill task : {} success", taskExecutionContext.getTaskName());
92139
}
93140

‎dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/executor/LogicTaskExecutorTest.java‎

Lines changed: 145 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,145 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.dolphinscheduler.server.master.engine.executor;
19+
20+
import static org.assertj.core.api.Assertions.assertThat;
21+
import static org.mockito.ArgumentMatchers.any;
22+
import static org.mockito.Mockito.mock;
23+
import static org.mockito.Mockito.mockConstruction;
24+
import static org.mockito.Mockito.verify;
25+
import static org.mockito.Mockito.verifyNoMoreInteractions;
26+
import static org.mockito.Mockito.when;
27+
28+
import org.apache.dolphinscheduler.common.enums.Flag;
29+
import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
30+
import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
31+
import org.apache.dolphinscheduler.plugin.task.api.parameters.AbstractParameters;
32+
import org.apache.dolphinscheduler.server.master.engine.executor.plugin.ILogicTask;
33+
import org.apache.dolphinscheduler.server.master.engine.executor.plugin.ILogicTaskPluginFactory;
34+
import org.apache.dolphinscheduler.server.master.engine.executor.plugin.LogicTaskPluginFactoryBuilder;
35+
import org.apache.dolphinscheduler.server.master.engine.executor.plugin.dependent.DependentLogicTask;
36+
import org.apache.dolphinscheduler.server.master.engine.executor.plugin.dependent.DependentTaskTracker;
37+
import org.apache.dolphinscheduler.task.executor.TaskExecutorState;
38+
39+
import java.util.ArrayList;
40+
import java.util.List;
41+
import java.util.concurrent.TimeUnit;
42+
43+
import org.junit.jupiter.api.AfterEach;
44+
import org.junit.jupiter.api.BeforeEach;
45+
import org.junit.jupiter.api.Test;
46+
import org.junit.jupiter.params.ParameterizedTest;
47+
import org.junit.jupiter.params.provider.NullSource;
48+
import org.junit.jupiter.params.provider.ValueSource;
49+
import org.mockito.MockedConstruction;
50+
51+
class LogicTaskExecutorTest {
52+
53+
private static final long FRAMEWORK_CHECK_INTERVAL_MILLIS = TimeUnit.SECONDS.toMillis(10);
54+
55+
private MockedConstruction<DependentTaskTracker> dependentTaskTrackers;
56+
57+
@BeforeEach
58+
void setUp() {
59+
dependentTaskTrackers = mockConstruction(DependentTaskTracker.class,
60+
(tracker, context) -> when(tracker.getDependentTaskStatus())
61+
.thenReturn(TaskExecutionStatus.RUNNING_EXECUTION));
62+
}
63+
64+
@AfterEach
65+
void tearDown() {
66+
dependentTaskTrackers.close();
67+
}
68+
69+
@ParameterizedTest
70+
@NullSource
71+
@ValueSource(ints = {1, 5, 10, 20})
72+
void frameworkUsesFixedCheckInterval(Integer checkInterval) throws Exception {
73+
LogicTaskExecutor taskExecutor = createDependentTaskExecutor(checkInterval);
74+
75+
assertThat(taskExecutor.getRemainingTrackDelay()).isZero();
76+
long beforeTrack = System.currentTimeMillis();
77+
assertThat(taskExecutor.trackTaskExecutorState()).isEqualTo(TaskExecutorState.RUNNING);
78+
long remainingDelay = taskExecutor.getRemainingTrackDelay();
79+
long afterTrack = System.currentTimeMillis();
80+
81+
assertThat(remainingDelay).isBetween(
82+
FRAMEWORK_CHECK_INTERVAL_MILLIS - (afterTrack - beforeTrack), FRAMEWORK_CHECK_INTERVAL_MILLIS);
83+
verify(dependentTaskTrackers.constructed().get(0)).getDependentTaskStatus();
84+
}
85+
86+
@Test
87+
void pauseIsReportedOnNextFrameworkCheck() throws Exception {
88+
LogicTaskExecutor taskExecutor = createDependentTaskExecutor(20);
89+
taskExecutor.trackTaskExecutorState();
90+
91+
taskExecutor.pause();
92+
93+
assertThat(taskExecutor.getRemainingTrackDelay()).isPositive()
94+
.isLessThanOrEqualTo(FRAMEWORK_CHECK_INTERVAL_MILLIS);
95+
assertThat(taskExecutor.trackTaskExecutorState()).isEqualTo(TaskExecutorState.PAUSED);
96+
DependentTaskTracker tracker = dependentTaskTrackers.constructed().get(0);
97+
verify(tracker).getDependentTaskStatus();
98+
verifyNoMoreInteractions(tracker);
99+
}
100+
101+
@Test
102+
void killIsReportedOnNextFrameworkCheck() throws Exception {
103+
LogicTaskExecutor taskExecutor = createDependentTaskExecutor(20);
104+
taskExecutor.trackTaskExecutorState();
105+
106+
taskExecutor.kill();
107+
108+
assertThat(taskExecutor.getRemainingTrackDelay()).isPositive()
109+
.isLessThanOrEqualTo(FRAMEWORK_CHECK_INTERVAL_MILLIS);
110+
assertThat(taskExecutor.trackTaskExecutorState()).isEqualTo(TaskExecutorState.KILLED);
111+
DependentTaskTracker tracker = dependentTaskTrackers.constructed().get(0);
112+
verify(tracker).getDependentTaskStatus();
113+
verifyNoMoreInteractions(tracker);
114+
}
115+
116+
private LogicTaskExecutor createDependentTaskExecutor(Integer checkInterval) throws Exception {
117+
TaskExecutionContext taskExecutionContext = new TaskExecutionContext();
118+
taskExecutionContext.setTaskType("DEPENDENT");
119+
taskExecutionContext.setTaskName("dependent");
120+
taskExecutionContext.setDryRun(Flag.NO.getCode());
121+
taskExecutionContext.setTaskParams(checkInterval == null ? "{\"dependence\":{}}"
122+
: "{\"dependence\":{\"checkInterval\":" + checkInterval + "}}");
123+
124+
return createTaskExecutor(taskExecutionContext, new DependentLogicTask(
125+
taskExecutionContext, null, null, null, null, null, null, null));
126+
}
127+
128+
@SuppressWarnings("unchecked")
129+
private LogicTaskExecutor createTaskExecutor(TaskExecutionContext taskExecutionContext,
130+
ILogicTask<? extends AbstractParameters> logicTask) throws Exception {
131+
ILogicTaskPluginFactory<? extends ILogicTask<? extends AbstractParameters>> taskPluginFactory =
132+
mock(ILogicTaskPluginFactory.class);
133+
when(taskPluginFactory.getTaskType()).thenReturn(taskExecutionContext.getTaskType());
134+
when(taskPluginFactory.createLogicTask(any())).thenAnswer(invocation -> logicTask);
135+
List<ILogicTaskPluginFactory<? extends ILogicTask<? extends AbstractParameters>>> taskPluginFactories =
136+
new ArrayList<>();
137+
taskPluginFactories.add(taskPluginFactory);
138+
LogicTaskExecutor taskExecutor = new LogicTaskExecutor(LogicTaskExecutorBuilder.builder()
139+
.taskExecutionContext(taskExecutionContext)
140+
.logicTaskPluginFactoryBuilder(new LogicTaskPluginFactoryBuilder(taskPluginFactories))
141+
.build());
142+
taskExecutor.start();
143+
return taskExecutor;
144+
}
145+
}

0 commit comments

Comments
 (0)