CNDB-15160: add counter to track num of skipped compaction aggregates due to insufficient disk space (#1962)

### What is the issue
#15160

### What does this PR fix and why was it fixed
count num of skipped compaction aggregates due to insufficient disk
space

Author: jasonstack

src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java

@@ -317,6 +317,11 @@ protected BackgroundCompactions getBackgroundCompactions()
         return backgroundCompactions;
     }
 
+    public long getSkippedAggregatesDueToDiskSpace()
+    {
+        return backgroundCompactions.getSkippedAggregatesDueToDiskSpace();
+    }
+
     public static Map<String, String> validateOptions(Map<String, String> options) throws ConfigurationException
     {
         return CompactionStrategyOptions.validateOptions(options);

src/java/org/apache/cassandra/db/compaction/BackgroundCompactions.java

@@ -24,6 +24,7 @@
 import java.util.TreeMap;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
 
 import com.google.common.collect.ImmutableList;
 import org.slf4j.Logger;
@@ -73,6 +74,11 @@ public class BackgroundCompactions
      */
     MovingAverage compactionRate = ExpMovingAverage.decayBy1000();
 
+    /**
+     * Track num of skipped compaction aggregates due to insufficient disk space
+     */
+    private final AtomicLong skippedAggregatesDueToDiskSpace = new AtomicLong(0);
+
     BackgroundCompactions(CompactionRealm realm)
     {
         this.metadata = realm.metadata();
@@ -283,6 +289,16 @@ private void updateCompactionRate(CompactionProgress progress)
         }
     }
 
+    public void incrementSkippedAggregatesDueToDiskSpace()
+    {
+        skippedAggregatesDueToDiskSpace.incrementAndGet();
+    }
+
+    public long getSkippedAggregatesDueToDiskSpace()
+    {
+        return skippedAggregatesDueToDiskSpace.get();
+    }
+
     public Collection<CompactionAggregate> getAggregates()
     {
         return aggregates;

src/java/org/apache/cassandra/db/compaction/UnifiedCompactionStrategy.java

@@ -1090,7 +1090,10 @@ List<CompactionAggregate> getSelection(List<CompactionAggregate.UnifiedAggregate
             ++proposed;
             long overheadSizeInBytes = pick.totalOverheadInBytes();
             if (overheadSizeInBytes > spaceAvailable)
+            {
+                getBackgroundCompactions().incrementSkippedAggregatesDueToDiskSpace();
                 continue; // compaction is too large for current cycle
+            }
 
             int currentLevel = levelOf(pick);
             boolean isAdaptive = controller.isRecentAdaptive(pick);

test/unit/org/apache/cassandra/db/compaction/UnifiedCompactionStrategyGetSelectionTest.java

@@ -190,6 +190,7 @@ public void testGetSelection(List<CompactionAggregate.UnifiedAggregate> compacti
         when(strategy.getController()).thenReturn(controller);
         when(strategy.getShardingStats(any())).thenReturn(stats);
         when(strategy.getSelection(any(), anyInt(), any(), anyLong(), anyInt())).thenCallRealMethod();
+        when(strategy.getBackgroundCompactions()).thenReturn(Mockito.mock(BackgroundCompactions.class));
 
         int[] perLevel = new int[levelCount];
         int maxReservations = totalCount / levelCount;

test/unit/org/apache/cassandra/db/compaction/UnifiedCompactionStrategyTest.java

@@ -20,6 +20,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.Comparator;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -2198,4 +2199,47 @@ public void testGetLevel()
         assertEquals(0.25d, level.min, 0);
         assertEquals(0.5d, level.max, 0);
     }
+
+    @Test
+    public void testSkippedAggregatesOnInsufficientDiskSpace()
+    {
+        long overheadSizeInBytes = 1000L;
+        long insufficientSpaceAvailable = 500L;
+
+        BackgroundCompactions backgroundCompactions = Mockito.mock(BackgroundCompactions.class);
+        Controller controller = Mockito.mock(Controller.class, Mockito.withSettings().stubOnly());
+
+        when(controller.prioritize(anyList())).thenCallRealMethod();
+        when(controller.getReservedThreads()).thenReturn(0);
+        when(controller.getReservationsType()).thenReturn(Reservations.Type.PER_LEVEL);
+        when(controller.getOverheadSizeInBytes(any(), anyLong())).thenReturn(overheadSizeInBytes);
+        when(controller.isRecentAdaptive(any())).thenReturn(false);
+        when(controller.overlapInclusionMethod()).thenReturn(Overlaps.InclusionMethod.TRANSITIVE);
+        when(controller.parallelizeOutputShards()).thenReturn(false);
+
+        CompactionSSTable mockSSTable = Mockito.mock(CompactionSSTable.class);
+        CompactionPick pick = CompactionPick.create(UUID.randomUUID(),
+                                                    0,
+                                                    ImmutableList.of(mockSSTable),
+                                                    Collections.emptySet(),
+                                                    1,
+                                                    overheadSizeInBytes,
+                                                    overheadSizeInBytes,
+                                                    overheadSizeInBytes);
+
+        CompactionAggregate.UnifiedAggregate aggregate = Mockito.mock(CompactionAggregate.UnifiedAggregate.class, Mockito.withSettings().stubOnly());
+        when(aggregate.getSelected()).thenReturn(pick);
+        when(aggregate.maxOverlap()).thenReturn(0);
+
+        UnifiedCompactionStrategy strategy = new UnifiedCompactionStrategy(strategyFactory, backgroundCompactions, controller);
+        List<CompactionAggregate.UnifiedAggregate> pending = Arrays.asList(aggregate);
+        int[] perLevel = new int[1];
+
+        List<CompactionAggregate> result = strategy.getSelection(pending, 1, perLevel, insufficientSpaceAvailable, 0);
+        assertEquals("No compactions should be selected when insufficient disk space", 0, result.size());
+        Mockito.verify(backgroundCompactions, Mockito.times(1)).incrementSkippedAggregatesDueToDiskSpace();
+
+        Mockito.when(backgroundCompactions.getSkippedAggregatesDueToDiskSpace()).thenReturn(1L);
+        assertThat(strategy.getSkippedAggregatesDueToDiskSpace()).isEqualTo(1);
+    }
 }