feat: handling for failed deployments during boot

Previously any failure during verticle deployment during boot could lead to the bootstrap of NeonBee failing / undeployment of all verticles. This commit introduces a new "bootDeploymentHandling" option in the NeonBeeConfig, which lets the user specify how to deal with failures during verticles / module deployments during boot. The change is compatible with how it is handled currently (FAIL_ON_ERROR), however also provides options to undeploy any failing modules (UNDEPLOY_FAILING) or even keeping modules that could only be partially deployed (KEEP_PARTIAL).

Author: kristian

src/generated/java/io/neonbee/config/NeonBeeConfigConverter.java

@@ -19,6 +19,12 @@ public class NeonBeeConfigConverter {
     static void fromJson(Iterable<java.util.Map.Entry<String, Object>> json, NeonBeeConfig obj) {
         for (java.util.Map.Entry<String, Object> member : json) {
             switch (member.getKey()) {
+            case "bootDeploymentHandling":
+                if (member.getValue() instanceof String) {
+                    obj.setBootDeploymentHandling(
+                            io.neonbee.config.NeonBeeConfig.BootDeploymentHandling.valueOf((String) member.getValue()));
+                }
+                break;
             case "defaultThreadingModel":
                 if (member.getValue() instanceof String) {
                     obj.setDefaultThreadingModel(io.vertx.core.ThreadingModel.valueOf((String) member.getValue()));
@@ -116,6 +122,9 @@ static void toJson(NeonBeeConfig obj, JsonObject json) {
     }
 
     static void toJson(NeonBeeConfig obj, java.util.Map<String, Object> json) {
+        if (obj.getBootDeploymentHandling() != null) {
+            json.put("bootDeploymentHandling", obj.getBootDeploymentHandling().name());
+        }
         if (obj.getDefaultThreadingModel() != null) {
             json.put("defaultThreadingModel", obj.getDefaultThreadingModel().name());
         }

src/main/java/io/neonbee/NeonBee.java

@@ -1,5 +1,7 @@
 package io.neonbee;
 
+import static io.neonbee.config.NeonBeeConfig.BootDeploymentHandling.FAIL_ON_ERROR;
+import static io.neonbee.config.NeonBeeConfig.BootDeploymentHandling.KEEP_PARTIAL;
 import static io.neonbee.internal.deploy.DeployableModule.fromJar;
 import static io.neonbee.internal.deploy.DeployableVerticle.fromClass;
 import static io.neonbee.internal.deploy.DeployableVerticle.fromVerticle;
@@ -44,6 +46,7 @@
 import io.neonbee.cluster.ClusterManagerFactory;
 import io.neonbee.config.HealthConfig;
 import io.neonbee.config.NeonBeeConfig;
+import io.neonbee.config.NeonBeeConfig.BootDeploymentHandling;
 import io.neonbee.config.ServerConfig;
 import io.neonbee.data.DataException;
 import io.neonbee.data.DataQuery;
@@ -102,6 +105,7 @@
 import io.vertx.core.eventbus.EventBusOptions;
 import io.vertx.core.eventbus.MessageCodec;
 import io.vertx.core.impl.ConcurrentHashSet;
+import io.vertx.core.impl.VertxInternal;
 import io.vertx.core.json.Json;
 import io.vertx.core.json.JsonObject;
 import io.vertx.core.net.PfxOptions;
@@ -566,21 +570,25 @@ private Future<Void> deploySystemVerticles() {
         List<Future<? extends Deployable>> requiredVerticles = new ArrayList<>();
         requiredVerticles.add(fromClass(vertx, ConsolidationVerticle.class, new JsonObject().put("instances", 1)));
         requiredVerticles.add(fromClass(vertx, LoggerManagerVerticle.class));
-
-        List<Future<Optional<? extends Deployable>>> optionalVerticles = new ArrayList<>();
         if (Optional.ofNullable(config.getHealthConfig()).map(HealthConfig::isEnabled).orElse(true)) {
             requiredVerticles.add(fromClass(vertx, HealthCheckVerticle.class));
         }
+
+        List<Future<Optional<? extends Deployable>>> optionalVerticles = new ArrayList<>();
         optionalVerticles.add(deployableWatchVerticle(options.getModelsDirectory(), ModelRefreshVerticle::new));
         optionalVerticles.add(deployableWatchVerticle(options.getModulesDirectory(), DeployerVerticle::new));
         optionalVerticles.add(deployableRedeployEntitiesJobVerticle(options));
 
         LOGGER.info("Deploying system verticles ...");
-        return all(List.of(fromDeployables(requiredVerticles).compose(allTo(this)),
-                all(optionalVerticles).map(CompositeFuture::list).map(optionals -> {
-                    return optionals.stream().map(Optional.class::cast).filter(Optional::isPresent).map(Optional::get)
-                            .map(Deployable.class::cast).toList();
-                }).map(Deployables::new).compose(anyTo(this)))).mapEmpty();
+        return all(fromDeployables(requiredVerticles).compose(allTo(this)).onFailure(throwable -> {
+            LOGGER.error("Failed to deploy (some / all) required system verticle(s)", throwable);
+        }), all(optionalVerticles).map(CompositeFuture::list).map(optionals -> {
+            return optionals.stream().map(Optional.class::cast).filter(Optional::isPresent).map(Optional::get)
+                    .map(Deployable.class::cast).toList();
+        }).map(Deployables::new).compose(anyTo(this)).onFailure(throwable -> {
+            LOGGER.error("Failed to deploy (some / all) optional system verticle(s), bootstrap will continue",
+                    throwable);
+        }).otherwiseEmpty()).mapEmpty();
     }
 
     private Future<Optional<? extends Deployable>> deployableWatchVerticle(
@@ -621,7 +629,9 @@ private Future<Optional<? extends Deployable>> deployableRedeployEntitiesJobVert
     private Future<Void> deployServerVerticle() {
         LOGGER.info("Deploying server verticle ...");
         return fromClass(vertx, ServerVerticle.class, new JsonObject().put("instances", NUMBER_DEFAULT_INSTANCES))
-                .compose(deployable -> deployable.deploy(this)).mapEmpty();
+                .compose(deployable -> deployable.deploy(this)).onFailure(throwable -> {
+                    LOGGER.error("Failed to deploy server verticle", throwable);
+                }).mapEmpty();
     }
 
     /**
@@ -638,11 +648,7 @@ private Future<Void> deployClassPathVerticles() {
         return scanForDeployableClasses(vertx).compose(deployableClasses -> fromDeployables(deployableClasses.stream()
                 .filter(verticleClass -> filterByAutoDeployAndProfiles(verticleClass, options.getActiveProfiles()))
                 .map(verticleClass -> fromClass(vertx, verticleClass)).collect(Collectors.toList())))
-                .onSuccess(deployables -> {
-                    if (LOGGER.isInfoEnabled()) {
-                        LOGGER.info("Deploy class path verticle(s) {}.", deployables.getIdentifier());
-                    }
-                }).compose(allTo(this)).mapEmpty();
+                .compose(handleBootDeployment("class path verticle(s)"));
     }
 
     @VisibleForTesting
@@ -665,7 +671,39 @@ private Future<Void> deployModules() {
 
         LOGGER.info("Deploying module(s) ...");
         return fromDeployables(moduleJarPaths.stream().map(moduleJarPath -> fromJar(vertx, moduleJarPath))
-                .collect(Collectors.toList())).compose(allTo(this)).mapEmpty();
+                .collect(Collectors.toList())).compose(handleBootDeployment("module(s)"));
+    }
+
+    private Function<Deployables, Future<Void>> handleBootDeployment(String deploymentType) {
+        BootDeploymentHandling handling = config.getBootDeploymentHandling();
+        return deployables -> {
+            // in case we should keep partial deployments, for every deployable that we are about to deploy
+            // set the keep partial deployment flag, so that in case there is an error we don't undeploy
+            if (handling == KEEP_PARTIAL) {
+                for (Deployable deployable : deployables.getDeployables()) {
+                    if (deployable instanceof Deployables) {
+                        ((Deployables) deployable).keepPartialDeployment();
+                    }
+                }
+            }
+
+            return (handling == FAIL_ON_ERROR ? allTo(this) : anyTo(this)).apply(deployables)
+                    .onSuccess(deployments -> {
+                        if (LOGGER.isInfoEnabled()) {
+                            LOGGER.info("Successfully deployed all {} {}",
+                                    deploymentType, deployments.getDeploymentId());
+                        }
+                    }).recover(throwable -> {
+                        if (LOGGER.isErrorEnabled()) {
+                            LOGGER.error("Failed to deploy (some / all) {}{}",
+                                    deploymentType, handling == FAIL_ON_ERROR ? "" : ", bootstrap will continue",
+                                    throwable);
+                        }
+
+                        // abort the boot process if any class path verticle failed to deploy
+                        return handling == FAIL_ON_ERROR ? failedFuture(throwable) : succeededFuture();
+                    }).mapEmpty();
+        };
     }
 
     @VisibleForTesting
@@ -721,7 +759,7 @@ private void registerCloseHandler(Vertx vertx) {
         try {
             // unfortunately the addCloseHook method is public, but hidden in VertxImpl. As we need to know when the
             // instance shuts down, register a close hook using reflections (might fail due to a SecurityManager)
-            vertx.getClass().getMethod("addCloseHook", Closeable.class).invoke(vertx, (Closeable) completion -> {
+            VertxInternal.class.getMethod("addCloseHook", Closeable.class).invoke(vertx, (Closeable) completion -> {
                 /*
                  * Called when Vert.x instance is closed, perform shut-down operations here
                  */

src/main/java/io/neonbee/config/NeonBeeConfig.java

@@ -1,5 +1,6 @@
 package io.neonbee.config;
 
+import static io.neonbee.config.NeonBeeConfig.BootDeploymentHandling.FAIL_ON_ERROR;
 import static io.neonbee.internal.helper.ConfigHelper.notFound;
 import static io.neonbee.internal.helper.ConfigHelper.readConfig;
 import static io.neonbee.internal.helper.ConfigHelper.rephraseConfigNames;
@@ -44,6 +45,31 @@
 @DataObject
 @JsonGen(publicConverter = false)
 public class NeonBeeConfig {
+    /**
+     * How (non-system related) deployments like verticles and modules are handled during booting up {@link NeonBee}.
+     */
+    public enum BootDeploymentHandling {
+        /**
+         * Abort the {@link NeonBee} boot, if any (non-system related) verticle / module deployment fails to be
+         * deployed.
+         */
+        FAIL_ON_ERROR,
+
+        /**
+         * Log any failure of a (non-system related) verticle / module deployments, but continue booting up
+         * {@link NeonBee} otherwise, while discarding / undeploying any partial deployments (i.e. if a module fails to
+         * deploy, all deployables related to that module will be undeployed again).
+         */
+        UNDEPLOY_FAILING,
+
+        /**
+         * Log any failure of a (non-system related) verticle / module deployments, but continue booting up
+         * {@link NeonBee} otherwise, while keeping partial deployments (i.e. if a module fails to deploy, some
+         * verticles might still stay deployed).
+         */
+        KEEP_PARTIAL
+    }
+
     /**
      * The default timeout for an event bus request.
      */
@@ -74,6 +100,8 @@ public class NeonBeeConfig {
 
     private int eventBusTimeout = DEFAULT_EVENT_BUS_TIMEOUT;
 
+    private BootDeploymentHandling bootDeploymentHandling = FAIL_ON_ERROR;
+
     private int deploymentTimeout = DEFAULT_DEPLOYMENT_TIMEOUT;
 
     private Integer modelsDeploymentTimeout;
@@ -258,6 +286,26 @@ public NeonBeeConfig setEventBusTimeout(int eventBusTimeout) {
         return this;
     }
 
+    /**
+     * Gets how {@link NeonBee} handles failures deploying verticles / modules during boot.
+     *
+     * @return the selected boot deployment handling method
+     */
+    public BootDeploymentHandling getBootDeploymentHandling() {
+        return bootDeploymentHandling;
+    }
+
+    /**
+     * Sets how {@link NeonBee} should handle failures deploying verticles / modules during boot.
+     *
+     * @param bootDeploymentHandling the selected boot deployment handling method
+     * @return the {@linkplain NeonBeeConfig} for fluent use
+     */
+    public NeonBeeConfig setBootDeploymentHandling(BootDeploymentHandling bootDeploymentHandling) {
+        this.bootDeploymentHandling = bootDeploymentHandling;
+        return this;
+    }
+
     /**
      * Returns the general deployment timeout for an individual deployment of any type in seconds. If unset / equal or
      * smaller than 0, no timeout applies to the deployment.

src/main/java/io/neonbee/internal/deploy/Deployables.java

@@ -167,10 +167,9 @@ protected Future<Void> undeploy(String deploymentId) {
 
         getDeployables().stream().map(deployable -> deployable.deploy(neonBee)).forEach(pendingDeployments::add);
         // when we should keep partial deployments use a joinComposite, so we wait for all deployments to finish
-        // independent if a single one fails or not. in case we should not keep partial deployments (default) use
-        // allComposite here, which will fail, when one deployment fails, and thus we can start undeploying all
-        // succeeded
-        // (or to be succeeded pending deployments) as unfortunately there is no way to cancel active deployments
+        // independent if a single one fails. in case we should not keep partial deployments (default) use allComposite
+        // here, which will fail, when one deployment fails, and thus we can start undeploying all succeeded (or to be
+        // succeeded pending deployments) as unfortunately there is no way to cancel active deployments
         (keepPartialDeployment ? Future.join(pendingDeployments) : Future.all(pendingDeployments))
                 .onComplete(deployPromise);