[Fix #722] Close executor service if created

Signed-off-by: fjtirado <[email protected]>

Author: fjtirado

impl/core/src/main/java/io/serverlessworkflow/impl/ExecutorServiceFactory.java

@@ -18,5 +18,6 @@
 import java.util.concurrent.ExecutorService;
 import java.util.function.Supplier;
 
-@FunctionalInterface
-public interface ExecutorServiceFactory extends Supplier<ExecutorService> {}
+public interface ExecutorServiceFactory extends Supplier<ExecutorService>, AutoCloseable {
+  void close();
+}

impl/core/src/main/java/io/serverlessworkflow/impl/DefaultExecutorServiceFactory.java renamed to impl/core/src/main/java/io/serverlessworkflow/impl/ExecutorServiceHolder.java

@@ -17,23 +17,31 @@
 
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 
-public class DefaultExecutorServiceFactory implements ExecutorServiceFactory {
+public class ExecutorServiceHolder implements ExecutorServiceFactory {
 
-  private static final ExecutorServiceFactory instance = new DefaultExecutorServiceFactory();
-
-  public static ExecutorServiceFactory instance() {
-    return instance;
-  }
-
-  private static class ExecutorServiceHolder {
-    private static ExecutorService instance = Executors.newCachedThreadPool();
-  }
+  private ExecutorService service;
+  private Lock serviceLock = new ReentrantLock();
 
   @Override
   public ExecutorService get() {
-    return ExecutorServiceHolder.instance;
+    try {
+      serviceLock.lock();
+      if (service == null) {
+        service = Executors.newCachedThreadPool();
+      }
+    } finally {
+      serviceLock.unlock();
+    }
+    return service;
   }
 
-  private DefaultExecutorServiceFactory() {}
+  @Override
+  public void close() {
+    if (service != null) {
+      service.shutdown();
+    }
+  }
 }

impl/core/src/main/java/io/serverlessworkflow/impl/WorkflowApplication.java

@@ -39,11 +39,14 @@
 import java.util.ServiceLoader.Provider;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
 import java.util.stream.Collectors;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class WorkflowApplication implements AutoCloseable {
 
+  private static final Logger logger = LoggerFactory.getLogger(WorkflowApplication.class);
+
   private final TaskExecutorFactory taskFactory;
   private final ExpressionFactory exprFactory;
   private final ResourceLoaderFactory resourceLoaderFactory;
@@ -137,7 +140,7 @@ public SchemaValidator getValidator(SchemaInline inline) {
     private SchemaValidatorFactory schemaValidatorFactory;
     private WorkflowPositionFactory positionFactory = () -> new QueueWorkflowPosition();
     private WorkflowIdFactory idFactory = () -> UlidCreator.getMonotonicUlid().toString();
-    private ExecutorServiceFactory executorFactory = () -> Executors.newCachedThreadPool();
+    private ExecutorServiceFactory executorFactory = new ExecutorServiceHolder();
     private EventConsumer<?, ?> eventConsumer = InMemoryEvents.get();
     private EventPublisher eventPublisher = InMemoryEvents.get();
     private RuntimeDescriptorFactory descriptorFactory =
@@ -236,12 +239,23 @@ public WorkflowDefinition workflowDefinition(Workflow workflow) {
 
   @Override
   public void close() {
+    safeClose(executorFactory);
+    safeClose(eventPublisher);
+    safeClose(eventConsumer);
     for (WorkflowDefinition definition : definitions.values()) {
-      definition.close();
+      safeClose(definition);
     }
     definitions.clear();
   }
 
+  private void safeClose(AutoCloseable closeable) {
+    try {
+      closeable.close();
+    } catch (Exception ex) {
+      logger.warn("Error closing resource {}", closeable.getClass().getName(), ex);
+    }
+  }
+
   public WorkflowPositionFactory positionFactory() {
     return positionFactory;
   }

impl/core/src/main/java/io/serverlessworkflow/impl/events/EventConsumer.java

@@ -21,7 +21,8 @@
 import java.util.Collection;
 import java.util.function.Consumer;
 
-public interface EventConsumer<T extends EventRegistration, V extends EventRegistrationBuilder> {
+public interface EventConsumer<T extends EventRegistration, V extends EventRegistrationBuilder>
+    extends AutoCloseable {
 
   V listen(EventFilter filter, WorkflowApplication workflowApplication);
 
@@ -30,4 +31,6 @@ public interface EventConsumer<T extends EventRegistration, V extends EventRegis
   T register(V builder, Consumer<CloudEvent> consumer);
 
   void unregister(T register);
+
+  void close();
 }

impl/core/src/main/java/io/serverlessworkflow/impl/events/EventPublisher.java

@@ -18,6 +18,8 @@
 import io.cloudevents.CloudEvent;
 import java.util.concurrent.CompletableFuture;
 
-public interface EventPublisher {
+public interface EventPublisher extends AutoCloseable {
   CompletableFuture<Void> publish(CloudEvent event);
+
+  void close();
 }

impl/core/src/main/java/io/serverlessworkflow/impl/events/InMemoryEvents.java

@@ -16,7 +16,8 @@
 package io.serverlessworkflow.impl.events;
 
 import io.cloudevents.CloudEvent;
-import io.serverlessworkflow.impl.DefaultExecutorServiceFactory;
+import io.serverlessworkflow.impl.ExecutorServiceFactory;
+import io.serverlessworkflow.impl.ExecutorServiceHolder;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
@@ -37,6 +38,8 @@ public static InMemoryEvents get() {
     return instance;
   }
 
+  private ExecutorServiceFactory serviceFactory = new ExecutorServiceHolder();
+
   private Map<String, Consumer<CloudEvent>> topicMap = new ConcurrentHashMap<>();
 
   private AtomicReference<Consumer<CloudEvent>> allConsumerRef = new AtomicReference<>();
@@ -64,7 +67,7 @@ public CompletableFuture<Void> publish(CloudEvent ce) {
             consumer.accept(ce);
           }
         },
-        DefaultExecutorServiceFactory.instance().get());
+        serviceFactory.get());
   }
 
   @Override
@@ -76,4 +79,9 @@ protected void registerToAll(Consumer<CloudEvent> consumer) {
   protected void unregisterFromAll() {
     allConsumerRef.set(null);
   }
+
+  @Override
+  public void close() {
+    serviceFactory.close();
+  }
 }