import org.apache.cassandra.db.compaction.AbstractCompactionStrategy; //导入依赖的package包/类
@SuppressWarnings("unchecked")
private static CompactionParams compactionParamsFromThrift(CfDef cf_def)
{
    Class<? extends AbstractCompactionStrategy> klass =
        CFMetaData.createCompactionStrategy(cf_def.compaction_strategy);
    Map<String, String> options = new HashMap<>(cf_def.compaction_strategy_options);

    int minThreshold = cf_def.min_compaction_threshold;
    int maxThreshold = cf_def.max_compaction_threshold;

    if (CompactionParams.supportsThresholdParams(klass))
    {
        options.putIfAbsent(CompactionParams.Option.MIN_THRESHOLD.toString(), Integer.toString(minThreshold));
        options.putIfAbsent(CompactionParams.Option.MAX_THRESHOLD.toString(), Integer.toString(maxThreshold));
    }

    return CompactionParams.create(klass, options);
}
 

import org.apache.cassandra.db.compaction.AbstractCompactionStrategy; //导入依赖的package包/类
@SuppressWarnings("unchecked")
public static boolean supportsThresholdParams(Class<? extends AbstractCompactionStrategy> klass)
{
    try
    {
        Map<String, String> unrecognizedOptions =
            (Map<String, String>) klass.getMethod("validateOptions", Map.class)
                                       .invoke(null, DEFAULT_THRESHOLDS);

        return unrecognizedOptions.isEmpty();
    }
    catch (Exception e)
    {
        throw new RuntimeException(e);
    }
}
 

import org.apache.cassandra.db.compaction.AbstractCompactionStrategy; //导入依赖的package包/类
private static void checkManifest(AbstractCompactionStrategy strategy, ColumnFamilyStore cfs, Collection<SSTableReader> sstables)
{
    WrappingCompactionStrategy wrappingStrategy = (WrappingCompactionStrategy)strategy;
    int maxSizeInMB = (int)((cfs.getCompactionStrategy().getMaxSSTableBytes()) / (1024L * 1024L));
    if (wrappingStrategy.getWrappedStrategies().size() == 2 && wrappingStrategy.getWrappedStrategies().iterator().next() instanceof LeveledCompactionStrategy)
    {
        System.out.println("Checking leveled manifest");
        Predicate<SSTableReader> repairedPredicate = new Predicate<SSTableReader>()
        {
            @Override
            public boolean apply(SSTableReader sstable)
            {
                return sstable.isRepaired();
            }
        };

        List<SSTableReader> repaired = Lists.newArrayList(Iterables.filter(sstables, repairedPredicate));
        List<SSTableReader> unRepaired = Lists.newArrayList(Iterables.filter(sstables, Predicates.not(repairedPredicate)));

        LeveledManifest repairedManifest = LeveledManifest.create(cfs, maxSizeInMB, repaired);
        for (int i = 1; i < repairedManifest.getLevelCount(); i++)
        {
            repairedManifest.repairOverlappingSSTables(i);
        }
        LeveledManifest unRepairedManifest = LeveledManifest.create(cfs, maxSizeInMB, unRepaired);
        for (int i = 1; i < unRepairedManifest.getLevelCount(); i++)
        {
            unRepairedManifest.repairOverlappingSSTables(i);
        }
    }
}
 

import org.apache.cassandra.db.compaction.AbstractCompactionStrategy; //导入依赖的package包/类
@Test
public void basicTest() throws InterruptedException
{
    Keyspace keyspace = Keyspace.open(KEYSPACE);
    ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF);
    cfs.truncateBlocking();
    for (int j = 0; j < 100; j ++)
    {
        ByteBuffer key = ByteBufferUtil.bytes(String.valueOf(j));
        Mutation rm = new Mutation(KEYSPACE, key);
        rm.add(CF, Util.cellname("0"), ByteBufferUtil.EMPTY_BYTE_BUFFER, j);
        rm.apply();
    }
    cfs.forceBlockingFlush();
    Set<SSTableReader> sstables = new HashSet<>(cfs.getSSTables());
    assertEquals(1, sstables.size());
    SSTableRewriter writer = new SSTableRewriter(cfs, sstables, 1000, false);
    try (AbstractCompactionStrategy.ScannerList scanners = cfs.getCompactionStrategy().getScanners(sstables);)
    {
        ISSTableScanner scanner = scanners.scanners.get(0);
        CompactionController controller = new CompactionController(cfs, sstables, cfs.gcBefore(System.currentTimeMillis()));
        writer.switchWriter(getWriter(cfs, sstables.iterator().next().descriptor.directory));
        while(scanner.hasNext())
        {
            AbstractCompactedRow row = new LazilyCompactedRow(controller, Arrays.asList(scanner.next()));
            writer.append(row);
        }
    }
    Collection<SSTableReader> newsstables = writer.finish();
    cfs.getDataTracker().markCompactedSSTablesReplaced(sstables, newsstables , OperationType.COMPACTION);
    Thread.sleep(100);
    validateCFS(cfs);
    int filecounts = assertFileCounts(sstables.iterator().next().descriptor.directory.list(), 0, 0);
    assertEquals(1, filecounts);

}
 

import org.apache.cassandra.db.compaction.AbstractCompactionStrategy; //导入依赖的package包/类
@Test
public void basicTest2() throws InterruptedException
{
    Keyspace keyspace = Keyspace.open(KEYSPACE);
    ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF);
    cfs.truncateBlocking();

    SSTableReader s = writeFile(cfs, 1000);
    cfs.addSSTable(s);
    Set<SSTableReader> sstables = new HashSet<>(cfs.getSSTables());
    assertEquals(1, sstables.size());
    SSTableRewriter.overrideOpenInterval(10000000);
    SSTableRewriter writer = new SSTableRewriter(cfs, sstables, 1000, false);
    try (AbstractCompactionStrategy.ScannerList scanners = cfs.getCompactionStrategy().getScanners(sstables);)
    {
        ISSTableScanner scanner = scanners.scanners.get(0);
        CompactionController controller = new CompactionController(cfs, sstables, cfs.gcBefore(System.currentTimeMillis()));
        writer.switchWriter(getWriter(cfs, sstables.iterator().next().descriptor.directory));
        while (scanner.hasNext())
        {
            AbstractCompactedRow row = new LazilyCompactedRow(controller, Arrays.asList(scanner.next()));
            writer.append(row);
        }
    }
    Collection<SSTableReader> newsstables = writer.finish();
    cfs.getDataTracker().markCompactedSSTablesReplaced(sstables, newsstables, OperationType.COMPACTION);
    Thread.sleep(100);
    validateCFS(cfs);
    int filecounts = assertFileCounts(sstables.iterator().next().descriptor.directory.list(), 0, 0);
    assertEquals(1, filecounts);
}
 

import org.apache.cassandra.db.compaction.AbstractCompactionStrategy; //导入依赖的package包/类
public static Class<? extends AbstractCompactionStrategy> createCompactionStrategy(String className) throws ConfigurationException
{
    className = className.contains(".") ? className : "org.apache.cassandra.db.compaction." + className;
    Class<AbstractCompactionStrategy> strategyClass = FBUtilities.classForName(className, "compaction strategy");
    if (!AbstractCompactionStrategy.class.isAssignableFrom(strategyClass))
        throw new ConfigurationException(String.format("Specified compaction strategy class (%s) is not derived from AbstractReplicationStrategy", className));

    return strategyClass;
}
 

import org.apache.cassandra.db.compaction.AbstractCompactionStrategy; //导入依赖的package包/类
public AbstractCompactionStrategy createCompactionStrategyInstance(ColumnFamilyStore cfs)
{
    try
    {
        Constructor<? extends AbstractCompactionStrategy> constructor = compactionStrategyClass.getConstructor(new Class[] {
            ColumnFamilyStore.class,
            Map.class // options
        });
        return constructor.newInstance(cfs, compactionStrategyOptions);
    }
    catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException | InstantiationException e)
    {
        throw new RuntimeException(e);
    }
}
 

import org.apache.cassandra.db.compaction.AbstractCompactionStrategy; //导入依赖的package包/类
public static AbstractCompactionStrategy createCompactionStrategyInstance(ColumnFamilyStore cfs,
                                                                          CompactionParams compactionParams)
{
    try
    {
        Constructor<? extends AbstractCompactionStrategy> constructor =
            compactionParams.klass().getConstructor(ColumnFamilyStore.class, Map.class);
        return constructor.newInstance(cfs, compactionParams.options());
    }
    catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException | InstantiationException e)
    {
        throw new RuntimeException(e);
    }
}
 

import org.apache.cassandra.db.compaction.AbstractCompactionStrategy; //导入依赖的package包/类
@SuppressWarnings("unchecked")
private static CompactionParams compactionFromRow(UntypedResultSet.Row row)
{
    Class<? extends AbstractCompactionStrategy> klass =
        CFMetaData.createCompactionStrategy(row.getString("compaction_strategy_class"));
    Map<String, String> options = fromJsonMap(row.getString("compaction_strategy_options"));

    int minThreshold = row.getInt("min_compaction_threshold");
    int maxThreshold = row.getInt("max_compaction_threshold");

    Map<String, String> optionsWithThresholds = new HashMap<>(options);
    optionsWithThresholds.putIfAbsent(CompactionParams.Option.MIN_THRESHOLD.toString(), Integer.toString(minThreshold));
    optionsWithThresholds.putIfAbsent(CompactionParams.Option.MAX_THRESHOLD.toString(), Integer.toString(maxThreshold));

    try
    {
        Map<String, String> unrecognizedOptions =
            (Map<String, String>) klass.getMethod("validateOptions", Map.class).invoke(null, optionsWithThresholds);

        if (unrecognizedOptions.isEmpty())
            options = optionsWithThresholds;
    }
    catch (Exception e)
    {
        throw new RuntimeException(e);
    }

    return CompactionParams.create(klass, options);
}
 

import org.apache.cassandra.db.compaction.AbstractCompactionStrategy; //导入依赖的package包/类
public static CompactionParams create(Class<? extends AbstractCompactionStrategy> klass, Map<String, String> options)
{
    boolean isEnabled = options.containsKey(Option.ENABLED.toString())
                      ? Boolean.parseBoolean(options.get(Option.ENABLED.toString()))
                      : DEFAULT_ENABLED;

    Map<String, String> allOptions = new HashMap<>(options);
    if (supportsThresholdParams(klass))
    {
        allOptions.putIfAbsent(Option.MIN_THRESHOLD.toString(), Integer.toString(DEFAULT_MIN_THRESHOLD));
        allOptions.putIfAbsent(Option.MAX_THRESHOLD.toString(), Integer.toString(DEFAULT_MAX_THRESHOLD));
    }

    return new CompactionParams(klass, allOptions, isEnabled);
}
 

import org.apache.cassandra.db.compaction.AbstractCompactionStrategy; //导入依赖的package包/类
private static Class<? extends AbstractCompactionStrategy> classFromName(String name)
{
    String className = name.contains(".")
                     ? name
                     : "org.apache.cassandra.db.compaction." + name;
    Class<AbstractCompactionStrategy> strategyClass = FBUtilities.classForName(className, "compaction strategy");

    if (!AbstractCompactionStrategy.class.isAssignableFrom(strategyClass))
    {
        throw new ConfigurationException(format("Compaction strategy class %s is not derived from AbstractReplicationStrategy",
                                                className));
    }

    return strategyClass;
}
 

import org.apache.cassandra.db.compaction.AbstractCompactionStrategy; //导入依赖的package包/类
@Test
public void basicTest() throws InterruptedException
{
    Keyspace keyspace = Keyspace.open(KEYSPACE);
    ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF);
    truncate(cfs);

    for (int j = 0; j < 100; j ++)
    {
        new RowUpdateBuilder(cfs.metadata, j, String.valueOf(j))
            .clustering("0")
            .add("val", ByteBufferUtil.EMPTY_BYTE_BUFFER)
            .build()
            .apply();
    }
    cfs.forceBlockingFlush();
    Set<SSTableReader> sstables = new HashSet<>(cfs.getLiveSSTables());
    assertEquals(1, sstables.size());
    assertEquals(sstables.iterator().next().bytesOnDisk(), cfs.metric.liveDiskSpaceUsed.getCount());
    int nowInSec = FBUtilities.nowInSeconds();
    try (AbstractCompactionStrategy.ScannerList scanners = cfs.getCompactionStrategyManager().getScanners(sstables);
         LifecycleTransaction txn = cfs.getTracker().tryModify(sstables, OperationType.UNKNOWN);
         SSTableRewriter writer = new SSTableRewriter(txn, 1000, false);
         CompactionController controller = new CompactionController(cfs, sstables, cfs.gcBefore(nowInSec));
         CompactionIterator ci = new CompactionIterator(OperationType.COMPACTION, scanners.scanners, controller, nowInSec, UUIDGen.getTimeUUID()))
    {
        writer.switchWriter(getWriter(cfs, sstables.iterator().next().descriptor.directory, txn));
        while(ci.hasNext())
        {
            writer.append(ci.next());
        }
        writer.finish();
    }
    LifecycleTransaction.waitForDeletions();
    validateCFS(cfs);
    int filecounts = assertFileCounts(sstables.iterator().next().descriptor.directory.list());
    assertEquals(1, filecounts);
    truncate(cfs);
}
 

import org.apache.cassandra.db.compaction.AbstractCompactionStrategy; //导入依赖的package包/类
@Test
public void basicTest2() throws InterruptedException
{
    Keyspace keyspace = Keyspace.open(KEYSPACE);
    ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF);
    truncate(cfs);

    SSTableReader s = writeFile(cfs, 1000);
    cfs.addSSTable(s);
    Set<SSTableReader> sstables = new HashSet<>(cfs.getLiveSSTables());
    assertEquals(1, sstables.size());

    int nowInSec = FBUtilities.nowInSeconds();
    try (AbstractCompactionStrategy.ScannerList scanners = cfs.getCompactionStrategyManager().getScanners(sstables);
         LifecycleTransaction txn = cfs.getTracker().tryModify(sstables, OperationType.UNKNOWN);
         SSTableRewriter writer = new SSTableRewriter(txn, 1000, false, 10000000, false);
         CompactionController controller = new CompactionController(cfs, sstables, cfs.gcBefore(nowInSec));
         CompactionIterator ci = new CompactionIterator(OperationType.COMPACTION, scanners.scanners, controller, nowInSec, UUIDGen.getTimeUUID()))
    {
        writer.switchWriter(getWriter(cfs, sstables.iterator().next().descriptor.directory, txn));
        while (ci.hasNext())
        {
            writer.append(ci.next());
        }
        writer.finish();
    }
    LifecycleTransaction.waitForDeletions();
    validateCFS(cfs);
    int filecounts = assertFileCounts(sstables.iterator().next().descriptor.directory.list());
    assertEquals(1, filecounts);
}
 

import org.apache.cassandra.db.compaction.AbstractCompactionStrategy; //导入依赖的package包/类
/**
 * emulates anticompaction - writing from one source sstable to two new sstables
 *
 * @throws IOException
 */
@Test
public void testTwoWriters() throws IOException
{
    Keyspace keyspace = Keyspace.open(KEYSPACE);
    ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF);
    truncate(cfs);

    SSTableReader s = writeFile(cfs, 1000);
    cfs.addSSTable(s);
    Set<SSTableReader> sstables = Sets.newHashSet(s);
    assertEquals(1, sstables.size());
    int nowInSec = FBUtilities.nowInSeconds();
    try (AbstractCompactionStrategy.ScannerList scanners = cfs.getCompactionStrategyManager().getScanners(sstables);
         LifecycleTransaction txn = cfs.getTracker().tryModify(sstables, OperationType.UNKNOWN);
         SSTableRewriter writer = new SSTableRewriter(txn, 1000, false, false);
         SSTableRewriter writer2 = new SSTableRewriter(txn, 1000, false, false);
         CompactionController controller = new CompactionController(cfs, sstables, cfs.gcBefore(nowInSec));
         CompactionIterator ci = new CompactionIterator(OperationType.COMPACTION, scanners.scanners, controller, nowInSec, UUIDGen.getTimeUUID())
         )
    {
        writer.switchWriter(getWriter(cfs, sstables.iterator().next().descriptor.directory, txn));
        writer2.switchWriter(getWriter(cfs, sstables.iterator().next().descriptor.directory, txn));
        while (ci.hasNext())
        {
            if (writer.currentWriter().getFilePointer() < 15000000)
                writer.append(ci.next());
            else
                writer2.append(ci.next());
        }
        for (int i = 0; i < 5000; i++)
            assertFalse(Util.getOnlyPartition(Util.cmd(cfs, ByteBufferUtil.bytes(i)).build()).isEmpty());
    }
    truncateCF();
    validateCFS(cfs);
}
 

import org.apache.cassandra.db.compaction.AbstractCompactionStrategy; //导入依赖的package包/类
public Class<? extends AbstractCompactionStrategy> getCompactionStrategy()
{
    return compactionStrategyClass;
}
 

import org.apache.cassandra.db.compaction.AbstractCompactionStrategy; //导入依赖的package包/类
@Test
public void getPositionsTest() throws InterruptedException
{
    Keyspace keyspace = Keyspace.open(KEYSPACE);
    ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF);
    cfs.truncateBlocking();

    SSTableReader s = writeFile(cfs, 1000);
    cfs.addSSTable(s);
    Set<SSTableReader> sstables = new HashSet<>(cfs.getSSTables());
    assertEquals(1, sstables.size());
    SSTableRewriter.overrideOpenInterval(10000000);
    SSTableRewriter writer = new SSTableRewriter(cfs, sstables, 1000, false);
    boolean checked = false;
    try (AbstractCompactionStrategy.ScannerList scanners = cfs.getCompactionStrategy().getScanners(sstables);)
    {
        ISSTableScanner scanner = scanners.scanners.get(0);
        CompactionController controller = new CompactionController(cfs, sstables, cfs.gcBefore(System.currentTimeMillis()));
        writer.switchWriter(getWriter(cfs, sstables.iterator().next().descriptor.directory));
        while (scanner.hasNext())
        {
            AbstractCompactedRow row = new LazilyCompactedRow(controller, Arrays.asList(scanner.next()));
            writer.append(row);
            if (!checked && writer.currentWriter().getFilePointer() > 15000000)
            {
                checked = true;
                for (SSTableReader sstable : cfs.getSSTables())
                {
                    if (sstable.openReason == SSTableReader.OpenReason.EARLY)
                    {
                        SSTableReader c = sstables.iterator().next();
                        Collection<Range<Token>> r = Arrays.asList(new Range<>(cfs.partitioner.getMinimumToken(), cfs.partitioner.getMinimumToken()));
                        List<Pair<Long, Long>> tmplinkPositions = sstable.getPositionsForRanges(r);
                        List<Pair<Long, Long>> compactingPositions = c.getPositionsForRanges(r);
                        assertEquals(1, tmplinkPositions.size());
                        assertEquals(1, compactingPositions.size());
                        assertEquals(0, tmplinkPositions.get(0).left.longValue());
                        // make sure we have no overlap between the early opened file and the compacting one:
                        assertEquals(tmplinkPositions.get(0).right.longValue(), compactingPositions.get(0).left.longValue());
                        assertEquals(c.uncompressedLength(), compactingPositions.get(0).right.longValue());
                    }
                }
            }
        }
    }
    assertTrue(checked);
    Collection<SSTableReader> newsstables = writer.finish();
    cfs.getDataTracker().markCompactedSSTablesReplaced(sstables, newsstables, OperationType.COMPACTION);
    Thread.sleep(100);
    validateCFS(cfs);
    int filecounts = assertFileCounts(sstables.iterator().next().descriptor.directory.list(), 0, 0);
    assertEquals(1, filecounts);
    cfs.truncateBlocking();
    Thread.sleep(1000); // make sure the deletion tasks have run etc
    validateCFS(cfs);
}
 

import org.apache.cassandra.db.compaction.AbstractCompactionStrategy; //导入依赖的package包/类
private CompactionParams(Class<? extends AbstractCompactionStrategy> klass, Map<String, String> options, boolean isEnabled)
{
    this.klass = klass;
    this.options = ImmutableMap.copyOf(options);
    this.isEnabled = isEnabled;
}
 

import org.apache.cassandra.db.compaction.AbstractCompactionStrategy; //导入依赖的package包/类
public Class<? extends AbstractCompactionStrategy> klass()
{
    return klass;
}
 

import org.apache.cassandra.db.compaction.AbstractCompactionStrategy; //导入依赖的package包/类
private SSTableReader replaceSSTable(ColumnFamilyStore cfs, LifecycleTransaction txn, boolean fail)
{
    List<SSTableReader> newsstables = null;
    int nowInSec = FBUtilities.nowInSeconds();
    try (CompactionController controller = new CompactionController(cfs, txn.originals(), cfs.gcBefore(FBUtilities.nowInSeconds())))
    {
        try (SSTableRewriter rewriter = new SSTableRewriter(txn, 1000, false);
             AbstractCompactionStrategy.ScannerList scanners = cfs.getCompactionStrategyManager().getScanners(txn.originals());
             CompactionIterator ci = new CompactionIterator(txn.opType(), scanners.scanners, controller, nowInSec, txn.opId())
        )
        {
            long lastCheckObsoletion = System.nanoTime();
            File directory = txn.originals().iterator().next().descriptor.directory;
            Descriptor desc = Descriptor.fromFilename(cfs.getSSTablePath(directory));
            CFMetaData metadata = Schema.instance.getCFMetaData(desc);
            rewriter.switchWriter(SSTableWriter.create(metadata,
                                                       desc,
                                                       0,
                                                       0,
                                                       0,
                                                       SerializationHeader.make(cfs.metadata, txn.originals()),
                                                       txn));
            while (ci.hasNext())
            {
                rewriter.append(ci.next());

                if (System.nanoTime() - lastCheckObsoletion > TimeUnit.MINUTES.toNanos(1L))
                {
                    controller.maybeRefreshOverlaps();
                    lastCheckObsoletion = System.nanoTime();
                }
            }

            if (!fail)
                newsstables = rewriter.finish();
            else
                rewriter.abort();
        }
    }

    assertTrue(fail || newsstables != null);

    if (newsstables != null)
    {
        Assert.assertEquals(1, newsstables.size());
        return newsstables.iterator().next();
    }

    return null;
}
 

import org.apache.cassandra.db.compaction.AbstractCompactionStrategy; //导入依赖的package包/类
@Test
public void getPositionsTest() throws InterruptedException
{
    Keyspace keyspace = Keyspace.open(KEYSPACE);
    ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF);
    truncate(cfs);

    SSTableReader s = writeFile(cfs, 1000);
    cfs.addSSTable(s);
    Set<SSTableReader> sstables = new HashSet<>(cfs.getLiveSSTables());
    assertEquals(1, sstables.size());

    int nowInSec = FBUtilities.nowInSeconds();
    boolean checked = false;
    try (AbstractCompactionStrategy.ScannerList scanners = cfs.getCompactionStrategyManager().getScanners(sstables);
         LifecycleTransaction txn = cfs.getTracker().tryModify(sstables, OperationType.UNKNOWN);
         SSTableRewriter writer = new SSTableRewriter(txn, 1000, false, 10000000, false);
         CompactionController controller = new CompactionController(cfs, sstables, cfs.gcBefore(nowInSec));
         CompactionIterator ci = new CompactionIterator(OperationType.COMPACTION, scanners.scanners, controller, nowInSec, UUIDGen.getTimeUUID()))
    {
        writer.switchWriter(getWriter(cfs, sstables.iterator().next().descriptor.directory, txn));
        while (ci.hasNext())
        {
            UnfilteredRowIterator row = ci.next();
            writer.append(row);
            if (!checked && writer.currentWriter().getFilePointer() > 1500000)
            {
                checked = true;
                for (SSTableReader sstable : cfs.getLiveSSTables())
                {
                    if (sstable.openReason == SSTableReader.OpenReason.EARLY)
                    {
                        SSTableReader c = txn.current(sstables.iterator().next());
                        Collection<Range<Token>> r = Arrays.asList(new Range<>(cfs.getPartitioner().getMinimumToken(), cfs.getPartitioner().getMinimumToken()));
                        List<Pair<Long, Long>> tmplinkPositions = sstable.getPositionsForRanges(r);
                        List<Pair<Long, Long>> compactingPositions = c.getPositionsForRanges(r);
                        assertEquals(1, tmplinkPositions.size());
                        assertEquals(1, compactingPositions.size());
                        assertEquals(0, tmplinkPositions.get(0).left.longValue());
                        // make sure we have no overlap between the early opened file and the compacting one:
                        assertEquals(tmplinkPositions.get(0).right.longValue(), compactingPositions.get(0).left.longValue());
                        assertEquals(c.uncompressedLength(), compactingPositions.get(0).right.longValue());
                    }
                }
            }
        }
        assertTrue(checked);
        writer.finish();
    }
    LifecycleTransaction.waitForDeletions();
    validateCFS(cfs);
    int filecounts = assertFileCounts(sstables.iterator().next().descriptor.directory.list());
    assertEquals(1, filecounts);
    truncate(cfs);
}