fix: don't fail on startup if disk usage monitor is disabled

broker/src/main/java/io/camunda/zeebe/broker/bootstrap/BrokerStartupProcess.java

@@ -48,10 +48,9 @@ private List<StartupStep<BrokerStartupContext>> buildStartupSteps(final BrokerCf
 
     result.add(new ClusterServicesStep());
 
-    if (config.getData().isDiskUsageMonitoringEnabled()) {
-      // must be executed before any disk space usage listeners are registered
-      result.add(new DiskSpaceUsageMonitorStep());
-    }
+    // must be executed before any disk space usage listeners are registered
+    result.add(new DiskSpaceUsageMonitorStep());
+
     result.add(new MonitoringServerStep());
     result.add(new BrokerAdminServiceStep());
 

broker/src/main/java/io/camunda/zeebe/broker/bootstrap/DiskSpaceUsageMonitorStep.java

@@ -7,12 +7,16 @@
  */
 package io.camunda.zeebe.broker.bootstrap;
 
+import io.camunda.zeebe.broker.system.configuration.DataCfg;
+import io.camunda.zeebe.broker.system.monitoring.DiskSpaceUsageListener;
 import io.camunda.zeebe.broker.system.monitoring.DiskSpaceUsageMonitor;
+import io.camunda.zeebe.broker.system.monitoring.DiskSpaceUsageMonitorActor;
 import io.camunda.zeebe.scheduler.ConcurrencyControl;
 import io.camunda.zeebe.scheduler.future.ActorFuture;
 import io.camunda.zeebe.util.FileUtil;
 import java.io.IOException;
 import java.nio.file.Paths;
+import java.util.function.LongSupplier;
 
 class DiskSpaceUsageMonitorStep extends AbstractBrokerStartupStep {
 
@@ -23,14 +27,43 @@ void startupInternal(
       final ActorFuture<BrokerStartupContext> startupFuture) {
 
     final var data = brokerStartupContext.getBrokerConfiguration().getData();
+
+    if (!data.isDiskUsageMonitoringEnabled()) {
+      brokerStartupContext.setDiskSpaceUsageMonitor(new DisabledDiskUsageMonitor());
+      startupFuture.complete(brokerStartupContext);
+      return;
+    }
+
+    startDiskUsageMonitorActor(brokerStartupContext, startupFuture, data);
+  }
+
+  @Override
+  void shutdownInternal(
+      final BrokerStartupContext brokerShutdownContext,
+      final ConcurrencyControl concurrencyControl,
+      final ActorFuture<BrokerStartupContext> shutdownFuture) {
+
+    final var diskSpaceUsageMonitor = brokerShutdownContext.getDiskSpaceUsageMonitor();
+    if (diskSpaceUsageMonitor instanceof DiskSpaceUsageMonitorActor actor) {
+      stopDiskUsageMonitorActor(brokerShutdownContext, concurrencyControl, shutdownFuture, actor);
+    } else {
+      brokerShutdownContext.setDiskSpaceUsageMonitor(null);
+      shutdownFuture.complete(brokerShutdownContext);
+    }
+  }
+
+  private static void startDiskUsageMonitorActor(
+      final BrokerStartupContext brokerStartupContext,
+      final ActorFuture<BrokerStartupContext> startupFuture,
+      final DataCfg data) {
     try {
       FileUtil.ensureDirectoryExists(Paths.get(data.getDirectory()));
     } catch (final IOException e) {
       startupFuture.completeExceptionally(e);
       return;
     }
 
-    final var diskSpaceUsageMonitor = new DiskSpaceUsageMonitor(data);
+    final var diskSpaceUsageMonitor = new DiskSpaceUsageMonitorActor(data);
 
     final var actorStartFuture =
         brokerStartupContext.getActorSchedulingService().submitActor(diskSpaceUsageMonitor);
@@ -50,43 +83,48 @@ void startupInternal(
             });
   }
 
-  @Override
-  void shutdownInternal(
+  private void stopDiskUsageMonitorActor(
       final BrokerStartupContext brokerShutdownContext,
       final ConcurrencyControl concurrencyControl,
-      final ActorFuture<BrokerStartupContext> shutdownFuture) {
-
-    final var diskSpaceUsageMonitor = brokerShutdownContext.getDiskSpaceUsageMonitor();
+      final ActorFuture<BrokerStartupContext> shutdownFuture,
+      final DiskSpaceUsageMonitorActor actor) {
+    final var closeFuture = actor.closeAsync();
+    concurrencyControl.runOnCompletion(
+        closeFuture,
+        (ok, error) -> {
+          if (error != null) {
+            shutdownFuture.completeExceptionally(error);
+            return;
+          }
 
-    if (diskSpaceUsageMonitor != null) {
-      final var closeFuture = diskSpaceUsageMonitor.closeAsync();
-      concurrencyControl.runOnCompletion(
-          closeFuture,
-          (ok, error) -> {
-            if (error != null) {
-              shutdownFuture.completeExceptionally(error);
-              return;
-            }
-
-            forwardExceptions(
-                () ->
-                    concurrencyControl.run(
-                        () ->
-                            forwardExceptions(
-                                () -> {
-                                  brokerShutdownContext.setDiskSpaceUsageMonitor(null);
-                                  shutdownFuture.complete(brokerShutdownContext);
-                                },
-                                shutdownFuture)),
-                shutdownFuture);
-          });
-    } else {
-      shutdownFuture.complete(brokerShutdownContext);
-    }
+          forwardExceptions(
+              () ->
+                  concurrencyControl.run(
+                      () ->
+                          forwardExceptions(
+                              () -> {
+                                brokerShutdownContext.setDiskSpaceUsageMonitor(null);
+                                shutdownFuture.complete(brokerShutdownContext);
+                              },
+                              shutdownFuture)),
+              shutdownFuture);
+        });
   }
 
   @Override
   public String getName() {
     return "Disk Space Usage Monitor";
   }
+
+  private static final class DisabledDiskUsageMonitor implements DiskSpaceUsageMonitor {
+
+    @Override
+    public void addDiskUsageListener(final DiskSpaceUsageListener listener) {}
+
+    @Override
+    public void removeDiskUsageListener(final DiskSpaceUsageListener listener) {}
+
+    @Override
+    public void setFreeDiskSpaceSupplier(final LongSupplier freeDiskSpaceSupplier) {}
+  }
 }

broker/src/main/java/io/camunda/zeebe/broker/system/monitoring/DiskSpaceUsageMonitor.java

@@ -7,78 +7,14 @@
  */
 package io.camunda.zeebe.broker.system.monitoring;
 
-import static io.camunda.zeebe.broker.Broker.LOG;
-
-import io.camunda.zeebe.broker.system.configuration.DataCfg;
-import io.camunda.zeebe.scheduler.Actor;
-import java.io.File;
-import java.io.IOException;
-import java.io.UncheckedIOException;
-import java.time.Duration;
-import java.util.HashSet;
-import java.util.Set;
 import java.util.function.LongSupplier;
 
-public class DiskSpaceUsageMonitor extends Actor {
-
-  private final Set<DiskSpaceUsageListener> diskSpaceUsageListeners = new HashSet<>();
-  private boolean currentDiskAvailableStatus = true;
-  private LongSupplier freeDiskSpaceSupplier;
-  private final Duration monitoringDelay;
-  private final long minFreeDiskSpaceRequired;
-
-  public DiskSpaceUsageMonitor(final DataCfg dataCfg) {
-    monitoringDelay = dataCfg.getDiskUsageMonitoringInterval();
-    minFreeDiskSpaceRequired = dataCfg.getFreeDiskSpaceCommandWatermark();
-    final var directory = new File(dataCfg.getDirectory());
-
-    if (!directory.exists()) {
-      throw new UncheckedIOException(new IOException("Folder '" + directory + "' does not exist."));
-    }
-    freeDiskSpaceSupplier = directory::getUsableSpace;
-  }
-
-  @Override
-  protected void onActorStarted() {
-    checkDiskUsageAndNotifyListeners();
-    actor.runAtFixedRate(monitoringDelay, this::checkDiskUsageAndNotifyListeners);
-  }
-
-  private void checkDiskUsageAndNotifyListeners() {
-    final long freeDiskSpaceAvailable = freeDiskSpaceSupplier.getAsLong();
-    final boolean previousStatus = currentDiskAvailableStatus;
-    currentDiskAvailableStatus = freeDiskSpaceAvailable >= minFreeDiskSpaceRequired;
-    if (currentDiskAvailableStatus != previousStatus) {
-      if (!currentDiskAvailableStatus) {
-        LOG.warn(
-            "Out of disk space. Current available {} bytes. Minimum needed {} bytes.",
-            freeDiskSpaceAvailable,
-            minFreeDiskSpaceRequired);
-        diskSpaceUsageListeners.forEach(DiskSpaceUsageListener::onDiskSpaceNotAvailable);
-      } else {
-        LOG.info("Disk space available again. Current available {} bytes", freeDiskSpaceAvailable);
-        diskSpaceUsageListeners.forEach(DiskSpaceUsageListener::onDiskSpaceAvailable);
-      }
-    }
-  }
+public interface DiskSpaceUsageMonitor {
 
-  public void addDiskUsageListener(final DiskSpaceUsageListener listener) {
-    actor.call(
-        () -> {
-          diskSpaceUsageListeners.add(listener);
-          // Listeners always assumes diskspace is available on start. So notify them if it is not.
-          if (!currentDiskAvailableStatus) {
-            listener.onDiskSpaceNotAvailable();
-          }
-        });
-  }
+  void addDiskUsageListener(DiskSpaceUsageListener listener);
 
-  public void removeDiskUsageListener(final DiskSpaceUsageListener listener) {
-    actor.call(() -> diskSpaceUsageListeners.remove(listener));
-  }
+  void removeDiskUsageListener(DiskSpaceUsageListener listener);
 
   // Used only for testing
-  public void setFreeDiskSpaceSupplier(final LongSupplier freeDiskSpaceSupplier) {
-    this.freeDiskSpaceSupplier = freeDiskSpaceSupplier;
-  }
+  void setFreeDiskSpaceSupplier(LongSupplier freeDiskSpaceSupplier);
 }

broker/src/main/java/io/camunda/zeebe/broker/system/monitoring/DiskSpaceUsageMonitorActor.java

@@ -0,0 +1,87 @@
+/*
+ * Copyright Camunda Services GmbH and/or licensed to Camunda Services GmbH under
+ * one or more contributor license agreements. See the NOTICE file distributed
+ * with this work for additional information regarding copyright ownership.
+ * Licensed under the Zeebe Community License 1.1. You may not use this file
+ * except in compliance with the Zeebe Community License 1.1.
+ */
+package io.camunda.zeebe.broker.system.monitoring;
+
+import static io.camunda.zeebe.broker.Broker.LOG;
+
+import io.camunda.zeebe.broker.system.configuration.DataCfg;
+import io.camunda.zeebe.scheduler.Actor;
+import java.io.File;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.time.Duration;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.function.LongSupplier;
+
+public class DiskSpaceUsageMonitorActor extends Actor implements DiskSpaceUsageMonitor {
+
+  private final Set<DiskSpaceUsageListener> diskSpaceUsageListeners = new HashSet<>();
+  private boolean currentDiskAvailableStatus = true;
+  private LongSupplier freeDiskSpaceSupplier;
+  private final Duration monitoringDelay;
+  private final long minFreeDiskSpaceRequired;
+
+  public DiskSpaceUsageMonitorActor(final DataCfg dataCfg) {
+    monitoringDelay = dataCfg.getDiskUsageMonitoringInterval();
+    minFreeDiskSpaceRequired = dataCfg.getFreeDiskSpaceCommandWatermark();
+    final var directory = new File(dataCfg.getDirectory());
+
+    if (!directory.exists()) {
+      throw new UncheckedIOException(new IOException("Folder '" + directory + "' does not exist."));
+    }
+    freeDiskSpaceSupplier = directory::getUsableSpace;
+  }
+
+  @Override
+  protected void onActorStarted() {
+    checkDiskUsageAndNotifyListeners();
+    actor.runAtFixedRate(monitoringDelay, this::checkDiskUsageAndNotifyListeners);
+  }
+
+  private void checkDiskUsageAndNotifyListeners() {
+    final long freeDiskSpaceAvailable = freeDiskSpaceSupplier.getAsLong();
+    final boolean previousStatus = currentDiskAvailableStatus;
+    currentDiskAvailableStatus = freeDiskSpaceAvailable >= minFreeDiskSpaceRequired;
+    if (currentDiskAvailableStatus != previousStatus) {
+      if (!currentDiskAvailableStatus) {
+        LOG.warn(
+            "Out of disk space. Current available {} bytes. Minimum needed {} bytes.",
+            freeDiskSpaceAvailable,
+            minFreeDiskSpaceRequired);
+        diskSpaceUsageListeners.forEach(DiskSpaceUsageListener::onDiskSpaceNotAvailable);
+      } else {
+        LOG.info("Disk space available again. Current available {} bytes", freeDiskSpaceAvailable);
+        diskSpaceUsageListeners.forEach(DiskSpaceUsageListener::onDiskSpaceAvailable);
+      }
+    }
+  }
+
+  @Override
+  public void addDiskUsageListener(final DiskSpaceUsageListener listener) {
+    actor.call(
+        () -> {
+          diskSpaceUsageListeners.add(listener);
+          // Listeners always assumes diskspace is available on start. So notify them if it is not.
+          if (!currentDiskAvailableStatus) {
+            listener.onDiskSpaceNotAvailable();
+          }
+        });
+  }
+
+  @Override
+  public void removeDiskUsageListener(final DiskSpaceUsageListener listener) {
+    actor.call(() -> diskSpaceUsageListeners.remove(listener));
+  }
+
+  // Used only for testing
+  @Override
+  public void setFreeDiskSpaceSupplier(final LongSupplier freeDiskSpaceSupplier) {
+    this.freeDiskSpaceSupplier = freeDiskSpaceSupplier;
+  }
+}

broker/src/test/java/io/camunda/zeebe/broker/bootstrap/CommandApiServiceStepTest.java

@@ -20,7 +20,7 @@
 import io.camunda.zeebe.broker.exporter.repo.ExporterRepository;
 import io.camunda.zeebe.broker.system.configuration.BrokerCfg;
 import io.camunda.zeebe.broker.system.monitoring.BrokerHealthCheckService;
-import io.camunda.zeebe.broker.system.monitoring.DiskSpaceUsageMonitor;
+import io.camunda.zeebe.broker.system.monitoring.DiskSpaceUsageMonitorActor;
 import io.camunda.zeebe.broker.transport.commandapi.CommandApiServiceImpl;
 import io.camunda.zeebe.protocol.impl.encoding.BrokerInfo;
 import io.camunda.zeebe.scheduler.ActorScheduler;
@@ -68,7 +68,7 @@ void setUp() {
             mock(ClusterServicesImpl.class, RETURNS_DEEP_STUBS),
             Collections.emptyList());
     testBrokerStartupContext.setConcurrencyControl(CONCURRENCY_CONTROL);
-    testBrokerStartupContext.setDiskSpaceUsageMonitor(mock(DiskSpaceUsageMonitor.class));
+    testBrokerStartupContext.setDiskSpaceUsageMonitor(mock(DiskSpaceUsageMonitorActor.class));
     testBrokerStartupContext.setGatewayBrokerTransport(mock(AtomixServerTransport.class));
   }
 
@@ -135,7 +135,7 @@ void shouldAddCommandApiServiceAsPartitionListener() {
     @Test
     void shouldAddCommandApiServiceAsDiskSpaceUsageListener() {
       // given
-      final var mockDiskSpaceUsageMonitor = mock(DiskSpaceUsageMonitor.class);
+      final var mockDiskSpaceUsageMonitor = mock(DiskSpaceUsageMonitorActor.class);
       testBrokerStartupContext.setDiskSpaceUsageMonitor(mockDiskSpaceUsageMonitor);
 
       // when
@@ -171,7 +171,7 @@ void setUp() {
       testBrokerStartupContext.setGatewayBrokerTransport(mockAtomixServerTransport);
       testBrokerStartupContext.setCommandApiService(mockCommandApiService);
       testBrokerStartupContext.addPartitionListener(mockCommandApiService);
-      testBrokerStartupContext.setDiskSpaceUsageMonitor(mock(DiskSpaceUsageMonitor.class));
+      testBrokerStartupContext.setDiskSpaceUsageMonitor(mock(DiskSpaceUsageMonitorActor.class));
       testBrokerStartupContext
           .getDiskSpaceUsageMonitor()
           .addDiskUsageListener(mockCommandApiService);
@@ -182,7 +182,7 @@ void setUp() {
     @Test
     void shouldRemoveCommandApiFromDiskSpaceUsageListenerList() {
       // given
-      final var mockDiskSpaceUsageMonitor = mock(DiskSpaceUsageMonitor.class);
+      final var mockDiskSpaceUsageMonitor = mock(DiskSpaceUsageMonitorActor.class);
       testBrokerStartupContext.setDiskSpaceUsageMonitor(mockDiskSpaceUsageMonitor);
 
       // when