protected boolean performCompaction(FileDetails fd, InternalScanner scanner, CellSink writer,
    long smallestReadPoint, boolean cleanSeqId, ThroughputController throughputController,
    CompactionRequestImpl request, CompactionProgress progress) throws IOException {
    assert writer instanceof ShipperListener;
    long bytesWrittenProgressForLog = 0;
    long bytesWrittenProgressForShippedCall = 0;
    // Since scanner.next() can return 'false' but still be delivering data,
    // we have to use a do/while loop.
    List<ExtendedCell> cells = new ArrayList<>();
    long currentTime = EnvironmentEdgeManager.currentTime();
    long lastMillis = 0;
    if (LOG.isDebugEnabled()) {
      lastMillis = currentTime;
    }
    CloseChecker closeChecker = new CloseChecker(conf, currentTime);
    String compactionName = ThroughputControlUtil.getNameForThrottling(store, "compaction");
    long now = 0;
    boolean hasMore;
    ScannerContext scannerContext = ScannerContext.newBuilder().setBatchLimit(compactionKVMax)
      .setSizeLimit(ScannerContext.LimitScope.BETWEEN_CELLS, Long.MAX_VALUE, Long.MAX_VALUE,
        compactScannerSizeLimit)
      .build();

    throughputController.start(compactionName);
    Shipper shipper = (scanner instanceof Shipper) ? (Shipper) scanner : null;
    long shippedCallSizeLimit =
      (long) request.getFiles().size() * this.store.getColumnFamilyDescriptor().getBlocksize();
    try {
      do {
        // InternalScanner is for CPs so we do not want to leak ExtendedCell to the interface, but
        // all the server side implementation should only add ExtendedCell to the List, otherwise it
        // will cause serious assertions in our code
        hasMore = scanner.next((List) cells, scannerContext);
        currentTime = EnvironmentEdgeManager.currentTime();
        if (LOG.isDebugEnabled()) {
          now = currentTime;
        }
        if (closeChecker.isTimeLimit(store, currentTime)) {
          progress.cancel();
          return false;
        }
        // output to writer:
        Cell lastCleanCell = null;
        long lastCleanCellSeqId = 0;
        for (ExtendedCell c : cells) {
          if (cleanSeqId && c.getSequenceId() <= smallestReadPoint) {
            lastCleanCell = c;
            lastCleanCellSeqId = c.getSequenceId();
            PrivateCellUtil.setSequenceId(c, 0);
          } else {
            lastCleanCell = null;
            lastCleanCellSeqId = 0;
          }
          int len = c.getSerializedSize();
          ++progress.currentCompactedKVs;
          progress.totalCompactedSize += len;
          bytesWrittenProgressForShippedCall += len;
          if (LOG.isDebugEnabled()) {
            bytesWrittenProgressForLog += len;
          }
          throughputController.control(compactionName, len);
          if (closeChecker.isSizeLimit(store, len)) {
            progress.cancel();
            return false;
          }
        }
        writer.appendAll(cells);
        if (shipper != null && bytesWrittenProgressForShippedCall > shippedCallSizeLimit) {
          if (lastCleanCell != null) {
            // HBASE-16931, set back sequence id to avoid affecting scan order unexpectedly.
            // ShipperListener will do a clone of the last cells it refer, so need to set back
            // sequence id before ShipperListener.beforeShipped
            PrivateCellUtil.setSequenceId(lastCleanCell, lastCleanCellSeqId);
          }
          // Clone the cells that are in the writer so that they are freed of references,
          // if they are holding any.
          ((ShipperListener) writer).beforeShipped();
          // The SHARED block references, being read for compaction, will be kept in prevBlocks
          // list(See HFileScannerImpl#prevBlocks). In case of scan flow, after each set of cells
          // being returned to client, we will call shipped() which can clear this list. Here by
          // we are doing the similar thing. In between the compaction (after every N cells
          // written with collective size of 'shippedCallSizeLimit') we will call shipped which
          // may clear prevBlocks list.
          shipper.shipped();
          bytesWrittenProgressForShippedCall = 0;
        }
        if (lastCleanCell != null) {
          // HBASE-16931, set back sequence id to avoid affecting scan order unexpectedly
          PrivateCellUtil.setSequenceId(lastCleanCell, lastCleanCellSeqId);
        }
        // Log the progress of long running compactions every minute if
        // logging at DEBUG level
        if (LOG.isDebugEnabled()) {
          if ((now - lastMillis) >= COMPACTION_PROGRESS_LOG_INTERVAL) {
            String rate = String.format("%.2f",
              (bytesWrittenProgressForLog / 1024.0) / ((now - lastMillis) / 1000.0));
            LOG.debug("Compaction progress: {} {}, rate={} KB/sec, throughputController is {}",
              compactionName, progress, rate, throughputController);
            lastMillis = now;
            bytesWrittenProgressForLog = 0;
          }
        }
        cells.clear();
      } while (hasMore);
    } catch (InterruptedException e) {
      progress.cancel();
      throw new InterruptedIOException(
        "Interrupted while control throughput of compacting " + compactionName);
    } finally {
      // Clone last cell in the final because writer will append last cell when committing. If
      // don't clone here and once the scanner get closed, then the memory of last cell will be
      // released. (HBASE-22582)
      ((ShipperListener) writer).beforeShipped();
      throughputController.finish(compactionName);
    }
    progress.complete();
    return true;
  }
Performs the compaction.
@param fd                FileDetails of cell sink writer
@param scanner           Where to read from.
@param writer            Where to write to.
@param smallestReadPoint Smallest read point.
@param cleanSeqId        When true, remove seqId(used to be mvcc) value which is &lt;=
                         smallestReadPoint
@param request           compaction request.
@param progress          Progress reporter.
@return Whether compaction ended; false if it was interrupted for some reason.
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/compactions/Compactor.java