Java 类org.apache.hadoop.hbase.regionserver.compactions.CompactionRequest 实例源码

private List<CompactionRequest> requestCompactionInternal(final Region r, final String why,
    int p, List<Pair<CompactionRequest, Store>> requests, boolean selectNow, User user)
        throws IOException {
  // not a special compaction request, so make our own list
  List<CompactionRequest> ret = null;
  if (requests == null) {
    ret = selectNow ? new ArrayList<CompactionRequest>(r.getStores().size()) : null;
    for (Store s : r.getStores()) {
      CompactionRequest cr = requestCompactionInternal(r, s, why, p, null, selectNow, user);
      if (selectNow) ret.add(cr);
    }
  } else {
    Preconditions.checkArgument(selectNow); // only system requests have selectNow == false
    ret = new ArrayList<CompactionRequest>(requests.size());
    for (Pair<CompactionRequest, Store> pair : requests) {
      ret.add(requestCompaction(r, pair.getSecond(), why, p, pair.getFirst(), user));
    }
  }
  return ret;
}

private CompactionContext selectCompaction(final Region r, final Store s,
    int priority, CompactionRequest request, User user) throws IOException {
  CompactionContext compaction = s.requestCompaction(priority, request, user);
  if (compaction == null) {
    if(LOG.isDebugEnabled() && r.getRegionInfo() != null) {
      LOG.debug("Not compacting " + r.getRegionInfo().getRegionNameAsString() +
          " because compaction request was cancelled");
    }
    return null;
  }
  assert compaction.hasSelection();
  if (priority != Store.NO_PRIORITY) {
    compaction.getRequest().setPriority(priority);
  }
  return compaction;
}

public void testCompactionEmptyHFile() throws IOException {
  // Set TTL
  ScanInfo oldScanInfo = store.getScanInfo();
  ScanInfo newScanInfo = new ScanInfo(oldScanInfo.getConfiguration(), oldScanInfo.getFamily(),
      oldScanInfo.getMinVersions(), oldScanInfo.getMaxVersions(), 600,
      oldScanInfo.getKeepDeletedCells(), oldScanInfo.getTimeToPurgeDeletes(),
      oldScanInfo.getComparator());
  store.setScanInfo(newScanInfo);
  // Do not compact empty store file
  List<StoreFile> candidates = sfCreate(0);
  for (StoreFile file : candidates) {
    if (file instanceof MockStoreFile) {
      MockStoreFile mockFile = (MockStoreFile) file;
      mockFile.setTimeRangeTracker(new TimeRangeTracker(-1, -1));
      mockFile.setEntries(0);
    }
  }
  // Test Default compactions
  CompactionRequest result = ((RatioBasedCompactionPolicy) store.storeEngine
      .getCompactionPolicy()).selectCompaction(candidates,
      new ArrayList<StoreFile>(), false, false, false);
  assertTrue(result.getFiles().size() == 0);
  store.setScanInfo(oldScanInfo);
}

/**
 * Test for HBASE-5920 - Test user requested major compactions always occurring
 */
@Test
public void testNonUserMajorCompactionRequest() throws Exception {
  Store store = r.getStore(COLUMN_FAMILY);
  createStoreFile(r);
  for (int i = 0; i < MAX_FILES_TO_COMPACT + 1; i++) {
    createStoreFile(r);
  }
  store.triggerMajorCompaction();

  CompactionRequest request = store.requestCompaction(Store.NO_PRIORITY, null).getRequest();
  assertNotNull("Expected to receive a compaction request", request);
  assertEquals(
    "System-requested major compaction should not occur if there are too many store files",
    false,
    request.isMajor());
}

/**
 * Test for HBASE-5920
 */
@Test
public void testUserMajorCompactionRequest() throws IOException{
  Store store = r.getStore(COLUMN_FAMILY);
  createStoreFile(r);
  for (int i = 0; i < MAX_FILES_TO_COMPACT + 1; i++) {
    createStoreFile(r);
  }
  store.triggerMajorCompaction();
  CompactionRequest request = store.requestCompaction(Store.PRIORITY_USER, null).getRequest();
  assertNotNull("Expected to receive a compaction request", request);
  assertEquals(
    "User-requested major compaction should always occur, even if there are too many store files",
    true, 
    request.isMajor());
}

@Override
public InternalScanner preCompactScannerOpen(ObserverContext<RegionCoprocessorEnvironment> c, Store store,
    List<? extends KeyValueScanner> scanners, ScanType scanType, long earliestPutTs, InternalScanner s,
    CompactionRequest request)
    throws IOException {
  // Get the latest tx snapshot state for the compaction
  TransactionVisibilityState snapshot = cache.getLatestState();

  // Record tx state before the compaction
  if (compactionState != null) {
    compactionState.record(request, snapshot);
  }

  // Also make sure to use the same snapshot for the compaction
  return createStoreScanner(c.getEnvironment(), "compaction", snapshot, store, scanners, scanType, earliestPutTs);
}

@Override
public InternalScanner preCompactScannerOpen(ObserverContext<RegionCoprocessorEnvironment> c, Store store,
    List<? extends KeyValueScanner> scanners, ScanType scanType, long earliestPutTs, InternalScanner s,
    CompactionRequest request)
    throws IOException {
  // Get the latest tx snapshot state for the compaction
  TransactionVisibilityState snapshot = cache.getLatestState();

  // Record tx state before the compaction
  if (compactionState != null) {
    compactionState.record(request, snapshot);
  }

  // Also make sure to use the same snapshot for the compaction
  return createStoreScanner(c.getEnvironment(), "compaction", snapshot, store, scanners, scanType, earliestPutTs);
}

@Override
public InternalScanner preCompactScannerOpen(ObserverContext<RegionCoprocessorEnvironment> c, Store store,
    List<? extends KeyValueScanner> scanners, ScanType scanType, long earliestPutTs, InternalScanner s,
    CompactionRequest request)
    throws IOException {
  // Get the latest tx snapshot state for the compaction
  TransactionVisibilityState snapshot = cache.getLatestState();

  // Record tx state before the compaction
  if (compactionState != null) {
    compactionState.record(request, snapshot);
  }

  // Also make sure to use the same snapshot for the compaction
  return createStoreScanner(c.getEnvironment(), "compaction", snapshot, store, scanners, scanType, earliestPutTs);
}

@Override
public InternalScanner preCompactScannerOpen(ObserverContext<RegionCoprocessorEnvironment> c, Store store,
    List<? extends KeyValueScanner> scanners, ScanType scanType, long earliestPutTs, InternalScanner s,
    CompactionRequest request)
    throws IOException {
  // Get the latest tx snapshot state for the compaction
  TransactionVisibilityState snapshot = cache.getLatestState();

  // Record tx state before the compaction
  if (compactionState != null) {
    compactionState.record(request, snapshot);
  }

  // Also make sure to use the same snapshot for the compaction
  return createStoreScanner(c.getEnvironment(), "compaction", snapshot, store, scanners, scanType, earliestPutTs);
}

@Override
public InternalScanner preCompactScannerOpen(ObserverContext<RegionCoprocessorEnvironment> c, Store store,
    List<? extends KeyValueScanner> scanners, ScanType scanType, long earliestPutTs, InternalScanner s,
    CompactionRequest request)
    throws IOException {
  // Get the latest tx snapshot state for the compaction
  TransactionVisibilityState snapshot = cache.getLatestState();

  // Record tx state before the compaction
  if (compactionState != null) {
    compactionState.record(request, snapshot);
  }

  // Also make sure to use the same snapshot for the compaction
  return createStoreScanner(c.getEnvironment(), "compaction", snapshot, store, scanners, scanType, earliestPutTs);
}

@Override
public InternalScanner preCompactScannerOpen(ObserverContext<RegionCoprocessorEnvironment> c, Store store,
    List<? extends KeyValueScanner> scanners, ScanType scanType, long earliestPutTs, InternalScanner s,
    CompactionRequest request)
    throws IOException {
  // Get the latest tx snapshot state for the compaction
  TransactionVisibilityState snapshot = cache.getLatestState();

  // Record tx state before the compaction
  if (compactionState != null) {
    compactionState.record(request, snapshot);
  }

  // Also make sure to use the same snapshot for the compaction
  return createStoreScanner(c.getEnvironment(), "compaction", snapshot, store, scanners, scanType, earliestPutTs);
}

@Override
public InternalScanner preCompactScannerOpen(ObserverContext<RegionCoprocessorEnvironment> c, Store store,
    List<? extends KeyValueScanner> scanners, ScanType scanType, long earliestPutTs, InternalScanner s,
    CompactionRequest request)
    throws IOException {
  // Get the latest tx snapshot state for the compaction
  TransactionVisibilityState snapshot = cache.getLatestState();

  // Record tx state before the compaction
  if (compactionState != null) {
    compactionState.record(request, snapshot);
  }

  // Also make sure to use the same snapshot for the compaction
  return createStoreScanner(c.getEnvironment(), "compaction", snapshot, store, scanners, scanType, earliestPutTs);
}

/**
 * Execute the actual compaction job.
 * If the compact once flag is not specified, execute the compaction until
 * no more compactions are needed. Uses the Configuration settings provided.
 */
private void compactStoreFiles(final HRegion region, final Path familyDir,
    final boolean compactOnce, final boolean major) throws IOException {
  LOG.info("Compact table=" + region.getTableDesc().getNameAsString() +
    " region=" + region.getRegionNameAsString() +
    " family=" + familyDir.getName());
  Store store = getStore(region, familyDir);
  if (major) {
    store.triggerMajorCompaction();
  }
  do {
    CompactionRequest cr = store.requestCompaction(Store.PRIORITY_USER, null);
    StoreFile storeFile = store.compact(cr);
    if (storeFile != null) {
      if (keepCompactedFiles && deleteCompacted) {
        fs.delete(storeFile.getPath(), false);
      }
    }
  } while (store.needsCompaction() && !compactOnce);
}

@Override
public synchronized List<CompactionRequest> requestCompaction(final HRegion r, final String why,
    int pri, final List<CompactionRequest> requests) throws IOException {
  List<CompactionRequest> ret;
  // not a special compaction request, so make out own list
  if (requests == null) {
    ret = new ArrayList<CompactionRequest>(r.getStores().size());
    for (Store s : r.getStores().values()) {
      ret.add(requestCompaction(r, s, why, pri, null));
    }
  } else {
    ret = new ArrayList<CompactionRequest>(requests.size());
    for (CompactionRequest request : requests) {
      ret.add(requestCompaction(r, request.getStore(), why, pri, request));
    }
  }
  return ret;
}

CompactJob mWinterCompactLCCIndexLocal(final CompactionRequest request, final Path writtenPath)
    throws IOException {
  // check reference file, not supported yet!
  boolean needToRebuild = false;
  for (StoreFile sf : request.getFiles()) {
    if (sf.getPath().getName().indexOf(".") != -1 || sf.isReference()) {
      needToRebuild = true;
      break;
    }
  }
  CompactJob job = null;
  if (needToRebuild) {
    job = new RebuildCompactJob(request.getStore(), request, writtenPath);
  } else {
    job = new NormalCompactJob(request.getStore(), request, writtenPath);
  }
  CompactJobQueue.getInstance().addJob(job);
  return job;
}

/**
 * Test for HBASE-5920
 */
@Test
public void testUserMajorCompactionRequest() throws IOException{
  Store store = r.getStore(COLUMN_FAMILY);
  createStoreFile(r);
  for (int i = 0; i < MAX_FILES_TO_COMPACT + 1; i++) {
    createStoreFile(r);
  }
  store.triggerMajorCompaction();
  CompactionRequest request = store.requestCompaction(Store.PRIORITY_USER, null).getRequest();
  assertNotNull("Expected to receive a compaction request", request);
  assertEquals(
    "User-requested major compaction should always occur, even if there are too many store files",
    true, 
    request.isMajor());
}

/**
 * See
 * {@link RegionObserver#preCompactScannerOpen(ObserverContext, Store, List, ScanType, long, InternalScanner, CompactionRequest)}
 */
public InternalScanner preCompactScannerOpen(Store store, List<StoreFileScanner> scanners,
    ScanType scanType, long earliestPutTs, CompactionRequest request) throws IOException {
  ObserverContext<RegionCoprocessorEnvironment> ctx = null;
  InternalScanner s = null;
  for (RegionEnvironment env: coprocessors) {
    if (env.getInstance() instanceof RegionObserver) {
      ctx = ObserverContext.createAndPrepare(env, ctx);
      try {
        s = ((RegionObserver) env.getInstance()).preCompactScannerOpen(ctx, store, scanners,
          scanType, earliestPutTs, s, request);
      } catch (Throwable e) {
        handleCoprocessorThrowable(env,e);
      }
      if (ctx.shouldComplete()) {
        break;
      }
    }
  }
  return s;
}

/**
 * Test for HBASE-5920 - Test user requested major compactions always occurring
 */
@Test
public void testNonUserMajorCompactionRequest() throws Exception {
  Store store = r.getStore(COLUMN_FAMILY);
  createStoreFile(r);
  for (int i = 0; i < MAX_FILES_TO_COMPACT + 1; i++) {
    createStoreFile(r);
  }
  store.triggerMajorCompaction();

  CompactionRequest request = store.requestCompaction(Store.NO_PRIORITY, null).getRequest();
  assertNotNull("Expected to receive a compaction request", request);
  assertEquals(
    "System-requested major compaction should not occur if there are too many store files",
    false,
    request.isMajor());
}

/**
 * Called after the {@link StoreFile}s to be compacted have been selected from the available
 * candidates.
 * @param store The store where compaction is being requested
 * @param selected The store files selected to compact
 * @param request custom compaction
 */
public void postCompactSelection(Store store, ImmutableList<StoreFile> selected,
    CompactionRequest request) {
  ObserverContext<RegionCoprocessorEnvironment> ctx = null;
  for (RegionEnvironment env: coprocessors) {
    if (env.getInstance() instanceof RegionObserver) {
      ctx = ObserverContext.createAndPrepare(env, ctx);
      try {
        ((RegionObserver) env.getInstance()).postCompactSelection(ctx, store, selected, request);
      } catch (Throwable e) {
        handleCoprocessorThrowableNoRethrow(env,e);
      }
      if (ctx.shouldComplete()) {
        break;
      }
    }
  }
}

/**
 * Called prior to rewriting the store files selected for compaction
 * @param store the store being compacted
 * @param scanner the scanner used to read store data during compaction
 * @param request the compaction that will be executed
 * @throws IOException
 */
public InternalScanner preCompact(Store store, InternalScanner scanner, 
    CompactionRequest request) throws IOException {
  ObserverContext<RegionCoprocessorEnvironment> ctx = null;
  boolean bypass = false;
  for (RegionEnvironment env : coprocessors) {
    if (env.getInstance() instanceof RegionObserver) {
      ctx = ObserverContext.createAndPrepare(env, ctx);
      try {
        scanner = ((RegionObserver) env.getInstance()).preCompact(ctx, store, scanner, request);
      } catch (Throwable e) {
        handleCoprocessorThrowable(env, e);
      }
      bypass |= ctx.shouldBypass();
      if (ctx.shouldComplete()) {
        break;
      }
    }
  }
  return bypass ? null : scanner;
}

/**
 * Called after the store compaction has completed.
 * @param store the store being compacted
 * @param resultFile the new store file written during compaction
 * @param request the compaction that is being executed
 * @throws IOException
 */
public void postCompact(Store store, StoreFile resultFile, CompactionRequest request)
    throws IOException {
  ObserverContext<RegionCoprocessorEnvironment> ctx = null;
  for (RegionEnvironment env: coprocessors) {
    if (env.getInstance() instanceof RegionObserver) {
      ctx = ObserverContext.createAndPrepare(env, ctx);
      try {
        ((RegionObserver) env.getInstance()).postCompact(ctx, store, resultFile, request);
      } catch (Throwable e) {
        handleCoprocessorThrowable(env, e);
      }
      if (ctx.shouldComplete()) {
        break;
      }
    }
  }
}

private List<CompactionRequest> requestCompactionInternal(final HRegion r, final String why,
    int p, List<Pair<CompactionRequest, Store>> requests, boolean selectNow) throws IOException {
  // not a special compaction request, so make our own list
  List<CompactionRequest> ret = null;
  if (requests == null) {
    ret = selectNow ? new ArrayList<CompactionRequest>(r.getStores().size()) : null;
    for (Store s : r.getStores().values()) {
      CompactionRequest cr = requestCompactionInternal(r, s, why, p, null, selectNow);
      if (selectNow) ret.add(cr);
    }
  } else {
    Preconditions.checkArgument(selectNow); // only system requests have selectNow == false
    ret = new ArrayList<CompactionRequest>(requests.size());
    for (Pair<CompactionRequest, Store> pair : requests) {
      ret.add(requestCompaction(r, pair.getSecond(), why, p, pair.getFirst()));
    }
  }
  return ret;
}

public void testCompactionEmptyHFile() throws IOException {
  // Set TTL
  ScanInfo oldScanInfo = store.getScanInfo();
  ScanInfo newScanInfo = new ScanInfo(oldScanInfo.getFamily(),
      oldScanInfo.getMinVersions(), oldScanInfo.getMaxVersions(), 600,
      oldScanInfo.getKeepDeletedCells(), oldScanInfo.getTimeToPurgeDeletes(),
      oldScanInfo.getComparator());
  store.setScanInfo(newScanInfo);
  // Do not compact empty store file
  List<StoreFile> candidates = sfCreate(0);
  for (StoreFile file : candidates) {
    if (file instanceof MockStoreFile) {
      MockStoreFile mockFile = (MockStoreFile) file;
      mockFile.setTimeRangeTracker(new TimeRangeTracker(-1, -1));
      mockFile.setEntries(0);
    }
  }
  // Test Default compactions
  CompactionRequest result = ((RatioBasedCompactionPolicy) store.storeEngine
      .getCompactionPolicy()).selectCompaction(candidates,
      new ArrayList<StoreFile>(), false, false, false);
  assertTrue(result.getFiles().size() == 0);
  store.setScanInfo(oldScanInfo);
}

@Override
public InternalScanner preCompactScannerOpen(ObserverContext<RegionCoprocessorEnvironment> c,
    Store store, List<? extends KeyValueScanner> scanners, ScanType scanType, long earliestPutTs,
    InternalScanner s, CompactionRequest request) throws IOException {
  HRegionServer rs = (HRegionServer) c.getEnvironment().getRegionServerServices();
  if (!IndexUtils.isIndexTable(store.getTableName())) {
    // Not an index table
    return null;
  }
  long smallestReadPoint = c.getEnvironment().getRegion().getSmallestReadPoint();
  String actualTableName = IndexUtils.getActualTableName(store.getTableName().getNameAsString());
  TTLStoreScanner ttlStoreScanner =
      new TTLStoreScanner(store, smallestReadPoint, earliestPutTs, scanType, scanners,
          new TTLExpiryChecker(), actualTableName, rs);
  return ttlStoreScanner;
}

private List<CompactionRequest> requestCompactionInternal(final HRegion r, final String why,
    int p, List<Pair<CompactionRequest, Store>> requests, boolean selectNow) throws IOException {
  // not a special compaction request, so make our own list
  List<CompactionRequest> ret = null;
  if (requests == null) {
    ret = selectNow ? new ArrayList<CompactionRequest>(r.getStores().size()) : null;
    for (Store s : r.getStores().values()) {
      CompactionRequest cr = requestCompactionInternal(r, s, why, p, null, selectNow);
      if (selectNow) ret.add(cr);
    }
  } else {
    Preconditions.checkArgument(selectNow); // only system requests have selectNow == false
    ret = new ArrayList<CompactionRequest>(requests.size());
    for (Pair<CompactionRequest, Store> pair : requests) {
      ret.add(requestCompaction(r, pair.getSecond(), why, p, pair.getFirst()));
    }
  }
  return ret;
}

private CompactionContext selectCompaction(final HRegion r, final Store s,
    int priority, CompactionRequest request) throws IOException {
  CompactionContext compaction = s.requestCompaction(priority, request);
  if (compaction == null) {
    if(LOG.isDebugEnabled()) {
      LOG.debug("Not compacting " + r.getRegionNameAsString() +
          " because compaction request was cancelled");
    }
    return null;
  }
  assert compaction.hasSelection();
  if (priority != Store.NO_PRIORITY) {
    compaction.getRequest().setPriority(priority);
  }
  return compaction;
}

/**
 * Called after the {@link StoreFile}s to be compacted have been selected from the available
 * candidates.
 * @param store The store where compaction is being requested
 * @param selected The store files selected to compact
 * @param request custom compaction
 */
public void postCompactSelection(final Store store, final ImmutableList<StoreFile> selected,
    final CompactionRequest request) {
  ObserverContext<RegionCoprocessorEnvironment> ctx = null;
  for (RegionEnvironment env: coprocessors) {
    if (env.getInstance() instanceof RegionObserver) {
      ctx = ObserverContext.createAndPrepare(env, ctx);
      Thread currentThread = Thread.currentThread();
      ClassLoader cl = currentThread.getContextClassLoader();
      try {
        currentThread.setContextClassLoader(env.getClassLoader());
        ((RegionObserver) env.getInstance()).postCompactSelection(ctx, store, selected,
          request);
      } catch (Throwable e) {
        handleCoprocessorThrowableNoRethrow(env,e);
      } finally {
        currentThread.setContextClassLoader(cl);
      }
      if (ctx.shouldComplete()) {
        break;
      }
    }
  }
}

public void testCompactionEmptyHFile() throws IOException {
  // Set TTL
  ScanInfo oldScanInfo = store.getScanInfo();
  ScanInfo newScanInfo = new ScanInfo(oldScanInfo.getFamily(),
      oldScanInfo.getMinVersions(), oldScanInfo.getMaxVersions(), 600,
      oldScanInfo.getKeepDeletedCells(), oldScanInfo.getTimeToPurgeDeletes(),
      oldScanInfo.getComparator());
  store.setScanInfo(newScanInfo);
  // Do not compact empty store file
  List<StoreFile> candidates = sfCreate(0);
  for (StoreFile file : candidates) {
    if (file instanceof MockStoreFile) {
      MockStoreFile mockFile = (MockStoreFile) file;
      mockFile.setTimeRangeTracker(new TimeRangeTracker(-1, -1));
      mockFile.setEntries(0);
    }
  }
  // Test Default compactions
  CompactionRequest result = ((RatioBasedCompactionPolicy) store.storeEngine
      .getCompactionPolicy()).selectCompaction(candidates,
      new ArrayList<StoreFile>(), false, false, false);
  assertTrue(result.getFiles().size() == 0);
  store.setScanInfo(oldScanInfo);
}

/**
 * Test for HBASE-5920 - Test user requested major compactions always occurring
 */
@Test
public void testNonUserMajorCompactionRequest() throws Exception {
  Store store = r.getStore(COLUMN_FAMILY);
  createStoreFile(r);
  for (int i = 0; i < MAX_FILES_TO_COMPACT + 1; i++) {
    createStoreFile(r);
  }
  store.triggerMajorCompaction();

  CompactionRequest request = store.requestCompaction(Store.NO_PRIORITY, null).getRequest();
  assertNotNull("Expected to receive a compaction request", request);
  assertEquals(
    "System-requested major compaction should not occur if there are too many store files",
    false,
    request.isMajor());
}

/**
 * Test for HBASE-5920
 */
@Test
public void testUserMajorCompactionRequest() throws IOException{
  Store store = r.getStore(COLUMN_FAMILY);
  createStoreFile(r);
  for (int i = 0; i < MAX_FILES_TO_COMPACT + 1; i++) {
    createStoreFile(r);
  }
  store.triggerMajorCompaction();
  CompactionRequest request = store.requestCompaction(Store.PRIORITY_USER, null).getRequest();
  assertNotNull("Expected to receive a compaction request", request);
  assertEquals(
    "User-requested major compaction should always occur, even if there are too many store files",
    true, 
    request.isMajor());
}

@Override
public InternalScanner preCompactScannerOpen(
    final ObserverContext<RegionCoprocessorEnvironment> c, final Store store,
    List<? extends KeyValueScanner> scanners, final ScanType scanType, final long earliestPutTs,
    final InternalScanner s, CompactionRequest request) throws IOException {
  return preCompactScannerOpen(c, store, scanners, scanType, earliestPutTs, s);
}

/**
 * @param r region store belongs to
 * @param s Store to request compaction on
 * @param why Why compaction requested -- used in debug messages
 * @param priority override the default priority (NO_PRIORITY == decide)
 * @param request custom compaction request. Can be <tt>null</tt> in which case a simple
 *          compaction will be used.
 */
private synchronized CompactionRequest requestCompactionInternal(final Region r, final Store s,
    final String why, int priority, CompactionRequest request, boolean selectNow, User user)
        throws IOException {
  if (this.server.isStopped()
      || (r.getTableDesc() != null && !r.getTableDesc().isCompactionEnabled())) {
    return null;
  }

  CompactionContext compaction = null;
  if (selectNow) {
    compaction = selectCompaction(r, s, priority, request, user);
    if (compaction == null) return null; // message logged inside
  }

  // We assume that most compactions are small. So, put system compactions into small
  // pool; we will do selection there, and move to large pool if necessary.
  ThreadPoolExecutor pool = (selectNow && s.throttleCompaction(compaction.getRequest().getSize()))
    ? longCompactions : shortCompactions;
  pool.execute(new CompactionRunner(s, r, compaction, pool, user));
  if (LOG.isDebugEnabled()) {
    String type = (pool == shortCompactions) ? "Small " : "Large ";
    LOG.debug(type + "Compaction requested: " + (selectNow ? compaction.toString() : "system")
        + (why != null && !why.isEmpty() ? "; Because: " + why : "") + "; " + this);
  }
  return selectNow ? compaction.getRequest() : null;
}

public RebuildCompactJob(HStore store, CompactionRequest request, Path writtenPath)
    throws IOException {
  super(store, request, writtenPath);
  StringBuilder sb = new StringBuilder();
  sb.append("RebuildCompactJob construction, hdfsPath: ").append(tmpHDFSPath);
  sb.append(", with ").append(request.getFiles().size())
      .append(" store files compacted, they are: ");
  for (StoreFile sf : request.getFiles()) {
    sb.append(sf.getPath()).append(", ");
  }
  printMessage(sb.toString());
}

public CompactJob(HStore store, final CompactionRequest request, final Path writtenPath)
    throws IOException {
  super(CompactJobQueue.getInstance().getJobQueueName());
  this.store = store;
  this.request = request;
  this.tmpHDFSPath = writtenPath;
}

@Override
public boolean select(List<StoreFile> filesCompacting, boolean isUserCompaction,
    boolean mayUseOffPeak, boolean forceMajor) throws IOException {
  this.stripeRequest = compactionPolicy.selectCompaction(
      storeFileManager, filesCompacting, mayUseOffPeak);
  this.request = (this.stripeRequest == null)
      ? new CompactionRequest(new ArrayList<StoreFile>()) : this.stripeRequest.getRequest();
  return this.stripeRequest != null;
}

@Override
public void forceSelect(CompactionRequest request) {
  super.forceSelect(request);
  if (this.stripeRequest != null) {
    this.stripeRequest.setRequest(this.request);
  } else {
    LOG.warn("Stripe store is forced to take an arbitrary file list and compact it.");
    this.stripeRequest = compactionPolicy.createEmptyRequest(storeFileManager, this.request);
  }
}

private void finishCompactionRequest(CompactionRequest cr) {
  this.region.reportCompactionRequestEnd(cr.isMajor(), cr.getFiles().size(), cr.getSize());
  if (cr.isOffPeak()) {
    offPeakCompactionTracker.set(false);
    cr.setOffPeak(false);
  }
  synchronized (filesCompacting) {
    filesCompacting.removeAll(cr.getFiles());
  }
}

/**
 * See
 * {@link RegionObserver#preCompactScannerOpen(ObserverContext, Store, List, ScanType, long, InternalScanner, CompactionRequest)}
 */
public InternalScanner preCompactScannerOpen(final Store store,
    final List<StoreFileScanner> scanners, final ScanType scanType, final long earliestPutTs,
    final CompactionRequest request) throws IOException {
  return execOperationWithResult(null,
      coprocessors.isEmpty() ? null : new RegionOperationWithResult<InternalScanner>() {
    @Override
    public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
        throws IOException {
      setResult(oserver.preCompactScannerOpen(ctx, store, scanners, scanType,
        earliestPutTs, getResult(), request));
    }
  });
}

/**
 * Called prior to selecting the {@link StoreFile}s for compaction from the list of currently
 * available candidates.
 * @param store The store where compaction is being requested
 * @param candidates The currently available store files
 * @param request custom compaction request
 * @return If {@code true}, skip the normal selection process and use the current list
 * @throws IOException
 */
public boolean preCompactSelection(final Store store, final List<StoreFile> candidates,
    final CompactionRequest request) throws IOException {
  return execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
    @Override
    public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
        throws IOException {
      oserver.preCompactSelection(ctx, store, candidates, request);
    }
  });
}

/**
 * Called after the {@link StoreFile}s to be compacted have been selected from the available
 * candidates.
 * @param store The store where compaction is being requested
 * @param selected The store files selected to compact
 * @param request custom compaction
 */
public void postCompactSelection(final Store store, final ImmutableList<StoreFile> selected,
    final CompactionRequest request) {
  try {
    execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
      @Override
      public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
          throws IOException {
        oserver.postCompactSelection(ctx, store, selected, request);
      }
    });
  } catch (IOException e) {
    LOG.warn(e);
  }
}