Skip to content

improve: lazy init workflow executor #2266

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 4 commits into from
Mar 7, 2024
Merged

improve: lazy init workflow executor #2266

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
b356d0b
improve: lazy init workflow executor
csviri Mar 7, 2024
5ddb46d
stop fix
csviri Mar 7, 2024
1bc4dd8
refactor: clean up
metacosm Mar 7, 2024
1559d7b
fix
csviri Mar 7, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 16 additions & 3 deletions ...ork-core/src/main/java/io/javaoperatorsdk/operator/api/config/ExecutorServiceManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@ public class ExecutorServiceManager {
private ExecutorService workflowExecutor;
private ExecutorService cachingExecutorService;
private boolean started;
private ConfigurationService configurationService;

ExecutorServiceManager(ConfigurationService configurationService) {
start(configurationService);
Expand Down Expand Up @@ -95,30 +96,38 @@ public ExecutorService reconcileExecutorService() {
}

public ExecutorService workflowExecutorService() {
lazyInitWorkflowExecutorService();
return workflowExecutor;
}

private synchronized void lazyInitWorkflowExecutorService() {
if (workflowExecutor == null) {
workflowExecutor =
new InstrumentedExecutorService(configurationService.getWorkflowExecutorService());
}
}

public ExecutorService cachingExecutorService() {
return cachingExecutorService;
}

public void start(ConfigurationService configurationService) {
if (!started) {
this.configurationService = configurationService; // used to lazy init workflow executor
this.cachingExecutorService = Executors.newCachedThreadPool();
this.executor = new InstrumentedExecutorService(configurationService.getExecutorService());
this.workflowExecutor =
new InstrumentedExecutorService(configurationService.getWorkflowExecutorService());
started = true;
}
}

public void stop(Duration gracefulShutdownTimeout) {
try {
var parallelExec = Executors.newFixedThreadPool(3);
log.debug("Closing executor");
var parallelExec = Executors.newFixedThreadPool(3);
parallelExec.invokeAll(List.of(shutdown(executor, gracefulShutdownTimeout),
shutdown(workflowExecutor, gracefulShutdownTimeout),
shutdown(cachingExecutorService, gracefulShutdownTimeout)));
workflowExecutor = null;
parallelExec.shutdownNow();
started = false;
} catch (InterruptedException e) {
Expand All @@ -130,6 +139,10 @@ public void stop(Duration gracefulShutdownTimeout) {
private static Callable<Void> shutdown(ExecutorService executorService,
Duration gracefulShutdownTimeout) {
return () -> {
// workflow executor can be null
if (executorService == null) {
return null;
}
executorService.shutdown();
if (!executorService.awaitTermination(gracefulShutdownTimeout.toMillis(),
TimeUnit.MILLISECONDS)) {
Expand Down