Add connection health monitoring and reconnect

Author: codepitbull

modules/hivemq-edge-module-opcua/src/main/java/com/hivemq/edge/adapters/opcua/OpcUaClientConnection.java

@@ -68,8 +68,6 @@ public class OpcUaClientConnection {
     private final @NotNull ProtocolAdapterMetricsService protocolAdapterMetricsService;
 
     private final @NotNull AtomicReference<ConnectionContext> context = new AtomicReference<>();
-    private final @NotNull ScheduledExecutorService healthCheckScheduler = Executors.newSingleThreadScheduledExecutor();
-    private final @NotNull AtomicReference<ScheduledFuture<?>> healthCheckFuture = new AtomicReference<>();
 
     OpcUaClientConnection(
             final @NotNull String adapterId,
@@ -208,19 +206,13 @@ synchronized boolean start(final ParsedConfig parsedConfig) {
         protocolAdapterState.setConnectionStatus(ProtocolAdapterState.ConnectionStatus.CONNECTED);
         client.addSessionActivityListener(activityListener);
 
-        // Schedule periodic health check to detect stale connections
-        scheduleHealthCheck();
-
         log.info("Client created and connected successfully");
         return true;
     }
 
     synchronized void stop() {
         log.info("Stopping OPC UA client");
 
-        // Cancel health check
-        cancelHealthCheck();
-
         final ConnectionContext ctx = context.get();
         if(ctx != null) {
             quietlyCloseClient(ctx.client(),true,  ctx.faultListener(), ctx.activityListener());
@@ -231,18 +223,6 @@ synchronized void stop() {
     void destroy() {
         log.info("Destroying OPC UA client");
 
-        // Cancel health check and shutdown scheduler
-        cancelHealthCheck();
-        healthCheckScheduler.shutdown();
-        try {
-            if (!healthCheckScheduler.awaitTermination(5, TimeUnit.SECONDS)) {
-                healthCheckScheduler.shutdownNow();
-            }
-        } catch (final InterruptedException e) {
-            Thread.currentThread().interrupt();
-            healthCheckScheduler.shutdownNow();
-        }
-
         final ConnectionContext ctx = context.get();
         if(ctx != null) {
             quietlyCloseClient(ctx.client(), false, ctx.faultListener(), ctx.activityListener());
@@ -343,58 +323,6 @@ private static void quietlyCloseClient(
         }
     }
 
-    /**
-     * Schedules periodic health checks to detect stale connections.
-     * Runs at configured health check interval to verify session is active.
-     */
-    private void scheduleHealthCheck() {
-        final int healthCheckIntervalSeconds = config.getHealthCheckInterval();
-        final ScheduledFuture<?> future = healthCheckScheduler.scheduleAtFixedRate(() -> {
-            try {
-                final ConnectionContext ctx = context.get();
-                if (ctx == null) {
-                    log.trace("Health check skipped - client not connected for adapter '{}'", adapterId);
-                    return;
-                }
-
-                final OpcUaClient client = ctx.client();
-                if (client.getSession().isEmpty()) {
-                    log.warn("Health check failed for adapter '{}' - session is not active", adapterId);
-                    eventService
-                            .createAdapterEvent(adapterId, PROTOCOL_ID_OPCUA)
-                            .withMessage("Connection health check failed for adapter '" + adapterId + "' - session inactive")
-                            .withSeverity(Event.SEVERITY.WARN)
-                            .fire();
-                    protocolAdapterState.setConnectionStatus(ProtocolAdapterState.ConnectionStatus.ERROR);
-                } else {
-                    log.trace("Health check passed for adapter '{}' - session is active", adapterId);
-                }
-            } catch (final Exception e) {
-                log.warn("Health check exception for adapter '{}'", adapterId, e);
-            }
-        }, healthCheckIntervalSeconds, healthCheckIntervalSeconds, TimeUnit.SECONDS);
-
-        // Store future and cancel any existing health check
-        final ScheduledFuture<?> oldFuture = healthCheckFuture.getAndSet(future);
-        if (oldFuture != null && !oldFuture.isDone()) {
-            oldFuture.cancel(false);
-        }
-
-        log.debug("Scheduled connection health check every {} seconds for adapter '{}'",
-                healthCheckIntervalSeconds, adapterId);
-    }
-
-    /**
-     * Cancels any pending health check.
-     */
-    private void cancelHealthCheck() {
-        final ScheduledFuture<?> future = healthCheckFuture.getAndSet(null);
-        if (future != null && !future.isDone()) {
-            future.cancel(false);
-            log.debug("Cancelled health check for adapter '{}'", adapterId);
-        }
-    }
-
     private record ConnectionContext(@NotNull OpcUaClient client,
                                      @NotNull ServiceFaultListener faultListener,
                                      @NotNull SessionActivityListener activityListener,

modules/hivemq-edge-module-opcua/src/main/java/com/hivemq/edge/adapters/opcua/OpcUaProtocolAdapter.java

@@ -78,6 +78,8 @@ public class OpcUaProtocolAdapter implements WritingProtocolAdapter {
     private final @NotNull OpcUaSpecificAdapterConfig config;
     private final @NotNull ScheduledExecutorService retryScheduler = Executors.newSingleThreadScheduledExecutor();
     private final @NotNull AtomicReference<ScheduledFuture<?>> retryFuture = new AtomicReference<>();
+    private final @NotNull ScheduledExecutorService healthCheckScheduler = Executors.newSingleThreadScheduledExecutor();
+    private final @NotNull AtomicReference<ScheduledFuture<?>> healthCheckFuture = new AtomicReference<>();
 
     // Stored for reconnection - set during start()
     private volatile ParsedConfig parsedConfig;
@@ -159,8 +161,9 @@ public synchronized void stop(
             final @NotNull ProtocolAdapterStopOutput output) {
         log.info("Stopping OPC UA protocol adapter {}", adapterId);
 
-        // Cancel any pending retries
+        // Cancel any pending retries and health checks
         cancelRetry();
+        cancelHealthCheck();
 
         // Clear stored configuration to prevent reconnection after stop
         this.parsedConfig = null;
@@ -189,8 +192,9 @@ private void reconnect() {
             return;
         }
 
-        // Cancel any pending retries
+        // Cancel any pending retries and health checks
         cancelRetry();
+        cancelHealthCheck();
 
         // Stop and clean up current connection
         final OpcUaClientConnection oldConn = opcUaClientConnection.getAndSet(null);
@@ -225,12 +229,54 @@ public ModuleServices moduleServices() {
         }
     }
 
+    /**
+     * Schedules periodic health check that monitors connection health and triggers reconnection if needed.
+     */
+    private void scheduleHealthCheck() {
+        final int healthCheckIntervalSeconds = config.getHealthCheckInterval();
+        final ScheduledFuture<?> future = healthCheckScheduler.scheduleAtFixedRate(() -> {
+            final OpcUaClientConnection conn = opcUaClientConnection.get();
+            if (conn == null) {
+                log.debug("Health check skipped - no active connection for adapter '{}'", adapterId);
+                return;
+            }
+
+            if (!conn.isHealthy()) {
+                log.warn("Health check failed for adapter '{}' - triggering reconnection", adapterId);
+                reconnect();
+            } else {
+                log.debug("Health check passed for adapter '{}'", adapterId);
+            }
+        }, healthCheckIntervalSeconds, healthCheckIntervalSeconds, TimeUnit.SECONDS);
+
+        // Store future so it can be cancelled if needed
+        final ScheduledFuture<?> oldFuture = healthCheckFuture.getAndSet(future);
+        if (oldFuture != null && !oldFuture.isDone()) {
+            oldFuture.cancel(false);
+        }
+
+        log.debug("Scheduled connection health check every {} seconds for adapter '{}'",
+                healthCheckIntervalSeconds, adapterId);
+    }
+
+    /**
+     * Cancels any pending health check.
+     */
+    private void cancelHealthCheck() {
+        final ScheduledFuture<?> future = healthCheckFuture.getAndSet(null);
+        if (future != null && !future.isDone()) {
+            future.cancel(false);
+            log.debug("Cancelled health check for adapter '{}'", adapterId);
+        }
+    }
+
     @Override
     public void destroy() {
         log.info("Destroying OPC UA protocol adapter {}", adapterId);
 
-        // Cancel any pending retries
+        // Cancel any pending retries and health checks
         cancelRetry();
+        cancelHealthCheck();
 
         // Shutdown retry scheduler
         retryScheduler.shutdown();
@@ -243,6 +289,17 @@ public void destroy() {
             retryScheduler.shutdownNow();
         }
 
+        // Shutdown health check scheduler
+        healthCheckScheduler.shutdown();
+        try {
+            if (!healthCheckScheduler.awaitTermination(5, TimeUnit.SECONDS)) {
+                healthCheckScheduler.shutdownNow();
+            }
+        } catch (final InterruptedException e) {
+            Thread.currentThread().interrupt();
+            healthCheckScheduler.shutdownNow();
+        }
+
         final OpcUaClientConnection conn = opcUaClientConnection.getAndSet(null);
         if (conn != null) {
             CompletableFuture.runAsync(() -> {
@@ -398,8 +455,9 @@ private void attemptConnection(
 
         CompletableFuture.supplyAsync(() -> conn.start(parsedConfig)).whenComplete((success, throwable) -> {
             if (success && throwable == null) {
-                // Connection succeeded - cancel any pending retries
+                // Connection succeeded - cancel any pending retries and start health check
                 cancelRetry();
+                scheduleHealthCheck();
                 log.info("OPC UA adapter '{}' connected successfully", adapterId);
             } else {
                 // Connection failed - clean up and schedule retry