diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/BaseRegionObserver.java hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/BaseRegionObserver.java index c861192..f145bf0 100644 --- hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/BaseRegionObserver.java +++ hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/BaseRegionObserver.java @@ -311,6 +311,12 @@ public abstract class BaseRegionObserver implements RegionObserver { } @Override + public void preGetForDeleteVersion(final ObserverContext e, + final Mutation delete, final Cell cell, final byte[] byteNow, final byte[] family, + final Get get) throws IOException { + } + + @Override public void postDelete(final ObserverContext e, final Delete delete, final WALEdit edit, final Durability durability) throws IOException { @@ -340,6 +346,15 @@ public abstract class BaseRegionObserver implements RegionObserver { } @Override + public boolean preCheckAndPutAfterRowLock( + final ObserverContext e, + final byte[] row, final byte[] family, final byte[] qualifier, final CompareOp compareOp, + final ByteArrayComparable comparator, final Put put, + final boolean result) throws IOException { + return result; + } + + @Override public boolean postCheckAndPut(final ObserverContext e, final byte [] row, final byte [] family, final byte [] qualifier, final CompareOp compareOp, final ByteArrayComparable comparator, @@ -356,6 +371,15 @@ public abstract class BaseRegionObserver implements RegionObserver { } @Override + public boolean preCheckAndDeleteAfterRowLock( + final ObserverContext e, + final byte[] row, final byte[] family, final byte[] qualifier, final CompareOp compareOp, + final ByteArrayComparable comparator, final Delete delete, + final boolean result) throws IOException { + return result; + } + + @Override public boolean postCheckAndDelete(final ObserverContext e, final byte [] row, final byte [] family, final byte [] qualifier, final CompareOp compareOp, final ByteArrayComparable comparator, @@ -370,6 +394,12 @@ public abstract class BaseRegionObserver implements RegionObserver { } @Override + public Result preAppendAfterRowLock(final ObserverContext e, + final Append append) throws IOException { + return null; + } + + @Override public Result postAppend(final ObserverContext e, final Append append, final Result result) throws IOException { return result; @@ -397,6 +427,12 @@ public abstract class BaseRegionObserver implements RegionObserver { } @Override + public Result preIncrementAfterRowLock(final ObserverContext e, + final Increment increment) throws IOException { + return null; + } + + @Override public Result postIncrement(final ObserverContext e, final Increment increment, final Result result) throws IOException { return result; diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/RegionObserver.java hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/RegionObserver.java index 96cc3bd..3867808 100644 --- hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/RegionObserver.java +++ hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/RegionObserver.java @@ -587,6 +587,25 @@ public interface RegionObserver extends Coprocessor { void preDelete(final ObserverContext c, final Delete delete, final WALEdit edit, final Durability durability) throws IOException; +/** + * Called before the client updates the timestamp for delete Columns with latest timestamp. + *

+ * Call CoprocessorEnvironment#bypass to skip default actions + *

+ * Call CoprocessorEnvironment#complete to skip any subsequent chained + * coprocessors + * @param c the environment provided by the region server + * @param mutation - the parent mutation associated with this delete cell + * @param cell - The deleteColumn with latest version cell + * @param byteNow - timestamp bytes + * @param family - family bytes + * @param get - the get formed using the current cell's row. + * Note that the get does not specify the family and qualifier + * @throws IOException + */ + void preGetForDeleteVersion(final ObserverContext c, + final Mutation mutation, final Cell cell, final byte[] byteNow, final byte[] family, + final Get get) throws IOException; /** * Called after the client deletes a value. @@ -657,7 +676,7 @@ public interface RegionObserver extends Coprocessor { MiniBatchOperationInProgress miniBatchOp, final boolean success) throws IOException; /** - * Called before checkAndPut + * Called before checkAndPut. *

* Call CoprocessorEnvironment#bypass to skip default actions *

@@ -682,6 +701,33 @@ public interface RegionObserver extends Coprocessor { throws IOException; /** + * Called before checkAndPut but after acquiring rowlock. + * Note: Caution to be taken for not doing any long time operation in this hook. Row will be + * locked for longer time. Trying to acquire lock on another row, within this, can lead to + * potential deadlock. + *

+ * Call CoprocessorEnvironment#bypass to skip default actions + *

+ * Call CoprocessorEnvironment#complete to skip any subsequent chained + * coprocessors + * @param c the environment provided by the region server + * @param row row to check + * @param family column family + * @param qualifier column qualifier + * @param compareOp the comparison operation + * @param comparator the comparator + * @param put data to put if check succeeds + * @param result + * @return the return value to return to client if bypassing default + * processing + * @throws IOException if an error occurred on the coprocessor + */ + boolean preCheckAndPutAfterRowLock(final ObserverContext c, + final byte[] row, final byte[] family, final byte[] qualifier, final CompareOp compareOp, + final ByteArrayComparable comparator, final Put put, + final boolean result) throws IOException; + + /** * Called after checkAndPut *

* Call CoprocessorEnvironment#complete to skip any subsequent chained @@ -704,7 +750,7 @@ public interface RegionObserver extends Coprocessor { throws IOException; /** - * Called before checkAndDelete + * Called before checkAndDelete. *

* Call CoprocessorEnvironment#bypass to skip default actions *

@@ -728,6 +774,32 @@ public interface RegionObserver extends Coprocessor { throws IOException; /** + * Called before checkAndDelete but after acquiring rowock. + * Note: Caution to be taken for not doing any long time operation in this hook. Row will be + * locked for longer time. Trying to acquire lock on another row, within this, can lead to + * potential deadlock. + *

+ * Call CoprocessorEnvironment#bypass to skip default actions + *

+ * Call CoprocessorEnvironment#complete to skip any subsequent chained + * coprocessors + * @param c the environment provided by the region server + * @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 + * @param result + * @return the value to return to client if bypassing default processing + * @throws IOException if an error occurred on the coprocessor + */ + boolean preCheckAndDeleteAfterRowLock(final ObserverContext c, + final byte[] row, final byte[] family, final byte[] qualifier, final CompareOp compareOp, + final ByteArrayComparable comparator, final Delete delete, + final boolean result) throws IOException; + + /** * Called after checkAndDelete *

* Call CoprocessorEnvironment#complete to skip any subsequent chained @@ -795,7 +867,7 @@ public interface RegionObserver extends Coprocessor { throws IOException; /** - * Called before Append + * Called before Append. *

* Call CoprocessorEnvironment#bypass to skip default actions *

@@ -811,6 +883,24 @@ public interface RegionObserver extends Coprocessor { throws IOException; /** + * Called before Append but after acquiring rowlock. + * Note: Caution to be taken for not doing any long time operation in this hook. Row will be + * locked for longer time. Trying to acquire lock on another row, within this, can lead to + * potential deadlock. + *

+ * Call CoprocessorEnvironment#bypass to skip default actions + *

+ * Call CoprocessorEnvironment#complete to skip any subsequent chained + * coprocessors + * @param c the environment provided by the region server + * @param append Append object + * @return result to return to the client if bypassing default processing + * @throws IOException if an error occurred on the coprocessor + */ + Result preAppendAfterRowLock(final ObserverContext c, + final Append append) throws IOException; + + /** * Called after Append *

* Call CoprocessorEnvironment#complete to skip any subsequent chained @@ -826,7 +916,7 @@ public interface RegionObserver extends Coprocessor { throws IOException; /** - * Called before Increment + * Called before Increment. *

* Call CoprocessorEnvironment#bypass to skip default actions *

@@ -842,6 +932,27 @@ public interface RegionObserver extends Coprocessor { throws IOException; /** + * Called before Increment but after acquiring rowlock. + * Note: Caution to be taken for not doing any long time operation in this hook. Row will be + * locked for longer time. Trying to acquire lock on another row, within this, can lead to + * potential deadlock. + *

+ * Call CoprocessorEnvironment#bypass to skip default actions + *

+ * Call CoprocessorEnvironment#complete to skip any subsequent chained coprocessors + * + * @param c + * the environment provided by the region server + * @param increment + * increment object + * @return result to return to the client if bypassing default processing + * @throws IOException + * if an error occurred on the coprocessor + */ + Result preIncrementAfterRowLock(final ObserverContext c, + final Increment increment) throws IOException; + + /** * Called after increment *

* Call CoprocessorEnvironment#complete to skip any subsequent chained diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java index dea1e81..2870a4f 100644 --- hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java +++ hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java @@ -2032,7 +2032,7 @@ public class HRegion implements HeapSize { // , Writable{ * Caller should have the row and region locks. * @throws IOException */ - void prepareDeleteTimestamps(Map> familyMap, byte[] byteNow) + void prepareDeleteTimestamps(Mutation mutation, Map> familyMap, byte[] byteNow) throws IOException { for (Map.Entry> e : familyMap.entrySet()) { @@ -2045,34 +2045,14 @@ public class HRegion implements HeapSize { // , Writable{ // Check if time is LATEST, change to time of most recent addition if so // This is expensive. if (kv.isLatestTimestamp() && kv.isDeleteType()) { - byte[] qual = CellUtil.cloneQualifier(kv); - if (qual == null) qual = HConstants.EMPTY_BYTE_ARRAY; - - Integer count = kvCount.get(qual); - if (count == null) { - kvCount.put(qual, 1); - } else { - kvCount.put(qual, count + 1); - } - count = kvCount.get(qual); - - Get get = new Get(CellUtil.cloneRow(kv)); - get.setMaxVersions(count); - get.addColumn(family, qual); - - List result = get(get, false); - - if (result.size() < count) { - // Nothing to delete - kv.updateLatestStamp(byteNow); - continue; - } - if (result.size() > count) { - throw new RuntimeException("Unexpected size: " + result.size()); + if (coprocessorHost != null) { + Get get = new Get(kv.getRow()); + if (!coprocessorHost.preGetForDeleteVersion(mutation, cell, byteNow, family, get)) { + prepareForDeleteVersion(byteNow, kv, kvCount, family); } - KeyValue getkv = KeyValueUtil.ensureKeyValue(result.get(count - 1)); - Bytes.putBytes(kv.getBuffer(), kv.getTimestampOffset(), - getkv.getBuffer(), getkv.getTimestampOffset(), Bytes.SIZEOF_LONG); + } else { + prepareForDeleteVersion(byteNow, kv, kvCount, family); + } } else { kv.updateLatestStamp(byteNow); } @@ -2080,6 +2060,39 @@ public class HRegion implements HeapSize { // , Writable{ } } + void prepareForDeleteVersion(byte[] byteNow, KeyValue kv, Map kvCount, + byte[] family) throws IOException { + byte[] qual = kv.getQualifier(); + if (qual == null) + qual = HConstants.EMPTY_BYTE_ARRAY; + + Integer count = kvCount.get(qual); + if (count == null) { + kvCount.put(qual, 1); + } else { + kvCount.put(qual, count + 1); + } + count = kvCount.get(qual); + + Get get = new Get(kv.getRow()); + get.setMaxVersions(count); + get.addColumn(family, qual); + + List result = get(get, false); + + if (result.size() < count) { + // Nothing to delete + kv.updateLatestStamp(byteNow); + return; + } + if (result.size() > count) { + throw new RuntimeException("Unexpected size: " + result.size()); + } + KeyValue getkv = KeyValueUtil.ensureKeyValue(result.get(count - 1)); + Bytes.putBytes(kv.getBuffer(), kv.getTimestampOffset(), getkv.getBuffer(), + getkv.getTimestampOffset(), Bytes.SIZEOF_LONG); + } + /** * @throws IOException */ @@ -2452,7 +2465,7 @@ public class HRegion implements HeapSize { // , Writable{ updateKVTimestamps(familyMaps[i].values(), byteNow); noOfPuts++; } else { - prepareDeleteTimestamps(familyMaps[i], byteNow); + prepareDeleteTimestamps(mutation, familyMaps[i], byteNow); noOfDeletes++; } } @@ -2712,9 +2725,21 @@ public class HRegion implements HeapSize { // , Writable{ RowLock rowLock = getRowLock(get.getRow()); // wait for all previous transactions to complete (with lock held) mvcc.completeMemstoreInsert(mvcc.beginMemstoreInsert()); - List result; try { - result = get(get, false); + if (this.getCoprocessorHost() != null) { + Boolean processed = null; + if (w instanceof Put) { + processed = this.getCoprocessorHost().preCheckAndPutAfterRowLock(row, family, + qualifier, compareOp, comparator, (Put) w); + } else if (w instanceof Delete) { + processed = this.getCoprocessorHost().preCheckAndDeleteAfterRowLock(row, family, + qualifier, compareOp, comparator, (Delete) w); + } + if (processed != null) { + return processed; + } + } + List result = get(get, false); boolean valueIsNull = comparator.getValue() == null || comparator.getValue().length == 0; @@ -5033,12 +5058,18 @@ public class HRegion implements HeapSize { // , Writable{ rowLock = getRowLock(row); try { lock(this.updatesLock.readLock()); - // wait for all prior MVCC transactions to finish - while we hold the row lock - // (so that we are guaranteed to see the latest state) - mvcc.completeMemstoreInsert(mvcc.beginMemstoreInsert()); - // now start my own transaction - w = mvcc.beginMemstoreInsert(); try { + // wait for all prior MVCC transactions to finish - while we hold the row lock + // (so that we are guaranteed to see the latest state) + mvcc.completeMemstoreInsert(mvcc.beginMemstoreInsert()); + if (this.coprocessorHost != null) { + Result r = this.coprocessorHost.preAppendAfterRowLock(append); + if(r!= null) { + return r; + } + } + // now start my own transaction + w = mvcc.beginMemstoreInsert(); long now = EnvironmentEdgeManager.currentTimeMillis(); // Process each family for (Map.Entry> family : append.getFamilyCellMap().entrySet()) { @@ -5221,12 +5252,18 @@ public class HRegion implements HeapSize { // , Writable{ RowLock rowLock = getRowLock(row); try { lock(this.updatesLock.readLock()); - // wait for all prior MVCC transactions to finish - while we hold the row lock - // (so that we are guaranteed to see the latest state) - mvcc.completeMemstoreInsert(mvcc.beginMemstoreInsert()); - // now start my own transaction - w = mvcc.beginMemstoreInsert(); try { + // wait for all prior MVCC transactions to finish - while we hold the row lock + // (so that we are guaranteed to see the latest state) + mvcc.completeMemstoreInsert(mvcc.beginMemstoreInsert()); + if (this.coprocessorHost != null) { + Result r = this.coprocessorHost.preIncrementAfterRowLock(increment); + if (r != null) { + return r; + } + } + // now start my own transaction + w = mvcc.beginMemstoreInsert(); long now = EnvironmentEdgeManager.currentTimeMillis(); // Process each family for (Map.Entry> family: diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MultiRowMutationProcessor.java hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MultiRowMutationProcessor.java index 847e6cb..ed70856 100644 --- hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MultiRowMutationProcessor.java +++ hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MultiRowMutationProcessor.java @@ -27,9 +27,9 @@ import org.apache.hadoop.hbase.DoNotRetryIOException; import org.apache.hadoop.hbase.KeyValue; import org.apache.hadoop.hbase.KeyValueUtil; import org.apache.hadoop.hbase.client.Delete; +import org.apache.hadoop.hbase.client.Durability; import org.apache.hadoop.hbase.client.Mutation; import org.apache.hadoop.hbase.client.Put; -import org.apache.hadoop.hbase.client.Durability; import org.apache.hadoop.hbase.protobuf.generated.MultiRowMutationProtos.MultiRowMutationProcessorRequest; import org.apache.hadoop.hbase.protobuf.generated.MultiRowMutationProtos.MultiRowMutationProcessorResponse; import org.apache.hadoop.hbase.regionserver.wal.WALEdit; @@ -69,6 +69,7 @@ MultiRowMutationProcessorResponse> { HRegion region, List mutationKvs, WALEdit walEdit) throws IOException { + RegionCoprocessorHost coprocessorHost = region.getCoprocessorHost(); byte[] byteNow = Bytes.toBytes(now); // Check mutations and apply edits to a single WALEdit for (Mutation m : mutations) { @@ -80,7 +81,7 @@ MultiRowMutationProcessorResponse> { } else if (m instanceof Delete) { Delete d = (Delete) m; region.prepareDelete(d); - region.prepareDeleteTimestamps(d.getFamilyCellMap(), byteNow); + region.prepareDeleteTimestamps(d, d.getFamilyCellMap(), byteNow); } else { throw new DoNotRetryIOException( "Action must be Put or Delete. But was: " diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java index 2e7b022..e0363b2 100644 --- hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java +++ hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java @@ -1730,7 +1730,7 @@ public class RSRpcServices implements HBaseRPCErrorHandler, ProtobufUtil.toComparator(condition.getComparator()); if (region.getCoprocessorHost() != null) { processed = region.getCoprocessorHost().preCheckAndPut( - row, family, qualifier, compareOp, comparator, put); + row, family, qualifier, compareOp, comparator, put); } if (processed == null) { boolean result = region.checkAndMutate(row, family, @@ -1758,7 +1758,7 @@ public class RSRpcServices implements HBaseRPCErrorHandler, ProtobufUtil.toComparator(condition.getComparator()); if (region.getCoprocessorHost() != null) { processed = region.getCoprocessorHost().preCheckAndDelete( - row, family, qualifier, compareOp, comparator, delete); + row, family, qualifier, compareOp, comparator, delete); } if (processed == null) { boolean result = region.checkAndMutate(row, family, diff --git hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RegionCoprocessorHost.java hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RegionCoprocessorHost.java index 177f153..a0f401d 100644 --- hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RegionCoprocessorHost.java +++ hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RegionCoprocessorHost.java @@ -1135,6 +1135,45 @@ public class RegionCoprocessorHost } /** + * @param mutation - the current mutation + * @param kv - the current cell + * @param byteNow - current timestamp in bytes + * @param family - the current family + * @param get - the get that could be used + * Note that the get only does not specify the family and qualifier that should be used + * @return true if default processing should be bypassed + * @exception IOException + * Exception + */ + public boolean preGetForDeleteVersion(Mutation mutation, + Cell kv, byte[] byteNow, byte[] family, Get get) throws IOException { + boolean bypass = false; + ObserverContext 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()) + .preGetForDeleteVersion(ctx, mutation, kv, + byteNow, family, get); + } catch (Throwable e) { + handleCoprocessorThrowable(env, e); + } finally { + currentThread.setContextClassLoader(cl); + } + bypass |= ctx.shouldBypass(); + if (ctx.shouldComplete()) { + break; + } + } + } + return bypass; + } + + /** * @param put The Put object * @param edit The WALEdit object. * @param durability The durability used @@ -1356,6 +1395,46 @@ public class RegionCoprocessorHost * @param compareOp the comparison operation * @param comparator the comparator * @param put data to put if check succeeds + * @return true or false to return to client if default processing should + * be bypassed, or null otherwise + * @throws IOException e + */ + public Boolean preCheckAndPutAfterRowLock(final byte[] row, final byte[] family, + final byte[] qualifier, final CompareOp compareOp, final ByteArrayComparable comparator, + final Put put) throws IOException { + boolean bypass = false; + boolean result = false; + ObserverContext 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()); + result = ((RegionObserver) env.getInstance()).preCheckAndPutAfterRowLock(ctx, row, + family, qualifier, compareOp, comparator, put, result); + } catch (Throwable e) { + handleCoprocessorThrowable(env, e); + } finally { + currentThread.setContextClassLoader(cl); + } + bypass |= ctx.shouldBypass(); + if (ctx.shouldComplete()) { + break; + } + } + } + return bypass ? result : null; + } + + /** + * @param row row to check + * @param family column family + * @param qualifier column qualifier + * @param compareOp the comparison operation + * @param comparator the comparator + * @param put data to put if check succeeds * @throws IOException e */ public boolean postCheckAndPut(final byte [] row, final byte [] family, @@ -1434,6 +1513,46 @@ public class RegionCoprocessorHost * @param compareOp the comparison operation * @param comparator the comparator * @param delete delete to commit if check succeeds + * @return true or false to return to client if default processing should + * be bypassed, or null otherwise + * @throws IOException e + */ + public Boolean preCheckAndDeleteAfterRowLock(final byte[] row, final byte[] family, + final byte[] qualifier, final CompareOp compareOp, final ByteArrayComparable comparator, + final Delete delete) throws IOException { + boolean bypass = false; + boolean result = false; + ObserverContext 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()); + result = ((RegionObserver) env.getInstance()).preCheckAndDeleteAfterRowLock(ctx, row, + family, qualifier, compareOp, comparator, delete, result); + } catch (Throwable e) { + handleCoprocessorThrowable(env, e); + } finally { + currentThread.setContextClassLoader(cl); + } + bypass |= ctx.shouldBypass(); + if (ctx.shouldComplete()) { + break; + } + } + } + return bypass ? result : null; + } + + /** + * @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, @@ -1496,6 +1615,38 @@ public class RegionCoprocessorHost } /** + * @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 preAppendAfterRowLock(final Append append) throws IOException { + boolean bypass = false; + Result result = null; + ObserverContext 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()); + result = ((RegionObserver) env.getInstance()).preAppendAfterRowLock(ctx, append); + } catch (Throwable e) { + handleCoprocessorThrowable(env, e); + } finally { + currentThread.setContextClassLoader(cl); + } + bypass |= ctx.shouldBypass(); + if (ctx.shouldComplete()) { + break; + } + } + } + return bypass ? result : null; + } + + /** * @param increment increment object * @return result to return to client if default operation should be * bypassed, null otherwise @@ -1528,6 +1679,38 @@ public class RegionCoprocessorHost } /** + * @param increment increment 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 preIncrementAfterRowLock(final Increment increment) throws IOException { + boolean bypass = false; + Result result = null; + ObserverContext 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()); + result = ((RegionObserver) env.getInstance()).preIncrementAfterRowLock(ctx, increment); + } catch (Throwable e) { + handleCoprocessorThrowable(env, e); + } finally { + currentThread.setContextClassLoader(cl); + } + bypass |= ctx.shouldBypass(); + if (ctx.shouldComplete()) { + break; + } + } + } + return bypass ? result : null; + } + + /** * @param append Append object * @param result the result returned by the append * @throws IOException if an error occurred on the coprocessor