ruanwenjun commented on code in PR #17037:
URL:
https://github.com/apache/dolphinscheduler/pull/17037#discussion_r1987346981
##########
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/GlobalTaskDispatchWaitingQueue.java:
##########
@@ -57,19 +59,22 @@ public synchronized void
dispatchTaskExecuteRunnable(ITaskExecutionRunnable ITas
public synchronized void
dispatchTaskExecuteRunnableWithDelay(ITaskExecutionRunnable
taskExecutionRunnable,
long
delayTimeMills) {
waitingTaskInstanceIds.add(taskExecutionRunnable.getTaskInstance().getId());
- priorityDelayQueue.add(new DelayEntry<>(delayTimeMills,
taskExecutionRunnable));
+ priorityQueue.add(new
TimeBasedTaskExecutionRunnableComparableEntry(delayTimeMills,
taskExecutionRunnable));
}
/**
- * Consume {@link ITaskExecutionRunnable} from the {@link
PriorityDelayQueue}, only the delay time <= 0 can be consumed.
+ * Consume {@link ITaskExecutionRunnable} from the {@link DelayQueue},
only the delay time <= 0 can be consumed.
*/
@SneakyThrows
- public ITaskExecutionRunnable takeTaskExecuteRunnable() {
- ITaskExecutionRunnable taskExecutionRunnable =
priorityDelayQueue.take().getData();
+ public TimeBasedTaskExecutionRunnableComparableEntry
takeTaskExecuteRunnable() {
Review Comment:
Why not return `ITaskExecutionRunnable`? Why expose
`TimeBasedTaskExecutionRunnableComparableEntry` to outside?
##########
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/GlobalTaskDispatchWaitingQueue.java:
##########
@@ -31,24 +32,25 @@
/**
* The class is used to store {@link ITaskExecutionRunnable} which needs to be
dispatched. The {@link ITaskExecutionRunnable}
- * will be stored in {@link PriorityDelayQueue}, if the {@link
ITaskExecutionRunnable}'s delay time is 0, then it will be
+ * will be stored in {@link DelayQueue}, if the {@link
ITaskExecutionRunnable}'s delay time is 0, then it will be
* consumed by {@link GlobalTaskDispatchWaitingQueueLooper}.
* <p>
- * The order of {@link ITaskExecutionRunnable} in the {@link
PriorityDelayQueue} is determined by {@link ITaskExecutionRunnable#compareTo}.
+ * The order of {@link ITaskExecutionRunnable} in the {@link DelayQueue} is
determined by {@link ITaskExecutionRunnable#compareTo}.
*/
@Slf4j
@Component
public class GlobalTaskDispatchWaitingQueue {
private final Set<Integer> waitingTaskInstanceIds =
ConcurrentHashMap.newKeySet();
- private final PriorityDelayQueue<DelayEntry<ITaskExecutionRunnable>>
priorityDelayQueue =
- new PriorityDelayQueue<>();
+
+ private final
PriorityBlockingQueue<TimeBasedTaskExecutionRunnableComparableEntry>
priorityQueue =
Review Comment:
```suggestion
private final
PriorityBlockingQueue<TimeBasedTaskExecutionRunnableComparableEntry> delayQueue
=
```
##########
dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/cluster/ClusterManager.java:
##########
@@ -48,9 +50,13 @@ public class ClusterManager {
@Autowired
private RegistryClient registryClient;
+ private final WorkerGroupTaskDispatchManager
workerGroupTaskDispatchManager;
+
public ClusterManager() {
this.masterClusters = new MasterClusters();
this.workerClusters = new WorkerClusters();
+ workerGroupTaskDispatchManager = new WorkerGroupTaskDispatchManager();
Review Comment:
Can this work? `WorkerGroupTaskDispatchManager` is a spring bean, did you
test this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
