CNDB-13524: Properly shutdown executors, use test CFS in PendingAntiCompactionTest

Author: driftx

test/unit/org/apache/cassandra/db/repair/PendingAntiCompactionTest.java

@@ -45,6 +45,7 @@
 
 import org.junit.Assert;
 import org.junit.Test;
+import org.awaitility.Awaitility;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -118,6 +119,18 @@ Future<Void> submitPendingAntiCompaction(PendingAntiCompaction.AcquireResult res
         }
     }
 
+    private void waitForExecutorShutdown(ExecutorService es)
+    {
+        es.shutdown();
+        try {
+            if (!es.awaitTermination(10, TimeUnit.SECONDS)) {
+                logger.warn("Executor did not terminate in time, forcing shutdown");
+                es.shutdownNow();
+            }
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+        }
+    }
     /**
      * verify the pending anti compaction happy path
      */
@@ -158,7 +171,7 @@ public void successCase() throws Exception
         }
         finally
         {
-            executor.shutdown();
+            waitForExecutorShutdown(executor);
         }
 
         assertEquals(3, cfs.getLiveSSTables().size());
@@ -491,7 +504,7 @@ public void testBlockedAcquisition() throws ExecutionException, InterruptedExcep
         }
         finally
         {
-            es.shutdown();
+            waitForExecutorShutdown(es);
             ISSTableScanner.closeAllAndPropagate(scanners, null);
         }
     }
@@ -568,44 +581,55 @@ public void onFailure(Throwable throwable)
         }
         finally
         {
-            es.shutdown();
+            waitForExecutorShutdown(es);
         }
     }
 
     @Test
-    public void testSSTablePredicateOngoingAntiCompaction()
+    public void testSSTablePredicateOngoingAntiCompaction() throws InterruptedException
     {
         ColumnFamilyStore cfs = MockSchema.newCFS();
-        cfs.disableAutoCompaction();
-        List<SSTableReader> sstables = new ArrayList<>();
-        List<SSTableReader> repairedSSTables = new ArrayList<>();
-        List<SSTableReader> pendingSSTables = new ArrayList<>();
-        for (int i = 1; i <= 10; i++)
-        {
-            SSTableReader sstable = MockSchema.sstable(i, i * 10, i * 10 + 9, cfs);
-            sstables.add(sstable);
-        }
-        for (int i = 1; i <= 10; i++)
+        try
         {
-            SSTableReader sstable = MockSchema.sstable(i + 10, i * 10, i * 10 + 9, cfs);
-            AbstractPendingRepairTest.mutateRepaired(sstable, System.currentTimeMillis());
-            repairedSSTables.add(sstable);
+            cfs.disableAutoCompaction();
+            List<SSTableReader> sstables = new ArrayList<>();
+            List<SSTableReader> repairedSSTables = new ArrayList<>();
+            List<SSTableReader> pendingSSTables = new ArrayList<>();
+            for (int i = 1; i <= 10; i++)
+            {
+                SSTableReader sstable = MockSchema.sstable(i, i * 10, i * 10 + 9, cfs);
+                sstables.add(sstable);
+            }
+            for (int i = 1; i <= 10; i++)
+            {
+                SSTableReader sstable = MockSchema.sstable(i + 10, i * 10, i * 10 + 9, cfs);
+                AbstractPendingRepairTest.mutateRepaired(sstable, System.currentTimeMillis());
+                repairedSSTables.add(sstable);
+            }
+            for (int i = 1; i <= 10; i++)
+            {
+                SSTableReader sstable = MockSchema.sstable(i + 20, i * 10, i * 10 + 9, cfs);
+                AbstractPendingRepairTest.mutateRepaired(sstable, nextTimeUUID(), false);
+                pendingSSTables.add(sstable);
+            }
+
+            cfs.addSSTables(sstables);
+            cfs.addSSTables(repairedSSTables);
+
+            // if we are compacting the non-repaired non-pending sstables, we should get an error
+            tryPredicate(cfs, sstables, null, true);
+            // make sure we don't try to grab pending or repaired sstables;
+            tryPredicate(cfs, repairedSSTables, sstables, false);
+            tryPredicate(cfs, pendingSSTables, sstables, false);
         }
-        for (int i = 1; i <= 10; i++)
+        finally
         {
-            SSTableReader sstable = MockSchema.sstable(i + 20, i * 10, i * 10 + 9, cfs);
-            AbstractPendingRepairTest.mutateRepaired(sstable, nextTimeUUID(), false);
-            pendingSSTables.add(sstable);
+            // Wait for any ongoing operations to complete
+            Awaitility.await()
+                    .atMost(10, TimeUnit.SECONDS)
+                    .pollInterval(100, TimeUnit.MILLISECONDS)
+                    .until(() -> getCompactionsFor(cfs).isEmpty());
         }
-
-        cfs.addSSTables(sstables);
-        cfs.addSSTables(repairedSSTables);
-
-        // if we are compacting the non-repaired non-pending sstables, we should get an error
-        tryPredicate(cfs, sstables, null, true);
-        // make sure we don't try to grab pending or repaired sstables;
-        tryPredicate(cfs, repairedSSTables, sstables, false);
-        tryPredicate(cfs, pendingSSTables, sstables, false);
     }
 
     private void tryPredicate(ColumnFamilyStore cfs, List<SSTableReader> compacting, List<SSTableReader> expectedLive, boolean shouldFail)
@@ -642,8 +666,8 @@ public boolean isGlobal()
     @Test
     public void testRetries() throws InterruptedException, ExecutionException, TimeoutException
     {
-        ColumnFamilyStore cfs = MockSchema.newCFS();
-        cfs.addSSTable(MockSchema.sstable(1, true, cfs));
+        makeSSTables(1);
+        
         CountDownLatch cdl = new CountDownLatch(5);
         ExecutorPlus es = executorFactory().sequential("test");
         AbstractTableOperation operation = new AbstractTableOperation()
@@ -687,15 +711,15 @@ protected PendingAntiCompaction.AcquireResult acquireSSTables()
         }
         finally
         {
-            es.shutdown();
+            waitForExecutorShutdown(es);
         }
     }
 
     @Test
     public void testRetriesTimeout() throws InterruptedException, ExecutionException, IOException, TimeoutException
     {
-        ColumnFamilyStore cfs = MockSchema.newCFS();
-        cfs.addSSTable(MockSchema.sstable(1, true, cfs));
+        makeSSTables(1);
+        
         ExecutorPlus es = executorFactory().sequential("test");
         TableOperation operation = new AbstractTableOperation()
         {
@@ -725,7 +749,7 @@ public boolean apply(SSTableReader sstable)
         }
         finally
         {
-            es.shutdown();
+            waitForExecutorShutdown(es);
         }
     }
 
@@ -754,7 +778,7 @@ public void testWith2i() throws ExecutionException, InterruptedException, Timeou
         }
         finally
         {
-            es.shutdown();
+            waitForExecutorShutdown(es);
         }
     }