strengthen tests

Author: marregui

hivemq-edge/src/main/java/com/hivemq/common/executors/ioc/ExecutorsModule.java

@@ -26,7 +26,6 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
 @Module
@@ -54,31 +53,6 @@ public abstract class ExecutorsModule {
         return Executors.newCachedThreadPool(new HiveMQEdgeThreadFactory(CACHED_WORKER_GROUP_NAME));
     }
 
-    public static void shutdownExecutor(
-            final @NotNull ExecutorService executor,
-            final @NotNull String name,
-            final int timeoutSeconds) {
-        log.debug("Shutting down executor service: {}", name);
-        if (!executor.isShutdown()) {
-            executor.shutdown();
-        }
-        try {
-            if (!executor.awaitTermination(timeoutSeconds, TimeUnit.SECONDS)) {
-                log.warn("Executor service {} did not terminate in {}s, forcing shutdown", name, timeoutSeconds);
-                executor.shutdownNow();
-                if (!executor.awaitTermination(2, TimeUnit.SECONDS)) {
-                    log.error("Executor service {} still has running tasks after forced shutdown", name);
-                }
-            } else {
-                log.debug("Executor service {} shut down successfully", name);
-            }
-        } catch (final InterruptedException e) {
-            Thread.currentThread().interrupt();
-            log.warn("Interrupted while waiting for executor service {} to terminate", name);
-            executor.shutdownNow();
-        }
-    }
-
     private static class HiveMQEdgeThreadFactory implements ThreadFactory {
         private final @NotNull String factoryName;
         private final @NotNull ThreadGroup group;

hivemq-edge/src/main/java/com/hivemq/protocols/ProtocolAdapterManager.java

@@ -63,7 +63,6 @@
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
-import static com.hivemq.common.executors.ioc.ExecutorsModule.shutdownExecutor;
 import static com.hivemq.persistence.domain.DomainTagAddResult.DomainTagPutStatus.ADAPTER_FAILED_TO_START;
 import static com.hivemq.persistence.domain.DomainTagAddResult.DomainTagPutStatus.ADAPTER_MISSING;
 import static com.hivemq.persistence.domain.DomainTagAddResult.DomainTagPutStatus.ALREADY_EXISTS;
@@ -88,7 +87,7 @@ public class ProtocolAdapterManager {
     private final @NotNull NorthboundConsumerFactory northboundConsumerFactory;
     private final @NotNull TagManager tagManager;
     private final @NotNull ProtocolAdapterExtractor config;
-    private final @NotNull ExecutorService singleThreadRefreshExecutor;
+    private final @NotNull ExecutorService refreshExecutor;
     private final @NotNull ExecutorService sharedAdapterExecutor;
     private final @NotNull AtomicBoolean shutdownInitiated;
 
@@ -124,7 +123,7 @@ public ProtocolAdapterManager(
         this.config = config;
         this.sharedAdapterExecutor = sharedAdapterExecutor;
         this.protocolAdapters = new ConcurrentHashMap<>();
-        this.singleThreadRefreshExecutor = Executors.newSingleThreadExecutor();
+        this.refreshExecutor = Executors.newSingleThreadExecutor();
         this.shutdownInitiated = new AtomicBoolean(false);
         shutdownHooks.add(new HiveMQShutdownHook() {
             @Override
@@ -161,7 +160,7 @@ public void start() {
     }
 
     public void refresh(final @NotNull List<ProtocolAdapterEntity> configs) {
-        singleThreadRefreshExecutor.submit(() -> {
+        refreshExecutor.submit(() -> {
             log.info("Refreshing adapters");
 
             final Map<String, ProtocolAdapterConfig> protocolAdapterConfigs = configs.stream()
@@ -365,7 +364,7 @@ public boolean isWritingEnabled() {
 
     private void shutdown() {
         if (shutdownInitiated.compareAndSet(false, true)) {
-            shutdownExecutor(singleThreadRefreshExecutor, "protocol-adapter-manager-refresh", 5);
+            shutdownRefreshExecutor();
 
             log.info("Initiating shutdown of Protocol Adapter Manager");
             final List<ProtocolAdapterWrapper> adaptersToStop = new ArrayList<>(protocolAdapters.values());
@@ -387,7 +386,7 @@ private void shutdown() {
             }
             // wait for all adapters to stop, with timeout
             try {
-                CompletableFuture.allOf(stopFutures.toArray(new CompletableFuture[0])).get(20, TimeUnit.SECONDS);
+                CompletableFuture.allOf(stopFutures.toArray(new CompletableFuture[0])).get(15, TimeUnit.SECONDS);
                 log.info("All adapters stopped successfully during shutdown");
             } catch (final TimeoutException e) {
                 log.warn("Timeout waiting for adapters to stop during shutdown");
@@ -406,6 +405,28 @@ private void shutdown() {
         }
     }
 
+    private void shutdownRefreshExecutor() {
+        final String name = "protocol-adapter-manager-refresh";
+        final int timeoutSeconds = 5;
+        log.debug("Shutting {} executor service", name);
+        refreshExecutor.shutdown();
+        try {
+            if (!refreshExecutor.awaitTermination(timeoutSeconds, TimeUnit.SECONDS)) {
+                log.warn("Executor service {} did not terminate in {}s, forcing shutdown", name, timeoutSeconds);
+                refreshExecutor.shutdownNow();
+                if (!refreshExecutor.awaitTermination(2, TimeUnit.SECONDS)) {
+                    log.error("Executor service {} still has running tasks after forced shutdown", name);
+                }
+            } else {
+                log.debug("Executor service {} shut down successfully", name);
+            }
+        } catch (final InterruptedException e) {
+            Thread.currentThread().interrupt();
+            log.warn("Interrupted while waiting for executor service {} to terminate", name);
+            refreshExecutor.shutdownNow();
+        }
+    }
+
     private @NotNull ProtocolAdapterWrapper createAdapterInternal(
             final @NotNull ProtocolAdapterConfig config,
             final @NotNull String version) {