remove stress over adapter by not restarting it and instead providing hot reload for mappings and tags (cheaper). will see how this improves stability.

marregui · marregui · commit a2535b49efd4 · 2025-10-29T13:59:38.000+01:00
diff --git a/hivemq-edge/src/main/java/com/hivemq/api/resources/impl/ProtocolAdaptersResourceImpl.java b/hivemq-edge/src/main/java/com/hivemq/api/resources/impl/ProtocolAdaptersResourceImpl.java
@@ -70,6 +70,8 @@
 import com.hivemq.edge.api.model.StatusTransitionResult;
 import com.hivemq.edge.api.model.TagSchema;
 import com.hivemq.edge.modules.adapters.impl.ProtocolAdapterDiscoveryOutputImpl;
+import com.hivemq.persistence.mappings.NorthboundMapping;
+import com.hivemq.persistence.mappings.SouthboundMapping;
 import com.hivemq.persistence.topicfilter.TopicFilterPersistence;
 import com.hivemq.persistence.topicfilter.TopicFilterPojo;
 import com.hivemq.protocols.InternalProtocolAdapterWritingService;
@@ -803,22 +805,31 @@ public int getDepth() {
                         missingTags));
             }
 
-            return configExtractor.getAdapterByAdapterId(adapterId)
-                    .map(cfg -> new ProtocolAdapterEntity(cfg.getAdapterId(),
-                            cfg.getProtocolId(),
-                            cfg.getConfigVersion(),
-                            cfg.getConfig(),
-                            converted,
-                            cfg.getSouthboundMappings(),
-                            cfg.getTags()))
-                    .map(newCfg -> {
-                        if (!configExtractor.updateAdapter(newCfg)) {
-                            return adapterCannotBeUpdatedError(adapterId);
-                        }
-                        log.info("Successfully updated northbound mappings for adapter '{}'.", adapterId);
-                        return Response.ok(northboundMappings).build();
-                    })
-                    .orElseGet(adapterNotUpdatedError(adapterId));
+            final List<NorthboundMapping> convertedMappings =
+                    converted.stream().map(NorthboundMappingEntity::toPersistence).toList();
+            if (protocolAdapterManager.updateNorthboundMappingsHotReload(adapterId, convertedMappings)) {
+                // update config persistence
+                return configExtractor.getAdapterByAdapterId(adapterId)
+                        .map(cfg -> new ProtocolAdapterEntity(cfg.getAdapterId(),
+                                cfg.getProtocolId(),
+                                cfg.getConfigVersion(),
+                                cfg.getConfig(),
+                                converted,
+                                cfg.getSouthboundMappings(),
+                                cfg.getTags()))
+                        .map(newCfg -> {
+                            if (!configExtractor.updateAdapter(newCfg)) {
+                                return adapterCannotBeUpdatedError(adapterId);
+                            }
+                            log.info("Successfully updated northbound mappings for adapter '{}' via hot-reload.",
+                                    adapterId);
+                            return Response.ok(northboundMappings).build();
+                        })
+                        .orElseGet(adapterNotUpdatedError(adapterId));
+            } else {
+                log.error("Hot-reload failed for northbound mappings on adapter '{}'", adapterId);
+                return errorResponse(new InternalServerError("Failed to hot-reload northbound mappings"));
+            }
         };
     }
 
@@ -841,22 +852,31 @@ public int getDepth() {
                         missingTags));
             }
 
-            return configExtractor.getAdapterByAdapterId(adapterId)
-                    .map(cfg -> new ProtocolAdapterEntity(cfg.getAdapterId(),
-                            cfg.getProtocolId(),
-                            cfg.getConfigVersion(),
-                            cfg.getConfig(),
-                            cfg.getNorthboundMappings(),
-                            converted,
-                            cfg.getTags()))
-                    .map(newCfg -> {
-                        if (!configExtractor.updateAdapter(newCfg)) {
-                            return adapterCannotBeUpdatedError(adapterId);
-                        }
-                        log.info("Successfully updated fromMappings for adapter '{}'.", adapterId);
-                        return Response.ok(southboundMappings).build();
-                    })
-                    .orElseGet(adapterNotUpdatedError(adapterId));
+            final List<SouthboundMapping> convertedMappings =
+                    converted.stream().map(entity -> entity.toPersistence(objectMapper)).toList();
+            if (protocolAdapterManager.updateSouthboundMappingsHotReload(adapterId, convertedMappings)) {
+                // update config persistence
+                return configExtractor.getAdapterByAdapterId(adapterId)
+                        .map(cfg -> new ProtocolAdapterEntity(cfg.getAdapterId(),
+                                cfg.getProtocolId(),
+                                cfg.getConfigVersion(),
+                                cfg.getConfig(),
+                                cfg.getNorthboundMappings(),
+                                converted,
+                                cfg.getTags()))
+                        .map(newCfg -> {
+                            if (!configExtractor.updateAdapter(newCfg)) {
+                                return adapterCannotBeUpdatedError(adapterId);
+                            }
+                            log.info("Successfully updated southbound mappings for adapter '{}' via hot-reload.",
+                                    adapterId);
+                            return Response.ok(southboundMappings).build();
+                        })
+                        .orElseGet(adapterNotUpdatedError(adapterId));
+            } else {
+                log.error("Hot-reload failed for southbound mappings on adapter '{}'", adapterId);
+                return errorResponse(new InternalServerError("Failed to hot-reload southbound mappings"));
+            }
         };
     }
 
@@ -911,11 +931,10 @@ private void validateAdapterSchema(
 
     private @NotNull Adapter toAdapter(final @NotNull ProtocolAdapterWrapper value) {
         final String adapterId = value.getId();
-        final Map<String, Object> config = runWithContextLoader(
-                value.getAdapterFactory().getClass().getClassLoader(),
-                () -> {
-                    final Map<String, Object> cfg = value.getAdapterFactory()
-                            .unconvertConfigObject(objectMapper, value.getConfigObject());
+        final Map<String, Object> config =
+                runWithContextLoader(value.getAdapterFactory().getClass().getClassLoader(), () -> {
+                    final Map<String, Object> cfg =
+                            value.getAdapterFactory().unconvertConfigObject(objectMapper, value.getConfigObject());
                     cfg.put("id", value.getId());
                     return cfg;
                 });
diff --git a/hivemq-edge/src/main/java/com/hivemq/protocols/ProtocolAdapterConfig.java b/hivemq-edge/src/main/java/com/hivemq/protocols/ProtocolAdapterConfig.java
@@ -30,12 +30,12 @@
 public class ProtocolAdapterConfig {
 
     private final @NotNull ProtocolSpecificAdapterConfig adapterConfig;
-    private final @NotNull List<? extends Tag> tags;
+    private @NotNull List<? extends Tag> tags;
     private final @NotNull String adapterId;
     private final @NotNull String protocolId;
     private final int configVersion;
-    private final @NotNull List<SouthboundMapping> southboundMappings;
-    private final @NotNull List<NorthboundMapping> northboundMappings;
+    private @NotNull List<SouthboundMapping> southboundMappings;
+    private @NotNull List<NorthboundMapping> northboundMappings;
 
     public ProtocolAdapterConfig(
             final @NotNull String adapterId,
@@ -99,6 +99,36 @@ public int getConfigVersion() {
         return configVersion;
     }
 
+    /**
+     * Updates the tags for hot-reload support.
+     * This method is used to update tags without restarting the adapter.
+     *
+     * @param tags the new tags
+     */
+    public void setTags(final @NotNull List<? extends Tag> tags) {
+        this.tags = tags;
+    }
+
+    /**
+     * Updates the northbound mappings for hot-reload support.
+     * This method is used to update northbound mappings without restarting the adapter.
+     *
+     * @param northboundMappings the new northbound mappings
+     */
+    public void setNorthboundMappings(final @NotNull List<NorthboundMapping> northboundMappings) {
+        this.northboundMappings = northboundMappings;
+    }
+
+    /**
+     * Updates the southbound mappings for hot-reload support.
+     * This method is used to update southbound mappings without restarting the adapter.
+     *
+     * @param southboundMappings the new southbound mappings
+     */
+    public void setSouthboundMappings(final @NotNull List<SouthboundMapping> southboundMappings) {
+        this.southboundMappings = southboundMappings;
+    }
+
     @Override
     public boolean equals(final Object o) {
         if (o == null || getClass() != o.getClass()) return false;
diff --git a/hivemq-edge/src/main/java/com/hivemq/protocols/ProtocolAdapterManager.java b/hivemq-edge/src/main/java/com/hivemq/protocols/ProtocolAdapterManager.java
@@ -40,6 +40,8 @@
 import com.hivemq.edge.modules.api.adapters.ProtocolAdapterPollingService;
 import com.hivemq.persistence.domain.DomainTag;
 import com.hivemq.persistence.domain.DomainTagAddResult;
+import com.hivemq.persistence.mappings.NorthboundMapping;
+import com.hivemq.persistence.mappings.SouthboundMapping;
 import com.hivemq.protocols.northbound.NorthboundConsumerFactory;
 import jakarta.inject.Inject;
 import jakarta.inject.Singleton;
@@ -305,28 +307,68 @@ public boolean isWritingEnabled() {
             if (alreadyExists) {
                 return DomainTagAddResult.failed(ALREADY_EXISTS, adapterId);
             }
-            deleteAdapterInternal(wrapper.getId());
+
             try {
-                tags.add(configConverter.domainTagToTag(wrapper.getProtocolAdapterInformation().getProtocolId(),
-                        domainTag));
-                startAsync(createAdapterInternal(new ProtocolAdapterConfig(wrapper.getId(),
-                        wrapper.getAdapterInformation().getProtocolId(),
-                        wrapper.getAdapterInformation().getCurrentConfigVersion(),
-                        wrapper.getConfigObject(),
-                        wrapper.getSouthboundMappings(),
-                        wrapper.getNorthboundMappings(),
-                        tags), versionProvider.getVersion())).get();
+                final var convertedTag =
+                        configConverter.domainTagToTag(wrapper.getProtocolAdapterInformation().getProtocolId(),
+                                domainTag);
+
+                // Use hot-reload to add tag without restarting the adapter
+                log.debug("Adding tag '{}' to adapter '{}' via hot-reload", domainTag.getTagName(), adapterId);
+                wrapper.addTagHotReload(convertedTag, eventService);
+
+                log.info("Successfully added tag '{}' to adapter '{}' via hot-reload",
+                        domainTag.getTagName(),
+                        adapterId);
                 return DomainTagAddResult.success();
-            } catch (final InterruptedException e) {
-                Thread.currentThread().interrupt();
-                log.error("Interrupted while async execution: ", e.getCause());
+            } catch (final IllegalStateException e) {
+                log.error("Cannot hot-reload tag, adapter not in correct state: {}", e.getMessage());
+                return DomainTagAddResult.failed(ADAPTER_FAILED_TO_START, adapterId);
             } catch (final Throwable e) {
-                log.error("Exception happened while async execution: ", e.getCause());
+                log.error("Exception happened while adding tag via hot-reload: ", e);
+                return DomainTagAddResult.failed(ADAPTER_FAILED_TO_START, adapterId);
             }
-            return DomainTagAddResult.failed(ADAPTER_FAILED_TO_START, adapterId);
         }).orElse(DomainTagAddResult.failed(ADAPTER_MISSING, adapterId));
     }
 
+    public boolean updateNorthboundMappingsHotReload(
+            final @NotNull String adapterId,
+            final @NotNull List<NorthboundMapping> northboundMappings) {
+        return getProtocolAdapterWrapperByAdapterId(adapterId).map(wrapper -> {
+            try {
+                log.debug("Updating northbound mappings for adapter '{}' via hot-reload", adapterId);
+                wrapper.updateMappingsHotReload(northboundMappings, null, eventService);
+                log.info("Successfully updated northbound mappings for adapter '{}' via hot-reload", adapterId);
+                return true;
+            } catch (final IllegalStateException e) {
+                log.error("Cannot hot-reload northbound mappings, adapter not in correct state: {}", e.getMessage());
+                return false;
+            } catch (final Throwable e) {
+                log.error("Exception happened while updating northbound mappings via hot-reload: ", e);
+                return false;
+            }
+        }).orElse(false);
+    }
+
+    public boolean updateSouthboundMappingsHotReload(
+            final @NotNull String adapterId,
+            final @NotNull List<SouthboundMapping> southboundMappings) {
+        return getProtocolAdapterWrapperByAdapterId(adapterId).map(wrapper -> {
+            try {
+                log.debug("Updating southbound mappings for adapter '{}' via hot-reload", adapterId);
+                wrapper.updateMappingsHotReload(null, southboundMappings, eventService);
+                log.info("Successfully updated southbound mappings for adapter '{}' via hot-reload", adapterId);
+                return true;
+            } catch (final IllegalStateException e) {
+                log.error("Cannot hot-reload southbound mappings, adapter not in correct state: {}", e.getMessage());
+                return false;
+            } catch (final Throwable e) {
+                log.error("Exception happened while updating southbound mappings via hot-reload: ", e);
+                return false;
+            }
+        }).orElse(false);
+    }
+
     public @NotNull List<DomainTag> getDomainTags() {
         return protocolAdapters.values()
                 .stream()
diff --git a/hivemq-edge/src/main/java/com/hivemq/protocols/ProtocolAdapterWrapper.java b/hivemq-edge/src/main/java/com/hivemq/protocols/ProtocolAdapterWrapper.java
