stop adapters orderly on shutdown of EDGE

Author: marregui

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

@@ -61,11 +61,21 @@ public abstract class ExecutorsModule {
             if (log.isDebugEnabled()) {
                 log.debug("Shutting down executor service: {}", name);
             }
-            executor.shutdown();
+            // Only initiate shutdown if not already shutting down
+            // This allows ProtocolAdapterManager to shut down executors first
+            if (!executor.isShutdown()) {
+                executor.shutdown();
+            }
             try {
-                if (!executor.awaitTermination(10, java.util.concurrent.TimeUnit.SECONDS)) {
+                // Reduced timeout since ProtocolAdapterManager should have already
+                // initiated shutdown for adapters
+                if (!executor.awaitTermination(5, java.util.concurrent.TimeUnit.SECONDS)) {
                     log.warn("Executor service {} did not terminate in time, forcing shutdown", name);
                     executor.shutdownNow();
+                    // Give a final grace period after forced shutdown
+                    if (!executor.awaitTermination(2, java.util.concurrent.TimeUnit.SECONDS)) {
+                        log.error("Executor service {} still has running tasks after forced shutdown", name);
+                    }
                 }
             } catch (final InterruptedException e) {
                 Thread.currentThread().interrupt();

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

@@ -57,6 +57,9 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Function;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
@@ -85,6 +88,7 @@ public class ProtocolAdapterManager {
     private final @NotNull ProtocolAdapterExtractor protocolAdapterConfig;
     private final @NotNull ExecutorService executorService;
     private final @NotNull ExecutorService sharedAdapterExecutor;
+    private final @NotNull AtomicBoolean shutdownInitiated;
 
     @Inject
     public ProtocolAdapterManager(
@@ -118,7 +122,18 @@ public ProtocolAdapterManager(
         this.sharedAdapterExecutor = sharedAdapterExecutor;
         this.protocolAdapters = new ConcurrentHashMap<>();
         this.executorService = Executors.newSingleThreadExecutor();
-        Runtime.getRuntime().addShutdownHook(new Thread(executorService::shutdown));
+        this.shutdownInitiated = new AtomicBoolean(false);
+
+        // Register coordinated shutdown hook that stops adapters BEFORE executors shutdown
+        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+            if (shutdownInitiated.compareAndSet(false, true)) {
+                log.info("Initiating coordinated shutdown of Protocol Adapter Manager");
+                stopAllAdaptersOnShutdown();
+                shutdownExecutorsGracefully();
+                log.info("Protocol Adapter Manager shutdown completed");
+            }
+        }, "protocol-adapter-manager-shutdown"));
+
         protocolAdapterWritingService.addWritingChangedCallback(() -> protocolAdapterFactoryManager.writingEnabledChanged(
                 protocolAdapterWritingService.writingEnabled()));
     }
@@ -519,4 +534,96 @@ public boolean writingEnabled() {
                         configConverter.convertTagDefinitionToJsonNode(tag.getDefinition())))
                 .toList());
     }
+
+    /**
+     * Stop all adapters during shutdown in a coordinated manner.
+     * This method is called by the shutdown hook BEFORE executors are shut down,
+     * ensuring adapters can complete their stop operations cleanly.
+     */
+    private void stopAllAdaptersOnShutdown() {
+        final List<ProtocolAdapterWrapper> adaptersToStop = new ArrayList<>(protocolAdapters.values());
+
+        if (adaptersToStop.isEmpty()) {
+            log.debug("No adapters to stop during shutdown");
+            return;
+        }
+
+        log.info("Stopping {} protocol adapters during shutdown", adaptersToStop.size());
+        final List<CompletableFuture<Void>> stopFutures = new ArrayList<>();
+
+        // Initiate stop for all adapters
+        for (final ProtocolAdapterWrapper wrapper : adaptersToStop) {
+            try {
+                log.debug("Initiating stop for adapter '{}'", wrapper.getId());
+                final CompletableFuture<Void> stopFuture = wrapper.stopAsync();
+                stopFutures.add(stopFuture);
+            } catch (final Exception e) {
+                log.error("Error initiating stop for adapter '{}' during shutdown", wrapper.getId(), e);
+            }
+        }
+
+        // Wait for all adapters to stop, with timeout
+        final CompletableFuture<Void> allStops = CompletableFuture.allOf(stopFutures.toArray(new CompletableFuture[0]));
+
+        try {
+            // Give adapters 20 seconds to stop gracefully
+            allStops.get(20, TimeUnit.SECONDS);
+            log.info("All adapters stopped successfully during shutdown");
+        } catch (final TimeoutException e) {
+            log.warn("Timeout waiting for adapters to stop during shutdown (waited 20s). Proceeding with executor shutdown.");
+            // Log which adapters failed to stop
+            for (int i = 0; i < stopFutures.size(); i++) {
+                if (!stopFutures.get(i).isDone()) {
+                    log.warn("Adapter '{}' did not complete stop operation within timeout", adaptersToStop.get(i).getId());
+                }
+            }
+        } catch (final InterruptedException e) {
+            Thread.currentThread().interrupt();
+            log.error("Interrupted while waiting for adapters to stop during shutdown", e);
+        } catch (final ExecutionException e) {
+            log.error("Error occurred while stopping adapters during shutdown", e.getCause());
+        }
+    }
+
+    /**
+     * Shutdown executors gracefully after adapters have stopped.
+     * This ensures a clean shutdown sequence.
+     */
+    private void shutdownExecutorsGracefully() {
+        log.debug("Shutting down protocol adapter manager executors");
+
+        // Shutdown the single-threaded executor used for adapter refresh
+        executorService.shutdown();
+        try {
+            if (!executorService.awaitTermination(5, TimeUnit.SECONDS)) {
+                log.warn("Executor service did not terminate in time, forcing shutdown");
+                executorService.shutdownNow();
+            }
+        } catch (final InterruptedException e) {
+            Thread.currentThread().interrupt();
+            log.warn("Interrupted while waiting for executor service to terminate");
+            executorService.shutdownNow();
+        }
+
+        // Shutdown the shared adapter executor
+        // Note: This may also be shut down by ExecutorsModule shutdown hook,
+        // but calling shutdown() multiple times is safe (idempotent)
+        sharedAdapterExecutor.shutdown();
+        try {
+            if (!sharedAdapterExecutor.awaitTermination(10, TimeUnit.SECONDS)) {
+                log.warn("Shared adapter executor did not terminate in time, forcing shutdown");
+                sharedAdapterExecutor.shutdownNow();
+                // Wait a bit more after forced shutdown
+                if (!sharedAdapterExecutor.awaitTermination(2, TimeUnit.SECONDS)) {
+                    log.error("Shared adapter executor still has running tasks after forced shutdown");
+                }
+            }
+        } catch (final InterruptedException e) {
+            Thread.currentThread().interrupt();
+            log.warn("Interrupted while waiting for shared adapter executor to terminate");
+            sharedAdapterExecutor.shutdownNow();
+        }
+
+        log.debug("Protocol adapter manager executors shutdown completed");
+    }
 }

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

@@ -336,38 +336,54 @@ private void cleanupConnectionStatusListener() {
             final var input = new ProtocolAdapterStopInputImpl();
             final var output = new ProtocolAdapterStopOutputImpl();
 
+            log.debug("Adapter '{}': Creating stop operation future", getId());
+
             final var stopFuture = CompletableFuture.supplyAsync(() -> {
+                        log.debug("Adapter '{}': Stop operation executing in thread '{}'", getId(), Thread.currentThread().getName());
+
                         // Signal FSM to stop (calls onStopping() internally)
+                        log.debug("Adapter '{}': Stopping adapter FSM", getId());
                         stopAdapter();
 
                         // Clean up listeners to prevent memory leaks
+                        log.debug("Adapter '{}': Cleaning up connection status listener", getId());
                         cleanupConnectionStatusListener();
 
                         // Remove consumers - must be done within async context
+                        log.debug("Adapter '{}': Removing {} consumers", getId(), consumers.size());
                         consumers.forEach(tagManager::removeConsumer);
 
+                        log.debug("Adapter '{}': Stopping polling", getId());
                         stopPolling(protocolAdapterPollingService);
+
+                        log.debug("Adapter '{}': Stopping writing", getId());
                         stopWriting(protocolAdapterWritingService);
 
                         try {
+                            log.debug("Adapter '{}': Calling adapter.stop()", getId());
                             adapter.stop(input, output);
                         } catch (final Throwable throwable) {
+                            log.error("Adapter '{}': Exception during adapter.stop()", getId(), throwable);
                             output.getOutputFuture().completeExceptionally(throwable);
                         }
+                        log.debug("Adapter '{}': Waiting for stop output future", getId());
                         return output.getOutputFuture();
                     }, sharedAdapterExecutor)  // Use shared executor to reduce thread overhead
                     .thenCompose(Function.identity()).whenComplete((result, throwable) -> {
+                        log.debug("Adapter '{}': Stop operation completed, starting cleanup", getId());
+
                         // Always call destroy() to ensure all resources are properly released
                         // This prevents resource leaks from underlying client libraries
                         try {
                             log.info("Destroying adapter with id '{}' to release all resources", getId());
                             adapter.destroy();
+                            log.debug("Adapter '{}': destroy() completed successfully", getId());
                         } catch (final Exception destroyException) {
                             log.error("Error destroying adapter with id {}", adapter.getId(), destroyException);
                         }
 
                         if (throwable == null) {
-                            log.info("Stopped adapter with id {}", adapter.getId());
+                            log.info("Stopped adapter with id '{}' successfully", adapter.getId());
                         } else {
                             log.error("Error stopping adapter with id {}", adapter.getId(), throwable);
                         }
@@ -376,6 +392,7 @@ private void cleanupConnectionStatusListener() {
                         operationLock.lock();
                         try {
                             currentStopFuture = null;
+                            log.debug("Adapter '{}': Cleared currentStopFuture reference", getId());
                         } finally {
                             operationLock.unlock();
                         }

modules/hivemq-edge-module-etherip/src/main/java/com/hivemq/edge/adapters/etherip/EipPollingProtocolAdapter.java

@@ -38,6 +38,8 @@
 
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.stream.Collectors;
 
 
@@ -50,7 +52,8 @@ public class EipPollingProtocolAdapter implements BatchPollingProtocolAdapter {
     private final @NotNull ProtocolAdapterState protocolAdapterState;
     protected final @NotNull AdapterFactories adapterFactories;
     private final @NotNull String adapterId;
-    private volatile @Nullable EtherNetIP etherNetIP;
+    private final @NotNull Lock connectionLock;
+    private @Nullable EtherNetIP etherNetIP;  // GuardedBy connectionLock
     private final @NotNull PublishChangedDataOnlyHandler lastSamples = new PublishChangedDataOnlyHandler();
     private final @NotNull DataPointFactory dataPointFactory;
 
@@ -69,6 +72,7 @@ public EipPollingProtocolAdapter(
                 .collect(Collectors.toMap(tag -> tag.getDefinition().getAddress(), tag -> tag));
         this.protocolAdapterState = input.getProtocolAdapterState();
         this.adapterFactories = input.adapterFactories();
+        this.connectionLock = new ReentrantLock();
     }
 
     @Override
@@ -80,35 +84,53 @@ public EipPollingProtocolAdapter(
     public void start(
             final @NotNull ProtocolAdapterStartInput input, final @NotNull ProtocolAdapterStartOutput output) {
         // any setup which should be done before the adapter starts polling comes here.
+        connectionLock.lock();
         try {
-            final EtherNetIP etherNetIP = new EtherNetIP(adapterConfig.getHost(), adapterConfig.getSlot());
-            etherNetIP.connectTcp();
-            this.etherNetIP = etherNetIP;
-            output.startedSuccessfully();
+            if (etherNetIP != null) {
+                log.warn("Adapter {} is already started, ignoring start request", adapterId);
+                output.startedSuccessfully();
+                return;
+            }
+
+            final EtherNetIP newConnection = new EtherNetIP(adapterConfig.getHost(), adapterConfig.getSlot());
+            newConnection.connectTcp();
+            this.etherNetIP = newConnection;
             protocolAdapterState.setConnectionStatus(ProtocolAdapterState.ConnectionStatus.CONNECTED);
+            output.startedSuccessfully();
         } catch (final Exception e) {
+            protocolAdapterState.setConnectionStatus(ProtocolAdapterState.ConnectionStatus.DISCONNECTED);
             output.failStart(e, null);
+        } finally {
+            connectionLock.unlock();
         }
     }
 
     @Override
     public void stop(
             final @NotNull ProtocolAdapterStopInput protocolAdapterStopInput,
             final @NotNull ProtocolAdapterStopOutput protocolAdapterStopOutput) {
+        connectionLock.lock();
         try {
             final EtherNetIP etherNetIPTemp = etherNetIP;
             etherNetIP = null;
+            protocolAdapterState.setConnectionStatus(ProtocolAdapterState.ConnectionStatus.DISCONNECTED);
+
             if (etherNetIPTemp != null) {
-                etherNetIPTemp.close();
-                protocolAdapterStopOutput.stoppedSuccessfully();
-                log.info("Stopped");
+                try {
+                    etherNetIPTemp.close();
+                    log.info("Stopped adapter {}", adapterId);
+                } catch (final Exception e) {
+                    log.warn("Error closing EtherNetIP connection for adapter {}", adapterId, e);
+                }
             } else {
-                protocolAdapterStopOutput.stoppedSuccessfully();
-                log.info("Stopped without an open connection");
+                log.info("Stopped adapter {} without an open connection", adapterId);
             }
+            protocolAdapterStopOutput.stoppedSuccessfully();
         } catch (final Exception e) {
             protocolAdapterStopOutput.failStop(e, "Unable to stop Ethernet IP connection");
-            log.error("Unable to stop", e);
+            log.error("Unable to stop adapter {}", adapterId, e);
+        } finally {
+            connectionLock.unlock();
         }
     }
 
@@ -121,10 +143,16 @@ public void stop(
     @Override
     public void poll(
             final @NotNull BatchPollingInput pollingInput, final @NotNull BatchPollingOutput pollingOutput) {
-        final var client = etherNetIP;
-        if (client == null) {
-            pollingOutput.fail("Polling failed because adapter wasn't started.");
-            return;
+        final EtherNetIP client;
+        connectionLock.lock();
+        try {
+            client = etherNetIP;
+            if (client == null) {
+                pollingOutput.fail("Polling failed because adapter wasn't started.");
+                return;
+            }
+        } finally {
+            connectionLock.unlock();
         }
 
         final var tagAddresses = tags.values().stream().map(v -> v.getDefinition().getAddress()).toArray(String[]::new);
@@ -148,14 +176,14 @@ public void poll(
             pollingOutput.finish();
         } catch (final CipException e) {
             if (e.getStatusCode() == 0x04) {
-                log.warn("A Tag doesn't exist on device.", e);
+                log.warn("A Tag doesn't exist on device for adapter {}", adapterId, e);
                 pollingOutput.fail(e, "Tag doesn't exist on device");
             } else {
-                log.warn("Problem accessing tag on device.", e);
+                log.warn("Problem accessing tag on device for adapter {}", adapterId, e);
                 pollingOutput.fail(e, "Problem accessing tag on device.");
             }
         } catch (final Exception e) {
-            log.warn("An exception occurred while reading tags '{}'.", tagAddresses, e);
+            log.warn("An exception occurred while reading tags '{}' for adapter {}", tagAddresses, adapterId, e);
             pollingOutput.fail(e, "An exception occurred while reading tags '" + tagAddresses + "'.");
         }
     }

modules/hivemq-edge-module-etherip/src/main/java/com/hivemq/edge/adapters/etherip/PublishChangedDataOnlyHandler.java

@@ -39,7 +39,9 @@ public boolean replaceIfValueIsNew(final @NotNull String tagName, final @NotNull
             }
         });
 
-        return newValue != computedValue;
+        // Return true if the value was actually replaced (i.e., computedValue is the new value)
+        // When values are equal, compute() returns the old value, so references will differ
+        return computedValue == newValue;
     }
 
     public void clear() {