make testCompactionProgress less racy
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/6f8d7371 Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/6f8d7371 Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/6f8d7371 Branch: refs/heads/trunk Commit: 6f8d73711a5d9d64947a1f5d51284bfda3d5d127 Parents: 2c35a5d Author: Jonathan Ellis <[email protected]> Authored: Tue Apr 9 21:10:58 2013 -0500 Committer: Jonathan Ellis <[email protected]> Committed: Tue Apr 9 21:10:58 2013 -0500 ---------------------------------------------------------------------- .../compaction/LeveledCompactionStrategyTest.java | 56 ++++++++------- 1 files changed, 31 insertions(+), 25 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/cassandra/blob/6f8d7371/test/unit/org/apache/cassandra/db/compaction/LeveledCompactionStrategyTest.java ---------------------------------------------------------------------- diff --git a/test/unit/org/apache/cassandra/db/compaction/LeveledCompactionStrategyTest.java b/test/unit/org/apache/cassandra/db/compaction/LeveledCompactionStrategyTest.java index 9760c7e..4c70fe0 100644 --- a/test/unit/org/apache/cassandra/db/compaction/LeveledCompactionStrategyTest.java +++ b/test/unit/org/apache/cassandra/db/compaction/LeveledCompactionStrategyTest.java @@ -20,7 +20,9 @@ package org.apache.cassandra.db.compaction; import java.nio.ByteBuffer; import java.util.Collection; import java.util.HashSet; +import java.util.List; import java.util.Set; +import java.util.concurrent.ExecutionException; import org.junit.Test; import org.junit.runner.RunWith; @@ -78,15 +80,11 @@ public class LeveledCompactionStrategyTest extends SchemaLoader cfs.forceBlockingFlush(); } - LeveledCompactionStrategy strat = (LeveledCompactionStrategy)cfs.getCompactionStrategy(); - while (strat.getLevelSize(0) > 1) - { - cfs.forceMajorCompaction(); - Thread.sleep(100); - } + waitForLeveling(cfs); + LeveledCompactionStrategy strategy = (LeveledCompactionStrategy) cfs.getCompactionStrategy(); // Checking we're not completely bad at math - assert strat.getLevelSize(1) > 0; - assert strat.getLevelSize(2) > 0; + assert strategy.getLevelSize(1) > 0; + assert strategy.getLevelSize(2) > 0; ActiveRepairService.CFPair p = new ActiveRepairService.CFPair(ksname, cfname); Range<Token> range = new Range<Token>(Util.token(""), Util.token("")); @@ -96,6 +94,17 @@ public class LeveledCompactionStrategyTest extends SchemaLoader CompactionManager.instance.submitValidation(cfs, validator).get(); } + /** + * wait for leveled compaction to quiesce on the given columnfamily + */ + private void waitForLeveling(ColumnFamilyStore cfs) throws InterruptedException, ExecutionException + { + LeveledCompactionStrategy strategy = (LeveledCompactionStrategy) cfs.getCompactionStrategy(); + // L0 is the lowest priority, so when that's done, we know everything is done + while (strategy.getLevelSize(0) > 1) + Thread.sleep(100); + } + @Test public void testCompactionProgress() throws Exception { @@ -119,14 +128,16 @@ public class LeveledCompactionStrategyTest extends SchemaLoader rm.apply(); cfs.forceBlockingFlush(); } - cfs.forceMajorCompaction(); - LeveledCompactionStrategy strat = (LeveledCompactionStrategy)cfs.getCompactionStrategy(); - assert strat.getLevelSize(1) > 0; + waitForLeveling(cfs); + LeveledCompactionStrategy strategy = (LeveledCompactionStrategy) cfs.getCompactionStrategy(); + assert strategy.getLevelSize(1) > 0; // get LeveledScanner for level 1 sstables - Collection<SSTableReader> sstables = strat.manifest.getLevel(1); - ICompactionScanner scanner = strat.getScanners(sstables).get(0); + Collection<SSTableReader> sstables = strategy.manifest.getLevel(1); + List<ICompactionScanner> scanners = strategy.getScanners(sstables); + assertEquals(1, scanners.size()); // should be one per level + ICompactionScanner scanner = scanners.get(0); // scan through to the end while (scanner.hasNext()) scanner.next(); @@ -162,35 +173,30 @@ public class LeveledCompactionStrategyTest extends SchemaLoader cfs.forceBlockingFlush(); } - cfs.disableAutoCompaction(); - LeveledCompactionStrategy strat = (LeveledCompactionStrategy)cfs.getCompactionStrategy(); - while (strat.getLevelSize(0) > 1) - { - cfs.forceMajorCompaction(); - Thread.sleep(100); - } + waitForLeveling(cfs); + LeveledCompactionStrategy strategy = (LeveledCompactionStrategy) cfs.getCompactionStrategy(); Set<SSTableReader> changedSSTables = new HashSet<SSTableReader>(); Collection<SSTableReader> sstables = cfs.getDataTracker().getUncompactingSSTables(); - cfs.getDataTracker().markCompacting(sstables); // dont touch my sstables! + assert cfs.getDataTracker().markCompacting(sstables); // dont touch my sstables! // change sstable level on all current sstables for (SSTableReader s : sstables) { assertTrue(s.getSSTableLevel() != 6); - strat.manifest.remove(s); + strategy.manifest.remove(s); LeveledManifest.mutateLevel(s.getSSTableMetadata(), s.descriptor, s.descriptor.filenameFor(Component.STATS), 6); s.reloadSSTableMetadata(); changedSSTables.add(s); - strat.manifest.add(s); + strategy.manifest.add(s); } // verify that all sstables in the changed set is level 6 - for(SSTableReader s : table.getColumnFamilyStore(cfname).getSSTables()) + for (SSTableReader s : table.getColumnFamilyStore(cfname).getSSTables()) { if (changedSSTables.contains(s)) assertTrue(s.getSSTableLevel() == 6); } - int [] levels = strat.manifest.getAllLevelSize(); + int[] levels = strategy.manifest.getAllLevelSize(); // verify that the manifest has correct amount of sstables assertEquals(changedSSTables.size(), levels[6]); }
