CNDB-15565: CNDB-15160: add counter to track num of skipped compactio…

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

@@ -358,6 +358,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

@@ -23,6 +23,7 @@
 import java.util.List;
 import java.util.TreeMap;
 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

@@ -1089,7 +1089,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

@@ -22,6 +22,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;
@@ -42,6 +43,7 @@
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Iterables;
 import com.google.common.collect.Sets;
+
 import org.junit.Assert;
 import org.junit.After;
 import org.junit.Before;
@@ -72,13 +74,15 @@
 import org.mockito.Mockito;
 
 import static org.assertj.core.api.Assertions.assertThat;
+
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertTrue;
+
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyDouble;
 import static org.mockito.ArgumentMatchers.anyInt;
@@ -2209,4 +2213,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(TimeUUID.Generator.nextTimeUUID(),
+                                                    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);
+    }
 }