Skip to content

Commit 4c348ed

Browse files
ruanwenjunruanwenjun
authored
Rename IWorkflowExecutionRunnable/ITaskExecutionRunnable to IWorkflowExecution/ITaskExecution (#18163)
1 parent ce0bc70 commit 4c348ed

141 files changed

Lines changed: 2000 additions & 2000 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/ILifecycleEventHandler.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -17,14 +17,14 @@
1717

1818
package org.apache.dolphinscheduler.server.master.engine;
1919

20-
import org.apache.dolphinscheduler.server.master.engine.workflow.runnable.IWorkflowExecutionRunnable;
20+
import org.apache.dolphinscheduler.server.master.engine.workflow.execution.IWorkflowExecution;
2121

2222
/**
2323
* The event handler interface, used to handle the event
2424
*/
2525
public interface ILifecycleEventHandler<T extends AbstractLifecycleEvent> {
2626

27-
void handle(final IWorkflowExecutionRunnable workflowExecutionRunnable,
27+
void handle(final IWorkflowExecution workflowExecution,
2828
final T event);
2929

3030
ILifecycleEventType matchEventType();

‎dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/IWorkflowRepository.java‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -17,19 +17,19 @@
1717

1818
package org.apache.dolphinscheduler.server.master.engine;
1919

20-
import org.apache.dolphinscheduler.server.master.engine.workflow.runnable.IWorkflowExecutionRunnable;
20+
import org.apache.dolphinscheduler.server.master.engine.workflow.execution.IWorkflowExecution;
2121

2222
import java.util.Collection;
2323

2424
import lombok.NonNull;
2525

2626
public interface IWorkflowRepository {
2727

28-
IWorkflowExecutionRunnable get(int workflowInstanceId);
28+
IWorkflowExecution get(int workflowInstanceId);
2929

30-
Collection<IWorkflowExecutionRunnable> getAll();
30+
Collection<IWorkflowExecution> getAll();
3131

32-
void put(@NonNull IWorkflowExecutionRunnable workflowExecuteThread);
32+
void put(@NonNull IWorkflowExecution workflowExecuteThread);
3333

3434
boolean contains(int workflowInstanceId);
3535

‎dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/WorkflowCacheRepository.java‎

Lines changed: 12 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,7 @@
1717

1818
package org.apache.dolphinscheduler.server.master.engine;
1919

20-
import org.apache.dolphinscheduler.server.master.engine.workflow.runnable.IWorkflowExecutionRunnable;
20+
import org.apache.dolphinscheduler.server.master.engine.workflow.execution.IWorkflowExecution;
2121
import org.apache.dolphinscheduler.server.master.metrics.WorkflowInstanceMetrics;
2222

2323
import java.util.Collection;
@@ -35,37 +35,37 @@
3535
@Component
3636
public class WorkflowCacheRepository implements IWorkflowRepository {
3737

38-
private final Map<Integer, IWorkflowExecutionRunnable> workflowExecutionRunnableMap = new ConcurrentHashMap<>();
38+
private final Map<Integer, IWorkflowExecution> workflowExecutionMap = new ConcurrentHashMap<>();
3939

4040
@PostConstruct
4141
public void registerMetrics() {
42-
WorkflowInstanceMetrics.registerWorkflowInstanceRunningGauge(workflowExecutionRunnableMap::size);
42+
WorkflowInstanceMetrics.registerWorkflowInstanceRunningGauge(workflowExecutionMap::size);
4343
}
4444

4545
@Override
46-
public IWorkflowExecutionRunnable get(final int workflowInstanceId) {
47-
return workflowExecutionRunnableMap.get(workflowInstanceId);
46+
public IWorkflowExecution get(final int workflowInstanceId) {
47+
return workflowExecutionMap.get(workflowInstanceId);
4848
}
4949

5050
@Override
5151
public boolean contains(final int workflowInstanceId) {
52-
return workflowExecutionRunnableMap.containsKey(workflowInstanceId);
52+
return workflowExecutionMap.containsKey(workflowInstanceId);
5353
}
5454

5555
@Override
5656
public void remove(final int workflowInstanceId) {
57-
workflowExecutionRunnableMap.remove(workflowInstanceId);
57+
workflowExecutionMap.remove(workflowInstanceId);
5858
}
5959

6060
@Override
61-
public void put(@NonNull final IWorkflowExecutionRunnable workflowExecutionRunnable) {
62-
final Integer workflowInstanceId = workflowExecutionRunnable.getId();
63-
workflowExecutionRunnableMap.put(workflowInstanceId, workflowExecutionRunnable);
61+
public void put(@NonNull final IWorkflowExecution workflowExecution) {
62+
final Integer workflowInstanceId = workflowExecution.getId();
63+
workflowExecutionMap.put(workflowInstanceId, workflowExecution);
6464
}
6565

6666
@Override
67-
public Collection<IWorkflowExecutionRunnable> getAll() {
68-
return ImmutableList.copyOf(workflowExecutionRunnableMap.values());
67+
public Collection<IWorkflowExecution> getAll() {
68+
return ImmutableList.copyOf(workflowExecutionMap.values());
6969
}
7070

7171
}

‎dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/WorkflowEventBusCoordinator.java‎

Lines changed: 9 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,7 @@
1818
package org.apache.dolphinscheduler.server.master.engine;
1919

2020
import org.apache.dolphinscheduler.dao.entity.WorkflowInstance;
21-
import org.apache.dolphinscheduler.server.master.engine.workflow.runnable.IWorkflowExecutionRunnable;
21+
import org.apache.dolphinscheduler.server.master.engine.workflow.execution.IWorkflowExecution;
2222
import org.apache.dolphinscheduler.server.master.runner.IWorkflowExecuteContext;
2323

2424
import lombok.extern.slf4j.Slf4j;
@@ -44,20 +44,20 @@ public void start() {
4444
* Register a WorkflowExecuteRunnable to the corresponding WorkflowEventBusFireWorker, once the WorkflowExecuteRunnable has been registered,
4545
* then the event will auto handler by the WorkflowEventBusFireWorker
4646
*/
47-
public void registerWorkflowEventBus(IWorkflowExecutionRunnable workflowExecutionRunnable) {
48-
final int workerSlot = calculateWorkflowEventBusFireWorkerSlot(workflowExecutionRunnable);
47+
public void registerWorkflowEventBus(IWorkflowExecution workflowExecution) {
48+
final int workerSlot = calculateWorkflowEventBusFireWorkerSlot(workflowExecution);
4949
final WorkflowEventBusFireWorker workflowEventBusFireWorker = workflowEventBusFireWorkers.getWorker(workerSlot);
50-
workflowEventBusFireWorker.registerWorkflowEventBus(workflowExecutionRunnable);
50+
workflowEventBusFireWorker.registerWorkflowEventBus(workflowExecution);
5151
}
5252

5353
/**
5454
* UeRegister a WorkflowExecuteRunnable to the corresponding WorkflowEventBusFireWorker, once the WorkflowExecuteRunnable has been deregistered,
5555
* then the EventBus will be removed from the WorkflowEventBusFireWorker.
5656
*/
57-
public void unRegisterWorkflowEventBus(IWorkflowExecutionRunnable workflowExecutionRunnable) {
58-
final int workerSlot = calculateWorkflowEventBusFireWorkerSlot(workflowExecutionRunnable);
57+
public void unRegisterWorkflowEventBus(IWorkflowExecution workflowExecution) {
58+
final int workerSlot = calculateWorkflowEventBusFireWorkerSlot(workflowExecution);
5959
final WorkflowEventBusFireWorker workflowEventBusFireWorker = workflowEventBusFireWorkers.getWorker(workerSlot);
60-
workflowEventBusFireWorker.unRegisterWorkflowEventBus(workflowExecutionRunnable);
60+
workflowEventBusFireWorker.unRegisterWorkflowEventBus(workflowExecution);
6161
}
6262

6363
/**
@@ -66,8 +66,8 @@ public void unRegisterWorkflowEventBus(IWorkflowExecutionRunnable workflowExecut
6666
* <p> e.g. If the workflowInstanceId is 1, and the workerSize is 3, then the slot is 1, the workflow will be registered to the worker[1].
6767
* <p> If the workflowInstanceIds are not consecutive numbers, these will cause some worker busy.
6868
*/
69-
private int calculateWorkflowEventBusFireWorkerSlot(IWorkflowExecutionRunnable workflowExecutionRunnable) {
70-
final IWorkflowExecuteContext workflowExecuteContext = workflowExecutionRunnable.getWorkflowExecuteContext();
69+
private int calculateWorkflowEventBusFireWorkerSlot(IWorkflowExecution workflowExecution) {
70+
final IWorkflowExecuteContext workflowExecuteContext = workflowExecution.getWorkflowExecuteContext();
7171
final WorkflowInstance workflowInstance = workflowExecuteContext.getWorkflowInstance();
7272
final Integer workflowInstanceId = workflowInstance.getId();
7373
return workflowInstanceId % workflowEventBusFireWorkers.getWorkerSize();

‎dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/WorkflowEventBusFireWorker.java‎

Lines changed: 21 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,7 @@
2626
import org.apache.dolphinscheduler.server.master.engine.exceptions.WorkflowEventFireException;
2727
import org.apache.dolphinscheduler.server.master.engine.task.lifecycle.AbstractTaskLifecycleEvent;
2828
import org.apache.dolphinscheduler.server.master.engine.task.lifecycle.event.TaskFatalLifecycleEvent;
29-
import org.apache.dolphinscheduler.server.master.engine.workflow.runnable.IWorkflowExecutionRunnable;
29+
import org.apache.dolphinscheduler.server.master.engine.workflow.execution.IWorkflowExecution;
3030
import org.apache.dolphinscheduler.server.master.runner.IWorkflowExecuteContext;
3131
import org.apache.dolphinscheduler.server.master.utils.ExceptionUtils;
3232

@@ -47,7 +47,7 @@
4747
@SuppressWarnings({"rawtypes", "unchecked"})
4848
public class WorkflowEventBusFireWorker {
4949

50-
private final Map<Integer, IWorkflowExecutionRunnable> registeredWorkflowExecuteRunnableMap =
50+
private final Map<Integer, IWorkflowExecution> registeredWorkflowExecuteRunnableMap =
5151
new ConcurrentHashMap<>();
5252

5353
private final Map<ILifecycleEventType, ILifecycleEventHandler> eventHandlerMap = new ConcurrentHashMap<>();
@@ -58,35 +58,35 @@ public void registerEventHandler(ILifecycleEventHandler eventHandler) {
5858
eventHandlerMap.put(eventHandler.matchEventType(), eventHandler);
5959
}
6060

61-
public void registerWorkflowEventBus(IWorkflowExecutionRunnable workflowExecutionRunnable) {
62-
final IWorkflowExecuteContext workflowExecuteContext = workflowExecutionRunnable.getWorkflowExecuteContext();
61+
public void registerWorkflowEventBus(IWorkflowExecution workflowExecution) {
62+
final IWorkflowExecuteContext workflowExecuteContext = workflowExecution.getWorkflowExecuteContext();
6363
final WorkflowInstance workflowInstance = workflowExecuteContext.getWorkflowInstance();
6464
final Integer workflowInstanceId = workflowInstance.getId();
6565
final String workflowInstanceName = workflowInstance.getName();
6666
checkState(!registeredWorkflowExecuteRunnableMap.containsKey(workflowInstanceId),
6767
"WorkflowExecuteRunnable(%s/%s already registered at WorkflowEventBusFireWorker", workflowInstanceId,
6868
workflowInstanceName);
69-
registeredWorkflowExecuteRunnableMap.put(workflowInstanceId, workflowExecutionRunnable);
69+
registeredWorkflowExecuteRunnableMap.put(workflowInstanceId, workflowExecution);
7070
}
7171

72-
public void unRegisterWorkflowEventBus(IWorkflowExecutionRunnable workflowExecutionRunnable) {
73-
final IWorkflowExecuteContext workflowExecuteContext = workflowExecutionRunnable.getWorkflowExecuteContext();
72+
public void unRegisterWorkflowEventBus(IWorkflowExecution workflowExecution) {
73+
final IWorkflowExecuteContext workflowExecuteContext = workflowExecution.getWorkflowExecuteContext();
7474
final WorkflowInstance workflowInstance = workflowExecuteContext.getWorkflowInstance();
7575
final Integer workflowInstanceId = workflowInstance.getId();
76-
registeredWorkflowExecuteRunnableMap.remove(workflowInstanceId, workflowExecutionRunnable);
76+
registeredWorkflowExecuteRunnableMap.remove(workflowInstanceId, workflowExecution);
7777
}
7878

7979
public void fireAllRegisteredEvent() {
80-
final List<IWorkflowExecutionRunnable> workflowExecutionRunnables = getWaitingFireWorkflowExecutionRunnables();
81-
if (CollectionUtils.isEmpty(workflowExecutionRunnables)) {
80+
final List<IWorkflowExecution> workflowExecutions = getWaitingFireWorkflowExecutions();
81+
if (CollectionUtils.isEmpty(workflowExecutions)) {
8282
return;
8383
}
84-
for (final IWorkflowExecutionRunnable workflowExecutionRunnable : workflowExecutionRunnables) {
85-
final Integer workflowInstanceId = workflowExecutionRunnable.getId();
86-
final String workflowInstanceName = workflowExecutionRunnable.getName();
84+
for (final IWorkflowExecution workflowExecution : workflowExecutions) {
85+
final Integer workflowInstanceId = workflowExecution.getId();
86+
final String workflowInstanceName = workflowExecution.getName();
8787
try {
8888
LogUtils.setWorkflowInstanceIdMDC(workflowInstanceId);
89-
doFireSingleWorkflowEventBus(workflowExecutionRunnable);
89+
doFireSingleWorkflowEventBus(workflowExecution);
9090
} catch (Exception ex) {
9191
log.error("Fire event failed for WorkflowExecuteRunnable: {}", workflowInstanceName, ex);
9292
} finally {
@@ -99,7 +99,7 @@ public int getRegisteredWorkflowExecuteRunnableSize() {
9999
return registeredWorkflowExecuteRunnableMap.size();
100100
}
101101

102-
private List<IWorkflowExecutionRunnable> getWaitingFireWorkflowExecutionRunnables() {
102+
private List<IWorkflowExecution> getWaitingFireWorkflowExecutions() {
103103
if (MapUtils.isEmpty(registeredWorkflowExecuteRunnableMap)) {
104104
return Collections.emptyList();
105105
}
@@ -109,8 +109,8 @@ private List<IWorkflowExecutionRunnable> getWaitingFireWorkflowExecutionRunnable
109109
.collect(Collectors.toList());
110110
}
111111

112-
private void doFireSingleWorkflowEventBus(final IWorkflowExecutionRunnable workflowExecutionRunnable) {
113-
final WorkflowEventBus workflowEventBus = workflowExecutionRunnable.getWorkflowEventBus();
112+
private void doFireSingleWorkflowEventBus(final IWorkflowExecution workflowExecution) {
113+
final WorkflowEventBus workflowEventBus = workflowExecution.getWorkflowEventBus();
114114
while (!workflowEventBus.isEmpty()) {
115115
Optional<AbstractLifecycleEvent> eventOptional = workflowEventBus.poll();
116116
if (!eventOptional.isPresent()) {
@@ -122,7 +122,7 @@ private void doFireSingleWorkflowEventBus(final IWorkflowExecutionRunnable workf
122122
// So we increase the event count before the event fired then we can get the correct event count
123123
// And if the event handle failed we will decrease the success event count
124124
workflowEventBus.getWorkflowEventBusSummary().increaseFireSuccessEventCount();
125-
doFireSingleEvent(workflowExecutionRunnable, lifecycleEvent);
125+
doFireSingleEvent(workflowExecution, lifecycleEvent);
126126
} catch (Exception ex) {
127127
// If the database connection is failed, do not remove the event from the event bus
128128
// so that the event can be fired again when the database connection is recovered
@@ -136,7 +136,7 @@ private void doFireSingleWorkflowEventBus(final IWorkflowExecutionRunnable workf
136136
if (ExceptionUtils.isTaskExecutionContextCreateException(ex)) {
137137
AbstractTaskLifecycleEvent taskLifecycleEvent = (AbstractTaskLifecycleEvent) lifecycleEvent;
138138
final TaskFatalLifecycleEvent taskFatalEvent = TaskFatalLifecycleEvent.builder()
139-
.taskExecutionRunnable(taskLifecycleEvent.getTaskExecutionRunnable())
139+
.taskExecution(taskLifecycleEvent.getTaskExecution())
140140
.endTime(new Date())
141141
.build();
142142
workflowEventBus.publish(taskFatalEvent);
@@ -149,13 +149,13 @@ private void doFireSingleWorkflowEventBus(final IWorkflowExecutionRunnable workf
149149
}
150150
}
151151

152-
private void doFireSingleEvent(final IWorkflowExecutionRunnable workflowExecutionRunnable,
152+
private void doFireSingleEvent(final IWorkflowExecution workflowExecution,
153153
final AbstractLifecycleEvent event) {
154154
final ILifecycleEventHandler lifecycleEventHandler = eventHandlerMap.get(event.getEventType());
155155
if (lifecycleEventHandler == null) {
156156
throw new RuntimeException("No EventHandler found for event: " + event.getEventType());
157157
}
158-
lifecycleEventHandler.handle(workflowExecutionRunnable, event);
158+
lifecycleEventHandler.handle(workflowExecution, event);
159159
}
160160

161161
}

‎dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/command/CommandEngine.java‎

Lines changed: 12 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -35,9 +35,9 @@
3535
import org.apache.dolphinscheduler.server.master.engine.IWorkflowRepository;
3636
import org.apache.dolphinscheduler.server.master.engine.WorkflowEventBusCoordinator;
3737
import org.apache.dolphinscheduler.server.master.engine.exceptions.CommandDuplicateHandleException;
38+
import org.apache.dolphinscheduler.server.master.engine.workflow.execution.IWorkflowExecution;
39+
import org.apache.dolphinscheduler.server.master.engine.workflow.execution.WorkflowExecutionFactory;
3840
import org.apache.dolphinscheduler.server.master.engine.workflow.lifecycle.event.WorkflowStartLifecycleEvent;
39-
import org.apache.dolphinscheduler.server.master.engine.workflow.runnable.IWorkflowExecutionRunnable;
40-
import org.apache.dolphinscheduler.server.master.engine.workflow.runnable.WorkflowExecutionRunnableFactory;
4141
import org.apache.dolphinscheduler.server.master.metrics.MasterServerMetrics;
4242
import org.apache.dolphinscheduler.server.master.metrics.WorkflowInstanceMetrics;
4343
import org.apache.dolphinscheduler.service.command.CommandService;
@@ -82,7 +82,7 @@ public class CommandEngine extends BaseDaemonThread implements AutoCloseable {
8282
private WorkflowInstanceDao workflowInstanceDao;
8383

8484
@Autowired
85-
private WorkflowExecutionRunnableFactory workflowExecutionRunnableFactory;
85+
private WorkflowExecutionFactory workflowExecutionFactory;
8686

8787
@Autowired
8888
private MetricsProvider metricsProvider;
@@ -144,7 +144,7 @@ public void run() {
144144
return command;
145145
}, commandHandleThreadPool)
146146
.thenApply(this::bootstrapCommand)
147-
.thenAccept(this::bootstrapWorkflowExecutionRunnable)
147+
.thenAccept(this::bootstrapWorkflowExecution)
148148
.thenAccept((unused) -> bootstrapSuccess(command))
149149
.exceptionally(throwable -> bootstrapError(command, throwable))
150150
.whenComplete((result, throwable) -> LogUtils.removeWorkflowInstanceIdMDC());
@@ -163,13 +163,13 @@ public void run() {
163163
}
164164
}
165165

166-
private IWorkflowExecutionRunnable bootstrapCommand(Command command) {
167-
return workflowExecutionRunnableFactory.createWorkflowExecuteRunnable(command);
166+
private IWorkflowExecution bootstrapCommand(Command command) {
167+
return workflowExecutionFactory.createWorkflowExecuteRunnable(command);
168168
}
169169

170-
private CompletableFuture<Void> bootstrapWorkflowExecutionRunnable(IWorkflowExecutionRunnable workflowExecutionRunnable) {
170+
private CompletableFuture<Void> bootstrapWorkflowExecution(IWorkflowExecution workflowExecution) {
171171
final WorkflowInstance workflowInstance =
172-
workflowExecutionRunnable.getWorkflowExecuteContext().getWorkflowInstance();
172+
workflowExecution.getWorkflowExecuteContext().getWorkflowInstance();
173173

174174
if (workflowInstance.getState() == WorkflowExecutionStatus.SERIAL_WAIT) {
175175
log.info("The workflow {} state is: {} will not be trigger now",
@@ -179,10 +179,10 @@ private CompletableFuture<Void> bootstrapWorkflowExecutionRunnable(IWorkflowExec
179179
}
180180

181181
WorkflowInstanceMetrics.recordWorkflowInstanceSubmit(workflowInstance.getWorkflowDefinitionCode());
182-
workflowRepository.put(workflowExecutionRunnable);
183-
workflowEventBusCoordinator.registerWorkflowEventBus(workflowExecutionRunnable);
184-
workflowExecutionRunnable.getWorkflowEventBus()
185-
.publish(WorkflowStartLifecycleEvent.of(workflowExecutionRunnable));
182+
workflowRepository.put(workflowExecution);
183+
workflowEventBusCoordinator.registerWorkflowEventBus(workflowExecution);
184+
workflowExecution.getWorkflowEventBus()
185+
.publish(WorkflowStartLifecycleEvent.of(workflowExecution));
186186
return CompletableFuture.completedFuture(null);
187187
}
188188

‎dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/command/ICommandHandler.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,7 @@
2222
import org.apache.dolphinscheduler.server.master.engine.command.handler.ReRunWorkflowCommandHandler;
2323
import org.apache.dolphinscheduler.server.master.engine.command.handler.RecoverFailureTaskCommandHandler;
2424
import org.apache.dolphinscheduler.server.master.engine.command.handler.RunWorkflowCommandHandler;
25-
import org.apache.dolphinscheduler.server.master.engine.workflow.runnable.WorkflowExecutionRunnable;
25+
import org.apache.dolphinscheduler.server.master.engine.workflow.execution.WorkflowExecution;
2626

2727
/**
2828
* The interface represent the handler used to handle the {@link Command}.
@@ -35,9 +35,9 @@
3535
public interface ICommandHandler {
3636

3737
/**
38-
* Handle the command and return the WorkflowExecutionRunnable.
38+
* Handle the command and return the WorkflowExecution.
3939
*/
40-
WorkflowExecutionRunnable handleCommand(final Command command);
40+
WorkflowExecution handleCommand(final Command command);
4141

4242
/**
4343
* The type of the command which should be handled by this handler.

0 commit comments

Comments
 (0)