diff --git a/CHANGES.txt b/CHANGES.txt index e3b23eea46..4d0e3e1d19 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,3 +1,7 @@ +3.0.19 + * Anti-compaction temporarily corrupts sstable state for readers (CASSANDRA-15004) + + 3.0.18 * Severe concurrency issues in STCS,DTCS,TWCS,TMD.Topology,TypeParser * Add a script to make running the cqlsh tests in cassandra repo easier (CASSANDRA-14951) diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java index 4fab7ccebe..4295c7a6ba 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java @@ -42,9 +42,11 @@ import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.config.Schema; import org.apache.cassandra.db.*; import org.apache.cassandra.db.compaction.CompactionInfo.Holder; +import org.apache.cassandra.db.lifecycle.ILifecycleTransaction; import org.apache.cassandra.db.lifecycle.LifecycleTransaction; import org.apache.cassandra.db.lifecycle.SSTableSet; import org.apache.cassandra.db.lifecycle.View; +import org.apache.cassandra.db.lifecycle.WrappedLifecycleTransaction; import org.apache.cassandra.db.rows.UnfilteredRowIterator; import org.apache.cassandra.db.view.ViewBuilder; import org.apache.cassandra.dht.Bounds; @@ -1058,7 +1060,7 @@ public class CompactionManager implements CompactionManagerMBean int expectedBloomFilterSize, long repairedAt, Collection sstables, - LifecycleTransaction txn) + ILifecycleTransaction txn) { FileUtils.createDirectory(compactionFileLocation); int minLevel = Integer.MAX_VALUE; @@ -1287,8 +1289,10 @@ public class CompactionManager implements CompactionManagerMBean logger.info(format, numAnticompact, antiCompactedSSTableCount); } - private int antiCompactGroup(ColumnFamilyStore cfs, Collection> ranges, - LifecycleTransaction anticompactionGroup, long repairedAt) + + @VisibleForTesting + int antiCompactGroup(ColumnFamilyStore cfs, Collection> ranges, + LifecycleTransaction anticompactionGroup, long repairedAt) { long groupMaxDataAge = -1; @@ -1313,17 +1317,52 @@ public class CompactionManager implements CompactionManagerMBean long unrepairedKeyCount = 0; int nowInSec = FBUtilities.nowInSeconds(); + /** + * HACK WARNING + * + * We have multiple writers operating over the same Transaction, producing different sets of sstables that all + * logically replace the transaction's originals. The SSTableRewriter assumes it has exclusive control over + * the transaction state, and this will lead to temporarily inconsistent sstable/tracker state if we do not + * take special measures to avoid it. + * + * Specifically, if a number of rewriter have prepareToCommit() invoked in sequence, then two problematic things happen: + * 1. The obsoleteOriginals() call of the first rewriter immediately remove the originals from the tracker, despite + * their having been only partially replaced. To avoid this, we must either avoid obsoleteOriginals() or checkpoint() + * 2. The LifecycleTransaction may only have prepareToCommit() invoked once, and this will checkpoint() also. + * + * Similarly commit() would finalise partially complete on-disk state. + * + * To avoid these problems, we introduce a SharedTxn that proxies all calls onto the underlying transaction + * except prepareToCommit(), checkpoint(), obsoleteOriginals(), and commit(). + * We then invoke these methods directly once each of the rewriter has updated the transaction + * with their share of replacements. + * + * Note that for the same essential reason we also explicitly disable early open. + * By noop-ing checkpoint we avoid any of the problems with early open, but by continuing to explicitly + * disable it we also prevent any of the extra associated work from being performed. + */ + class SharedTxn extends WrappedLifecycleTransaction + { + public SharedTxn(ILifecycleTransaction delegate) { super(delegate); } + public Throwable commit(Throwable accumulate) { return accumulate; } + public void prepareToCommit() {} + public void checkpoint() {} + public void obsoleteOriginals() {} + public void close() {} + } + CompactionStrategyManager strategy = cfs.getCompactionStrategyManager(); - try (SSTableRewriter repairedSSTableWriter = new SSTableRewriter(anticompactionGroup, groupMaxDataAge, false, false); - SSTableRewriter unRepairedSSTableWriter = new SSTableRewriter(anticompactionGroup, groupMaxDataAge, false, false); + try (SharedTxn sharedTxn = new SharedTxn(anticompactionGroup); + SSTableRewriter repairedSSTableWriter = new SSTableRewriter(sharedTxn, groupMaxDataAge, false, false); + SSTableRewriter unRepairedSSTableWriter = new SSTableRewriter(sharedTxn, groupMaxDataAge, false, false); AbstractCompactionStrategy.ScannerList scanners = strategy.getScanners(anticompactionGroup.originals()); CompactionController controller = new CompactionController(cfs, sstableAsSet, getDefaultGcBefore(cfs, nowInSec)); CompactionIterator ci = new CompactionIterator(OperationType.ANTICOMPACTION, scanners.scanners, controller, nowInSec, UUIDGen.getTimeUUID(), metrics)) { int expectedBloomFilterSize = Math.max(cfs.metadata.params.minIndexInterval, (int)(SSTableReader.getApproximateKeyCount(sstableAsSet))); - repairedSSTableWriter.switchWriter(CompactionManager.createWriterForAntiCompaction(cfs, destination, expectedBloomFilterSize, repairedAt, sstableAsSet, anticompactionGroup)); - unRepairedSSTableWriter.switchWriter(CompactionManager.createWriterForAntiCompaction(cfs, destination, expectedBloomFilterSize, ActiveRepairService.UNREPAIRED_SSTABLE, sstableAsSet, anticompactionGroup)); + repairedSSTableWriter.switchWriter(CompactionManager.createWriterForAntiCompaction(cfs, destination, expectedBloomFilterSize, repairedAt, sstableAsSet, sharedTxn)); + unRepairedSSTableWriter.switchWriter(CompactionManager.createWriterForAntiCompaction(cfs, destination, expectedBloomFilterSize, ActiveRepairService.UNREPAIRED_SSTABLE, sstableAsSet, sharedTxn)); Range.OrderedRangeContainmentChecker containmentChecker = new Range.OrderedRangeContainmentChecker(ranges); while (ci.hasNext()) { @@ -1345,16 +1384,17 @@ public class CompactionManager implements CompactionManagerMBean } List anticompactedSSTables = new ArrayList<>(); - // since both writers are operating over the same Transaction, we cannot use the convenience Transactional.finish() method, - // as on the second finish() we would prepareToCommit() on a Transaction that has already been committed, which is forbidden by the API - // (since it indicates misuse). We call permitRedundantTransitions so that calls that transition to a state already occupied are permitted. - anticompactionGroup.permitRedundantTransitions(); + repairedSSTableWriter.setRepairedAt(repairedAt).prepareToCommit(); unRepairedSSTableWriter.prepareToCommit(); + anticompactionGroup.checkpoint(); + anticompactionGroup.obsoleteOriginals(); + anticompactionGroup.prepareToCommit(); anticompactedSSTables.addAll(repairedSSTableWriter.finished()); anticompactedSSTables.addAll(unRepairedSSTableWriter.finished()); repairedSSTableWriter.commit(); unRepairedSSTableWriter.commit(); + Throwables.maybeFail(anticompactionGroup.commit(null)); logger.trace("Repaired {} keys out of {} for {}/{} in {}", repairedKeyCount, repairedKeyCount + unrepairedKeyCount, diff --git a/src/java/org/apache/cassandra/db/lifecycle/ILifecycleTransaction.java b/src/java/org/apache/cassandra/db/lifecycle/ILifecycleTransaction.java new file mode 100644 index 0000000000..d694a86a50 --- /dev/null +++ b/src/java/org/apache/cassandra/db/lifecycle/ILifecycleTransaction.java @@ -0,0 +1,38 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.db.lifecycle; + +import java.util.Collection; +import java.util.Set; + +import org.apache.cassandra.io.sstable.SSTable; +import org.apache.cassandra.io.sstable.format.SSTableReader; +import org.apache.cassandra.utils.concurrent.Transactional; + +public interface ILifecycleTransaction extends Transactional, LifecycleNewTracker +{ + void checkpoint(); + void update(SSTableReader reader, boolean original); + void update(Collection readers, boolean original); + public SSTableReader current(SSTableReader reader); + void obsolete(SSTableReader reader); + void obsoleteOriginals(); + Set originals(); + boolean isObsolete(SSTableReader reader); +} diff --git a/src/java/org/apache/cassandra/db/lifecycle/LifecycleTransaction.java b/src/java/org/apache/cassandra/db/lifecycle/LifecycleTransaction.java index af9a80a718..7ecaa38e1c 100644 --- a/src/java/org/apache/cassandra/db/lifecycle/LifecycleTransaction.java +++ b/src/java/org/apache/cassandra/db/lifecycle/LifecycleTransaction.java @@ -55,7 +55,7 @@ import static org.apache.cassandra.utils.concurrent.Refs.selfRefs; * action to occur at the beginning of the commit phase, but also *requires* that the prepareToCommit() phase only take * actions that can be rolled back. */ -public class LifecycleTransaction extends Transactional.AbstractTransactional implements LifecycleNewTracker +public class LifecycleTransaction extends Transactional.AbstractTransactional implements ILifecycleTransaction { private static final Logger logger = LoggerFactory.getLogger(LifecycleTransaction.class); @@ -282,11 +282,6 @@ public class LifecycleTransaction extends Transactional.AbstractTransactional im return tracker.isDummy(); } - public void permitRedundantTransitions() - { - super.permitRedundantTransitions(); - } - /** * call when a consistent batch of changes is ready to be made atomically visible * these will be exposed in the Tracker atomically, or an exception will be thrown; in this case diff --git a/src/java/org/apache/cassandra/db/lifecycle/WrappedLifecycleTransaction.java b/src/java/org/apache/cassandra/db/lifecycle/WrappedLifecycleTransaction.java new file mode 100644 index 0000000000..ff84208bde --- /dev/null +++ b/src/java/org/apache/cassandra/db/lifecycle/WrappedLifecycleTransaction.java @@ -0,0 +1,112 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.db.lifecycle; + +import java.util.Collection; +import java.util.Set; + +import org.apache.cassandra.db.compaction.OperationType; +import org.apache.cassandra.io.sstable.SSTable; +import org.apache.cassandra.io.sstable.format.SSTableReader; + +public class WrappedLifecycleTransaction implements ILifecycleTransaction +{ + + final ILifecycleTransaction delegate; + public WrappedLifecycleTransaction(ILifecycleTransaction delegate) + { + this.delegate = delegate; + } + + public void checkpoint() + { + delegate.checkpoint(); + } + + public void update(SSTableReader reader, boolean original) + { + delegate.update(reader, original); + } + + public void update(Collection readers, boolean original) + { + delegate.update(readers, original); + } + + public SSTableReader current(SSTableReader reader) + { + return delegate.current(reader); + } + + public void obsolete(SSTableReader reader) + { + delegate.obsolete(reader); + } + + public void obsoleteOriginals() + { + delegate.obsoleteOriginals(); + } + + public Set originals() + { + return delegate.originals(); + } + + public boolean isObsolete(SSTableReader reader) + { + return delegate.isObsolete(reader); + } + + public Throwable commit(Throwable accumulate) + { + return delegate.commit(accumulate); + } + + public Throwable abort(Throwable accumulate) + { + return delegate.abort(accumulate); + } + + public void prepareToCommit() + { + delegate.prepareToCommit(); + } + + public void close() + { + delegate.close(); + } + + public void trackNew(SSTable table) + { + delegate.trackNew(table); + } + + public void untrackNew(SSTable table) + { + delegate.untrackNew(table); + } + + public OperationType opType() + { + return delegate.opType(); + } + +} diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableRewriter.java b/src/java/org/apache/cassandra/io/sstable/SSTableRewriter.java index 0ea28d728e..b2fbcb1da0 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableRewriter.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableRewriter.java @@ -28,6 +28,7 @@ import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.db.DecoratedKey; import org.apache.cassandra.db.RowIndexEntry; +import org.apache.cassandra.db.lifecycle.ILifecycleTransaction; import org.apache.cassandra.db.rows.UnfilteredRowIterator; import org.apache.cassandra.io.sstable.format.SSTableReader; import org.apache.cassandra.io.sstable.format.SSTableWriter; @@ -59,7 +60,7 @@ public class SSTableRewriter extends Transactional.AbstractTransactional impleme private final long maxAge; private long repairedAt = -1; // the set of final readers we will expose on commit - private final LifecycleTransaction transaction; // the readers we are rewriting (updated as they are replaced) + private final ILifecycleTransaction transaction; // the readers we are rewriting (updated as they are replaced) private final List preparedForCommit = new ArrayList<>(); private long currentlyOpenedEarlyAt; // the position (in MB) in the target file we last (re)opened at @@ -74,18 +75,18 @@ public class SSTableRewriter extends Transactional.AbstractTransactional impleme // for testing (TODO: remove when have byteman setup) private boolean throwEarly, throwLate; - public SSTableRewriter(LifecycleTransaction transaction, long maxAge, boolean isOffline) + public SSTableRewriter(ILifecycleTransaction transaction, long maxAge, boolean isOffline) { this(transaction, maxAge, isOffline, true); } - public SSTableRewriter(LifecycleTransaction transaction, long maxAge, boolean isOffline, boolean shouldOpenEarly) + public SSTableRewriter(ILifecycleTransaction transaction, long maxAge, boolean isOffline, boolean shouldOpenEarly) { this(transaction, maxAge, isOffline, calculateOpenInterval(shouldOpenEarly), false); } @VisibleForTesting - public SSTableRewriter(LifecycleTransaction transaction, long maxAge, boolean isOffline, long preemptiveOpenInterval, boolean keepOriginals) + public SSTableRewriter(ILifecycleTransaction transaction, long maxAge, boolean isOffline, long preemptiveOpenInterval, boolean keepOriginals) { this.transaction = transaction; this.maxAge = maxAge; @@ -94,12 +95,12 @@ public class SSTableRewriter extends Transactional.AbstractTransactional impleme this.preemptiveOpenInterval = preemptiveOpenInterval; } - public static SSTableRewriter constructKeepingOriginals(LifecycleTransaction transaction, boolean keepOriginals, long maxAge, boolean isOffline) + public static SSTableRewriter constructKeepingOriginals(ILifecycleTransaction transaction, boolean keepOriginals, long maxAge, boolean isOffline) { return new SSTableRewriter(transaction, maxAge, isOffline, calculateOpenInterval(true), keepOriginals); } - public static SSTableRewriter construct(ColumnFamilyStore cfs, LifecycleTransaction transaction, boolean keepOriginals, long maxAge, boolean isOffline) + public static SSTableRewriter construct(ColumnFamilyStore cfs, ILifecycleTransaction transaction, boolean keepOriginals, long maxAge, boolean isOffline) { return new SSTableRewriter(transaction, maxAge, isOffline, calculateOpenInterval(cfs.supportsEarlyOpen()), keepOriginals); } diff --git a/src/java/org/apache/cassandra/utils/concurrent/Transactional.java b/src/java/org/apache/cassandra/utils/concurrent/Transactional.java index abb876c221..afc4fdf466 100644 --- a/src/java/org/apache/cassandra/utils/concurrent/Transactional.java +++ b/src/java/org/apache/cassandra/utils/concurrent/Transactional.java @@ -74,7 +74,6 @@ public interface Transactional extends AutoCloseable ABORTED; } - private boolean permitRedundantTransitions; private State state = State.IN_PROGRESS; // the methods for actually performing the necessary behaviours, that are themselves protected against @@ -109,8 +108,6 @@ public interface Transactional extends AutoCloseable */ public final Throwable commit(Throwable accumulate) { - if (permitRedundantTransitions && state == State.COMMITTED) - return accumulate; if (state != State.READY_TO_COMMIT) throw new IllegalStateException("Cannot commit unless READY_TO_COMMIT; state is " + state); accumulate = doCommit(accumulate); @@ -165,8 +162,6 @@ public interface Transactional extends AutoCloseable */ public final void prepareToCommit() { - if (permitRedundantTransitions && state == State.READY_TO_COMMIT) - return; if (state != State.IN_PROGRESS) throw new IllegalStateException("Cannot prepare to commit unless IN_PROGRESS; state is " + state); @@ -204,11 +199,6 @@ public interface Transactional extends AutoCloseable { return state; } - - protected void permitRedundantTransitions() - { - permitRedundantTransitions = true; - } } // commit should generally never throw an exception, and preferably never generate one, diff --git a/test/unit/org/apache/cassandra/db/compaction/AntiCompactionBytemanTest.java b/test/unit/org/apache/cassandra/db/compaction/AntiCompactionBytemanTest.java new file mode 100644 index 0000000000..ba6f3a1a5a --- /dev/null +++ b/test/unit/org/apache/cassandra/db/compaction/AntiCompactionBytemanTest.java @@ -0,0 +1,125 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.db.compaction; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; +import java.util.Iterator; +import java.util.List; +import java.util.Set; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import com.google.common.collect.Sets; +import com.google.common.util.concurrent.Uninterruptibles; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.apache.cassandra.cql3.CQLTester; +import org.apache.cassandra.cql3.UntypedResultSet; +import org.apache.cassandra.db.lifecycle.LifecycleTransaction; +import org.apache.cassandra.dht.Murmur3Partitioner; +import org.apache.cassandra.dht.Range; +import org.apache.cassandra.dht.Token; +import org.apache.cassandra.io.sstable.format.SSTableReader; +import org.jboss.byteman.contrib.bmunit.BMRule; +import org.jboss.byteman.contrib.bmunit.BMRules; +import org.jboss.byteman.contrib.bmunit.BMUnitRunner; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; + +@RunWith(BMUnitRunner.class) +public class AntiCompactionBytemanTest extends CQLTester +{ + @Test + @BMRules(rules = { @BMRule(name = "Insert delay after first prepareToCommit", + targetClass = "CompactionManager", + targetMethod = "antiCompactGroup", + condition = "not flagged(\"done\")", + targetLocation = "AFTER INVOKE prepareToCommit", + action = "Thread.sleep(2000);") } ) + public void testRedundantTransitions() throws Throwable + { + createTable("create table %s (id int primary key, i int)"); + execute("insert into %s (id, i) values (1, 1)"); + execute("insert into %s (id, i) values (2, 1)"); + getCurrentColumnFamilyStore().forceBlockingFlush(); + UntypedResultSet res = execute("select token(id) as tok from %s"); + Iterator it = res.iterator(); + List tokens = new ArrayList<>(); + while (it.hasNext()) + { + UntypedResultSet.Row r = it.next(); + tokens.add(r.getLong("tok")); + } + tokens.sort(Long::compareTo); + + long first = tokens.get(0) - 10; + long last = tokens.get(0) + 10; + Range toRepair = new Range<>(new Murmur3Partitioner.LongToken(first), new Murmur3Partitioner.LongToken(last)); + + AtomicBoolean failed = new AtomicBoolean(false); + AtomicBoolean finished = new AtomicBoolean(false); + + Thread t = new Thread(() -> { + while (!finished.get()) + { + UntypedResultSet result = null; + try + { + result = execute("select id from %s"); + } + catch (Throwable throwable) + { + failed.set(true); + throw new RuntimeException(throwable); + } + + Iterator rowIter = result.iterator(); + Set ids = new HashSet<>(); + while (rowIter.hasNext()) + { + UntypedResultSet.Row r = rowIter.next(); + ids.add(r.getInt("id")); + } + if (!Sets.newHashSet(1,2).equals(ids)) + { + failed.set(true); + return; + } + Uninterruptibles.sleepUninterruptibly(10, TimeUnit.MILLISECONDS); + } + }); + t.start(); + assertEquals(1, getCurrentColumnFamilyStore().getLiveSSTables().size()); + SSTableReader sstableBefore = getCurrentColumnFamilyStore().getLiveSSTables().iterator().next(); + + try (LifecycleTransaction txn = getCurrentColumnFamilyStore().getTracker().tryModify(getCurrentColumnFamilyStore().getLiveSSTables(), OperationType.ANTICOMPACTION)) + { + CompactionManager.instance.antiCompactGroup(getCurrentColumnFamilyStore(), Collections.singleton(toRepair), txn, 123); + } + finished.set(true); + t.join(); + assertFalse(failed.get()); + assertFalse(getCurrentColumnFamilyStore().getLiveSSTables().contains(sstableBefore)); + AntiCompactionTest.assertOnDiskState(getCurrentColumnFamilyStore(), 2); + } +} \ No newline at end of file diff --git a/test/unit/org/apache/cassandra/db/compaction/AntiCompactionTest.java b/test/unit/org/apache/cassandra/db/compaction/AntiCompactionTest.java index 8991f88a64..e2b17e8788 100644 --- a/test/unit/org/apache/cassandra/db/compaction/AntiCompactionTest.java +++ b/test/unit/org/apache/cassandra/db/compaction/AntiCompactionTest.java @@ -27,6 +27,7 @@ import java.util.SortedSet; import java.util.TreeSet; import java.util.UUID; import java.util.concurrent.ExecutionException; +import java.util.stream.Collectors; import com.google.common.collect.ImmutableSet; import com.google.common.collect.Iterables; @@ -137,6 +138,7 @@ public class AntiCompactionTest assertEquals(0, store.getTracker().getCompacting().size()); assertEquals(repairedKeys, 4); assertEquals(nonRepairedKeys, 6); + assertOnDiskState(store, 2); } @Test @@ -164,6 +166,7 @@ public class AntiCompactionTest } assertEquals(sum, cfs.metric.liveDiskSpaceUsed.getCount()); assertEquals(rows, 1000 * (1000 * 5));//See writeFile for how this number is derived + assertOnDiskState(cfs, 2); } private SSTableReader writeFile(ColumnFamilyStore cfs, int count) @@ -259,6 +262,7 @@ public class AntiCompactionTest } assertEquals(repairedKeys, 40); assertEquals(nonRepairedKeys, 60); + assertOnDiskState(store, 10); } @Test @@ -282,6 +286,7 @@ public class AntiCompactionTest assertThat(sstable.isRepaired(), is(true)); assertThat(sstable.selfRef().globalCount(), is(1)); assertThat(store.getTracker().getCompacting().size(), is(0)); + assertOnDiskState(store, 1); } @Test @@ -381,6 +386,7 @@ public class AntiCompactionTest assertThat(sstablesSorted.first().selfRef().globalCount(), is(1)); assertThat(sstablesSorted.last().selfRef().globalCount(), is(1)); assertThat(store.getTracker().getCompacting().size(), is(0)); + assertOnDiskState(store, 2); } @@ -410,6 +416,7 @@ public class AntiCompactionTest assertThat(store.getLiveSSTables().size(), is(10)); assertThat(Iterables.get(store.getLiveSSTables(), 0).isRepaired(), is(false)); + assertOnDiskState(store, 10); } private ColumnFamilyStore prepareColumnFamilyStore() @@ -442,4 +449,25 @@ public class AntiCompactionTest return ImmutableSet.copyOf(cfs.getTracker().getView().sstables(SSTableSet.LIVE, (s) -> !s.isRepaired())); } + static void assertOnDiskState(ColumnFamilyStore cfs, int expectedSSTableCount) + { + LifecycleTransaction.waitForDeletions(); + assertEquals(expectedSSTableCount, cfs.getLiveSSTables().size()); + Set liveGenerations = cfs.getLiveSSTables().stream().map(sstable -> sstable.descriptor.generation).collect(Collectors.toSet()); + int fileCount = 0; + for (File f : cfs.getDirectories().getCFDirectories()) + { + for (File sst : f.listFiles()) + { + if (sst.getName().contains("Data")) + { + Descriptor d = Descriptor.fromFilename(sst.getAbsolutePath()); + assertTrue(liveGenerations.contains(d.generation)); + fileCount++; + } + } + } + assertEquals(expectedSSTableCount, fileCount); + } + }