natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fixed bug where UnifiedCompactionTask constructor was calling the wrong base constructor of CompactionTask 

patch by Ethan Brown; reviewed by Branimir Lambov, Stefan Miklosovic for CASSANDRA-18757
This commit is contained in:
Ethan Brown 2023-11-23 16:18:18 +01:00 committed by Stefan Miklosovic
parent c26f2a3d69
commit 9c6e671434
No known key found for this signature in database
GPG Key ID: 32F35CB2F546D93E
3 changed files with 179 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -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)

2
src/java/org/apache/cassandra/db/compaction/unified/UnifiedCompactionTask.java
View File

@ -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;
}

179
test/unit/org/apache/cassandra/db/compaction/CompactionControllerTest.java
View File

@ -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()));