HBASE-17924 Consider sorting the row order when processing multi() ops before taking rowlocks

(Allan Yang)

Conflicts:
	hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java

@@ -32,6 +32,7 @@ import java.net.BindException;
 import java.net.InetSocketAddress;
 import java.net.UnknownHostException;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Iterator;
@@ -804,6 +805,14 @@ public class RSRpcServices implements HBaseRPCErrorHandler,
     long before = EnvironmentEdgeManager.currentTime();
     boolean batchContainsPuts = false, batchContainsDelete = false;
     try {
+      /** HBASE-17924
+       * mutationActionMap is a map to map the relation between mutations and actions
+       * since mutation array may have been reoredered.In order to return the right
+       * result or exception to the corresponding actions, We need to know which action
+       * is the mutation belong to. We can't sort ClientProtos.Action array, since they
+       * are bonded to cellscanners.
+       */
+      Map<Mutation, ClientProtos.Action> mutationActionMap = new HashMap<Mutation, ClientProtos.Action>();
       int i = 0;
       for (ClientProtos.Action action: mutations) {
         MutationProto m = action.getMutation();
@@ -815,6 +824,7 @@ public class RSRpcServices implements HBaseRPCErrorHandler,
           mutation = ProtobufUtil.toDelete(m, cells);
           batchContainsDelete = true;
         }
+        mutationActionMap.put(mutation, action);
         mArray[i++] = mutation;
         quota.addMutation(mutation);
       }
@@ -822,11 +832,15 @@ public class RSRpcServices implements HBaseRPCErrorHandler,
       if (!region.getRegionInfo().isMetaTable()) {
         regionServer.cacheFlusher.reclaimMemStoreMemory();
       }
-
+      // HBASE-17924
+      // sort to improve lock efficiency
+      Arrays.sort(mArray);
       OperationStatus[] codes = region.batchMutate(mArray, HConstants.NO_NONCE,
         HConstants.NO_NONCE);
       for (i = 0; i < codes.length; i++) {
-        int index = mutations.get(i).getIndex();
+        Mutation currentMutation = mArray[i];
+        ClientProtos.Action currentAction = mutationActionMap.get(currentMutation);
+        int index = currentAction.getIndex();
         Exception e = null;
         switch (codes[i].getOperationStatusCode()) {
           case BAD_FAMILY:
@@ -1846,6 +1860,9 @@ public class RSRpcServices implements HBaseRPCErrorHandler,
           walEntries.add(walEntry);
         }
         if(edits!=null && !edits.isEmpty()) {
+          // HBASE-17924
+          // sort to improve lock efficiency
+          Collections.sort(edits);
           long replaySeqId = (entry.getKey().hasOrigSequenceNumber()) ?
             entry.getKey().getOrigSequenceNumber() : entry.getKey().getLogSequenceNumber();
           OperationStatus[] result = doReplayBatchOp(region, edits, replaySeqId);

hbase-server/src/main/java/org/apache/hadoop/hbase/wal/WALSplitter.java

@@ -2266,7 +2266,7 @@ public class WALSplitter {
   }
 
   /** A struct used by getMutationsFromWALEntry */
-  public static class MutationReplay {
+  public static class MutationReplay implements Comparable<MutationReplay> {
     public MutationReplay(MutationType type, Mutation mutation, long nonceGroup, long nonce) {
       this.type = type;
       this.mutation = mutation;
@@ -2282,6 +2282,25 @@ public class WALSplitter {
     public final Mutation mutation;
     public final long nonceGroup;
     public final long nonce;
+
+    @Override
+    public int compareTo(final MutationReplay d) {
+      return this.mutation.compareTo(d.mutation);
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+      if(!(obj instanceof MutationReplay)) {
+        return false;
+      } else {
+        return this.compareTo((MutationReplay)obj) == 0;
+      }
+    }
+
+    @Override
+    public int hashCode() {
+      return this.mutation.hashCode();
+    }
   }
 
   /**