Java 类org.apache.hadoop.hbase.coprocessor.RegionObserver 实例源码

/**
 * Invoked after a region is closed
 * @param abortRequested true if the server is aborting
 */
public void postClose(final boolean abortRequested) {
  try {
    execOperation(false, new RegionOperation() {
      @Override
      public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
          throws IOException {
        oserver.postClose(ctx, abortRequested);
      }
      public void postEnvCall(RegionEnvironment env) {
        shutdown(env);
      }
    });
  } catch (IOException e) {
    LOG.warn(e);
  }
}

/**
 * @param info
 * @param logKey
 * @param logEdit
 * @return true if default behavior should be bypassed, false otherwise
 * @throws IOException
 */
public boolean preWALRestore(final HRegionInfo info, final WALKey logKey,
    final WALEdit logEdit) throws IOException {
  return execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
    @Override
    public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
        throws IOException {
      // Once we don't need to support the legacy call, replace RegionOperation with a version
      // that's ObserverContext<RegionEnvironment> and avoid this cast.
      final RegionEnvironment env = (RegionEnvironment)ctx.getEnvironment();
      if (env.useLegacyPre) {
        if (logKey instanceof HLogKey) {
          oserver.preWALRestore(ctx, info, (HLogKey)logKey, logEdit);
        } else {
          legacyWarning(oserver.getClass(), "There are wal keys present that are not HLogKey.");
        }
      } else {
        oserver.preWALRestore(ctx, info, logKey, logEdit);
      }
    }
  });
}

/**
 * @param info
 * @param logKey
 * @param logEdit
 * @throws IOException
 */
public void postWALRestore(final HRegionInfo info, final WALKey logKey, final WALEdit logEdit)
    throws IOException {
  execOperation(coprocessors.isEmpty() ? null : new RegionOperation() {
    @Override
    public void call(RegionObserver oserver, ObserverContext<RegionCoprocessorEnvironment> ctx)
        throws IOException {
      // Once we don't need to support the legacy call, replace RegionOperation with a version
      // that's ObserverContext<RegionEnvironment> and avoid this cast.
      final RegionEnvironment env = (RegionEnvironment)ctx.getEnvironment();
      if (env.useLegacyPost) {
        if (logKey instanceof HLogKey) {
          oserver.postWALRestore(ctx, info, (HLogKey)logKey, logEdit);
        } else {
          legacyWarning(oserver.getClass(), "There are wal keys present that are not HLogKey.");
        }
      } else {
        oserver.postWALRestore(ctx, info, logKey, logEdit);
      }
    }
  });
}

static void setupOnce() throws Exception {
  // Using the our load balancer to control region plans
  conf.setClass(HConstants.HBASE_MASTER_LOADBALANCER_CLASS,
    MyLoadBalancer.class, LoadBalancer.class);
  conf.setClass(CoprocessorHost.REGION_COPROCESSOR_CONF_KEY,
    MyRegionObserver.class, RegionObserver.class);
  // Reduce the maximum attempts to speed up the test
  conf.setInt("hbase.assignment.maximum.attempts", 3);
  // Put meta on master to avoid meta server shutdown handling
  conf.set("hbase.balancer.tablesOnMaster", "hbase:meta");
  conf.setInt("hbase.master.maximum.ping.server.attempts", 3);
  conf.setInt("hbase.master.ping.server.retry.sleep.interval", 1);

  TEST_UTIL.startMiniCluster(1, 4, null, MyMaster.class, MyRegionServer.class);
  admin = TEST_UTIL.getHBaseAdmin();
}

/**
 * Invoked before a region open
 */
public void preOpen(){
  ObserverContext<RegionCoprocessorEnvironment> ctx = null;
  for (RegionEnvironment env: coprocessors) {
    if (env.getInstance() instanceof RegionObserver) {
      ctx = ObserverContext.createAndPrepare(env, ctx);
       try {
        ((RegionObserver)env.getInstance()).preOpen(ctx);
       } catch (Throwable e) {
         handleCoprocessorThrowableNoRethrow(env, e);
       }
      if (ctx.shouldComplete()) {
        break;
      }
    }
  }
}

/**
 * Invoked after a region open
 */
public void postOpen(){
  ObserverContext<RegionCoprocessorEnvironment> ctx = null;
  for (RegionEnvironment env: coprocessors) {
    if (env.getInstance() instanceof RegionObserver) {
      ctx = ObserverContext.createAndPrepare(env, ctx);
      try {
        ((RegionObserver)env.getInstance()).postOpen(ctx);
      } catch (Throwable e) {
        handleCoprocessorThrowableNoRethrow(env, e);
      }
      if (ctx.shouldComplete()) {
        break;
      }
    }
  }
}

/**
 * @param miniBatchOp
 * @throws IOException
 */
public void postBatchMutate(
    final MiniBatchOperationInProgress<Mutation> miniBatchOp) throws IOException {
  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()).postBatchMutate(ctx, miniBatchOp);
      } catch (Throwable e) {
        handleCoprocessorThrowable(env, e);
      } finally {
        currentThread.setContextClassLoader(cl);
      }
      if (ctx.shouldComplete()) {
        break;
      }
    }
  }
}

/**
 * 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;
}

/**
 * Invoked just before the rollback of a failed split is started
 * @throws IOException
 */
public void preRollBackSplit() throws IOException {
  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()).preRollBackSplit(ctx);
      } catch (Throwable e) {
        handleCoprocessorThrowable(env, e);
      } finally {
        currentThread.setContextClassLoader(cl);
      }
      if (ctx.shouldComplete()) {
        break;
      }
    }
  }
}

/**
 * @param row the row key
 * @param family the family
 * @param result the result set from the region
 * @exception IOException Exception
 */
public void postGetClosestRowBefore(final byte[] row, final byte[] family,
    final Result result) throws IOException {
  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()).postGetClosestRowBefore(ctx, row, family, result);
      } catch (Throwable e) {
        handleCoprocessorThrowable(env, e);
      } finally {
        currentThread.setContextClassLoader(cl);
      }
      if (ctx.shouldComplete()) {
        break;
      }
    }
  }
}

/**
 * Invoked before a memstore flush
 * @throws IOException
 */
public InternalScanner preFlush(Store store, InternalScanner scanner) 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()).preFlush(
            ctx, store, scanner);
      } catch (Throwable e) {
        handleCoprocessorThrowable(env,e);
      }
      bypass |= ctx.shouldBypass();
      if (ctx.shouldComplete()) {
        break;
      }
    }
  }
  return bypass ? null : scanner;
}

/**
 * Invoked just after the rollback of a failed split is done
 * @throws IOException
 */
public void postRollBackSplit() throws IOException {
  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()).postRollBackSplit(ctx);
      } catch (Throwable e) {
        handleCoprocessorThrowable(env, e);
      } finally {
        currentThread.setContextClassLoader(cl);
      }
      if (ctx.shouldComplete()) {
        break;
      }
    }
  }
}

/**
 * See
 * {@link RegionObserver#preFlush(ObserverContext, Store, KeyValueScanner)}
 */
public InternalScanner preFlushScannerOpen(Store store, KeyValueScanner memstoreScanner) 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()).preFlushScannerOpen(ctx, store, memstoreScanner, s);
      } catch (Throwable e) {
        handleCoprocessorThrowable(env, e);
      }
      if (ctx.shouldComplete()) {
        break;
      }
    }
  }
  return s;
}

/**
 * Invoked after a region is closed
 * @param abortRequested true if the server is aborting
 */
public void postClose(final boolean abortRequested) {
  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()).postClose(ctx, abortRequested);
      } catch (Throwable e) {
        handleCoprocessorThrowableNoRethrow(env, e);
      } finally {
        currentThread.setContextClassLoader(cl);
      }
    }
    shutdown(env);
  }
}

/**
 * Invoked just before a split
 * @throws IOException
 */
public void preSplit() throws IOException {
  ObserverContext<RegionCoprocessorEnvironment> ctx = null;
  for (RegionEnvironment env: coprocessors) {
    if (env.getInstance() instanceof RegionObserver) {
      ctx = ObserverContext.createAndPrepare(env, ctx);
      try {
        ((RegionObserver)env.getInstance()).preSplit(ctx);
      } catch (Throwable e) {
        handleCoprocessorThrowable(env, e);
      }
      if (ctx.shouldComplete()) {
        break;
      }
    }
  }
}

/**
 * Invoked after a region open
 */
public void postOpen() {
  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()).postOpen(ctx);
      } catch (Throwable e) {
        handleCoprocessorThrowableNoRethrow(env, e);
      } finally {
        currentThread.setContextClassLoader(cl);
      }
      if (ctx.shouldComplete()) {
        break;
      }
    }
  }
}

/**
 * @param row the row key
 * @param family the family
 * @param result the result set from the region
 * @return true if default processing should be bypassed
 * @exception IOException Exception
 */
public boolean preGetClosestRowBefore(final byte[] row, final byte[] family,
    final Result result) throws IOException {
  boolean bypass = false;
  ObserverContext<RegionCoprocessorEnvironment> ctx = null;
  for (RegionEnvironment env: coprocessors) {
    if (env.getInstance() instanceof RegionObserver) {
      ctx = ObserverContext.createAndPrepare(env, ctx);
      try {
        ((RegionObserver)env.getInstance()).preGetClosestRowBefore(ctx, row,
            family, result);
      } catch (Throwable e) {
        handleCoprocessorThrowable(env, e);
      }
      bypass |= ctx.shouldBypass();
      if (ctx.shouldComplete()) {
        break;
      }
    }
  }
  return bypass;
}

/**
 * @param row the row key
 * @param family the family
 * @param result the result set from the region
 * @exception IOException Exception
 */
public void postGetClosestRowBefore(final byte[] row, final byte[] family,
    final Result result) throws IOException {
  ObserverContext<RegionCoprocessorEnvironment> ctx = null;
  for (RegionEnvironment env: coprocessors) {
    if (env.getInstance() instanceof RegionObserver) {
      ctx = ObserverContext.createAndPrepare(env, ctx);
      try {
        ((RegionObserver)env.getInstance()).postGetClosestRowBefore(ctx, row,
            family, result);
      } catch (Throwable e) {
        handleCoprocessorThrowable(env, e);
      }
      if (ctx.shouldComplete()) {
        break;
      }
    }
  }
}

/**
 * @param get the Get request
 * @param results the result set
 * @exception IOException Exception
 */
public void postGet(final Get get, final List<KeyValue> results)
throws IOException {
  ObserverContext<RegionCoprocessorEnvironment> ctx = null;
  for (RegionEnvironment env: coprocessors) {
    if (env.getInstance() instanceof RegionObserver) {
      ctx = ObserverContext.createAndPrepare(env, ctx);
      try {
        ((RegionObserver)env.getInstance()).postGet(ctx, get, results);
      } catch (Throwable e) {
        handleCoprocessorThrowable(env, e);
      }
      if (ctx.shouldComplete()) {
        break;
      }
    }
  }
}

/**
 * @param s the scanner
 * @exception IOException Exception
 */
public void postScannerClose(final InternalScanner s) throws IOException {
  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()).postScannerClose(ctx, s);
      } catch (Throwable e) {
        handleCoprocessorThrowable(env, e);
      } finally {
        currentThread.setContextClassLoader(cl);
      }
      if (ctx.shouldComplete()) {
        break;
      }
    }
  }
}

/**
 * @param get the Get request
 * @param exists the result returned by the region server
 * @return the result to return to the client
 * @exception IOException Exception
 */
public boolean postExists(final Get get, boolean exists)
    throws IOException {
  ObserverContext<RegionCoprocessorEnvironment> ctx = null;
  for (RegionEnvironment env: coprocessors) {
    if (env.getInstance() instanceof RegionObserver) {
      ctx = ObserverContext.createAndPrepare(env, ctx);
      try {
        exists = ((RegionObserver)env.getInstance()).postExists(ctx, get, exists);
      } catch (Throwable e) {
        handleCoprocessorThrowable(env, e);
      }
      if (ctx.shouldComplete()) {
        break;
      }
    }
  }
  return exists;
}

/**
 * @param put The Put object
 * @param edit The WALEdit object.
 * @param writeToWAL true if the change should be written to the WAL
 * @exception IOException Exception
 */
public void postPut(Put put, WALEdit edit,
    final boolean writeToWAL) throws IOException {
  ObserverContext<RegionCoprocessorEnvironment> ctx = null;
  for (RegionEnvironment env: coprocessors) {
    if (env.getInstance() instanceof RegionObserver) {
      ctx = ObserverContext.createAndPrepare(env, ctx);
      try {
        ((RegionObserver)env.getInstance()).postPut(ctx, put, edit, writeToWAL);
      } catch (Throwable e) {
        handleCoprocessorThrowable(env, e);
      }
      if (ctx.shouldComplete()) {
        break;
      }
    }
  }
}

/**
 * Invoked after a memstore flush
 * @throws IOException
 */
public void postFlush(final Store store, final StoreFile storeFile) throws IOException {
  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()).postFlush(ctx, store, storeFile);
      } catch (Throwable e) {
        handleCoprocessorThrowable(env, e);
      } finally {
        currentThread.setContextClassLoader(cl);
      }
      if (ctx.shouldComplete()) {
        break;
      }
    }
  }
}

/**
 * @param miniBatchOp
 * @return true if default processing should be bypassed
 * @throws IOException
 */
public boolean preBatchMutate(
    final MiniBatchOperationInProgress<Pair<Mutation, Integer>> miniBatchOp) throws IOException {
  boolean bypass = false;
  ObserverContext<RegionCoprocessorEnvironment> ctx = null;
  for (RegionEnvironment env : coprocessors) {
    if (env.getInstance() instanceof RegionObserver) {
      ctx = ObserverContext.createAndPrepare(env, ctx);
      try {
        ((RegionObserver) env.getInstance()).preBatchMutate(ctx, miniBatchOp);
      } catch (Throwable e) {
        handleCoprocessorThrowable(env, e);
      }
      bypass |= ctx.shouldBypass();
      if (ctx.shouldComplete()) {
        break;
      }
    }
  }
  return bypass;
}

/**
 * @param miniBatchOp
 * @throws IOException
 */
public void postBatchMutate(
    final MiniBatchOperationInProgress<Pair<Mutation, Integer>> miniBatchOp) throws IOException {
  ObserverContext<RegionCoprocessorEnvironment> ctx = null;
  for (RegionEnvironment env : coprocessors) {
    if (env.getInstance() instanceof RegionObserver) {
      ctx = ObserverContext.createAndPrepare(env, ctx);
      try {
        ((RegionObserver) env.getInstance()).postBatchMutate(ctx, miniBatchOp);
      } catch (Throwable e) {
        handleCoprocessorThrowable(env, e);
      }
      if (ctx.shouldComplete()) {
        break;
      }
    }
  }
}

public void postCloseRegionOperation(final Operation op) throws IOException {
  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()).postCloseRegionOperation(ctx, op);
      } catch (Throwable e) {
        handleCoprocessorThrowable(env, e);
      } finally {
        currentThread.setContextClassLoader(cl);
      }
      if (ctx.shouldComplete()) {
        break;
      }
    }
  }
}

/**
 * @param row row to check
 * @param family column family
 * @param qualifier column qualifier
 * @param compareOp the comparison operation
 * @param comparator the comparator
 * @param delete delete to commit if check succeeds
 * @throws IOException e
 */
public boolean postCheckAndDelete(final byte [] row, final byte [] family,
    final byte [] qualifier, final CompareOp compareOp,
    final WritableByteArrayComparable comparator, final Delete delete,
    boolean result)
  throws IOException {
  ObserverContext<RegionCoprocessorEnvironment> ctx = null;
  for (RegionEnvironment env: coprocessors) {
    if (env.getInstance() instanceof RegionObserver) {
      ctx = ObserverContext.createAndPrepare(env, ctx);
      try {
        result = ((RegionObserver)env.getInstance())
          .postCheckAndDelete(ctx, row, family, qualifier, compareOp,
            comparator, delete, result);
      } catch (Throwable e) {
        handleCoprocessorThrowable(env, e);
      }
      if (ctx.shouldComplete()) {
        break;
      }
    }
  }
  return result;
}

/**
 * @param put The Put object
 * @param edit The WALEdit object.
 * @param durability The durability used
 * @exception IOException Exception
 */
public void postPut(final Put put, final WALEdit edit, final Durability durability)
    throws IOException {
  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()).postPut(ctx, put, edit, durability);
      } catch (Throwable e) {
        handleCoprocessorThrowable(env, e);
      } finally {
        currentThread.setContextClassLoader(cl);
      }
      if (ctx.shouldComplete()) {
        break;
      }
    }
  }
}

/**
 * @param append append object
 * @return result to return to client if default operation should be
 * bypassed, null otherwise
 * @throws IOException if an error occurred on the coprocessor
 */
public Result preAppend(Append append)
    throws IOException {
  boolean bypass = false;
  Result result = null;
  ObserverContext<RegionCoprocessorEnvironment> ctx = null;
  for (RegionEnvironment env: coprocessors) {
    if (env.getInstance() instanceof RegionObserver) {
      ctx = ObserverContext.createAndPrepare(env, ctx);
      try {
        result = ((RegionObserver)env.getInstance()).preAppend(ctx, append);
      } catch (Throwable e) {
        handleCoprocessorThrowable(env, e);
      }
      bypass |= ctx.shouldBypass();
      if (ctx.shouldComplete()) {
        break;
      }
    }
  }
  return bypass ? result : null;
}

/**
 * @param append Append object
 * @param result the result returned by postAppend
 * @throws IOException if an error occurred on the coprocessor
 */
public void postAppend(final Append append, Result result)
    throws IOException {
  ObserverContext<RegionCoprocessorEnvironment> ctx = null;
  for (RegionEnvironment env: coprocessors) {
    if (env.getInstance() instanceof RegionObserver) {
      ctx = ObserverContext.createAndPrepare(env, ctx);
      try {
        ((RegionObserver)env.getInstance()).postAppend(ctx, append, result);
      } catch (Throwable e) {
        handleCoprocessorThrowable(env, e);
      }
      if (ctx.shouldComplete()) {
        break;
      }
    }
  }
}

/**
 * @param familyPaths pairs of { CF, file path } submitted for bulk load
 * @param hasLoaded whether load was successful or not
 * @return the possibly modified value of hasLoaded
 * @throws IOException
 */
public boolean postBulkLoadHFile(final List<Pair<byte[], String>> familyPaths,
    boolean hasLoaded) throws IOException {
  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());
        hasLoaded = ((RegionObserver)env.getInstance()).postBulkLoadHFile(ctx, familyPaths,
          hasLoaded);
      } catch (Throwable e) {
        handleCoprocessorThrowable(env, e);
      } finally {
        currentThread.setContextClassLoader(cl);
      }
      if (ctx.shouldComplete()) {
        break;
      }
    }
  }
  return hasLoaded;
}

/**
 * @param scan the Scan specification
 * @return scanner id to return to client if default operation should be
 * bypassed, false otherwise
 * @exception IOException Exception
 */
public RegionScanner preScannerOpen(Scan scan) throws IOException {
  boolean bypass = false;
  RegionScanner s = null;
  ObserverContext<RegionCoprocessorEnvironment> ctx = null;
  for (RegionEnvironment env: coprocessors) {
    if (env.getInstance() instanceof RegionObserver) {
      ctx = ObserverContext.createAndPrepare(env, ctx);
      try {
        s = ((RegionObserver)env.getInstance()).preScannerOpen(ctx, scan, s);
      } catch (Throwable e) {
        handleCoprocessorThrowable(env, e);
      }
      bypass |= ctx.shouldBypass();
      if (ctx.shouldComplete()) {
        break;
      }
    }
  }
  return bypass ? s : null;
}

/**
 * @param scan the Scan specification
 * @param s the scanner
 * @return the scanner instance to use
 * @exception IOException Exception
 */
public RegionScanner postScannerOpen(final Scan scan, RegionScanner s) throws IOException {
  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());
        s = ((RegionObserver)env.getInstance()).postScannerOpen(ctx, scan, s);
      } catch (Throwable e) {
        handleCoprocessorThrowable(env, e);
      } finally {
        currentThread.setContextClassLoader(cl);
      }
      if (ctx.shouldComplete()) {
        break;
      }
    }
  }
  return s;
}

/**
 * @param s the scanner
 * @param results the result set returned by the region server
 * @param limit the maximum number of results to return
 * @return 'has next' indication to client if bypassing default behavior, or
 * null otherwise
 * @exception IOException Exception
 */
public Boolean preScannerNext(final InternalScanner s,
    final List<Result> results, int limit) throws IOException {
  boolean bypass = false;
  boolean hasNext = false;
  ObserverContext<RegionCoprocessorEnvironment> ctx = null;
  for (RegionEnvironment env: coprocessors) {
    if (env.getInstance() instanceof RegionObserver) {
      ctx = ObserverContext.createAndPrepare(env, ctx);
      try {
        hasNext = ((RegionObserver)env.getInstance()).preScannerNext(ctx, s, results,
          limit, hasNext);
      } catch (Throwable e) {
        handleCoprocessorThrowable(env, e);
      }
      bypass |= ctx.shouldBypass();
      if (ctx.shouldComplete()) {
        break;
      }
    }
  }
  return bypass ? hasNext : null;
}

/**
 * @param s the scanner
 * @param results the result set returned by the region server
 * @param limit the maximum number of results to return
 * @param hasMore
 * @return 'has more' indication to give to client
 * @exception IOException Exception
 */
public boolean postScannerNext(final InternalScanner s,
    final List<Result> results, final int limit, boolean hasMore)
    throws IOException {
  ObserverContext<RegionCoprocessorEnvironment> ctx = null;
  for (RegionEnvironment env: coprocessors) {
    if (env.getInstance() instanceof RegionObserver) {
      ctx = ObserverContext.createAndPrepare(env, ctx);
      try {
        hasMore = ((RegionObserver)env.getInstance()).postScannerNext(ctx, s,
          results, limit, hasMore);
      } catch (Throwable e) {
        handleCoprocessorThrowable(env, e);
      }
      if (ctx.shouldComplete()) {
        break;
      }
    }
  }
  return hasMore;
}

/**
 * This will be called by the scan flow when the current scanned row is being filtered out by the
 * filter.
 * @param s the scanner
 * @param currentRow The current rowkey which got filtered out
 * @param offset offset to rowkey
 * @param length length of rowkey
 * @return whether more rows are available for the scanner or not
 * @throws IOException
 */
public boolean postScannerFilterRow(final InternalScanner s, final byte[] currentRow, int offset,
    short length) throws IOException {
  boolean hasMore = true; // By default assume more rows there.
  ObserverContext<RegionCoprocessorEnvironment> ctx = null;
  for (RegionEnvironment env : coprocessors) {
    if (env.getInstance() instanceof RegionObserver) {
      ctx = ObserverContext.createAndPrepare(env, ctx);
      try {
        hasMore = ((RegionObserver) env.getInstance()).postScannerFilterRow(ctx, s, currentRow,
            offset, length, hasMore);
      } catch (Throwable e) {
        handleCoprocessorThrowable(env, e);
      }
      if (ctx.shouldComplete()) {
        break;
      }
    }
  }
  return hasMore;
}

/**
 * @param put The Put object
 * @param edit The WALEdit object.
 * @param durability The durability used
 * @return true if default processing should be bypassed
 * @exception IOException Exception
 */
public boolean prePut(final Put put, final WALEdit edit, final Durability durability)
    throws IOException {
  boolean bypass = false;
  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()).prePut(ctx, put, edit, durability);
      } catch (Throwable e) {
        handleCoprocessorThrowable(env, e);
      } finally {
        currentThread.setContextClassLoader(cl);
      }
      bypass |= ctx.shouldBypass();
      if (ctx.shouldComplete()) {
        break;
      }
    }
  }
  return bypass;
}

/**
 * @param info
 * @param logKey
 * @param logEdit
 * @return true if default behavior should be bypassed, false otherwise
 * @throws IOException
 */
public boolean preWALRestore(HRegionInfo info, HLogKey logKey,
    WALEdit logEdit) throws IOException {
  boolean bypass = false;
  ObserverContext<RegionCoprocessorEnvironment> ctx = null;
  for (RegionEnvironment env: coprocessors) {
    if (env.getInstance() instanceof RegionObserver) {
      ctx = ObserverContext.createAndPrepare(env, ctx);
      try {
        ((RegionObserver)env.getInstance()).preWALRestore(ctx, info, logKey,
            logEdit);
      } catch (Throwable e) {
        handleCoprocessorThrowable(env, e);
      }
      bypass |= ctx.shouldBypass();
      if (ctx.shouldComplete()) {
        break;
      }
    }
  }
  return bypass;
}

/**
 * @param info
 * @param logKey
 * @param logEdit
 * @throws IOException
 */
public void postWALRestore(HRegionInfo info, HLogKey logKey,
    WALEdit logEdit) throws IOException {
  ObserverContext<RegionCoprocessorEnvironment> ctx = null;
  for (RegionEnvironment env: coprocessors) {
    if (env.getInstance() instanceof RegionObserver) {
      ctx = ObserverContext.createAndPrepare(env, ctx);
      try {
        ((RegionObserver)env.getInstance()).postWALRestore(ctx, info,
            logKey, logEdit);
      } catch (Throwable e) {
        handleCoprocessorThrowable(env, e);
      }
      if (ctx.shouldComplete()) {
        break;
      }
    }
  }
}

/**
 * @param get the Get request
 * @param results the result sett
 * @exception IOException Exception
 */
public void postGet(final Get get, final List<Cell> results)
throws IOException {
  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()).postGetOp(ctx, get, results);
      } catch (Throwable e) {
        handleCoprocessorThrowable(env, e);
      } finally {
        currentThread.setContextClassLoader(cl);
      }
      if (ctx.shouldComplete()) {
        break;
      }
    }
  }
}