CNDB-14353: fetch non-compacting sstables instead of live sstables for UCS (#1861)

This is to fix a race in UCS:
1. sstable-1 is compacting
2. 2nd compaction thread fetches live sstable which includes sstable-1
3. sstable-1 is now compacted, no longer as compacting in Tracker
4. 2nd compaction thread continues: UCS#getCompactionArenas fetches list
of compacting sstables which doesn't include sstable-1
5. sstable-1 is now included for next compaction but failed to create
txn because it's already compacted

### What is the issue
#14353:  UCS failed to create txn because sstable is already compacted on non-compactor

### What does this PR fix and why was it fixed

use non-compacting sstables instead of live sstables, so there won't be race

Author: jasonstack

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

@@ -57,6 +57,7 @@
 import org.apache.cassandra.db.lifecycle.LifecycleNewTracker;
 import org.apache.cassandra.db.lifecycle.LifecycleTransaction;
 import org.apache.cassandra.db.lifecycle.PartialLifecycleTransaction;
+import org.apache.cassandra.db.lifecycle.SSTableSet;
 import org.apache.cassandra.dht.Range;
 import org.apache.cassandra.dht.Token;
 import org.apache.cassandra.exceptions.ConfigurationException;
@@ -217,7 +218,7 @@ public synchronized CompactionTasks getUserDefinedTasks(Collection<? extends Com
     /// same effect as compacting all of the sstables in the arena together in one operation.
     public synchronized List<CompactionAggregate.UnifiedAggregate> getMaximalAggregates()
     {
-        return getMaximalAggregates(realm.getLiveSSTables());
+        return getMaximalAggregates(Sets.newHashSet(realm.getSSTables(SSTableSet.NONCOMPACTING)));
     }
 
     public synchronized List<CompactionAggregate.UnifiedAggregate> getMaximalAggregates(Collection<? extends CompactionSSTable> sstables)
@@ -1209,7 +1210,7 @@ public Level getLevel(int index, double min, double max)
     @VisibleForTesting
     Map<Arena, List<Level>> getLevels()
     {
-        return getLevels(realm.getLiveSSTables(), UnifiedCompactionStrategy::isSuitableForCompaction);
+        return getLevels(Sets.newHashSet(realm.getSSTables(SSTableSet.NONCOMPACTING)), UnifiedCompactionStrategy::isSuitableForCompaction);
     }
 
     private static boolean isSuitableForCompaction(CompactionSSTable sstable, boolean isCompacting)
@@ -1224,8 +1225,7 @@ Iterable<? extends CompactionSSTable> getSuitableSSTables()
 
     Iterable<? extends CompactionSSTable> getFilteredSSTables(BiPredicate<CompactionSSTable, Boolean> predicate)
     {
-        Set<? extends CompactionSSTable> compacting = realm.getCompactingSSTables();
-        return Iterables.filter(realm.getLiveSSTables(), s -> predicate.test(s, compacting.contains(s)));
+        return Iterables.filter(realm.getSSTables(SSTableSet.NONCOMPACTING), s -> predicate.test(s, false));
     }
 
     /// Groups the sstables passed in into arenas and buckets. This is used by the strategy to determine
@@ -1314,7 +1314,7 @@ public Map<Arena, List<Level>> getLevels(Collection<? extends CompactionSSTable>
     @Override
     public Map<String, String> getMaxOverlapsMap()
     {
-        final Set<? extends CompactionSSTable> liveSSTables = realm.getLiveSSTables();
+        final Set<? extends CompactionSSTable> liveSSTables = Sets.newHashSet(realm.getSSTables(SSTableSet.NONCOMPACTING));
         Map<UnifiedCompactionStrategy.Arena, List<UnifiedCompactionStrategy.Level>> arenas =
                 getLevels(liveSSTables, (i1, i2) -> true); // take all sstables
 

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

@@ -400,6 +400,52 @@ private void testGetMultipleBucketsOneArenaNonOverlappingAggregates(Map<Integer,
         System.out.println(strategy.getMaxOverlapsMap());
     }
 
+    @Test
+    public void testGetLevelsWithCompactingSSTables()
+    {
+        final int m = 2; // minimal sorted run size in MB m
+        long minimalSizeBytes = m << 20;
+
+        Controller controller = Mockito.mock(Controller.class);
+        when(controller.getMinSstableSizeBytes()).thenReturn(minimalSizeBytes);
+        when(controller.getNumShards(anyDouble())).thenReturn(1);
+        when(controller.getBaseSstableSize(anyInt())).thenReturn((double) minimalSizeBytes);
+        when(controller.maxConcurrentCompactions()).thenReturn(1000); // let it generate as many candidates as it can
+        when(controller.maxThroughput()).thenReturn(Double.MAX_VALUE);
+        when(controller.maxSSTablesToCompact()).thenReturn(1000);
+        when(controller.prioritize(anyList())).thenAnswer(answ -> answ.getArgument(0));
+        when(controller.getReservedThreads()).thenReturn(Integer.MAX_VALUE);
+        when(controller.getReservationsType()).thenReturn(Reservations.Type.PER_LEVEL);
+        when(controller.parallelizeOutputShards()).thenReturn(true);
+        when(controller.getScalingParameter(anyInt())).thenReturn(2);
+        when(controller.getFanout(anyInt())).thenCallRealMethod();
+        when(controller.getThreshold(anyInt())).thenCallRealMethod();
+        when(controller.getMaxLevelDensity(anyInt(), anyDouble())).thenCallRealMethod();
+        when(controller.getSurvivalFactor(anyInt())).thenReturn(1.0);
+        when(controller.random()).thenCallRealMethod();
+
+        UnifiedCompactionStrategy strategy = new UnifiedCompactionStrategy(strategyFactory, controller);
+
+        int sstableCount = 4;
+        List<SSTableReader> sstables = mockSSTables(sstableCount, 100, 0, System.currentTimeMillis(), 0, true, null);
+        dataTracker.addInitialSSTables(sstables);
+
+        // all sstables are non-compacting
+        assertEquals(sstableCount, strategy.getLevels().values().stream().flatMap(Collection::stream).mapToInt(l -> l.sstables.size()).sum());
+
+        // mark one sstable as compacting
+        try (LifecycleTransaction txn = dataTracker.tryModify(List.of(sstables.get(0)), OperationType.COMPACTION))
+        {
+            assertEquals(sstableCount - 1, strategy.getLevels().values().stream().flatMap(Collection::stream).mapToInt(l -> l.sstables.size()).sum());
+
+            txn.obsoleteOriginals();
+            txn.abort(); // unmark compacting
+        }
+
+        // all sstables are non-compacting
+        assertEquals(sstableCount, strategy.getLevels().values().stream().flatMap(Collection::stream).mapToInt(l -> l.sstables.size()).sum());
+    }
+
     private BufferDecoratedKey key(long token)
     {
         return new BufferDecoratedKey(new Murmur3Partitioner.LongToken(token), ByteBuffer.allocate(0));