Apache
/
hbase
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

HBASE-11126-Add RegionObserver pre hooks that operate under row lock (Ram)

This commit is contained in:
Ramkrishna 2014-06-03 17:54:59 +05:30
parent 369141b795
commit 6a2467bbf2
7 changed files with 644 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/BaseRegionObserver.java
View File

@ -310,6 +310,12 @@ public abstract class BaseRegionObserver implements RegionObserver {
final WALEdit edit, final Durability durability) throws IOException {
}
@Override
public void prePrepareTimeStampForDeleteVersion(
final ObserverContext<RegionCoprocessorEnvironment> e, final Mutation delete,
final Cell cell, final byte[] byteNow, final Get get) throws IOException {
}
@Override
public void postDelete(final ObserverContext<RegionCoprocessorEnvironment> e,
final Delete delete, final WALEdit edit, final Durability durability)
@ -339,6 +345,15 @@ public abstract class BaseRegionObserver implements RegionObserver {
return result;
}
@Override
public boolean preCheckAndPutAfterRowLock(
final ObserverContext<RegionCoprocessorEnvironment> 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<RegionCoprocessorEnvironment> e,
final byte [] row, final byte [] family, final byte [] qualifier,
@ -355,6 +370,15 @@ public abstract class BaseRegionObserver implements RegionObserver {
return result;
}
@Override
public boolean preCheckAndDeleteAfterRowLock(
final ObserverContext<RegionCoprocessorEnvironment> 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<RegionCoprocessorEnvironment> e,
final byte [] row, final byte [] family, final byte [] qualifier,
@ -369,6 +393,12 @@ public abstract class BaseRegionObserver implements RegionObserver {
return null;
}
@Override
public Result preAppendAfterRowLock(final ObserverContext<RegionCoprocessorEnvironment> e,
final Append append) throws IOException {
return null;
}
@Override
public Result postAppend(final ObserverContext<RegionCoprocessorEnvironment> e,
final Append append, final Result result) throws IOException {
@ -396,6 +426,12 @@ public abstract class BaseRegionObserver implements RegionObserver {
return null;
}
@Override
public Result preIncrementAfterRowLock(final ObserverContext<RegionCoprocessorEnvironment> e,
final Increment increment) throws IOException {
return null;
}
@Override
public Result postIncrement(final ObserverContext<RegionCoprocessorEnvironment> e,
final Increment increment, final Result result) throws IOException {

122
hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/RegionObserver.java
View File

@ -587,6 +587,24 @@ public interface RegionObserver extends Coprocessor {
void preDelete(final ObserverContext<RegionCoprocessorEnvironment> c,
final Delete delete, final WALEdit edit, final Durability durability)
throws IOException;
/**
* Called before the server updates the timestamp for version delete with latest timestamp.
* <p>
* Call CoprocessorEnvironment#bypass to skip default actions
* <p>
* 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 get - the get formed using the current cell's row.
* Note that the get does not specify the family and qualifier
* @throws IOException
*/
void prePrepareTimeStampForDeleteVersion(final ObserverContext<RegionCoprocessorEnvironment> c,
final Mutation mutation, final Cell cell, final byte[] byteNow,
final Get get) throws IOException;
/**
* Called after the client deletes a value.
@ -657,7 +675,7 @@ public interface RegionObserver extends Coprocessor {
MiniBatchOperationInProgress<Mutation> miniBatchOp, final boolean success) throws IOException;
/**
* Called before checkAndPut
* Called before checkAndPut.
* <p>
* Call CoprocessorEnvironment#bypass to skip default actions
* <p>
@ -681,6 +699,34 @@ public interface RegionObserver extends Coprocessor {
final Put put, final boolean result)
throws IOException;
/**
* Called before checkAndPut but after acquiring rowlock.
* <p>
* <b>Note:</b> 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.
* <p>
* Call CoprocessorEnvironment#bypass to skip default actions
* <p>
* 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<RegionCoprocessorEnvironment> 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
* <p>
@ -704,7 +750,7 @@ public interface RegionObserver extends Coprocessor {
throws IOException;
/**
* Called before checkAndDelete
* Called before checkAndDelete.
* <p>
* Call CoprocessorEnvironment#bypass to skip default actions
* <p>
@ -727,6 +773,33 @@ public interface RegionObserver extends Coprocessor {
final Delete delete, final boolean result)
throws IOException;
/**
* Called before checkAndDelete but after acquiring rowock.
* <p>
* <b>Note:</b> 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.
* <p>
* Call CoprocessorEnvironment#bypass to skip default actions
* <p>
* 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<RegionCoprocessorEnvironment> 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
* <p>
@ -795,7 +868,7 @@ public interface RegionObserver extends Coprocessor {
throws IOException;
/**
* Called before Append
* Called before Append.
* <p>
* Call CoprocessorEnvironment#bypass to skip default actions
* <p>
@ -810,6 +883,25 @@ public interface RegionObserver extends Coprocessor {
final Append append)
throws IOException;
/**
* Called before Append but after acquiring rowlock.
* <p>
* <b>Note:</b> 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.
* <p>
* Call CoprocessorEnvironment#bypass to skip default actions
* <p>
* 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<RegionCoprocessorEnvironment> c,
final Append append) throws IOException;
/**
* Called after Append
* <p>
@ -826,7 +918,7 @@ public interface RegionObserver extends Coprocessor {
throws IOException;
/**
* Called before Increment
* Called before Increment.
* <p>
* Call CoprocessorEnvironment#bypass to skip default actions
* <p>
@ -841,6 +933,28 @@ public interface RegionObserver extends Coprocessor {
final Increment increment)
throws IOException;
/**
* Called before Increment but after acquiring rowlock.
* <p>
* <b>Note:</b> 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.
* <p>
* Call CoprocessorEnvironment#bypass to skip default actions
* <p>
* 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<RegionCoprocessorEnvironment> c,
final Increment increment) throws IOException;
/**
* Called after increment
* <p>

96
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
View File

@ -2030,10 +2030,13 @@ public class HRegion implements HeapSize { // , Writable{
/**
* Setup correct timestamps in the KVs in Delete object.
* Caller should have the row and region locks.
* @param mutation
* @param familyMap
* @param byteNow
* @throws IOException
*/
void prepareDeleteTimestamps(Map<byte[], List<Cell>> familyMap, byte[] byteNow)
throws IOException {
void prepareDeleteTimestamps(Mutation mutation, Map<byte[], List<Cell>> familyMap,
byte[] byteNow) throws IOException {
for (Map.Entry<byte[], List<Cell>> e : familyMap.entrySet()) {
byte[] family = e.getKey();
@ -2059,20 +2062,14 @@ public class HRegion implements HeapSize { // , Writable{
Get get = new Get(CellUtil.cloneRow(kv));
get.setMaxVersions(count);
get.addColumn(family, qual);
List<Cell> result = get(get, false);
if (result.size() < count) {
// Nothing to delete
kv.updateLatestStamp(byteNow);
continue;
if (coprocessorHost != null) {
if (!coprocessorHost.prePrepareTimeStampForDeleteVersion(mutation, cell, byteNow,
get)) {
updateDeleteLatestVersionTimeStamp(kv, get, count, byteNow);
}
} else {
updateDeleteLatestVersionTimeStamp(kv, get, count, byteNow);
}
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);
} else {
kv.updateLatestStamp(byteNow);
}
@ -2080,6 +2077,23 @@ public class HRegion implements HeapSize { // , Writable{
}
}
void updateDeleteLatestVersionTimeStamp(KeyValue kv, Get get, int count, byte[] byteNow)
throws IOException {
List<Cell> 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 +2466,9 @@ public class HRegion implements HeapSize { // , Writable{
updateKVTimestamps(familyMaps[i].values(), byteNow);
noOfPuts++;
} else {
prepareDeleteTimestamps(familyMaps[i], byteNow);
if (!isInReplay) {
prepareDeleteTimestamps(mutation, familyMaps[i], byteNow);
}
noOfDeletes++;
}
}
@ -2712,9 +2728,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<Cell> 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<Cell> result = get(get, false);
boolean valueIsNull = comparator.getValue() == null ||
comparator.getValue().length == 0;
@ -5033,12 +5061,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<byte[], List<Cell>> family : append.getFamilyCellMap().entrySet()) {
@ -5221,12 +5255,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<byte [], List<Cell>> family:

4
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/MultiRowMutationProcessor.java
View File

@ -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;
@ -80,7 +80,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: "

182
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RegionCoprocessorHost.java
View File

@ -1134,6 +1134,44 @@ public class RegionCoprocessorHost
return bypass;
}
/**
* @param mutation - the current mutation
* @param kv - the current cell
* @param byteNow - current timestamp in bytes
* @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 prePrepareTimeStampForDeleteVersion(Mutation mutation,
Cell kv, byte[] byteNow, Get get) 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())
.prePrepareTimeStampForDeleteVersion(ctx, mutation, kv,
byteNow, 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.
@ -1349,6 +1387,46 @@ public class RegionCoprocessorHost
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
* @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<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());
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
@ -1427,6 +1505,46 @@ public class RegionCoprocessorHost
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
* @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<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());
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
@ -1495,6 +1613,38 @@ public class RegionCoprocessorHost
return bypass ? result : null;
}
/**
* @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<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());
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
@ -1527,6 +1677,38 @@ public class RegionCoprocessorHost
return bypass ? result : null;
}
/**
* @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<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());
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

140
hbase-server/src/test/java/org/apache/hadoop/hbase/coprocessor/SimpleRegionObserver.java
View File

@ -40,6 +40,7 @@ import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.CoprocessorEnvironment;
import org.apache.hadoop.hbase.HRegionInfo;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Append;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.client.Get;
@ -48,6 +49,8 @@ import org.apache.hadoop.hbase.client.Mutation;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.ByteArrayComparable;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.io.FSDataInputStreamWrapper;
import org.apache.hadoop.hbase.io.Reference;
import org.apache.hadoop.hbase.io.hfile.CacheConfig;
@ -96,11 +99,22 @@ public class SimpleRegionObserver extends BaseRegionObserver {
final AtomicInteger ctPrePut = new AtomicInteger(0);
final AtomicInteger ctPostPut = new AtomicInteger(0);
final AtomicInteger ctPreDeleted = new AtomicInteger(0);
final AtomicInteger ctPrePrepareDeleteTS = new AtomicInteger(0);
final AtomicInteger ctPostDeleted = new AtomicInteger(0);
final AtomicInteger ctPreGetClosestRowBefore = new AtomicInteger(0);
final AtomicInteger ctPostGetClosestRowBefore = new AtomicInteger(0);
final AtomicInteger ctPreIncrement = new AtomicInteger(0);
final AtomicInteger ctPreIncrementAfterRowLock = new AtomicInteger(0);
final AtomicInteger ctPreAppend = new AtomicInteger(0);
final AtomicInteger ctPreAppendAfterRowLock = new AtomicInteger(0);
final AtomicInteger ctPostIncrement = new AtomicInteger(0);
final AtomicInteger ctPostAppend = new AtomicInteger(0);
final AtomicInteger ctPreCheckAndPut = new AtomicInteger(0);
final AtomicInteger ctPreCheckAndPutAfterRowLock = new AtomicInteger(0);
final AtomicInteger ctPostCheckAndPut = new AtomicInteger(0);
final AtomicInteger ctPreCheckAndDelete = new AtomicInteger(0);
final AtomicInteger ctPreCheckAndDeleteAfterRowLock = new AtomicInteger(0);
final AtomicInteger ctPostCheckAndDelete = new AtomicInteger(0);
final AtomicInteger ctPreWALRestored = new AtomicInteger(0);
final AtomicInteger ctPostWALRestored = new AtomicInteger(0);
final AtomicInteger ctPreScannerNext = new AtomicInteger(0);
@ -434,6 +448,12 @@ public class SimpleRegionObserver extends BaseRegionObserver {
}
}
@Override
public void prePrepareTimeStampForDeleteVersion(ObserverContext<RegionCoprocessorEnvironment> e,
Mutation delete, Cell cell, byte[] byteNow, Get get) throws IOException {
ctPrePrepareDeleteTS.incrementAndGet();
}
@Override
public void postDelete(final ObserverContext<RegionCoprocessorEnvironment> c,
final Delete delete, final WALEdit edit,
@ -520,6 +540,13 @@ public class SimpleRegionObserver extends BaseRegionObserver {
return null;
}
@Override
public Result preIncrementAfterRowLock(ObserverContext<RegionCoprocessorEnvironment> e,
Increment increment) throws IOException {
ctPreIncrementAfterRowLock.incrementAndGet();
return null;
}
@Override
public Result postIncrement(final ObserverContext<RegionCoprocessorEnvironment> c,
final Increment increment, final Result result) throws IOException {
@ -527,6 +554,75 @@ public class SimpleRegionObserver extends BaseRegionObserver {
return result;
}
@Override
public boolean preCheckAndPut(ObserverContext<RegionCoprocessorEnvironment> e, byte[] row,
byte[] family, byte[] qualifier, CompareOp compareOp, ByteArrayComparable comparator,
Put put, boolean result) throws IOException {
ctPreCheckAndPut.incrementAndGet();
return true;
}
@Override
public boolean preCheckAndPutAfterRowLock(ObserverContext<RegionCoprocessorEnvironment> e,
byte[] row, byte[] family, byte[] qualifier, CompareOp compareOp,
ByteArrayComparable comparator, Put put, boolean result) throws IOException {
ctPreCheckAndPutAfterRowLock.incrementAndGet();
return true;
}
@Override
public boolean postCheckAndPut(ObserverContext<RegionCoprocessorEnvironment> e, byte[] row,
byte[] family, byte[] qualifier, CompareOp compareOp, ByteArrayComparable comparator,
Put put, boolean result) throws IOException {
ctPostCheckAndPut.incrementAndGet();
return true;
}
@Override
public boolean preCheckAndDelete(ObserverContext<RegionCoprocessorEnvironment> e, byte[] row,
byte[] family, byte[] qualifier, CompareOp compareOp, ByteArrayComparable comparator,
Delete delete, boolean result) throws IOException {
ctPreCheckAndDelete.incrementAndGet();
return true;
}
@Override
public boolean preCheckAndDeleteAfterRowLock(ObserverContext<RegionCoprocessorEnvironment> e,
byte[] row, byte[] family, byte[] qualifier, CompareOp compareOp,
ByteArrayComparable comparator, Delete delete, boolean result) throws IOException {
ctPreCheckAndDeleteAfterRowLock.incrementAndGet();
return true;
}
@Override
public boolean postCheckAndDelete(ObserverContext<RegionCoprocessorEnvironment> e, byte[] row,
byte[] family, byte[] qualifier, CompareOp compareOp, ByteArrayComparable comparator,
Delete delete, boolean result) throws IOException {
ctPostCheckAndDelete.incrementAndGet();
return true;
}
@Override
public Result preAppendAfterRowLock(ObserverContext<RegionCoprocessorEnvironment> e,
Append append) throws IOException {
ctPreAppendAfterRowLock.incrementAndGet();
return null;
}
@Override
public Result preAppend(ObserverContext<RegionCoprocessorEnvironment> e, Append append)
throws IOException {
ctPreAppend.incrementAndGet();
return null;
}
@Override
public Result postAppend(ObserverContext<RegionCoprocessorEnvironment> e, Append append,
Result result) throws IOException {
ctPostAppend.incrementAndGet();
return null;
}
@Override
public void preBulkLoadHFile(ObserverContext<RegionCoprocessorEnvironment> ctx,
List<Pair<byte[], String>> familyPaths) throws IOException {
@ -646,14 +742,58 @@ public class SimpleRegionObserver extends BaseRegionObserver {
return ctPostCloseRegionOperation.get();
}
public boolean hadPreCheckAndPut() {
return ctPreCheckAndPut.get() > 0;
}
public boolean hadPreCheckAndPutAfterRowLock() {
return ctPreCheckAndPutAfterRowLock.get() > 0;
}
public boolean hadPostCheckAndPut() {
return ctPostCheckAndPut.get() > 0;
}
public boolean hadPreCheckAndDelete() {
return ctPreCheckAndDelete.get() > 0;
}
public boolean hadPreCheckAndDeleteAfterRowLock() {
return ctPreCheckAndDeleteAfterRowLock.get() > 0;
}
public boolean hadPostCheckAndDelete() {
return ctPostCheckAndDelete.get() > 0;
}
public boolean hadPreIncrement() {
return ctPreIncrement.get() > 0;
}
public boolean hadPreIncrementAfterRowLock() {
return ctPreIncrementAfterRowLock.get() > 0;
}
public boolean hadPostIncrement() {
return ctPostIncrement.get() > 0;
}
public boolean hadPreAppend() {
return ctPreAppend.get() > 0;
}
public boolean hadPreAppendAfterRowLock() {
return ctPreAppendAfterRowLock.get() > 0;
}
public boolean hadPostAppend() {
return ctPostAppend.get() > 0;
}
public boolean hadPrePreparedDeleteTS() {
return ctPrePrepareDeleteTS.get() > 0;
}
public boolean hadPreWALRestored() {
return ctPreWALRestored.get() > 0;
}

106
hbase-server/src/test/java/org/apache/hadoop/hbase/coprocessor/TestRegionObserverInterface.java
View File

@ -46,6 +46,7 @@ import org.apache.hadoop.hbase.MediumTests;
import org.apache.hadoop.hbase.MiniHBaseCluster;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Append;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.client.Get;
@ -145,9 +146,9 @@ public class TestRegionObserverInterface {
verifyMethodResult(SimpleRegionObserver.class,
new String[] {"hadPreGet", "hadPostGet", "hadPrePut", "hadPostPut",
"hadDelete"},
"hadDelete", "hadPrePreparedDeleteTS"},
tableName,
new Boolean[] {true, true, true, true, false}
new Boolean[] {true, true, true, true, false, false}
);
Delete delete = new Delete(ROW);
@ -158,9 +159,9 @@ public class TestRegionObserverInterface {
verifyMethodResult(SimpleRegionObserver.class,
new String[] {"hadPreGet", "hadPostGet", "hadPrePut", "hadPostPut",
"hadPreBatchMutate", "hadPostBatchMutate", "hadDelete"},
"hadPreBatchMutate", "hadPostBatchMutate", "hadDelete", "hadPrePreparedDeleteTS"},
tableName,
new Boolean[] {true, true, true, true, true, true, true}
new Boolean[] {true, true, true, true, true, true, true, true}
);
} finally {
util.deleteTable(tableName);
@ -218,17 +219,106 @@ public class TestRegionObserverInterface {
inc.addColumn(A, A, 1);
verifyMethodResult(SimpleRegionObserver.class,
new String[] {"hadPreIncrement", "hadPostIncrement"},
new String[] {"hadPreIncrement", "hadPostIncrement", "hadPreIncrementAfterRowLock"},
tableName,
new Boolean[] {false, false}
new Boolean[] {false, false, false}
);
table.increment(inc);
verifyMethodResult(SimpleRegionObserver.class,
new String[] {"hadPreIncrement", "hadPostIncrement"},
new String[] {"hadPreIncrement", "hadPostIncrement", "hadPreIncrementAfterRowLock"},
tableName,
new Boolean[] {true, true}
new Boolean[] {true, true, true}
);
} finally {
util.deleteTable(tableName);
table.close();
}
}
@Test
public void testCheckAndPutHooks() throws IOException {
TableName tableName =
TableName.valueOf(TEST_TABLE.getNameAsString() + ".testCheckAndPutHooks");
HTable table = util.createTable(tableName, new byte[][] {A, B, C});
try {
Put p = new Put(Bytes.toBytes(0));
p.add(A, A, A);
table.put(p);
table.flushCommits();
p = new Put(Bytes.toBytes(0));
p.add(A, A, A);
verifyMethodResult(SimpleRegionObserver.class,
new String[] {"hadPreCheckAndPut",
"hadPreCheckAndPutAfterRowLock", "hadPostCheckAndPut"},
tableName,
new Boolean[] {false, false, false}
);
table.checkAndPut(Bytes.toBytes(0), A, A, A, p);
verifyMethodResult(SimpleRegionObserver.class,
new String[] {"hadPreCheckAndPut",
"hadPreCheckAndPutAfterRowLock", "hadPostCheckAndPut"},
tableName,
new Boolean[] {true, true, true}
);
} finally {
util.deleteTable(tableName);
table.close();
}
}
@Test
public void testCheckAndDeleteHooks() throws IOException {
TableName tableName =
TableName.valueOf(TEST_TABLE.getNameAsString() + ".testCheckAndDeleteHooks");
HTable table = util.createTable(tableName, new byte[][] {A, B, C});
try {
Put p = new Put(Bytes.toBytes(0));
p.add(A, A, A);
table.put(p);
table.flushCommits();
Delete d = new Delete(Bytes.toBytes(0));
table.delete(d);
verifyMethodResult(SimpleRegionObserver.class,
new String[] {"hadPreCheckAndDelete",
"hadPreCheckAndDeleteAfterRowLock", "hadPostCheckAndDelete"},
tableName,
new Boolean[] {false, false, false}
);
table.checkAndDelete(Bytes.toBytes(0), A, A, A, d);
verifyMethodResult(SimpleRegionObserver.class,
new String[] {"hadPreCheckAndDelete",
"hadPreCheckAndDeleteAfterRowLock", "hadPostCheckAndDelete"},
tableName,
new Boolean[] {true, true, true}
);
} finally {
util.deleteTable(tableName);
table.close();
}
}
@Test
public void testAppendHook() throws IOException {
TableName tableName = TableName.valueOf(TEST_TABLE.getNameAsString() + ".testAppendHook");
HTable table = util.createTable(tableName, new byte[][] {A, B, C});
try {
Append app = new Append(Bytes.toBytes(0));
app.add(A, A, A);
verifyMethodResult(SimpleRegionObserver.class,
new String[] {"hadPreAppend", "hadPostAppend", "hadPreAppendAfterRowLock"},
tableName,
new Boolean[] {false, false, false}
);
table.append(app);
verifyMethodResult(SimpleRegionObserver.class,
new String[] {"hadPreAppend", "hadPostAppend", "hadPreAppendAfterRowLock"},
tableName,
new Boolean[] {true, true, true}
);
} finally {
util.deleteTable(tableName);