You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by ra...@apache.org on 2013/12/15 03:02:30 UTC
svn commit: r1550990 - in /hbase/trunk/hbase-server/src:
main/java/org/apache/hadoop/hbase/client/
main/java/org/apache/hadoop/hbase/coprocessor/
main/java/org/apache/hadoop/hbase/regionserver/
test/java/org/apache/hadoop/hbase/coprocessor/
Author: rajeshbabu
Date: Sun Dec 15 02:02:29 2013
New Revision: 1550990
URL: http://svn.apache.org/r1550990
Log:
HBASE-9261 Add cp hooks after {start|close}RegionOperation
Modified:
hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/client/ClientSideRegionScanner.java
hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/BaseRegionObserver.java
hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/RegionObserver.java
hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RegionCoprocessorHost.java
hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/coprocessor/SimpleRegionObserver.java
hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/coprocessor/TestRegionObserverInterface.java
Modified: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/client/ClientSideRegionScanner.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/client/ClientSideRegionScanner.java?rev=1550990&r1=1550989&r2=1550990&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/client/ClientSideRegionScanner.java (original)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/client/ClientSideRegionScanner.java Sun Dec 15 02:02:29 2013
@@ -104,8 +104,8 @@ public class ClientSideRegionScanner ext
}
}
if (this.region != null) {
- this.region.closeRegionOperation();
try {
+ this.region.closeRegionOperation();
this.region.close(true);
this.region = null;
} catch (IOException ex) {
Modified: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/BaseRegionObserver.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/BaseRegionObserver.java?rev=1550990&r1=1550989&r2=1550990&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/BaseRegionObserver.java (original)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/BaseRegionObserver.java Sun Dec 15 02:02:29 2013
@@ -45,6 +45,7 @@ import org.apache.hadoop.hbase.io.FSData
import org.apache.hadoop.hbase.io.Reference;
import org.apache.hadoop.hbase.io.hfile.CacheConfig;
import org.apache.hadoop.hbase.regionserver.HRegion;
+import org.apache.hadoop.hbase.regionserver.HRegion.Operation;
import org.apache.hadoop.hbase.regionserver.InternalScanner;
import org.apache.hadoop.hbase.regionserver.KeyValueScanner;
import org.apache.hadoop.hbase.regionserver.MiniBatchOperationInProgress;
@@ -322,6 +323,11 @@ public abstract class BaseRegionObserver
}
@Override
+ public void postBatchMutateIndispensably(final ObserverContext<RegionCoprocessorEnvironment> ctx,
+ MiniBatchOperationInProgress<Mutation> miniBatchOp, final boolean success) throws IOException {
+ }
+
+ @Override
public boolean preCheckAndPut(final ObserverContext<RegionCoprocessorEnvironment> e,
final byte [] row, final byte [] family, final byte [] qualifier,
final CompareOp compareOp, final ByteArrayComparable comparator,
@@ -481,4 +487,14 @@ public abstract class BaseRegionObserver
MutationType opType, Mutation mutation, Cell oldCell, Cell newCell) throws IOException {
return newCell;
}
+
+ @Override
+ public void postStartRegionOperation(final ObserverContext<RegionCoprocessorEnvironment> ctx,
+ Operation op) throws IOException {
+ }
+
+ @Override
+ public void postCloseRegionOperation(final ObserverContext<RegionCoprocessorEnvironment> ctx,
+ Operation op) throws IOException {
+ }
}
Modified: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/RegionObserver.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/RegionObserver.java?rev=1550990&r1=1550989&r2=1550990&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/RegionObserver.java (original)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/RegionObserver.java Sun Dec 15 02:02:29 2013
@@ -43,6 +43,7 @@ import org.apache.hadoop.hbase.io.FSData
import org.apache.hadoop.hbase.io.Reference;
import org.apache.hadoop.hbase.io.hfile.CacheConfig;
import org.apache.hadoop.hbase.regionserver.HRegion;
+import org.apache.hadoop.hbase.regionserver.HRegion.Operation;
import org.apache.hadoop.hbase.regionserver.InternalScanner;
import org.apache.hadoop.hbase.regionserver.KeyValueScanner;
import org.apache.hadoop.hbase.regionserver.MiniBatchOperationInProgress;
@@ -619,6 +620,36 @@ public interface RegionObserver extends
final MiniBatchOperationInProgress<Mutation> miniBatchOp) throws IOException;
/**
+ * This will be called for region operations where read lock is acquired in
+ * {@link HRegion#startRegionOperation()}.
+ * @param ctx
+ * @param operation The operation is about to be taken on the region
+ * @throws IOException
+ */
+ void postStartRegionOperation(final ObserverContext<RegionCoprocessorEnvironment> ctx,
+ Operation operation) throws IOException;
+
+ /**
+ * Called after releasing read lock in {@link HRegion#closeRegionOperation(Operation)}.
+ * @param ctx
+ * @param operation
+ * @throws IOException
+ */
+ void postCloseRegionOperation(final ObserverContext<RegionCoprocessorEnvironment> ctx,
+ Operation operation) throws IOException;
+
+ /**
+ * Called after the completion of batch put/delete and will be called even if the batch operation
+ * fails
+ * @param ctx
+ * @param miniBatchOp
+ * @param success true if batch operation is successful otherwise false.
+ * @throws IOException
+ */
+ void postBatchMutateIndispensably(final ObserverContext<RegionCoprocessorEnvironment> ctx,
+ MiniBatchOperationInProgress<Mutation> miniBatchOp, final boolean success) throws IOException;
+
+ /**
* Called before checkAndPut
* <p>
* Call CoprocessorEnvironment#bypass to skip default actions
Modified: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java?rev=1550990&r1=1550989&r2=1550990&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java (original)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java Sun Dec 15 02:02:29 2013
@@ -225,7 +225,7 @@ public class HRegion implements HeapSize
* operations have to be defined here. It's only needed when a special check is need in
* startRegionOperation
*/
- protected enum Operation {
+ public enum Operation {
ANY, GET, PUT, DELETE, SCAN, APPEND, INCREMENT, SPLIT_REGION, MERGE_REGION, BATCH_MUTATE,
REPLAY_BATCH_MUTATE, COMPACT_REGION
}
@@ -1738,7 +1738,7 @@ public class HRegion implements HeapSize
}
return result;
} finally {
- closeRegionOperation();
+ closeRegionOperation(Operation.GET);
}
}
@@ -1778,7 +1778,7 @@ public class HRegion implements HeapSize
}
return instantiateRegionScanner(scan, additionalScanners);
} finally {
- closeRegionOperation();
+ closeRegionOperation(Operation.SCAN);
}
}
@@ -1831,7 +1831,7 @@ public class HRegion implements HeapSize
// All edits for the given row (across all column families) must happen atomically.
doBatchMutate(delete);
} finally {
- closeRegionOperation();
+ closeRegionOperation(Operation.DELETE);
}
}
@@ -1927,7 +1927,7 @@ public class HRegion implements HeapSize
// All edits for the given row (across all column families) must happen atomically.
doBatchMutate(put);
} finally {
- closeRegionOperation();
+ closeRegionOperation(Operation.PUT);
}
}
@@ -2077,11 +2077,9 @@ public class HRegion implements HeapSize
checkResources();
long newSize;
- if (batchOp.isInReplay()) {
- startRegionOperation(Operation.REPLAY_BATCH_MUTATE);
- } else {
- startRegionOperation(Operation.BATCH_MUTATE);
- }
+ Operation op = Operation.BATCH_MUTATE;
+ if (batchOp.isInReplay()) op = Operation.REPLAY_BATCH_MUTATE;
+ startRegionOperation(op);
try {
if (!initialized) {
@@ -2094,7 +2092,7 @@ public class HRegion implements HeapSize
long addedSize = doMiniBatchMutation(batchOp);
newSize = this.addAndGetGlobalMemstoreSize(addedSize);
} finally {
- closeRegionOperation();
+ closeRegionOperation(op);
}
if (isFlushSize(newSize)) {
requestFlush();
@@ -2476,6 +2474,16 @@ public class HRegion implements HeapSize
}
}
}
+ if (coprocessorHost != null && !batchOp.isInReplay()) {
+ // call the coprocessor hook to do any finalization steps
+ // after the put is done
+ MiniBatchOperationInProgress<Mutation> miniBatchOp =
+ new MiniBatchOperationInProgress<Mutation>(batchOp.getMutationsForCoprocs(),
+ batchOp.retCodeDetails, batchOp.walEditsFromCoprocessors, firstIndex,
+ lastIndexExclusive);
+ coprocessorHost.postBatchMutateIndispensably(miniBatchOp, success);
+ }
+
batchOp.nextIndexToProcess = lastIndexExclusive;
}
}
@@ -3655,7 +3663,7 @@ public class HRegion implements HeapSize
try {
return nextRaw(outResults, limit);
} finally {
- closeRegionOperation();
+ closeRegionOperation(Operation.SCAN);
}
}
@@ -5015,7 +5023,7 @@ public class HRegion implements HeapSize
if (w != null) {
mvcc.completeMemstoreInsert(w);
}
- closeRegionOperation();
+ closeRegionOperation(Operation.APPEND);
}
if (this.metricsRegion != null) {
@@ -5193,7 +5201,7 @@ public class HRegion implements HeapSize
if (w != null) {
mvcc.completeMemstoreInsert(w);
}
- closeRegionOperation();
+ closeRegionOperation(Operation.INCREMENT);
if (this.metricsRegion != null) {
this.metricsRegion.updateIncrement();
}
@@ -5515,23 +5523,17 @@ public class HRegion implements HeapSize
* modifies data. It has to be called just before a try.
* #closeRegionOperation needs to be called in the try's finally block
* Acquires a read lock and checks if the region is closing or closed.
- * @throws NotServingRegionException when the region is closing or closed
- * @throws RegionTooBusyException if failed to get the lock in time
- * @throws InterruptedIOException if interrupted while waiting for a lock
+ * @throws IOException
*/
- public void startRegionOperation()
- throws NotServingRegionException, RegionTooBusyException, InterruptedIOException {
+ public void startRegionOperation() throws IOException {
startRegionOperation(Operation.ANY);
}
/**
* @param op The operation is about to be taken on the region
- * @throws NotServingRegionException
- * @throws RegionTooBusyException
- * @throws InterruptedIOException
+ * @throws IOException
*/
- protected void startRegionOperation(Operation op) throws NotServingRegionException,
- RegionTooBusyException, InterruptedIOException {
+ protected void startRegionOperation(Operation op) throws IOException {
switch (op) {
case INCREMENT:
case APPEND:
@@ -5566,14 +5568,36 @@ public class HRegion implements HeapSize
lock.readLock().unlock();
throw new NotServingRegionException(getRegionNameAsString() + " is closed");
}
+ try {
+ if (coprocessorHost != null) {
+ coprocessorHost.postStartRegionOperation(op);
+ }
+ } catch (Exception e) {
+ lock.readLock().unlock();
+ throw new IOException(e);
+ }
}
/**
* Closes the lock. This needs to be called in the finally block corresponding
* to the try block of #startRegionOperation
+ * @throws IOException
*/
- public void closeRegionOperation() {
+ public void closeRegionOperation() throws IOException {
+ closeRegionOperation(Operation.ANY);
+ }
+
+ /**
+ * Closes the lock. This needs to be called in the finally block corresponding
+ * to the try block of {@link #startRegionOperation(Operation)}
+ * @param operation
+ * @throws IOException
+ */
+ public void closeRegionOperation(Operation operation) throws IOException {
lock.readLock().unlock();
+ if (coprocessorHost != null) {
+ coprocessorHost.postCloseRegionOperation(operation);
+ }
}
/**
Modified: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RegionCoprocessorHost.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RegionCoprocessorHost.java?rev=1550990&r1=1550989&r2=1550990&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RegionCoprocessorHost.java (original)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RegionCoprocessorHost.java Sun Dec 15 02:02:29 2013
@@ -62,6 +62,7 @@ import org.apache.hadoop.hbase.io.FSData
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.io.Reference;
import org.apache.hadoop.hbase.io.hfile.CacheConfig;
+import org.apache.hadoop.hbase.regionserver.HRegion.Operation;
import org.apache.hadoop.hbase.regionserver.compactions.CompactionRequest;
import org.apache.hadoop.hbase.regionserver.wal.HLogKey;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
@@ -1077,6 +1078,26 @@ public class RegionCoprocessorHost
}
}
+ public void postBatchMutateIndispensably(
+ final MiniBatchOperationInProgress<Mutation> miniBatchOp, final boolean success)
+ throws IOException {
+ ObserverContext<RegionCoprocessorEnvironment> ctx = null;
+ for (RegionEnvironment env : coprocessors) {
+ if (env.getInstance() instanceof RegionObserver) {
+ ctx = ObserverContext.createAndPrepare(env, ctx);
+ try {
+ ((RegionObserver) env.getInstance()).postBatchMutateIndispensably(ctx, miniBatchOp,
+ success);
+ } catch (Throwable e) {
+ handleCoprocessorThrowable(env, e);
+ }
+ if (ctx.shouldComplete()) {
+ break;
+ }
+ }
+ }
+ }
+
/**
* @param row row to check
* @param family column family
@@ -1641,6 +1662,40 @@ public class RegionCoprocessorHost
return hasLoaded;
}
+ public void postStartRegionOperation(Operation op) throws IOException {
+ ObserverContext<RegionCoprocessorEnvironment> ctx = null;
+ for (RegionEnvironment env : coprocessors) {
+ if (env.getInstance() instanceof RegionObserver) {
+ ctx = ObserverContext.createAndPrepare(env, ctx);
+ try {
+ ((RegionObserver) env.getInstance()).postStartRegionOperation(ctx, op);
+ } catch (Throwable e) {
+ handleCoprocessorThrowable(env, e);
+ }
+ if (ctx.shouldComplete()) {
+ break;
+ }
+ }
+ }
+ }
+
+ public void postCloseRegionOperation(Operation op) throws IOException {
+ ObserverContext<RegionCoprocessorEnvironment> ctx = null;
+ for (RegionEnvironment env : coprocessors) {
+ if (env.getInstance() instanceof RegionObserver) {
+ ctx = ObserverContext.createAndPrepare(env, ctx);
+ try {
+ ((RegionObserver) env.getInstance()).postCloseRegionOperation(ctx, op);
+ } catch (Throwable e) {
+ handleCoprocessorThrowable(env, e);
+ }
+ if (ctx.shouldComplete()) {
+ break;
+ }
+ }
+ }
+ }
+
/**
* @param fs fileystem to read from
* @param p path to the file
Modified: hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/coprocessor/SimpleRegionObserver.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/coprocessor/SimpleRegionObserver.java?rev=1550990&r1=1550989&r2=1550990&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/coprocessor/SimpleRegionObserver.java (original)
+++ hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/coprocessor/SimpleRegionObserver.java Sun Dec 15 02:02:29 2013
@@ -52,6 +52,7 @@ import org.apache.hadoop.hbase.io.FSData
import org.apache.hadoop.hbase.io.Reference;
import org.apache.hadoop.hbase.io.hfile.CacheConfig;
import org.apache.hadoop.hbase.regionserver.HRegion;
+import org.apache.hadoop.hbase.regionserver.HRegion.Operation;
import org.apache.hadoop.hbase.regionserver.InternalScanner;
import org.apache.hadoop.hbase.regionserver.KeyValueScanner;
import org.apache.hadoop.hbase.regionserver.Leases;
@@ -119,6 +120,9 @@ public class SimpleRegionObserver extend
final AtomicInteger ctPreSplitAfterPONR = new AtomicInteger(0);
final AtomicInteger ctPreStoreFileReaderOpen = new AtomicInteger(0);
final AtomicInteger ctPostStoreFileReaderOpen = new AtomicInteger(0);
+ final AtomicInteger ctPostBatchMutateIndispensably = new AtomicInteger(0);
+ final AtomicInteger ctPostStartRegionOperation = new AtomicInteger(0);
+ final AtomicInteger ctPostCloseRegionOperation = new AtomicInteger(0);
final AtomicBoolean throwOnPostFlush = new AtomicBoolean(false);
static final String TABLE_SKIPPED = "SKIPPED_BY_PREWALRESTORE";
@@ -464,6 +468,26 @@ public class SimpleRegionObserver extend
}
@Override
+ public void postStartRegionOperation(final ObserverContext<RegionCoprocessorEnvironment> ctx,
+ Operation op) throws IOException {
+ ctPostStartRegionOperation.incrementAndGet();
+ }
+
+ @Override
+ public void postCloseRegionOperation(final ObserverContext<RegionCoprocessorEnvironment> ctx,
+ Operation op) throws IOException {
+ if (ctPostStartRegionOperation.get() > 0) {
+ ctPostCloseRegionOperation.incrementAndGet();
+ }
+ }
+
+ @Override
+ public void postBatchMutateIndispensably(final ObserverContext<RegionCoprocessorEnvironment> ctx,
+ MiniBatchOperationInProgress<Mutation> miniBatchOp, final boolean success) throws IOException {
+ ctPostBatchMutateIndispensably.incrementAndGet();
+ }
+
+ @Override
public void preGetClosestRowBefore(final ObserverContext<RegionCoprocessorEnvironment> c,
final byte[] row, final byte[] family, final Result result)
throws IOException {
@@ -597,11 +621,31 @@ public class SimpleRegionObserver extend
public boolean hadPostBatchMutate() {
return ctPostBatchMutate.get() > 0;
}
-
+
+ public boolean hadPostBatchMutateIndispensably() {
+ return ctPostBatchMutateIndispensably.get() > 0;
+ }
+
+ public boolean hadPostStartRegionOperation() {
+ return ctPostStartRegionOperation.get() > 0;
+ }
+
+ public boolean hadPostCloseRegionOperation() {
+ return ctPostCloseRegionOperation.get() > 0;
+ }
+
public boolean hadDelete() {
return !(ctBeforeDelete.get() > 0);
}
+ public int getCtPostStartRegionOperation() {
+ return ctPostStartRegionOperation.get();
+ }
+
+ public int getCtPostCloseRegionOperation() {
+ return ctPostCloseRegionOperation.get();
+ }
+
public boolean hadPreIncrement() {
return ctPreIncrement.get() > 0;
}
Modified: hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/coprocessor/TestRegionObserverInterface.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/coprocessor/TestRegionObserverInterface.java?rev=1550990&r1=1550989&r2=1550990&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/coprocessor/TestRegionObserverInterface.java (original)
+++ hbase/trunk/hbase-server/src/test/java/org/apache/hadoop/hbase/coprocessor/TestRegionObserverInterface.java Sun Dec 15 02:02:29 2013
@@ -115,11 +115,10 @@ public class TestRegionObserverInterface
// coprocessor.
HTable table = util.createTable(tableName, new byte[][] {A, B, C});
try {
- verifyMethodResult(SimpleRegionObserver.class,
- new String[] {"hadPreGet", "hadPostGet", "hadPrePut", "hadPostPut",
- "hadDelete"},
- tableName,
- new Boolean[] {false, false, false, false, false});
+ verifyMethodResult(SimpleRegionObserver.class, new String[] { "hadPreGet", "hadPostGet",
+ "hadPrePut", "hadPostPut", "hadDelete", "hadPostStartRegionOperation",
+ "hadPostCloseRegionOperation", "hadPostBatchMutateIndispensably" }, tableName,
+ new Boolean[] { false, false, false, false, false, false, false, false });
Put put = new Put(ROW);
put.add(A, A, A);
@@ -127,12 +126,11 @@ public class TestRegionObserverInterface
put.add(C, C, C);
table.put(put);
- verifyMethodResult(SimpleRegionObserver.class,
- new String[] {"hadPreGet", "hadPostGet", "hadPrePut", "hadPostPut",
- "hadPreBatchMutate", "hadPostBatchMutate", "hadDelete"},
- tableName,
- new Boolean[] {false, false, true, true, true, true, false}
- );
+ verifyMethodResult(SimpleRegionObserver.class, new String[] { "hadPreGet", "hadPostGet",
+ "hadPrePut", "hadPostPut", "hadPreBatchMutate", "hadPostBatchMutate", "hadDelete",
+ "hadPostStartRegionOperation", "hadPostCloseRegionOperation",
+ "hadPostBatchMutateIndispensably" }, TEST_TABLE, new Boolean[] { false, false, true,
+ true, true, true, false, true, true, true });
verifyMethodResult(SimpleRegionObserver.class,
new String[] {"getCtPreOpen", "getCtPostOpen", "getCtPreClose", "getCtPostClose"},