diff --git a/CHANGES.txt b/CHANGES.txt index 19c16fde48..32f4c5efbd 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 5.0-beta1 + * 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) diff --git a/src/java/org/apache/cassandra/db/compaction/unified/UnifiedCompactionTask.java b/src/java/org/apache/cassandra/db/compaction/unified/UnifiedCompactionTask.java index de62fc9113..5a729f6432 100644 --- a/src/java/org/apache/cassandra/db/compaction/unified/UnifiedCompactionTask.java +++ b/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; } diff --git a/test/unit/org/apache/cassandra/db/compaction/CompactionControllerTest.java b/test/unit/org/apache/cassandra/db/compaction/CompactionControllerTest.java index ff76ce2575..f1777cca09 100644 --- a/test/unit/org/apache/cassandra/db/compaction/CompactionControllerTest.java +++ b/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 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 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()));