remove race condition

Author: marregui

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

@@ -199,6 +199,12 @@ public void start() {
     }
 
     public void refresh(final @NotNull List<ProtocolAdapterEntity> configs) {
+        // Don't submit refresh if shutdown initiated or executor is shutting down
+        if (shutdownInitiated.get() || refreshExecutor.isShutdown()) {
+            log.debug("Skipping refresh because manager is shutting down");
+            return;
+        }
+
         refreshExecutor.submit(() -> {
             // Atomically check and clear skip flag (hot-reload in progress)
             if (skipRefreshForAdapter.getAndSet(false)) {
@@ -264,18 +270,33 @@ public void refresh(final @NotNull List<ProtocolAdapterEntity> configs) {
                         log.error(
                                 "Existing adapters were modified while a refresh was ongoing, adapter with name '{}' was deleted and could not be updated",
                                 name);
+                        return;
                     }
-                    if (wrapper != null && !protocolAdapterConfigs.get(name).equals(wrapper.getConfig())) {
-                        if (log.isDebugEnabled()) {
-                            log.debug("Updating adapter '{}'", name);
+
+                    if (!protocolAdapterConfigs.get(name).equals(wrapper.getConfig())) {
+                        final boolean isStarted =
+                                wrapper.getRuntimeStatus() == ProtocolAdapterState.RuntimeStatus.STARTED;
+
+                        if (!isStarted) {
+                            // Adapter is stopped - update config by recreating wrapper but don't start
+                            if (log.isDebugEnabled()) {
+                                log.debug("Updating config for stopped adapter '{}' without starting", name);
+                            }
+                            deleteAdapterInternal(name);
+                            createAdapterInternal(protocolAdapterConfigs.get(name), versionProvider.getVersion());
+                        } else {
+                            // Adapter is started - do full stop->delete->create->start cycle
+                            if (log.isDebugEnabled()) {
+                                log.debug("Updating adapter '{}'", name);
+                            }
+                            stopAsync(name).thenApply(v -> {
+                                        deleteAdapterInternal(name);
+                                        return null;
+                                    })
+                                    .thenCompose(ignored -> startAsync(createAdapterInternal(protocolAdapterConfigs.get(
+                                            name), versionProvider.getVersion())))
+                                    .get();
                         }
-                        stopAsync(name).thenApply(v -> {
-                                    deleteAdapterInternal(name);
-                                    return null;
-                                })
-                                .thenCompose(ignored -> startAsync(createAdapterInternal(protocolAdapterConfigs.get(name),
-                                        versionProvider.getVersion())))
-                                .get();
                     } else {
                         if (log.isDebugEnabled()) {
                             log.debug("Not-updating adapter '{}' since the config is unchanged", name);

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

@@ -157,7 +157,20 @@ public boolean startSouthbound() {
                         .map(InternalWritingContextImpl::new)
                         .collect(Collectors.<InternalWritingContext>toList()));
         if (started) {
-            log.info("Southbound started for adapter {}", adapter.getId());
+            // Allow time for writing capabilities to initialize after hot-reload
+            // This prevents data loss by ensuring ReactiveWriters complete MQTT subscription
+            // and are ready to process messages before hot-reload completes.
+            // TODO: Replace with proper event-based initialization by:
+            //       1. Making ReactiveWriter.start() return CompletableFuture<Void>
+            //       2. Having ProtocolAdapterWritingServiceImpl.startWriting() wait on all futures
+            //       3. Removing this sleep in favor of the future-based approach
+            try {
+                Thread.sleep(1000);
+                log.info("Southbound started for adapter {}", adapter.getId());
+            } catch (final InterruptedException e) {
+                Thread.currentThread().interrupt();
+                log.warn("Interrupted while waiting for southbound initialization after hot-reload for adapter '{}'", getId());
+            }
             transitionSouthboundState(StateEnum.CONNECTED);
         } else {
             log.error("Southbound start failed for adapter {}", adapter.getId());