HDDS-10614. Avoid decreasing cached space usage below zero

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/fs/CachingSpaceUsageSource.java

@@ -25,6 +25,7 @@
 import org.slf4j.LoggerFactory;
 
 import jakarta.annotation.Nullable;
+
 import java.time.Duration;
 import java.util.OptionalLong;
 import java.util.concurrent.Executors;
@@ -94,7 +95,17 @@ public void incrementUsedSpace(long usedSpace) {
   }
 
   public void decrementUsedSpace(long reclaimedSpace) {
-    cachedValue.addAndGet(-1 * reclaimedSpace);
+    cachedValue.updateAndGet(current -> {
+      long newValue = current - reclaimedSpace;
+      if (newValue < 0) {
+        LOG.warn(
+            "Attempted to decrement used space to a negative value. Current: {}, Decrement: {}",
+            current, reclaimedSpace);
+        return 0; // Reset to zero
+      } else {
+        return newValue;
+      }
+    });
   }
 
   public void start() {

hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/fs/TestCachingSpaceUsageSource.java

@@ -142,6 +142,21 @@ public void savesValueOnShutdown() {
     verify(executor).shutdown();
   }
 
+  @Test
+  public void testDecrementDoesNotGoNegative() {
+    SpaceUsageCheckParams params = paramsBuilder(new AtomicLong(50))
+        .withRefresh(Duration.ZERO)
+        .build();
+    CachingSpaceUsageSource subject = new CachingSpaceUsageSource(params);
+    AtomicLong cachedValue = new AtomicLong(50);
+
+    // Try to decrement more than the current value
+    subject.decrementUsedSpace(100);
+
+    // Check that the value has been set to 0
+    assertEquals(0, subject.getUsedSpace());
+  }
+
   private static long missingInitialValue() {
     return 0L;
   }