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

@@ -31,6 +31,7 @@
 import java.util.concurrent.atomic.AtomicLong;
 
 import static org.apache.hadoop.hdds.fs.MockSpaceUsageCheckParams.newBuilder;
+import static org.apache.ratis.util.Preconditions.assertTrue;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.mockito.Mockito.any;
 import static org.mockito.Mockito.anyLong;
@@ -142,6 +143,24 @@ public void savesValueOnShutdown() {
     verify(executor).shutdown();
   }
 
+  @Test
+  public void testDecrementDoesNotGoNegative() {
+    CachingSpaceUsageSource subject;
+    SpaceUsageCheckParams params;
+    AtomicLong cachedValue;
+    cachedValue = new AtomicLong(50);
+    params = paramsBuilder(cachedValue)
+        .withRefresh(Duration.ZERO)
+        .build();
+    subject = new CachingSpaceUsageSource(params);
+    // Try to decrement more than the current value
+    subject.decrementUsedSpace(100);
+
+    // Check that the value has not gone negative and has been reset to 0
+    assertTrue(subject.getUsedSpace() >= 0,
+        "Cached used space should not be negative");
+  }
+
   private static long missingInitialValue() {
     return 0L;
   }