Reapply "refactor: Let attemptStartingConsumers() return CompletableFuture<Boolean> instead of Optional<Throwable>"

This reverts commit 8a51c29.

Author: caoccao

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

@@ -150,25 +150,32 @@ public ProtocolAdapterWrapper(
                 })
                 .thenCompose(Function.identity())
                 .handle((ignored, error) -> {
-                    if(error != null) {
+                    if (error != null) {
                         log.error("Error starting adapter", error);
                         stopAfterFailedStart();
                         protocolAdapterState.setRuntimeStatus(ProtocolAdapterState.RuntimeStatus.STOPPED);
                         //we still return the initial error since that's the most significant information
                         return CompletableFuture.failedFuture(error);
                     } else {
-                        return attemptStartingConsumers(writingEnabled, moduleServices.eventService())
-                            .map(startException -> {
+                        return attemptStartingConsumers(writingEnabled,
+                                moduleServices.eventService()).handle((success, startException) -> {
+                            if (startException == null) {
+                                protocolAdapterState.setRuntimeStatus(ProtocolAdapterState.RuntimeStatus.STARTED);
+                                if (success) {
+                                    log.debug("Successfully started adapter with id {}", adapter.getId());
+                                } else {
+                                    log.debug("Partially started adapter with id {}", adapter.getId());
+                                }
+                            } else {
                                 log.error("Failed to start adapter with id {}", adapter.getId(), startException);
                                 stopAfterFailedStart();
-                                protocolAdapterState.setRuntimeStatus(ProtocolAdapterState.RuntimeStatus.STOPPED);
                                 //we still return the initial error since that's the most significant information
-                                return CompletableFuture.failedFuture(startException);
-                            })
-                            .orElseGet(() -> {
-                                protocolAdapterState.setRuntimeStatus(ProtocolAdapterState.RuntimeStatus.STARTED);
-                                return CompletableFuture.completedFuture(null);
-                            });
+                                protocolAdapterState.setRuntimeStatus(ProtocolAdapterState.RuntimeStatus.STOPPED);
+                                throw new RuntimeException("Failed to start adapter with id " + adapter.getId(),
+                                        startException);
+                            }
+                            return success;
+                        });
                     }
                 })
                 .thenApply(ignored -> (Void)null)
@@ -201,7 +208,10 @@ private void stopAfterFailedStart() {
         }
     }
 
-    private @NotNull Optional<Throwable> attemptStartingConsumers(final boolean writingEnabled, final @NotNull EventService eventService) {
+    private @NotNull CompletableFuture<Boolean> attemptStartingConsumers(
+            final boolean writingEnabled,
+            final @NotNull EventService eventService) {
+        final CompletableFuture<Boolean> future = new CompletableFuture<>();
         try {
             //Adapter started successfully, now start the consumers
             createAndSubscribeTagConsumer();
@@ -215,21 +225,24 @@ private void stopAfterFailedStart() {
                             try {
                                 if (startWriting(protocolAdapterWritingService).get()) {
                                     log.info("Successfully started adapter with id {}", adapter.getId());
+                                    future.complete(true);
                                 } else {
                                     log.error("Protocol adapter start failed as data hub is not available.");
+                                    future.complete(false);
                                 }
                             } catch (final Exception e) {
                                 log.error("Failed to start writing for adapter with id {}.", adapter.getId(), e);
+                                future.completeExceptionally(e);
                             }
                         }
                     }
                 });
             }
         } catch (final Throwable e) {
             log.error("Protocol adapter start failed");
-            return Optional.of(e);
+            future.completeExceptionally(e);
         }
-        return Optional.empty();
+        return future;
     }
 
     public @NotNull CompletableFuture<Void> stopAsync(final boolean destroy) {