protected final List<Path> compact(final CompactionRequestImpl request,
    InternalScannerFactory scannerFactory, CellSinkFactory<T> sinkFactory,
    ThroughputController throughputController, User user) throws IOException {
    FileDetails fd = getFileDetails(request.getFiles(), request.isAllFiles(), request.isMajor());

    // Find the smallest read point across all the Scanners.
    long smallestReadPoint = getSmallestReadPoint();

    boolean dropCache;
    if (request.isMajor() || request.isAllFiles()) {
      dropCache = this.dropCacheMajor;
    } else {
      dropCache = this.dropCacheMinor;
    }

    InternalScanner scanner = null;
    boolean finished = false;
    List<StoreFileScanner> scanners =
      createFileScanners(request.getFiles(), smallestReadPoint, dropCache);
    T writer = null;
    CompactionProgress progress = new CompactionProgress(fd.maxKeyCount);
    progressSet.add(progress);
    try {
      /* Include deletes, unless we are doing a major compaction */
      ScanType scanType = scannerFactory.getScanType(request);
      ScanInfo scanInfo = preCompactScannerOpen(request, scanType, user);
      scanner = postCompactScannerOpen(request, scanType,
        scannerFactory.createScanner(scanInfo, scanners, scanType, fd, smallestReadPoint), user);
      boolean cleanSeqId = false;
      if (fd.minSeqIdToKeep > 0 && !store.getColumnFamilyDescriptor().isNewVersionBehavior()) {
        // For mvcc-sensitive family, we never set mvcc to 0.
        smallestReadPoint = Math.min(fd.minSeqIdToKeep, smallestReadPoint);
        cleanSeqId = true;
      }
      writer = sinkFactory.createWriter(scanner, fd, dropCache, request.isMajor(),
        request.getWriterCreationTracker());
      finished = performCompaction(fd, scanner, writer, smallestReadPoint, cleanSeqId,
        throughputController, request, progress);
      if (!finished) {
        throw new InterruptedIOException("Aborting compaction of store " + store + " in region "
          + store.getRegionInfo().getRegionNameAsString() + " because it was interrupted.");
      }
    } finally {
      // createScanner may fail when seeking hfiles encounter Exception, e.g. even only one hfile
      // reader encounters java.io.IOException: Invalid HFile block magic:
      // \x00\x00\x00\x00\x00\x00\x00\x00
      // and then scanner will be null, but scanners for all the hfiles should be closed,
      // or else we will find leak of ESTABLISHED sockets.
      if (scanner == null) {
        for (StoreFileScanner sfs : scanners) {
          sfs.close();
        }
      } else {
        Closeables.close(scanner, true);
      }
      if (!finished) {
        if (writer != null) {
          abortWriter(writer);
        }
      } else {
        store.updateCompactedMetrics(request.isMajor(), progress);
      }
      progressSet.remove(progress);
    }
    assert finished : "We should have exited the method on all error paths";
    assert writer != null : "Writer should be non-null if no error";
    return commitWriter(writer, fd, request);
  }
Calls coprocessor, if any, to create scanners - after normal scanner creation.
@param request  Compaction request.
@param scanType Scan type.
@param scanner  The default scanner created for compaction.
@return Scanner scanner to use (usually the default); null if compaction should not proceed.
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/compactions/Compactor.java