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,16 @@ public void incrementUsedSpace(long usedSpace) {
   }
 
   public void decrementUsedSpace(long reclaimedSpace) {
-    cachedValue.addAndGet(-1 * reclaimedSpace);
+    long current = cachedValue.get();
+    long newValue = current - reclaimedSpace;
+    if (newValue < 0) {
+      LOG.warn(
+          "Attempted to decrement used space to a negative value. Current: {}, Decrement: {}",
+          current, reclaimedSpace);
+      cachedValue.set(current); // Retain the previous value
+    } else {
+      cachedValue.set(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,28 @@ 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
+    assertTrue(subject.getUsedSpace() >= 0,
+        "Cached used space should not be negative");
+
+    // Check that it has retained the previous value
+    assertEquals(50, subject.getUsedSpace(),
+        "Cached used space should remain at the initial value");
+  }
+
   private static long missingInitialValue() {
     return 0L;
   }