This is an automated email from the ASF dual-hosted git repository.
zihaoxiang pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/dolphinscheduler.git
The following commit(s) were added to refs/heads/dev by this push:
new 2cdf39fc09 [Fix-17555] [Master] TaskDispatchableEvent might block when
the queue exist hight priority delay event (#17556)
2cdf39fc09 is described below
commit 2cdf39fc099d9f25c1c2a7b696738ff0f4348a10
Author: Wenjun Ruan <[email protected]>
AuthorDate: Thu Oct 9 18:33:48 2025 +0800
[Fix-17555] [Master] TaskDispatchableEvent might block when the queue exist
hight priority delay event (#17556)
---
.../task/dispatcher/WorkerGroupDispatcher.java | 3 +-
.../dispatcher/event/TaskDispatchableEvent.java | 27 ++++++++++++++----
.../WorkerGroupDispatcherCoordinatorTest.java | 4 +++
.../task/dispatcher/WorkerGroupDispatcherTest.java | 32 ++++++++++++++++++++++
.../event/TaskDispatchableEventTest.java | 2 +-
5 files changed, 60 insertions(+), 8 deletions(-)
diff --git
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/dispatcher/WorkerGroupDispatcher.java
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/dispatcher/WorkerGroupDispatcher.java
index 7cb5f8d472..28834e27e7 100644
---
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/dispatcher/WorkerGroupDispatcher.java
+++
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/dispatcher/WorkerGroupDispatcher.java
@@ -115,7 +115,8 @@ public class WorkerGroupDispatcher extends BaseDaemonThread
{
*/
public void dispatchTask(final ITaskExecutionRunnable
taskExecutionRunnable, final long delayTimeMills) {
waitingDispatchTaskIds.add(taskExecutionRunnable.getId());
- workerGroupEventBus.add(new TaskDispatchableEvent<>(delayTimeMills,
taskExecutionRunnable));
+ workerGroupEventBus.add(new TaskDispatchableEvent<>(delayTimeMills,
taskExecutionRunnable,
+
taskExecutionRunnable.getTaskExecutionContext().getDispatchFailTimes()));
}
public boolean removeTask(ITaskExecutionRunnable taskExecutionRunnable) {
diff --git
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/dispatcher/event/TaskDispatchableEvent.java
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/dispatcher/event/TaskDispatchableEvent.java
index 9d99e4b958..02a1c7b788 100644
---
a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/dispatcher/event/TaskDispatchableEvent.java
+++
b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/dispatcher/event/TaskDispatchableEvent.java
@@ -30,9 +30,16 @@ public class TaskDispatchableEvent<V extends Comparable<V>>
extends AbstractDela
protected final V data;
+ protected final int dispatchTimes;
+
public TaskDispatchableEvent(long delayTimeMills, V data) {
+ this(delayTimeMills, data, 0);
+ }
+
+ public TaskDispatchableEvent(long delayTimeMills, V data, int
dispatchTimes) {
super(delayTimeMills);
this.data = checkNotNull(data, "data is null");
+ this.dispatchTimes = dispatchTimes;
}
@Override
@@ -44,14 +51,22 @@ public class TaskDispatchableEvent<V extends Comparable<V>>
extends AbstractDela
@SuppressWarnings("unchecked")
final TaskDispatchableEvent<V> otherEvent = (TaskDispatchableEvent<V>)
other;
- // there should compare data first for priority
- if (data != null && otherEvent.data != null) {
- final int compareResult = data.compareTo(otherEvent.data);
- if (compareResult != 0) {
- return compareResult;
+ // For two retry events, we should compare the priority first, since
the task delay time has already been
+ // expired.
+ if (dispatchTimes > 0 && otherEvent.dispatchTimes > 0) {
+ int priorityCompareResult = data.compareTo(otherEvent.data);
+ if (priorityCompareResult != 0) {
+ return priorityCompareResult;
}
+ return super.compareTo(otherEvent);
}
- return super.compareTo(other);
+ // For two new events, we should compare the delay time first, since
the delay time is not expired.
+ // For two evens, if one is new another is retry, we should compare
the delay time first
+ int delayTimeCompareResult = super.compareTo(otherEvent);
+ if (delayTimeCompareResult != 0) {
+ return delayTimeCompareResult;
+ }
+ return data.compareTo(otherEvent.data);
}
}
diff --git
a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/task/dispatcher/WorkerGroupDispatcherCoordinatorTest.java
b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/task/dispatcher/WorkerGroupDispatcherCoordinatorTest.java
index 9d39de3a20..e2e96a9614 100644
---
a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/task/dispatcher/WorkerGroupDispatcherCoordinatorTest.java
+++
b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/task/dispatcher/WorkerGroupDispatcherCoordinatorTest.java
@@ -23,6 +23,7 @@ import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
+import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
import
org.apache.dolphinscheduler.server.master.engine.task.client.ITaskExecutorClient;
import
org.apache.dolphinscheduler.server.master.engine.task.runnable.ITaskExecutionRunnable;
@@ -49,9 +50,12 @@ class WorkerGroupDispatcherCoordinatorTest {
ITaskExecutionRunnable taskExecutionRunnable =
Mockito.mock(ITaskExecutionRunnable.class);
TaskInstance taskInstance = mock(TaskInstance.class);
+ TaskExecutionContext taskExecutionContext =
mock(TaskExecutionContext.class);
when(taskExecutionRunnable.getTaskInstance()).thenReturn(taskInstance);
when(taskInstance.getWorkerGroup()).thenReturn(workerGroup);
+
when(taskExecutionRunnable.getTaskExecutionContext()).thenReturn(taskExecutionContext);
+ when(taskExecutionContext.getDispatchFailTimes()).thenReturn(1);
assertFalse(workerGroupDispatcherCoordinator.existWorkerGroup(workerGroup));
diff --git
a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/task/dispatcher/WorkerGroupDispatcherTest.java
b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/task/dispatcher/WorkerGroupDispatcherTest.java
index 7f5b0f4c8c..77525cb181 100644
---
a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/task/dispatcher/WorkerGroupDispatcherTest.java
+++
b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/task/dispatcher/WorkerGroupDispatcherTest.java
@@ -20,6 +20,7 @@ package
org.apache.dolphinscheduler.server.master.engine.task.dispatcher;
import static org.awaitility.Awaitility.await;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
@@ -35,6 +36,7 @@ import java.time.Duration;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.mockito.InOrder;
class WorkerGroupDispatcherTest {
@@ -52,6 +54,7 @@ class WorkerGroupDispatcherTest {
ITaskExecutionRunnable taskExecutionRunnable =
mock(ITaskExecutionRunnable.class);
TaskInstance taskInstance = mock(TaskInstance.class);
when(taskExecutionRunnable.getTaskInstance()).thenReturn(taskInstance);
+ when(taskExecutionRunnable.getTaskExecutionContext()).thenReturn(new
TaskExecutionContext());
dispatcher.start();
dispatcher.dispatchTask(taskExecutionRunnable, 0);
@@ -65,6 +68,7 @@ class WorkerGroupDispatcherTest {
ITaskExecutionRunnable taskExecutionRunnable =
mock(ITaskExecutionRunnable.class);
TaskInstance taskInstance = mock(TaskInstance.class);
when(taskExecutionRunnable.getTaskInstance()).thenReturn(taskInstance);
+ when(taskExecutionRunnable.getTaskExecutionContext()).thenReturn(new
TaskExecutionContext());
dispatcher.start();
dispatcher.dispatchTask(taskExecutionRunnable, 2000);
@@ -73,11 +77,39 @@ class WorkerGroupDispatcherTest {
.untilAsserted(() -> verify(taskExecutorClient,
times(1)).dispatch(taskExecutionRunnable));
}
+ @Test
+ void dispatchTask_withOneDelayTaskAnotherIsDispatchRetryTask() throws
TaskDispatchException {
+ final ITaskExecutionRunnable delayTaskExecutionRunnable =
mock(ITaskExecutionRunnable.class);
+ final TaskInstance delayTaskInstance = mock(TaskInstance.class);
+
when(delayTaskExecutionRunnable.getTaskInstance()).thenReturn(delayTaskInstance);
+
when(delayTaskExecutionRunnable.getTaskExecutionContext()).thenReturn(TaskExecutionContext.builder().build());
+ when(delayTaskExecutionRunnable.getId()).thenReturn(1);
+
+ final ITaskExecutionRunnable dispatchRetryTaskExecutionRunnable =
mock(ITaskExecutionRunnable.class);
+ final TaskInstance dispatchRetryTaskInstance =
mock(TaskInstance.class);
+
when(dispatchRetryTaskExecutionRunnable.getTaskInstance()).thenReturn(dispatchRetryTaskInstance);
+ when(dispatchRetryTaskExecutionRunnable.getTaskExecutionContext())
+
.thenReturn(TaskExecutionContext.builder().dispatchFailTimes(1).build());
+ when(dispatchRetryTaskExecutionRunnable.getId()).thenReturn(2);
+ dispatcher.start();
+
+ dispatcher.dispatchTask(delayTaskExecutionRunnable, 2000);
+ dispatcher.dispatchTask(dispatchRetryTaskExecutionRunnable, 100);
+ await()
+ .atLeast(Duration.ofMillis(1500))
+ .untilAsserted(() -> verify(taskExecutorClient,
times(2)).dispatch(any()));
+
+ InOrder inOrder = inOrder(taskExecutorClient);
+
inOrder.verify(taskExecutorClient).dispatch(dispatchRetryTaskExecutionRunnable);
+
inOrder.verify(taskExecutorClient).dispatch(delayTaskExecutionRunnable);
+ }
+
@Test
void dispatchTask_HasBeenRemoved() {
ITaskExecutionRunnable taskExecutionRunnable =
mock(ITaskExecutionRunnable.class);
TaskInstance taskInstance = mock(TaskInstance.class);
when(taskExecutionRunnable.getTaskInstance()).thenReturn(taskInstance);
+ when(taskExecutionRunnable.getTaskExecutionContext()).thenReturn(new
TaskExecutionContext());
dispatcher.dispatchTask(taskExecutionRunnable, 0);
dispatcher.removeTask(taskExecutionRunnable);
diff --git
a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/task/dispatcher/event/TaskDispatchableEventTest.java
b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/task/dispatcher/event/TaskDispatchableEventTest.java
index 315ab0df30..54f6705264 100644
---
a/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/task/dispatcher/event/TaskDispatchableEventTest.java
+++
b/dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/engine/task/dispatcher/event/TaskDispatchableEventTest.java
@@ -38,6 +38,6 @@ class TaskDispatchableEventTest {
TaskDispatchableEvent<String> highPriorityEntry =
new TaskDispatchableEvent<>(15_000L, "1_HIGH");
TaskDispatchableEvent<String> lowPriorityEntry = new
TaskDispatchableEvent<>(5_000L, "3_LOW");
- assertThat(highPriorityEntry.compareTo(lowPriorityEntry) < 0).isTrue();
+ assertThat(highPriorityEntry.compareTo(lowPriorityEntry) > 0).isTrue();
}
}
