STAR-1352 Port remainder of CNDB-4337 - expose all compaction picks in an aggregate, and their status (apache#472)

Co-authored-by: Stefania Alborghetti <[email protected]>

Author: djatnieks

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

@@ -204,7 +204,7 @@ CompactionAggregateStatistics getCommonStatistics(boolean trackHotness)
                 numCompactingSSTables += compaction.sstables().size();
             }
 
-            if (compaction.progress() != null)
+            if (compaction.inProgress())
             {
                 read += compaction.progress().uncompressedBytesRead();
                 written += compaction.progress().uncompressedBytesWritten();
@@ -420,7 +420,7 @@ public CompactionAggregateStatistics getStatistics()
             long readLevel = 0L;
 
             for (CompactionPick compaction : compactions)
-                if (!compaction.completed() && compaction.progress() != null)
+                if (!compaction.completed() && compaction.inProgress())
                     readLevel += compaction.progress().uncompressedBytesRead(level);
 
             return new LeveledCompactionStatistics(stats, level, score, pendingCompactions, readLevel);

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

@@ -198,6 +198,11 @@ public CompactionProgress progress()
         return progress;
     }
 
+    public boolean inProgress()
+    {
+        return progress != null;
+    }
+    
     public boolean completed()
     {
         return completed;

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

@@ -1007,6 +1007,9 @@ CompactionAggregate.UnifiedAggregate getCompactionAggregate(Shard shard,
                 bipartitionSSTables(sstables, allExpiredSSTables, liveSet, expiredSet);
             }
 
+            if (logger.isTraceEnabled())
+                logger.trace("Creating compaction aggregate with live set {}, and expired set {}", liveSet, expiredSet);
+
             List<CompactionPick> pending = ImmutableList.of();
             CompactionPick selected;
             int count = liveSet.size();
@@ -1072,6 +1075,9 @@ else if (count <= fanout * controller.getFanout(index + 1))
             else if (hasExpiredSSTables)
                 selected = selected.withExpiredSSTables(expiredSet);
 
+            if (logger.isTraceEnabled())
+                logger.trace("Returning compaction aggregate with selected compaction {}, and pending {}, for shard {}",
+                             selected, pending, shard);
             return CompactionAggregate.createUnified(sstables, selected, pending, shard, this);
         }
 

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

@@ -574,7 +574,7 @@ void testLimitCompactionsThroughput(int maxCount, int maxThroughput)
         assertEquals(0, strategy.getNextBackgroundTasks(FBUtilities.nowInSeconds()).size());
 
         for (CompactionPick pick : strategy.backgroundCompactions.getCompactionsInProgress())
-            if (pick.progress() == null)
+            if (!pick.inProgress())
                 strategy.backgroundCompactions.onInProgress(mockProgress(strategy, pick.id()));
 
         // and also when they do