Merge branch 'cassandra-5.0' into trunk

CHANGES.txt

@@ -3,6 +3,7 @@
  * Add the ability to disable bulk loading of SSTables (CASSANDRA-18781)
  * Clean up obsolete functions and simplify cql_version handling in cqlsh (CASSANDRA-18787)
 Merged from 5.0:
+ * Fixed bug where UnifiedCompactionTask constructor was calling the wrong base constructor of CompactionTask (CASSANDRA-18757)
  * Fix SAI unindexed contexts not considering CONTAINS KEY (CASSANDRA-19040)
  * Ensure that empty SAI column indexes do not fail on validation after full-SSTable streaming (CASSANDRA-19017)
  * SAI in-memory index should check max term size (CASSANDRA-18926)

src/java/org/apache/cassandra/db/compaction/unified/UnifiedCompactionTask.java

@@ -43,7 +43,7 @@ public class UnifiedCompactionTask extends CompactionTask
                                  long gcBefore,
                                  ShardManager shardManager)
     {
-        super(cfs, txn, gcBefore, strategy.getController().getIgnoreOverlapsInExpirationCheck());
+        super(cfs, txn, gcBefore);
         this.controller = strategy.getController();
         this.shardManager = shardManager;
     }

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

@@ -224,15 +224,109 @@ public class CompactionControllerTest extends SchemaLoader
     condition = "Thread.currentThread().getName().equals(\"compaction1\")",
     action = "org.apache.cassandra.db.compaction.CompactionControllerTest.incrementOverlapRefreshCounter();")
     })
-    public void testIgnoreOverlaps() throws Exception
+    public void testIgnoreOverlapsTrue() throws Exception
     {
+        resetCounters();
         testOverlapIterator(true);
+    }
+
+    @Test
+    @BMRules(rules = {
+    @BMRule(name = "Pause compaction",
+    targetClass = "CompactionTask",
+    targetMethod = "runMayThrow",
+    targetLocation = "INVOKE getCompactionAwareWriter",
+    condition = "Thread.currentThread().getName().equals(\"compaction1\")",
+    action = "org.apache.cassandra.db.compaction.CompactionControllerTest.createCompactionControllerLatch.countDown();" +
+             "com.google.common.util.concurrent.Uninterruptibles.awaitUninterruptibly" +
+             "(org.apache.cassandra.db.compaction.CompactionControllerTest.compaction2FinishLatch);"),
+    @BMRule(name = "Check overlaps",
+    targetClass = "CompactionTask",
+    targetMethod = "runMayThrow",
+    targetLocation = "INVOKE finish",
+    condition = "Thread.currentThread().getName().equals(\"compaction1\")",
+    action = "org.apache.cassandra.db.compaction.CompactionControllerTest.compaction1RefreshLatch.countDown();" +
+             "com.google.common.util.concurrent.Uninterruptibles.awaitUninterruptibly" +
+             "(org.apache.cassandra.db.compaction.CompactionControllerTest.refreshCheckLatch);"),
+    @BMRule(name = "Increment overlap refresh counter",
+    targetClass = "ColumnFamilyStore",
+    targetMethod = "getAndReferenceOverlappingLiveSSTables",
+    condition = "Thread.currentThread().getName().equals(\"compaction1\")",
+    action = "org.apache.cassandra.db.compaction.CompactionControllerTest.incrementOverlapRefreshCounter();")
+    })
+    public void testIgnoreOverlapsFalse() throws Exception
+    {
+        resetCounters();
+        testOverlapIterator(false);
+    }
+
+    @Test
+    @BMRules(rules = {
+    @BMRule(name = "Pause compaction",
+    targetClass = "CompactionTask",
+    targetMethod = "runMayThrow",
+    targetLocation = "INVOKE getCompactionAwareWriter",
+    condition = "Thread.currentThread().getName().equals(\"compaction1\")",
+    action = "org.apache.cassandra.db.compaction.CompactionControllerTest.createCompactionControllerLatch.countDown();" +
+             "com.google.common.util.concurrent.Uninterruptibles.awaitUninterruptibly" +
+             "(org.apache.cassandra.db.compaction.CompactionControllerTest.compaction2FinishLatch);"),
+    @BMRule(name = "Check overlaps",
+    targetClass = "CompactionTask",
+    targetMethod = "runMayThrow",
+    targetLocation = "INVOKE finish",
+    condition = "Thread.currentThread().getName().equals(\"compaction1\")",
+    action = "org.apache.cassandra.db.compaction.CompactionControllerTest.compaction1RefreshLatch.countDown();" +
+             "com.google.common.util.concurrent.Uninterruptibles.awaitUninterruptibly" +
+             "(org.apache.cassandra.db.compaction.CompactionControllerTest.refreshCheckLatch);"),
+    @BMRule(name = "Increment overlap refresh counter",
+    targetClass = "ColumnFamilyStore",
+    targetMethod = "getAndReferenceOverlappingLiveSSTables",
+    condition = "Thread.currentThread().getName().equals(\"compaction1\")",
+    action = "org.apache.cassandra.db.compaction.CompactionControllerTest.incrementOverlapRefreshCounter();")
+    })
+    public void testIgnoreOverlapsUCSTrue() throws Exception
+    {
+        resetCounters();
+        testOverlapIteratorUCS(true);
+    }
+
+    @Test
+    @BMRules(rules = {
+    @BMRule(name = "Pause compaction",
+    targetClass = "CompactionTask",
+    targetMethod = "runMayThrow",
+    targetLocation = "INVOKE getCompactionAwareWriter",
+    condition = "Thread.currentThread().getName().equals(\"compaction1\")",
+    action = "org.apache.cassandra.db.compaction.CompactionControllerTest.createCompactionControllerLatch.countDown();" +
+             "com.google.common.util.concurrent.Uninterruptibles.awaitUninterruptibly" +
+             "(org.apache.cassandra.db.compaction.CompactionControllerTest.compaction2FinishLatch);"),
+    @BMRule(name = "Check overlaps",
+    targetClass = "CompactionTask",
+    targetMethod = "runMayThrow",
+    targetLocation = "INVOKE finish",
+    condition = "Thread.currentThread().getName().equals(\"compaction1\")",
+    action = "org.apache.cassandra.db.compaction.CompactionControllerTest.compaction1RefreshLatch.countDown();" +
+             "com.google.common.util.concurrent.Uninterruptibles.awaitUninterruptibly" +
+             "(org.apache.cassandra.db.compaction.CompactionControllerTest.refreshCheckLatch);"),
+    @BMRule(name = "Increment overlap refresh counter",
+    targetClass = "ColumnFamilyStore",
+    targetMethod = "getAndReferenceOverlappingLiveSSTables",
+    condition = "Thread.currentThread().getName().equals(\"compaction1\")",
+    action = "org.apache.cassandra.db.compaction.CompactionControllerTest.incrementOverlapRefreshCounter();")
+    })
+    public void testIgnoreOverlapsUCSFalse() throws Exception
+    {
+        resetCounters();
+        testOverlapIteratorUCS(false);
+    }
+
+    private void resetCounters()
+    {
         overlapRefreshCounter = 0;
         compaction2FinishLatch = new CountDownLatch(1);
         createCompactionControllerLatch = new CountDownLatch(1);
         compaction1RefreshLatch = new CountDownLatch(1);
         refreshCheckLatch = new CountDownLatch(1);
-        testOverlapIterator(false);
     }
 
     public void testOverlapIterator(boolean ignoreOverlaps) throws Exception
@@ -271,6 +365,87 @@ public class CompactionControllerTest extends SchemaLoader
         twcs.startup();
 
         CompactionTask task = (CompactionTask)twcs.getUserDefinedTask(sstables, 0);
+        assertFalse(task.keepOriginals);
+
+        assertNotNull(task);
+        assertEquals(1, Iterables.size(task.transaction.originals()));
+
+        //start a compaction for the first sstable (compaction1)
+        //the overlap iterator should contain sstable2
+        //this compaction will be paused by the BMRule
+        Thread t = new Thread(() -> {
+            task.execute(null);
+        });
+
+        //start a compaction for the second sstable (compaction2)
+        //the overlap iterator should contain sstable1
+        //this compaction should complete as normal
+        Thread t2 = new Thread(() -> {
+            Uninterruptibles.awaitUninterruptibly(createCompactionControllerLatch);
+            assertEquals(1, overlapRefreshCounter);
+            CompactionManager.instance.forceUserDefinedCompaction(sstable2);
+
+            //after compaction2 is finished, wait 1 minute and then resume compaction1 (this gives enough time for the overlapIterator to be refreshed)
+            //after resuming, the overlap iterator for compaction1 should be updated to include the new sstable created by compaction2,
+            //and it should not contain sstable2
+            try
+            {
+                TimeUnit.MINUTES.sleep(1);
+            }
+            catch (InterruptedException e)
+            {
+                throw new RuntimeException(e);
+            }
+            compaction2FinishLatch.countDown();
+        });
+
+        t.setName("compaction1");
+        t.start();
+        t2.start();
+
+        compaction1RefreshLatch.await();
+        //at this point, the overlap iterator for compaction1 should be refreshed
+
+        //verify that the overlap iterator for compaction1 is refreshed twice, (once during the constructor, and again after compaction2 finishes)
+        assertEquals(2, overlapRefreshCounter);
+
+        refreshCheckLatch.countDown();
+        t.join();
+    }
+
+    public void testOverlapIteratorUCS(boolean ignoreOverlaps) throws Exception
+    {
+
+        Keyspace keyspace = Keyspace.open(KEYSPACE);
+        ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF1);
+        cfs.truncateBlocking();
+        cfs.disableAutoCompaction();
+
+        //create 2 overlapping sstables
+        DecoratedKey key = Util.dk("k1");
+        long timestamp1 = FBUtilities.timestampMicros();
+        long timestamp2 = timestamp1 - 5;
+        applyMutation(cfs.metadata(), key, timestamp1);
+        cfs.forceBlockingFlush(ColumnFamilyStore.FlushReason.UNIT_TESTS);
+        assertEquals(cfs.getLiveSSTables().size(), 1);
+        Set<SSTableReader> sstables = cfs.getLiveSSTables();
+
+        applyMutation(cfs.metadata(), key, timestamp2);
+        cfs.forceBlockingFlush(ColumnFamilyStore.FlushReason.UNIT_TESTS);
+        assertEquals(cfs.getLiveSSTables().size(), 2);
+        String sstable2 = cfs.getLiveSSTables().iterator().next().getFilename();
+
+        CassandraRelevantProperties.ALLOW_UNSAFE_AGGRESSIVE_SSTABLE_EXPIRATION.setBoolean(true);
+        Map<String, String> options = new HashMap<>();
+        options.put(TimeWindowCompactionStrategyOptions.UNSAFE_AGGRESSIVE_SSTABLE_EXPIRATION_KEY, Boolean.toString(ignoreOverlaps));
+        UnifiedCompactionStrategy ucs = new UnifiedCompactionStrategy(cfs, options);
+        for (SSTableReader sstable : cfs.getLiveSSTables())
+            ucs.addSSTable(sstable);
+
+        ucs.startup();
+
+        CompactionTask task = (CompactionTask)ucs.getUserDefinedTask(sstables, 0);
+        assertFalse(task.keepOriginals);
 
         assertNotNull(task);
         assertEquals(1, Iterables.size(task.transaction.originals()));