hbase-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From st...@apache.org
Subject hbase git commit: HBASE-18960 A few bug fixes and minor improvements around batchMutate
Date Tue, 17 Oct 2017 20:57:10 GMT
Repository: hbase
Updated Branches:
  refs/heads/master 70f4c5da4 -> e1941aa6d


HBASE-18960 A few bug fixes and minor improvements around batchMutate

* batch validation and preparation is done before we start iterating over operations for writes
* durability, familyCellMaps and observedExceptions are batch wide and are now sotred in BatchOperation,
  as a result durability is consistent across all operations in a batch
* for all operations done by preBatchMutate() CP hook, operation status is updated to success
* doWALAppend() is modified to habdle replay and is used from doMiniBatchMutate()
* minor improvements

Signed-off-by: Michael Stack <stack@apache.org>


Project: http://git-wip-us.apache.org/repos/asf/hbase/repo
Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/e1941aa6
Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/e1941aa6
Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/e1941aa6

Branch: refs/heads/master
Commit: e1941aa6d14afd116a555fc93a3149f3e7c20af2
Parents: 70f4c5d
Author: Umesh Agashe <uagashe@cloudera.com>
Authored: Fri Oct 6 15:40:05 2017 -0700
Committer: Michael Stack <stack@apache.org>
Committed: Tue Oct 17 13:57:00 2017 -0700

----------------------------------------------------------------------
 .../hadoop/hbase/regionserver/HRegion.java      | 420 ++++++++-----------
 .../regionserver/TestHRegionReplayEvents.java   |  21 +
 2 files changed, 207 insertions(+), 234 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hbase/blob/e1941aa6/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
index 0bef925..1cbb689 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegion.java
@@ -661,7 +661,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver,
Regi
 
   private final MetricsRegion metricsRegion;
   private final MetricsRegionWrapperImpl metricsRegionWrapper;
-  private final Durability durability;
+  private final Durability regionDurability;
   private final boolean regionStatsEnabled;
   // Stores the replication scope of the various column families of the table
   // that has non-default scope
@@ -787,9 +787,8 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver,
Regi
      */
     this.rowProcessorTimeout = conf.getLong(
         "hbase.hregion.row.processor.timeout", DEFAULT_ROW_PROCESSOR_TIMEOUT);
-    this.durability = htd.getDurability() == Durability.USE_DEFAULT
-        ? DEFAULT_DURABILITY
-        : htd.getDurability();
+    this.regionDurability = htd.getDurability() == Durability.USE_DEFAULT ?
+        DEFAULT_DURABILITY : htd.getDurability();
     if (rsServices != null) {
       this.rsAccounting = this.rsServices.getRegionServerAccounting();
       // don't initialize coprocessors if not running within a regionserver
@@ -1945,13 +1944,6 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver,
Regi
   // upkeep.
   //////////////////////////////////////////////////////////////////////////////
   /**
-   * @return returns size of largest HStore.
-   */
-  public long getLargestHStoreSize() {
-    return stores.values().stream().mapToLong(HStore::getSize).max().orElse(0L);
-  }
-
-  /**
    * Do preparation for pending compaction.
    * @throws IOException
    */
@@ -3018,21 +3010,28 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver,
Regi
   }
 
   /**
-   * Struct-like class that tracks the progress of a batch operation,
-   * accumulating status codes and tracking the index at which processing
-   * is proceeding.
+   * Struct-like class that tracks the progress of a batch operation, accumulating status
codes
+   * and tracking the index at which processing is proceeding. These batch operations may
get
+   * split into mini-batches for processing.
    */
   private abstract static class BatchOperation<T> {
     T[] operations;
     int nextIndexToProcess = 0;
     OperationStatus[] retCodeDetails;
     WALEdit[] walEditsFromCoprocessors;
+    // reference family cell maps directly so coprocessors can mutate them if desired
+    Map<byte[], List<Cell>>[] familyCellMaps;
+    ObservedExceptionsInBatch observedExceptions;
+    Durability durability;  //Durability of the batch (highest durability of all operations)
 
     public BatchOperation(T[] operations) {
       this.operations = operations;
       this.retCodeDetails = new OperationStatus[operations.length];
       this.walEditsFromCoprocessors = new WALEdit[operations.length];
       Arrays.fill(this.retCodeDetails, OperationStatus.NOT_RUN);
+      familyCellMaps = new Map[operations.length];
+      observedExceptions = new ObservedExceptionsInBatch();
+      durability = Durability.USE_DEFAULT;
     }
 
     public abstract Mutation getMutation(int index);
@@ -3046,12 +3045,69 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver,
Regi
     public boolean isDone() {
       return nextIndexToProcess == operations.length;
     }
+
+    /**
+     * Validates each mutation and prepares a batch for write.
+     * NOTE: As CP prePut()/ preDelete() hooks may modify mutations, this method should be
called
+     * after prePut()/ preDelete() CP hooks are run for all mutations in a batch.
+     */
+    public void checkAndPrepare(final HRegion region) throws IOException {
+      long now = EnvironmentEdgeManager.currentTime();
+      for (int i = 0 ; i < operations.length; i++) {
+        // Skip anything that "ran" already
+        if (retCodeDetails[i].getOperationStatusCode() == OperationStatusCode.NOT_RUN) {
+          Mutation mutation = getMutation(i);
+
+          try {
+            region.checkAndPrepareMutation(mutation, isInReplay(), now);
+
+            // store the family map reference to allow for mutations
+            familyCellMaps[i] = mutation.getFamilyCellMap();
+            // store durability for the batch (highest durability of all operations in the
batch)
+            Durability tmpDur = region.getEffectiveDurability(mutation.getDurability());
+            if (tmpDur.ordinal() > durability.ordinal()) {
+              durability = tmpDur;
+            }
+          } catch (NoSuchColumnFamilyException nscf) {
+            final String msg = "No such column family in batch mutation. ";
+            if (observedExceptions.hasSeenNoSuchFamily()) {
+              LOG.warn(msg + nscf.getMessage());
+            } else {
+              LOG.warn(msg, nscf);
+              observedExceptions.sawNoSuchFamily();
+            }
+            retCodeDetails[i] = new OperationStatus(
+                OperationStatusCode.BAD_FAMILY, nscf.getMessage());
+          } catch (FailedSanityCheckException fsce) {
+            final String msg = "Batch Mutation did not pass sanity check. ";
+            if (observedExceptions.hasSeenFailedSanityCheck()) {
+              LOG.warn(msg + fsce.getMessage());
+            } else {
+              LOG.warn(msg, fsce);
+              observedExceptions.sawFailedSanityCheck();
+            }
+            retCodeDetails[i] = new OperationStatus(
+                OperationStatusCode.SANITY_CHECK_FAILURE, fsce.getMessage());
+          } catch (WrongRegionException we) {
+            final String msg = "Batch mutation had a row that does not belong to this region.
";
+            if (observedExceptions.hasSeenWrongRegion()) {
+              LOG.warn(msg + we.getMessage());
+            } else {
+              LOG.warn(msg, we);
+              observedExceptions.sawWrongRegion();
+            }
+            retCodeDetails[i] = new OperationStatus(
+                OperationStatusCode.SANITY_CHECK_FAILURE, we.getMessage());
+          }
+        }
+      }
+    }
   }
 
-  private static class MutationBatch extends BatchOperation<Mutation> {
+  private static class MutationBatchOperation extends BatchOperation<Mutation> {
     private long nonceGroup;
     private long nonce;
-    public MutationBatch(Mutation[] operations, long nonceGroup, long nonce) {
+    public MutationBatchOperation(Mutation[] operations, long nonceGroup, long nonce) {
       super(operations);
       this.nonceGroup = nonceGroup;
       this.nonce = nonce;
@@ -3088,9 +3144,9 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver,
Regi
     }
   }
 
-  private static class ReplayBatch extends BatchOperation<MutationReplay> {
+  private static class ReplayBatchOperation extends BatchOperation<MutationReplay>
{
     private long replaySeqId = 0;
-    public ReplayBatch(MutationReplay[] operations, long seqId) {
+    public ReplayBatchOperation(MutationReplay[] operations, long seqId) {
       super(operations);
       this.replaySeqId = seqId;
     }
@@ -3133,10 +3189,9 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver,
Regi
     //  * batchMutate with single mutation - put/delete, separate or from checkAndMutate.
     //  * coprocessor calls (see ex. BulkDeleteEndpoint).
     // So nonces are not really ever used by HBase. They could be by coprocs, and checkAnd...
-    return batchMutate(new MutationBatch(mutations, nonceGroup, nonce));
+    return batchMutate(new MutationBatchOperation(mutations, nonceGroup, nonce));
   }
 
-  @Override
   public OperationStatus[] batchMutate(Mutation[] mutations) throws IOException {
     return batchMutate(mutations, HConstants.NO_NONCE, HConstants.NO_NONCE);
   }
@@ -3162,12 +3217,14 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver,
Regi
       }
       return statuses;
     }
-    return batchMutate(new ReplayBatch(mutations, replaySeqId));
+    return batchMutate(new ReplayBatchOperation(mutations, replaySeqId));
   }
 
   /**
    * Perform a batch of mutations.
-   * It supports only Put and Delete mutations and will ignore other types passed.
+   * It supports only Put and Delete mutations and will ignore other types passed. Operations
in
+   * a batch are stored with highest durability specified of for all operations in a batch,
+   * except for {@link Durability#SKIP_WAL}.
    * @param batchOp contains the list of mutations
    * @return an array of OperationStatus which internally contains the
    *         OperationStatusCode and the exceptionMessage if any.
@@ -3187,8 +3244,10 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver,
Regi
         if (!initialized) {
           this.writeRequestsCount.add(batchOp.operations.length);
           if (!batchOp.isInReplay()) {
-            doPreBatchMutateHook(batchOp);
+            callPreMutateCPHooks(batchOp);
           }
+          // validate and prepare batch for write, after CP pre-hooks
+          batchOp.checkAndPrepare(this);
           initialized = true;
         }
         doMiniBatchMutate(batchOp);
@@ -3201,8 +3260,11 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver,
Regi
     return batchOp.retCodeDetails;
   }
 
-  private void doPreBatchMutateHook(BatchOperation<?> batchOp)
-      throws IOException {
+  /**
+   * Runs prePut/ preDelete coprocessor hooks for each mutation in a batch.
+   * @param batchOp
+   */
+  private void callPreMutateCPHooks(BatchOperation<?> batchOp) throws IOException {
     /* Run coprocessor pre hook outside of locks to avoid deadlock */
     WALEdit walEdit = new WALEdit();
     if (coprocessorHost != null) {
@@ -3252,27 +3314,24 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver,
Regi
     long currentNonce = HConstants.NO_NONCE;
     WALEdit walEdit = null;
     boolean locked = false;
-    // reference family maps directly so coprocessors can mutate them if desired
-    Map<byte[], List<Cell>>[] familyMaps = new Map[batchOp.operations.length];
     // We try to set up a batch in the range [firstIndex,lastIndexExclusive)
     int firstIndex = batchOp.nextIndexToProcess;
     int lastIndexExclusive = firstIndex;
     boolean success = false;
+    boolean doneByCoprocessor = false;
     int noOfPuts = 0;
     int noOfDeletes = 0;
     WriteEntry writeEntry = null;
     int cellCount = 0;
     /** Keep track of the locks we hold so we can release them in finally clause */
     List<RowLock> acquiredRowLocks = Lists.newArrayListWithCapacity(batchOp.operations.length);
-    MemStoreSize memstoreSize = new MemStoreSize();
-    final ObservedExceptionsInBatch observedExceptions = new ObservedExceptionsInBatch();
+    MemStoreSize memStoreSize = new MemStoreSize();
     try {
       // STEP 1. Try to acquire as many locks as we can, and ensure we acquire at least one.
       int numReadyToWrite = 0;
-      long now = EnvironmentEdgeManager.currentTime();
-      while (lastIndexExclusive < batchOp.operations.length) {
-        if (checkBatchOp(batchOp, lastIndexExclusive, familyMaps, now, observedExceptions))
{
-          lastIndexExclusive++;
+      for (; lastIndexExclusive < batchOp.operations.length; lastIndexExclusive++) {
+        if (batchOp.retCodeDetails[lastIndexExclusive].getOperationStatusCode()
+            != OperationStatusCode.NOT_RUN) {
           continue;
         }
         Mutation mutation = batchOp.getMutation(lastIndexExclusive);
@@ -3293,9 +3352,8 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver,
Regi
           acquiredRowLocks.add(rowLock);
         }
 
-        lastIndexExclusive++;
         numReadyToWrite++;
-        if (replay) {
+        if (replay || getEffectiveDurability(mutation.getDurability()) != Durability.SKIP_WAL)
{
           for (List<Cell> cells : mutation.getFamilyCellMap().values()) {
             cellCount += cells.size();
           }
@@ -3303,42 +3361,36 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver,
Regi
       }
 
       // We've now grabbed as many mutations off the list as we can
-
-      // STEP 2. Update any LATEST_TIMESTAMP timestamps
-      // We should record the timestamp only after we have acquired the rowLock,
-      // otherwise, newer puts/deletes are not guaranteed to have a newer timestamp
-      now = EnvironmentEdgeManager.currentTime();
-      byte[] byteNow = Bytes.toBytes(now);
-
       // Nothing to put/delete -- an exception in the above such as NoSuchColumnFamily?
       if (numReadyToWrite <= 0) {
         return;
       }
 
-      for (int i = firstIndex; !replay && i < lastIndexExclusive; i++) {
-        // skip invalid
-        if (batchOp.retCodeDetails[i].getOperationStatusCode()
-            != OperationStatusCode.NOT_RUN) {
-          // lastIndexExclusive was incremented above.
-          continue;
-        }
+      // STEP 2. Update any LATEST_TIMESTAMP timestamps
+      // We should record the timestamp only after we have acquired the rowLock,
+      // otherwise, newer puts/deletes are not guaranteed to have a newer timestamp
+      long now = EnvironmentEdgeManager.currentTime();
+      if (!replay) {
+        byte[] byteNow = Bytes.toBytes(now);
+        for (int i = firstIndex; i < lastIndexExclusive; i++) {
+          // skip invalid
+          if (batchOp.retCodeDetails[i].getOperationStatusCode() != OperationStatusCode.NOT_RUN)
{
+            // lastIndexExclusive was incremented above.
+            continue;
+          }
 
-        Mutation mutation = batchOp.getMutation(i);
-        if (mutation instanceof Put) {
-          updateCellTimestamps(familyMaps[i].values(), byteNow);
-          noOfPuts++;
-        } else {
-          prepareDeleteTimestamps(mutation, familyMaps[i], byteNow);
-          noOfDeletes++;
-        }
-        rewriteCellTags(familyMaps[i], mutation);
-        WALEdit fromCP = batchOp.walEditsFromCoprocessors[i];
-        if (fromCP != null) {
-          cellCount += fromCP.size();
-        }
-        if (getEffectiveDurability(mutation.getDurability()) != Durability.SKIP_WAL) {
-          for (List<Cell> cells : familyMaps[i].values()) {
-            cellCount += cells.size();
+          Mutation mutation = batchOp.getMutation(i);
+          if (mutation instanceof Put) {
+            updateCellTimestamps(batchOp.familyCellMaps[i].values(), byteNow);
+            noOfPuts++;
+          } else {
+            prepareDeleteTimestamps(mutation, batchOp.familyCellMaps[i], byteNow);
+            noOfDeletes++;
+          }
+          rewriteCellTags(batchOp.familyCellMaps[i], mutation);
+          WALEdit fromCP = batchOp.walEditsFromCoprocessors[i];
+          if (fromCP != null) {
+            cellCount += fromCP.size();
           }
         }
       }
@@ -3351,6 +3403,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver,
Regi
           new MiniBatchOperationInProgress<>(batchOp.getMutationsForCoprocs(),
           batchOp.retCodeDetails, batchOp.walEditsFromCoprocessors, firstIndex, lastIndexExclusive);
         if (coprocessorHost.preBatchMutate(miniBatchOp)) {
+          doneByCoprocessor = true;
           return;
         } else {
           for (int i = firstIndex; i < lastIndexExclusive; i++) {
@@ -3368,15 +3421,16 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver,
Regi
             // Else Coprocessor added more Mutations corresponding to the Mutation at this
index.
             for (int j = 0; j < cpMutations.length; j++) {
               Mutation cpMutation = cpMutations[j];
-              Map<byte[], List<Cell>> cpFamilyMap = cpMutation.getFamilyCellMap();
-              checkAndPrepareMutation(cpMutation, replay, cpFamilyMap, now);
+              checkAndPrepareMutation(cpMutation, replay, now);
 
               // Acquire row locks. If not, the whole batch will fail.
               acquiredRowLocks.add(getRowLockInternal(cpMutation.getRow(), true));
 
               // Returned mutations from coprocessor correspond to the Mutation at index
i. We can
               // directly add the cells from those mutations to the familyMaps of this mutation.
-              mergeFamilyMaps(familyMaps[i], cpFamilyMap); // will get added to the memstore
later
+              Map<byte[], List<Cell>> cpFamilyMap = cpMutation.getFamilyCellMap();
+              // will get added to the memStore later
+              mergeFamilyMaps(batchOp.familyCellMaps[i], cpFamilyMap);
 
               // The durability of returned mutation is replaced by the corresponding mutation.
               // If the corresponding mutation contains the SKIP_WAL, we shouldn't count
the
@@ -3393,7 +3447,6 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver,
Regi
 
       // STEP 3. Build WAL edit
       walEdit = new WALEdit(cellCount, replay);
-      Durability durability = Durability.USE_DEFAULT;
       for (int i = firstIndex; i < lastIndexExclusive; i++) {
         // Skip puts that were determined to be invalid during preprocessing
         if (batchOp.retCodeDetails[i].getOperationStatusCode() != OperationStatusCode.NOT_RUN)
{
@@ -3401,12 +3454,8 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver,
Regi
         }
 
         Mutation m = batchOp.getMutation(i);
-        Durability tmpDur = getEffectiveDurability(m.getDurability());
-        if (tmpDur.ordinal() > durability.ordinal()) {
-          durability = tmpDur;
-        }
         // we use durability of the original mutation for the mutation passed by CP.
-        if (tmpDur == Durability.SKIP_WAL) {
+        if (getEffectiveDurability(m.getDurability()) == Durability.SKIP_WAL) {
           recordMutationWithoutWal(m.getFamilyCellMap());
           continue;
         }
@@ -3431,58 +3480,21 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver,
Regi
             walEdit.add(cell);
           }
         }
-        addFamilyMapToWALEdit(familyMaps[i], walEdit);
+        addFamilyMapToWALEdit(batchOp.familyCellMaps[i], walEdit);
       }
 
       // STEP 4. Append the final edit to WAL and sync.
       Mutation mutation = batchOp.getMutation(firstIndex);
-      WALKey walKey = null;
-      long txid;
-      if (replay) {
-        // use wal key from the original
-        walKey = new WALKey(this.getRegionInfo().getEncodedNameAsBytes(),
-          this.htableDescriptor.getTableName(), WALKey.NO_SEQUENCE_ID, now,
-          mutation.getClusterIds(), currentNonceGroup, currentNonce, mvcc);
-        walKey.setOrigLogSeqNum(batchOp.getReplaySequenceId());
-        if (!walEdit.isEmpty()) {
-          txid = this.wal.append(this.getRegionInfo(), walKey, walEdit, true);
-          if (txid != 0) {
-            sync(txid, durability);
-          }
-        }
-      } else {
-        try {
-          if (!walEdit.isEmpty()) {
-            // we use HLogKey here instead of WALKey directly to support legacy coprocessors.
-            walKey = new WALKey(this.getRegionInfo().getEncodedNameAsBytes(),
-                this.htableDescriptor.getTableName(), WALKey.NO_SEQUENCE_ID, now,
-                mutation.getClusterIds(), currentNonceGroup, currentNonce, mvcc,
-                this.getReplicationScope());
-            // TODO: Use the doAppend methods below... complicated by the replay stuff above.
-            txid = this.wal.append(this.getRegionInfo(), walKey, walEdit, true);
-            if (txid != 0) {
-              sync(txid, durability);
-            }
-            if (writeEntry == null) {
-              // if MVCC not preassigned, wait here until assigned
-              writeEntry = walKey.getWriteEntry();
-            }
-          }
-        } catch (IOException ioe) {
-          if (walKey != null && writeEntry == null) {
-            // the writeEntry is not preassigned and error occurred during append or sync
-            mvcc.complete(walKey.getWriteEntry());
-          }
-          throw ioe;
-        }
-      }
-      if (walKey == null) {
-        // If no walKey, then not in replay and skipping WAL or some such. Begin an MVCC
transaction
-        // to get sequence id.
+      writeEntry = doWALAppend(walEdit, batchOp.durability, mutation.getClusterIds(), now,
+          currentNonceGroup, currentNonce,
+          replay ? batchOp.getReplaySequenceId() : WALKey.NO_SEQUENCE_ID);
+      if (!replay && writeEntry == null) {
+        // If no writeEntry, then not in replay and skipping WAL or some such. Begin an MVCC
+        // transaction to get sequence id.
         writeEntry = mvcc.begin();
       }
 
-      // STEP 5. Write back to memstore
+      // STEP 5. Write back to memStore
       for (int i = firstIndex; i < lastIndexExclusive; i++) {
         if (batchOp.retCodeDetails[i].getOperationStatusCode() != OperationStatusCode.NOT_RUN)
{
           continue;
@@ -3493,14 +3505,14 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver,
Regi
         // we use durability of the original mutation for the mutation passed by CP.
         boolean updateSeqId = replay || batchOp.getMutation(i).getDurability() == Durability.SKIP_WAL;
         if (updateSeqId) {
-          this.updateSequenceId(familyMaps[i].values(),
+          this.updateSequenceId(batchOp.familyCellMaps[i].values(),
             replay? batchOp.getReplaySequenceId(): writeEntry.getWriteNumber());
         }
-        applyFamilyMapToMemStore(familyMaps[i], memstoreSize);
+        applyFamilyMapToMemStore(batchOp.familyCellMaps[i], memStoreSize);
       }
 
       // update memstore size
-      this.addAndGetMemStoreSize(memstoreSize);
+      this.addAndGetMemStoreSize(memStoreSize);
 
       // calling the post CP hook for batch mutation
       if (!replay && coprocessorHost != null) {
@@ -3511,30 +3523,33 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver,
Regi
       }
 
       // STEP 6. Complete mvcc.
-      if (replay) {
-        this.mvcc.advanceTo(batchOp.getReplaySequenceId());
-      } else {
-        // writeEntry won't be empty if not in replay mode
+      if (writeEntry != null) {
         mvcc.completeAndWait(writeEntry);
         writeEntry = null;
       }
+      if (replay) {
+        this.mvcc.advanceTo(batchOp.getReplaySequenceId());
+      }
+
+      success = true;
+    } finally {
+      // Call complete rather than completeAndWait because we probably had error if walKey
!= null
+      if (writeEntry != null) mvcc.complete(writeEntry);
 
-      // STEP 7. Release row locks, etc.
       if (locked) {
         this.updatesLock.readLock().unlock();
-        locked = false;
       }
       releaseRowLocks(acquiredRowLocks);
 
-      for (int i = firstIndex; i < lastIndexExclusive; i ++) {
+      for (int i = firstIndex; i < lastIndexExclusive; i++) {
         if (batchOp.retCodeDetails[i] == OperationStatus.NOT_RUN) {
-          batchOp.retCodeDetails[i] = OperationStatus.SUCCESS;
+          batchOp.retCodeDetails[i] =
+              success || doneByCoprocessor ? OperationStatus.SUCCESS : OperationStatus.FAILURE;
         }
       }
 
-      // STEP 8. Run coprocessor post hooks. This should be done after the wal is
       // synced so that the coprocessor contract is adhered to.
-      if (!replay && coprocessorHost != null) {
+      if (!replay && coprocessorHost != null && !doneByCoprocessor) {
         for (int i = firstIndex; i < lastIndexExclusive; i++) {
           // only for successful puts
           if (batchOp.retCodeDetails[i].getOperationStatusCode()
@@ -3550,15 +3565,6 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver,
Regi
         }
       }
 
-      success = true;
-    } finally {
-      // Call complete rather than completeAndWait because we probably had error if walKey
!= null
-      if (writeEntry != null) mvcc.complete(writeEntry);
-      if (locked) {
-        this.updatesLock.readLock().unlock();
-      }
-      releaseRowLocks(acquiredRowLocks);
-
       // See if the column families were consistent through the whole thing.
       // if they were then keep them. If they were not then pass a null.
       // null will be treated as unknown.
@@ -3577,13 +3583,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver,
Regi
           this.metricsRegion.updateDelete();
         }
       }
-      if (!success) {
-        for (int i = firstIndex; i < lastIndexExclusive; i++) {
-          if (batchOp.retCodeDetails[i].getOperationStatusCode() == OperationStatusCode.NOT_RUN)
{
-            batchOp.retCodeDetails[i] = OperationStatus.FAILURE;
-          }
-        }
-      }
+
       if (coprocessorHost != null && !batchOp.isInReplay()) {
         // call the coprocessor hook to do any finalization steps
         // after the put is done
@@ -3622,75 +3622,20 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver,
Regi
     this.mvcc.complete(walKey.getWriteEntry());
   }
 
-  private boolean checkBatchOp(BatchOperation<?> batchOp, final int lastIndexExclusive,
-      final Map<byte[], List<Cell>>[] familyMaps, final long now,
-      final ObservedExceptionsInBatch observedExceptions)
-  throws IOException {
-    boolean skip = false;
-    // Skip anything that "ran" already
-    if (batchOp.retCodeDetails[lastIndexExclusive].getOperationStatusCode()
-        != OperationStatusCode.NOT_RUN) {
-      return true;
-    }
-    Mutation mutation = batchOp.getMutation(lastIndexExclusive);
-    Map<byte[], List<Cell>> familyMap = mutation.getFamilyCellMap();
-    // store the family map reference to allow for mutations
-    familyMaps[lastIndexExclusive] = familyMap;
-
-    try {
-      checkAndPrepareMutation(mutation, batchOp.isInReplay(), familyMap, now);
-    } catch (NoSuchColumnFamilyException nscf) {
-      final String msg = "No such column family in batch mutation. ";
-      if (observedExceptions.hasSeenNoSuchFamily()) {
-        LOG.warn(msg + nscf.getMessage());
-      } else {
-        LOG.warn(msg, nscf);
-        observedExceptions.sawNoSuchFamily();
-      }
-      batchOp.retCodeDetails[lastIndexExclusive] = new OperationStatus(
-          OperationStatusCode.BAD_FAMILY, nscf.getMessage());
-      skip = true;
-    } catch (FailedSanityCheckException fsce) {
-      final String msg = "Batch Mutation did not pass sanity check. ";
-      if (observedExceptions.hasSeenFailedSanityCheck()) {
-        LOG.warn(msg + fsce.getMessage());
-      } else {
-        LOG.warn(msg, fsce);
-        observedExceptions.sawFailedSanityCheck();
-      }
-      batchOp.retCodeDetails[lastIndexExclusive] = new OperationStatus(
-          OperationStatusCode.SANITY_CHECK_FAILURE, fsce.getMessage());
-      skip = true;
-    } catch (WrongRegionException we) {
-      final String msg = "Batch mutation had a row that does not belong to this region. ";
-      if (observedExceptions.hasSeenWrongRegion()) {
-        LOG.warn(msg + we.getMessage());
-      } else {
-        LOG.warn(msg, we);
-        observedExceptions.sawWrongRegion();
-      }
-      batchOp.retCodeDetails[lastIndexExclusive] = new OperationStatus(
-          OperationStatusCode.SANITY_CHECK_FAILURE, we.getMessage());
-      skip = true;
-    }
-    return skip;
-  }
-
-  private void checkAndPrepareMutation(Mutation mutation, boolean replay,
-      final Map<byte[], List<Cell>> familyMap, final long now)
-          throws IOException {
+  private void checkAndPrepareMutation(Mutation mutation, boolean replay, final long now)
+      throws IOException {
+    checkRow(mutation.getRow(), "doMiniBatchMutation");
     if (mutation instanceof Put) {
       // Check the families in the put. If bad, skip this one.
       if (replay) {
-        removeNonExistentColumnFamilyForReplay(familyMap);
+        removeNonExistentColumnFamilyForReplay(mutation.getFamilyCellMap());
       } else {
-        checkFamilies(familyMap.keySet());
+        checkFamilies(mutation.getFamilyCellMap().keySet());
       }
       checkTimestamps(mutation.getFamilyCellMap(), now);
     } else {
       prepareDelete((Delete)mutation);
     }
-    checkRow(mutation.getRow(), "doMiniBatchMutation");
   }
 
   /**
@@ -3721,7 +3666,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver,
Regi
    * the table descriptor.
    */
   protected Durability getEffectiveDurability(Durability d) {
-    return d == Durability.USE_DEFAULT ? this.durability : d;
+    return d == Durability.USE_DEFAULT ? this.regionDurability : d;
   }
 
   @Override
@@ -7430,28 +7375,42 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver,
Regi
       nonceGroup, nonce);
   }
 
+  private WriteEntry doWALAppend(WALEdit walEdit, Durability durability, List<UUID>
clusterIds,
+      long now, long nonceGroup, long nonce) throws IOException {
+    return doWALAppend(walEdit, durability, clusterIds, now, nonceGroup, nonce,
+        WALKey.NO_SEQUENCE_ID);
+  }
+
   /**
    * @return writeEntry associated with this append
    */
   private WriteEntry doWALAppend(WALEdit walEdit, Durability durability, List<UUID>
clusterIds,
-      long now, long nonceGroup, long nonce)
-  throws IOException {
+      long now, long nonceGroup, long nonce, long replaySeqId) throws IOException {
+    Preconditions.checkArgument(!walEdit.isReplay() || replaySeqId != WALKey.NO_SEQUENCE_ID,
+        "Invalid replay sequence Id for replay WALEdit!");
     WriteEntry writeEntry = null;
-    // Using default cluster id, as this can only happen in the originating cluster.
-    // A slave cluster receives the final value (not the delta) as a Put. We use HLogKey
-    // here instead of WALKey directly to support legacy coprocessors.
-    WALKey walKey = new WALKey(this.getRegionInfo().getEncodedNameAsBytes(),
-      this.htableDescriptor.getTableName(), WALKey.NO_SEQUENCE_ID, now, clusterIds,
-      nonceGroup, nonce, mvcc, this.getReplicationScope());
-    try {
-      long txid =
-        this.wal.append(this.getRegionInfo(), walKey, walEdit, true);
-      // Call sync on our edit.
-      if (txid != 0) sync(txid, durability);
-      writeEntry = walKey.getWriteEntry();
-    } catch (IOException ioe) {
-      if (walKey != null) mvcc.complete(walKey.getWriteEntry());
-      throw ioe;
+    if (!walEdit.isEmpty()) {
+      // Using default cluster id, as this can only happen in the originating cluster.
+      // A slave cluster receives the final value (not the delta) as a Put. We use HLogKey
+      // here instead of WALKey directly to support legacy coprocessors.
+      WALKey walKey = walEdit.isReplay() ? new WALKey(this.getRegionInfo().getEncodedNameAsBytes(),
+          this.htableDescriptor.getTableName(), WALKey.NO_SEQUENCE_ID, now, clusterIds, nonceGroup,
+          nonce, mvcc) :
+          new WALKey(this.getRegionInfo().getEncodedNameAsBytes(),
+          this.htableDescriptor.getTableName(), WALKey.NO_SEQUENCE_ID, now, clusterIds,
+          nonceGroup, nonce, mvcc, this.getReplicationScope());
+      if (walEdit.isReplay()) {
+        walKey.setOrigLogSeqNum(replaySeqId);
+      }
+      try {
+        long txid = this.wal.append(this.getRegionInfo(), walKey, walEdit, true);
+        // Call sync on our edit.
+        if (txid != 0) sync(txid, durability);
+        writeEntry = walKey.getWriteEntry();
+      } catch (IOException ioe) {
+        if (walKey != null) mvcc.complete(walKey.getWriteEntry());
+        throw ioe;
+      }
     }
     return writeEntry;
   }
@@ -7846,13 +7805,6 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver,
Regi
   }
 
   /**
-   * Give the region a chance to prepare before it is split.
-   */
-  protected void prepareToSplit() {
-    // nothing
-  }
-
-  /**
    * Return the splitpoint. null indicates the region isn't splittable
    * If the splitpoint isn't explicitly specified, it will go over the stores
    * to find the best splitpoint. Currently the criteria of best splitpoint
@@ -8115,7 +8067,7 @@ public class HRegion implements HeapSize, PropagatingConfigurationObserver,
Regi
    * Check whether we should sync the wal from the table's durability settings
    */
   private boolean shouldSyncWAL() {
-    return durability.ordinal() >  Durability.ASYNC_WAL.ordinal();
+    return regionDurability.ordinal() >  Durability.ASYNC_WAL.ordinal();
   }
 
   /**

http://git-wip-us.apache.org/repos/asf/hbase/blob/e1941aa6/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegionReplayEvents.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegionReplayEvents.java
b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegionReplayEvents.java
index 7cb28f9..f1d9475 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegionReplayEvents.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHRegionReplayEvents.java
@@ -310,6 +310,27 @@ public class TestHRegionReplayEvents {
   }
 
   @Test
+  public void testBatchReplayWithMultipleNonces() throws IOException {
+    try {
+      MutationReplay[] mutations = new MutationReplay[100];
+      for (int i = 0; i < 100; i++) {
+        Put put = new Put(Bytes.toBytes(i));
+        put.setDurability(Durability.SYNC_WAL);
+        for (byte[] familly : this.families) {
+          put.addColumn(familly, this.cq, null);
+          long nonceNum = i / 10;
+          mutations[i] = new MutationReplay(MutationType.PUT, put, nonceNum, nonceNum);
+        }
+      }
+      primaryRegion.batchReplay(mutations, 20);
+    } catch (Exception e) {
+      String msg = "Error while replay of batch with multiple nonces. ";
+      LOG.error(msg, e);
+      fail(msg + e.getMessage());
+    }
+  }
+
+  @Test
   public void testReplayFlushesAndCompactions() throws IOException {
     // initiate a secondary region with some data.
 


Mime
View raw message