mirror of https://github.com/apache/cassandra
Anti-compaction temporarily corrupts sstable state
This may result in queries being answered with incorrect results. patch by Blake Eggleston, Marcus Eriksson and Benedict for CASSANDRA-15004 Co-authored-by: Marcus Eriksson <marcuse@apache.org> Co-authored-by: Blake Eggleston <bdeggleston@gmail.com>
This commit is contained in:
Benedict Elliott Smith
parent
aa2a5c6994
commit
44785dd2ee
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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<SSTableReader> 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<Range<Token>> ranges,
|
||||
LifecycleTransaction anticompactionGroup, long repairedAt)
|
||||
|
||||
@VisibleForTesting
|
||||
int antiCompactGroup(ColumnFamilyStore cfs, Collection<Range<Token>> 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<SSTableReader> 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,
|
||||
|
|
|
|||
|
|
@ -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<SSTableReader> readers, boolean original);
|
||||
public SSTableReader current(SSTableReader reader);
|
||||
void obsolete(SSTableReader reader);
|
||||
void obsoleteOriginals();
|
||||
Set<SSTableReader> originals();
|
||||
boolean isObsolete(SSTableReader reader);
|
||||
}
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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<SSTableReader> 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<SSTableReader> 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();
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -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<SSTableReader> 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);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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<UntypedResultSet.Row> it = res.iterator();
|
||||
List<Long> 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<Token> 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<UntypedResultSet.Row> rowIter = result.iterator();
|
||||
Set<Integer> 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);
|
||||
}
|
||||
}
|
||||
|
|
@ -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<Integer> 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);
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue