import org.apache.cassandra.metrics.CompactionMetrics; //导入依赖的package包/类
@SuppressWarnings("resource") // We make sure to close mergedIterator in close() and CompactionIterator is itself an AutoCloseable
public CompactionIterator(OperationType type, List<ISSTableScanner> scanners, CompactionController controller, int nowInSec, UUID compactionId, CompactionMetrics metrics)
{
    this.controller = controller;
    this.type = type;
    this.scanners = scanners;
    this.nowInSec = nowInSec;
    this.compactionId = compactionId;
    this.bytesRead = 0;

    long bytes = 0;
    for (ISSTableScanner scanner : scanners)
        bytes += scanner.getLengthInBytes();
    this.totalBytes = bytes;
    this.mergeCounters = new long[scanners.size()];
    this.metrics = metrics;

    if (metrics != null)
        metrics.beginCompaction(this);

    UnfilteredPartitionIterator merged = scanners.isEmpty()
                                         ? EmptyIterators.unfilteredPartition(controller.cfs.metadata, false)
                                         : UnfilteredPartitionIterators.merge(scanners, nowInSec, listener());
    boolean isForThrift = merged.isForThrift(); // to stop capture of iterator in Purger, which is confusing for debug
    this.compacted = Transformation.apply(merged, new Purger(isForThrift, controller, nowInSec));
}
 

import org.apache.cassandra.metrics.CompactionMetrics; //导入依赖的package包/类
@After
public void afterTest()
{
    for (CompactionInfo.Holder holder: CompactionMetrics.getCompactions())
    {
        holder.stop();
    }

    String ksname = KEYSPACE1;
    String cfname = CF_STANDARDLOWiINTERVAL; // index interval of 8, no key caching
    Keyspace keyspace = Keyspace.open(ksname);
    ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(cfname);
    cfs.metadata.minIndexInterval(originalMinIndexInterval);
    cfs.metadata.maxIndexInterval(originalMaxIndexInterval);
    IndexSummaryManager.instance.setMemoryPoolCapacityInMB(originalCapacity);
}
 

import org.apache.cassandra.metrics.CompactionMetrics; //导入依赖的package包/类
public List<Map<String, String>> getCompactions()
{
    List<Holder> compactionHolders = CompactionMetrics.getCompactions();
    List<Map<String, String>> out = new ArrayList<Map<String, String>>(compactionHolders.size());
    for (CompactionInfo.Holder ci : compactionHolders)
        out.add(ci.getCompactionInfo().asMap());
    return out;
}
 

import org.apache.cassandra.metrics.CompactionMetrics; //导入依赖的package包/类
public List<String> getCompactionSummary()
{
    List<Holder> compactionHolders = CompactionMetrics.getCompactions();
    List<String> out = new ArrayList<String>(compactionHolders.size());
    for (CompactionInfo.Holder ci : compactionHolders)
        out.add(ci.getCompactionInfo().toString());
    return out;
}
 

import org.apache.cassandra.metrics.CompactionMetrics; //导入依赖的package包/类
public void stopCompaction(String type)
{
    OperationType operation = OperationType.valueOf(type);
    for (Holder holder : CompactionMetrics.getCompactions())
    {
        if (holder.getCompactionInfo().getTaskType() == operation)
            holder.stop();
    }
}
 

import org.apache.cassandra.metrics.CompactionMetrics; //导入依赖的package包/类
/**
 * Shutdowns both compaction and validation executors, cancels running compaction / validation,
 * and waits for tasks to complete if tasks were not cancelable.
 */
public void forceShutdown()
{
    // shutdown executors to prevent further submission
    executor.shutdown();
    validationExecutor.shutdown();

    // interrupt compactions and validations
    for (Holder compactionHolder : CompactionMetrics.getCompactions())
    {
        compactionHolder.stop();
    }

    // wait for tasks to terminate
    // compaction tasks are interrupted above, so it shuold be fairy quick
    // until not interrupted tasks to complete.
    for (ExecutorService exec : Arrays.asList(executor, validationExecutor))
    {
        try
        {
            exec.awaitTermination(1, TimeUnit.MINUTES);
        }
        catch (InterruptedException e)
        {
            logger.error("Interrupted while waiting for tasks to be terminated", e);
        }
    }
}
 

import org.apache.cassandra.metrics.CompactionMetrics; //导入依赖的package包/类
public void stopCompactionById(String compactionId)
{
    for (Holder holder : CompactionMetrics.getCompactions())
    {
        UUID holderId = holder.getCompactionInfo().compactionId();
        if (holderId != null && holderId.equals(UUID.fromString(compactionId)))
            holder.stop();
    }
}
 

import org.apache.cassandra.metrics.CompactionMetrics; //导入依赖的package包/类
public CompactionManager(APIClient client) {
    super(MBEAN_OBJECT_NAME, client, new CompactionMetrics());
}
 

import org.apache.cassandra.metrics.CompactionMetrics; //导入依赖的package包/类
public int getActiveCompactions()
{
    return CompactionMetrics.getCompactions().size();
}
 

import org.apache.cassandra.metrics.CompactionMetrics; //导入依赖的package包/类
public ValidationCompactionIterator(List<ISSTableScanner> scanners, ValidationCompactionController controller, int nowInSec, CompactionMetrics metrics)
{
    super(OperationType.VALIDATION, scanners, controller, nowInSec, UUIDGen.getTimeUUID(), metrics);
}
 

import org.apache.cassandra.metrics.CompactionMetrics; //导入依赖的package包/类
@Test
public void testCancelIndex() throws Exception
{
    String ksname = KEYSPACE1;
    String cfname = CF_STANDARDLOWiINTERVAL; // index interval of 8, no key caching
    Keyspace keyspace = Keyspace.open(ksname);
    final ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(cfname);
    final int numSSTables = 4;
    int numRows = 256;
    createSSTables(ksname, cfname, numSSTables, numRows);

    final List<SSTableReader> sstables = new ArrayList<>(cfs.getLiveSSTables());
    for (SSTableReader sstable : sstables)
        sstable.overrideReadMeter(new RestorableMeter(100.0, 100.0));

    final long singleSummaryOffHeapSpace = sstables.get(0).getIndexSummaryOffHeapSize();

    // everything should get cut in half
    final AtomicReference<CompactionInterruptedException> exception = new AtomicReference<>();

    Thread t = new Thread(new Runnable()
    {
        public void run()
        {
            try
            {
                // Don't leave enough space for even the minimal index summaries
                try (LifecycleTransaction txn = cfs.getTracker().tryModify(sstables, OperationType.UNKNOWN))
                {
                    redistributeSummaries(Collections.EMPTY_LIST, of(cfs.metadata.cfId, txn), singleSummaryOffHeapSpace);
                }
            }
            catch (CompactionInterruptedException ex)
            {
                exception.set(ex);
            }
            catch (IOException ignored)
            {
            }
        }
    });
    t.start();
    while (CompactionManager.instance.getActiveCompactions() == 0 && t.isAlive())
        Thread.sleep(1);
    CompactionManager.instance.stopCompaction("INDEX_SUMMARY");
    t.join();

    assertNotNull("Expected compaction interrupted exception", exception.get());
    assertTrue("Expected no active compactions", CompactionMetrics.getCompactions().isEmpty());

    Set<SSTableReader> beforeRedistributionSSTables = new HashSet<>(sstables);
    Set<SSTableReader> afterCancelSSTables = new HashSet<>(cfs.getLiveSSTables());
    Set<SSTableReader> disjoint = Sets.symmetricDifference(beforeRedistributionSSTables, afterCancelSSTables);
    assertTrue(String.format("Mismatched files before and after cancelling redistribution: %s",
                             Joiner.on(",").join(disjoint)),
               disjoint.isEmpty());

    validateData(cfs, numRows);
}