hbase-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ser...@apache.org
Subject svn commit: r1542168 [2/3] - in /hbase/trunk: hbase-client/src/main/java/org/apache/hadoop/hbase/client/ hbase-client/src/main/java/org/apache/hadoop/hbase/exceptions/ hbase-client/src/main/java/org/apache/hadoop/hbase/protobuf/ hbase-common/src/main/j...
Date Fri, 15 Nov 2013 04:36:31 GMT
Modified: hbase/trunk/hbase-protocol/src/main/java/org/apache/hadoop/hbase/protobuf/generated/MultiRowMutationProtos.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-protocol/src/main/java/org/apache/hadoop/hbase/protobuf/generated/MultiRowMutationProtos.java?rev=1542168&r1=1542167&r2=1542168&view=diff
==============================================================================
--- hbase/trunk/hbase-protocol/src/main/java/org/apache/hadoop/hbase/protobuf/generated/MultiRowMutationProtos.java (original)
+++ hbase/trunk/hbase-protocol/src/main/java/org/apache/hadoop/hbase/protobuf/generated/MultiRowMutationProtos.java Fri Nov 15 04:36:30 2013
@@ -711,6 +711,26 @@ public final class MultiRowMutationProto
      */
     org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MutationProtoOrBuilder getMutationRequestOrBuilder(
         int index);
+
+    // optional uint64 nonce_group = 2;
+    /**
+     * <code>optional uint64 nonce_group = 2;</code>
+     */
+    boolean hasNonceGroup();
+    /**
+     * <code>optional uint64 nonce_group = 2;</code>
+     */
+    long getNonceGroup();
+
+    // optional uint64 nonce = 3;
+    /**
+     * <code>optional uint64 nonce = 3;</code>
+     */
+    boolean hasNonce();
+    /**
+     * <code>optional uint64 nonce = 3;</code>
+     */
+    long getNonce();
   }
   /**
    * Protobuf type {@code MutateRowsRequest}
@@ -771,6 +791,16 @@ public final class MultiRowMutationProto
               mutationRequest_.add(input.readMessage(org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MutationProto.PARSER, extensionRegistry));
               break;
             }
+            case 16: {
+              bitField0_ |= 0x00000001;
+              nonceGroup_ = input.readUInt64();
+              break;
+            }
+            case 24: {
+              bitField0_ |= 0x00000002;
+              nonce_ = input.readUInt64();
+              break;
+            }
           }
         }
       } catch (com.google.protobuf.InvalidProtocolBufferException e) {
@@ -813,6 +843,7 @@ public final class MultiRowMutationProto
       return PARSER;
     }
 
+    private int bitField0_;
     // repeated .MutationProto mutation_request = 1;
     public static final int MUTATION_REQUEST_FIELD_NUMBER = 1;
     private java.util.List<org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MutationProto> mutationRequest_;
@@ -849,8 +880,42 @@ public final class MultiRowMutationProto
       return mutationRequest_.get(index);
     }
 
+    // optional uint64 nonce_group = 2;
+    public static final int NONCE_GROUP_FIELD_NUMBER = 2;
+    private long nonceGroup_;
+    /**
+     * <code>optional uint64 nonce_group = 2;</code>
+     */
+    public boolean hasNonceGroup() {
+      return ((bitField0_ & 0x00000001) == 0x00000001);
+    }
+    /**
+     * <code>optional uint64 nonce_group = 2;</code>
+     */
+    public long getNonceGroup() {
+      return nonceGroup_;
+    }
+
+    // optional uint64 nonce = 3;
+    public static final int NONCE_FIELD_NUMBER = 3;
+    private long nonce_;
+    /**
+     * <code>optional uint64 nonce = 3;</code>
+     */
+    public boolean hasNonce() {
+      return ((bitField0_ & 0x00000002) == 0x00000002);
+    }
+    /**
+     * <code>optional uint64 nonce = 3;</code>
+     */
+    public long getNonce() {
+      return nonce_;
+    }
+
     private void initFields() {
       mutationRequest_ = java.util.Collections.emptyList();
+      nonceGroup_ = 0L;
+      nonce_ = 0L;
     }
     private byte memoizedIsInitialized = -1;
     public final boolean isInitialized() {
@@ -873,6 +938,12 @@ public final class MultiRowMutationProto
       for (int i = 0; i < mutationRequest_.size(); i++) {
         output.writeMessage(1, mutationRequest_.get(i));
       }
+      if (((bitField0_ & 0x00000001) == 0x00000001)) {
+        output.writeUInt64(2, nonceGroup_);
+      }
+      if (((bitField0_ & 0x00000002) == 0x00000002)) {
+        output.writeUInt64(3, nonce_);
+      }
       getUnknownFields().writeTo(output);
     }
 
@@ -886,6 +957,14 @@ public final class MultiRowMutationProto
         size += com.google.protobuf.CodedOutputStream
           .computeMessageSize(1, mutationRequest_.get(i));
       }
+      if (((bitField0_ & 0x00000001) == 0x00000001)) {
+        size += com.google.protobuf.CodedOutputStream
+          .computeUInt64Size(2, nonceGroup_);
+      }
+      if (((bitField0_ & 0x00000002) == 0x00000002)) {
+        size += com.google.protobuf.CodedOutputStream
+          .computeUInt64Size(3, nonce_);
+      }
       size += getUnknownFields().getSerializedSize();
       memoizedSerializedSize = size;
       return size;
@@ -911,6 +990,16 @@ public final class MultiRowMutationProto
       boolean result = true;
       result = result && getMutationRequestList()
           .equals(other.getMutationRequestList());
+      result = result && (hasNonceGroup() == other.hasNonceGroup());
+      if (hasNonceGroup()) {
+        result = result && (getNonceGroup()
+            == other.getNonceGroup());
+      }
+      result = result && (hasNonce() == other.hasNonce());
+      if (hasNonce()) {
+        result = result && (getNonce()
+            == other.getNonce());
+      }
       result = result &&
           getUnknownFields().equals(other.getUnknownFields());
       return result;
@@ -928,6 +1017,14 @@ public final class MultiRowMutationProto
         hash = (37 * hash) + MUTATION_REQUEST_FIELD_NUMBER;
         hash = (53 * hash) + getMutationRequestList().hashCode();
       }
+      if (hasNonceGroup()) {
+        hash = (37 * hash) + NONCE_GROUP_FIELD_NUMBER;
+        hash = (53 * hash) + hashLong(getNonceGroup());
+      }
+      if (hasNonce()) {
+        hash = (37 * hash) + NONCE_FIELD_NUMBER;
+        hash = (53 * hash) + hashLong(getNonce());
+      }
       hash = (29 * hash) + getUnknownFields().hashCode();
       memoizedHashCode = hash;
       return hash;
@@ -1044,6 +1141,10 @@ public final class MultiRowMutationProto
         } else {
           mutationRequestBuilder_.clear();
         }
+        nonceGroup_ = 0L;
+        bitField0_ = (bitField0_ & ~0x00000002);
+        nonce_ = 0L;
+        bitField0_ = (bitField0_ & ~0x00000004);
         return this;
       }
 
@@ -1071,6 +1172,7 @@ public final class MultiRowMutationProto
       public org.apache.hadoop.hbase.protobuf.generated.MultiRowMutationProtos.MutateRowsRequest buildPartial() {
         org.apache.hadoop.hbase.protobuf.generated.MultiRowMutationProtos.MutateRowsRequest result = new org.apache.hadoop.hbase.protobuf.generated.MultiRowMutationProtos.MutateRowsRequest(this);
         int from_bitField0_ = bitField0_;
+        int to_bitField0_ = 0;
         if (mutationRequestBuilder_ == null) {
           if (((bitField0_ & 0x00000001) == 0x00000001)) {
             mutationRequest_ = java.util.Collections.unmodifiableList(mutationRequest_);
@@ -1080,6 +1182,15 @@ public final class MultiRowMutationProto
         } else {
           result.mutationRequest_ = mutationRequestBuilder_.build();
         }
+        if (((from_bitField0_ & 0x00000002) == 0x00000002)) {
+          to_bitField0_ |= 0x00000001;
+        }
+        result.nonceGroup_ = nonceGroup_;
+        if (((from_bitField0_ & 0x00000004) == 0x00000004)) {
+          to_bitField0_ |= 0x00000002;
+        }
+        result.nonce_ = nonce_;
+        result.bitField0_ = to_bitField0_;
         onBuilt();
         return result;
       }
@@ -1121,6 +1232,12 @@ public final class MultiRowMutationProto
             }
           }
         }
+        if (other.hasNonceGroup()) {
+          setNonceGroup(other.getNonceGroup());
+        }
+        if (other.hasNonce()) {
+          setNonce(other.getNonce());
+        }
         this.mergeUnknownFields(other.getUnknownFields());
         return this;
       }
@@ -1394,6 +1511,72 @@ public final class MultiRowMutationProto
         return mutationRequestBuilder_;
       }
 
+      // optional uint64 nonce_group = 2;
+      private long nonceGroup_ ;
+      /**
+       * <code>optional uint64 nonce_group = 2;</code>
+       */
+      public boolean hasNonceGroup() {
+        return ((bitField0_ & 0x00000002) == 0x00000002);
+      }
+      /**
+       * <code>optional uint64 nonce_group = 2;</code>
+       */
+      public long getNonceGroup() {
+        return nonceGroup_;
+      }
+      /**
+       * <code>optional uint64 nonce_group = 2;</code>
+       */
+      public Builder setNonceGroup(long value) {
+        bitField0_ |= 0x00000002;
+        nonceGroup_ = value;
+        onChanged();
+        return this;
+      }
+      /**
+       * <code>optional uint64 nonce_group = 2;</code>
+       */
+      public Builder clearNonceGroup() {
+        bitField0_ = (bitField0_ & ~0x00000002);
+        nonceGroup_ = 0L;
+        onChanged();
+        return this;
+      }
+
+      // optional uint64 nonce = 3;
+      private long nonce_ ;
+      /**
+       * <code>optional uint64 nonce = 3;</code>
+       */
+      public boolean hasNonce() {
+        return ((bitField0_ & 0x00000004) == 0x00000004);
+      }
+      /**
+       * <code>optional uint64 nonce = 3;</code>
+       */
+      public long getNonce() {
+        return nonce_;
+      }
+      /**
+       * <code>optional uint64 nonce = 3;</code>
+       */
+      public Builder setNonce(long value) {
+        bitField0_ |= 0x00000004;
+        nonce_ = value;
+        onChanged();
+        return this;
+      }
+      /**
+       * <code>optional uint64 nonce = 3;</code>
+       */
+      public Builder clearNonce() {
+        bitField0_ = (bitField0_ & ~0x00000004);
+        nonce_ = 0L;
+        onChanged();
+        return this;
+      }
+
       // @@protoc_insertion_point(builder_scope:MutateRowsRequest)
     }
 
@@ -2006,13 +2189,14 @@ public final class MultiRowMutationProto
     java.lang.String[] descriptorData = {
       "\n\026MultiRowMutation.proto\032\014Client.proto\"\"" +
       "\n MultiRowMutationProcessorRequest\"#\n!Mu" +
-      "ltiRowMutationProcessorResponse\"=\n\021Mutat" +
+      "ltiRowMutationProcessorResponse\"a\n\021Mutat" +
       "eRowsRequest\022(\n\020mutation_request\030\001 \003(\0132\016" +
-      ".MutationProto\"\024\n\022MutateRowsResponse2P\n\027" +
-      "MultiRowMutationService\0225\n\nMutateRows\022\022." +
-      "MutateRowsRequest\032\023.MutateRowsResponseBL" +
-      "\n*org.apache.hadoop.hbase.protobuf.gener" +
-      "atedB\026MultiRowMutationProtosH\001\210\001\001\240\001\001"
+      ".MutationProto\022\023\n\013nonce_group\030\002 \001(\004\022\r\n\005n" +
+      "once\030\003 \001(\004\"\024\n\022MutateRowsResponse2P\n\027Mult" +
+      "iRowMutationService\0225\n\nMutateRows\022\022.Muta" +
+      "teRowsRequest\032\023.MutateRowsResponseBL\n*or" +
+      "g.apache.hadoop.hbase.protobuf.generated" +
+      "B\026MultiRowMutationProtosH\001\210\001\001\240\001\001"
     };
     com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner assigner =
       new com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner() {
@@ -2036,7 +2220,7 @@ public final class MultiRowMutationProto
           internal_static_MutateRowsRequest_fieldAccessorTable = new
             com.google.protobuf.GeneratedMessage.FieldAccessorTable(
               internal_static_MutateRowsRequest_descriptor,
-              new java.lang.String[] { "MutationRequest", });
+              new java.lang.String[] { "MutationRequest", "NonceGroup", "Nonce", });
           internal_static_MutateRowsResponse_descriptor =
             getDescriptor().getMessageTypes().get(3);
           internal_static_MutateRowsResponse_fieldAccessorTable = new

Modified: hbase/trunk/hbase-protocol/src/main/java/org/apache/hadoop/hbase/protobuf/generated/RowProcessorProtos.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-protocol/src/main/java/org/apache/hadoop/hbase/protobuf/generated/RowProcessorProtos.java?rev=1542168&r1=1542167&r2=1542168&view=diff
==============================================================================
--- hbase/trunk/hbase-protocol/src/main/java/org/apache/hadoop/hbase/protobuf/generated/RowProcessorProtos.java (original)
+++ hbase/trunk/hbase-protocol/src/main/java/org/apache/hadoop/hbase/protobuf/generated/RowProcessorProtos.java Fri Nov 15 04:36:30 2013
@@ -50,6 +50,26 @@ public final class RowProcessorProtos {
      * <code>optional bytes row_processor_initializer_message = 3;</code>
      */
     com.google.protobuf.ByteString getRowProcessorInitializerMessage();
+
+    // optional uint64 nonce_group = 4;
+    /**
+     * <code>optional uint64 nonce_group = 4;</code>
+     */
+    boolean hasNonceGroup();
+    /**
+     * <code>optional uint64 nonce_group = 4;</code>
+     */
+    long getNonceGroup();
+
+    // optional uint64 nonce = 5;
+    /**
+     * <code>optional uint64 nonce = 5;</code>
+     */
+    boolean hasNonce();
+    /**
+     * <code>optional uint64 nonce = 5;</code>
+     */
+    long getNonce();
   }
   /**
    * Protobuf type {@code ProcessRequest}
@@ -117,6 +137,16 @@ public final class RowProcessorProtos {
               rowProcessorInitializerMessage_ = input.readBytes();
               break;
             }
+            case 32: {
+              bitField0_ |= 0x00000008;
+              nonceGroup_ = input.readUInt64();
+              break;
+            }
+            case 40: {
+              bitField0_ |= 0x00000010;
+              nonce_ = input.readUInt64();
+              break;
+            }
           }
         }
       } catch (com.google.protobuf.InvalidProtocolBufferException e) {
@@ -259,10 +289,44 @@ public final class RowProcessorProtos {
       return rowProcessorInitializerMessage_;
     }
 
+    // optional uint64 nonce_group = 4;
+    public static final int NONCE_GROUP_FIELD_NUMBER = 4;
+    private long nonceGroup_;
+    /**
+     * <code>optional uint64 nonce_group = 4;</code>
+     */
+    public boolean hasNonceGroup() {
+      return ((bitField0_ & 0x00000008) == 0x00000008);
+    }
+    /**
+     * <code>optional uint64 nonce_group = 4;</code>
+     */
+    public long getNonceGroup() {
+      return nonceGroup_;
+    }
+
+    // optional uint64 nonce = 5;
+    public static final int NONCE_FIELD_NUMBER = 5;
+    private long nonce_;
+    /**
+     * <code>optional uint64 nonce = 5;</code>
+     */
+    public boolean hasNonce() {
+      return ((bitField0_ & 0x00000010) == 0x00000010);
+    }
+    /**
+     * <code>optional uint64 nonce = 5;</code>
+     */
+    public long getNonce() {
+      return nonce_;
+    }
+
     private void initFields() {
       rowProcessorClassName_ = "";
       rowProcessorInitializerMessageName_ = "";
       rowProcessorInitializerMessage_ = com.google.protobuf.ByteString.EMPTY;
+      nonceGroup_ = 0L;
+      nonce_ = 0L;
     }
     private byte memoizedIsInitialized = -1;
     public final boolean isInitialized() {
@@ -289,6 +353,12 @@ public final class RowProcessorProtos {
       if (((bitField0_ & 0x00000004) == 0x00000004)) {
         output.writeBytes(3, rowProcessorInitializerMessage_);
       }
+      if (((bitField0_ & 0x00000008) == 0x00000008)) {
+        output.writeUInt64(4, nonceGroup_);
+      }
+      if (((bitField0_ & 0x00000010) == 0x00000010)) {
+        output.writeUInt64(5, nonce_);
+      }
       getUnknownFields().writeTo(output);
     }
 
@@ -310,6 +380,14 @@ public final class RowProcessorProtos {
         size += com.google.protobuf.CodedOutputStream
           .computeBytesSize(3, rowProcessorInitializerMessage_);
       }
+      if (((bitField0_ & 0x00000008) == 0x00000008)) {
+        size += com.google.protobuf.CodedOutputStream
+          .computeUInt64Size(4, nonceGroup_);
+      }
+      if (((bitField0_ & 0x00000010) == 0x00000010)) {
+        size += com.google.protobuf.CodedOutputStream
+          .computeUInt64Size(5, nonce_);
+      }
       size += getUnknownFields().getSerializedSize();
       memoizedSerializedSize = size;
       return size;
@@ -348,6 +426,16 @@ public final class RowProcessorProtos {
         result = result && getRowProcessorInitializerMessage()
             .equals(other.getRowProcessorInitializerMessage());
       }
+      result = result && (hasNonceGroup() == other.hasNonceGroup());
+      if (hasNonceGroup()) {
+        result = result && (getNonceGroup()
+            == other.getNonceGroup());
+      }
+      result = result && (hasNonce() == other.hasNonce());
+      if (hasNonce()) {
+        result = result && (getNonce()
+            == other.getNonce());
+      }
       result = result &&
           getUnknownFields().equals(other.getUnknownFields());
       return result;
@@ -373,6 +461,14 @@ public final class RowProcessorProtos {
         hash = (37 * hash) + ROW_PROCESSOR_INITIALIZER_MESSAGE_FIELD_NUMBER;
         hash = (53 * hash) + getRowProcessorInitializerMessage().hashCode();
       }
+      if (hasNonceGroup()) {
+        hash = (37 * hash) + NONCE_GROUP_FIELD_NUMBER;
+        hash = (53 * hash) + hashLong(getNonceGroup());
+      }
+      if (hasNonce()) {
+        hash = (37 * hash) + NONCE_FIELD_NUMBER;
+        hash = (53 * hash) + hashLong(getNonce());
+      }
       hash = (29 * hash) + getUnknownFields().hashCode();
       memoizedHashCode = hash;
       return hash;
@@ -488,6 +584,10 @@ public final class RowProcessorProtos {
         bitField0_ = (bitField0_ & ~0x00000002);
         rowProcessorInitializerMessage_ = com.google.protobuf.ByteString.EMPTY;
         bitField0_ = (bitField0_ & ~0x00000004);
+        nonceGroup_ = 0L;
+        bitField0_ = (bitField0_ & ~0x00000008);
+        nonce_ = 0L;
+        bitField0_ = (bitField0_ & ~0x00000010);
         return this;
       }
 
@@ -528,6 +628,14 @@ public final class RowProcessorProtos {
           to_bitField0_ |= 0x00000004;
         }
         result.rowProcessorInitializerMessage_ = rowProcessorInitializerMessage_;
+        if (((from_bitField0_ & 0x00000008) == 0x00000008)) {
+          to_bitField0_ |= 0x00000008;
+        }
+        result.nonceGroup_ = nonceGroup_;
+        if (((from_bitField0_ & 0x00000010) == 0x00000010)) {
+          to_bitField0_ |= 0x00000010;
+        }
+        result.nonce_ = nonce_;
         result.bitField0_ = to_bitField0_;
         onBuilt();
         return result;
@@ -557,6 +665,12 @@ public final class RowProcessorProtos {
         if (other.hasRowProcessorInitializerMessage()) {
           setRowProcessorInitializerMessage(other.getRowProcessorInitializerMessage());
         }
+        if (other.hasNonceGroup()) {
+          setNonceGroup(other.getNonceGroup());
+        }
+        if (other.hasNonce()) {
+          setNonce(other.getNonce());
+        }
         this.mergeUnknownFields(other.getUnknownFields());
         return this;
       }
@@ -772,6 +886,72 @@ public final class RowProcessorProtos {
         return this;
       }
 
+      // optional uint64 nonce_group = 4;
+      private long nonceGroup_ ;
+      /**
+       * <code>optional uint64 nonce_group = 4;</code>
+       */
+      public boolean hasNonceGroup() {
+        return ((bitField0_ & 0x00000008) == 0x00000008);
+      }
+      /**
+       * <code>optional uint64 nonce_group = 4;</code>
+       */
+      public long getNonceGroup() {
+        return nonceGroup_;
+      }
+      /**
+       * <code>optional uint64 nonce_group = 4;</code>
+       */
+      public Builder setNonceGroup(long value) {
+        bitField0_ |= 0x00000008;
+        nonceGroup_ = value;
+        onChanged();
+        return this;
+      }
+      /**
+       * <code>optional uint64 nonce_group = 4;</code>
+       */
+      public Builder clearNonceGroup() {
+        bitField0_ = (bitField0_ & ~0x00000008);
+        nonceGroup_ = 0L;
+        onChanged();
+        return this;
+      }
+
+      // optional uint64 nonce = 5;
+      private long nonce_ ;
+      /**
+       * <code>optional uint64 nonce = 5;</code>
+       */
+      public boolean hasNonce() {
+        return ((bitField0_ & 0x00000010) == 0x00000010);
+      }
+      /**
+       * <code>optional uint64 nonce = 5;</code>
+       */
+      public long getNonce() {
+        return nonce_;
+      }
+      /**
+       * <code>optional uint64 nonce = 5;</code>
+       */
+      public Builder setNonce(long value) {
+        bitField0_ |= 0x00000010;
+        nonce_ = value;
+        onChanged();
+        return this;
+      }
+      /**
+       * <code>optional uint64 nonce = 5;</code>
+       */
+      public Builder clearNonce() {
+        bitField0_ = (bitField0_ & ~0x00000010);
+        nonce_ = 0L;
+        onChanged();
+        return this;
+      }
+
       // @@protoc_insertion_point(builder_scope:ProcessRequest)
     }
 
@@ -1479,15 +1659,16 @@ public final class RowProcessorProtos {
       descriptor;
   static {
     java.lang.String[] descriptorData = {
-      "\n\022RowProcessor.proto\"\215\001\n\016ProcessRequest\022" +
+      "\n\022RowProcessor.proto\"\261\001\n\016ProcessRequest\022" +
       " \n\030row_processor_class_name\030\001 \002(\t\022.\n&row" +
       "_processor_initializer_message_name\030\002 \001(" +
       "\t\022)\n!row_processor_initializer_message\030\003" +
-      " \001(\014\"/\n\017ProcessResponse\022\034\n\024row_processor" +
-      "_result\030\001 \002(\0142C\n\023RowProcessorService\022,\n\007" +
-      "Process\022\017.ProcessRequest\032\020.ProcessRespon" +
-      "seBH\n*org.apache.hadoop.hbase.protobuf.g" +
-      "eneratedB\022RowProcessorProtosH\001\210\001\001\240\001\001"
+      " \001(\014\022\023\n\013nonce_group\030\004 \001(\004\022\r\n\005nonce\030\005 \001(\004" +
+      "\"/\n\017ProcessResponse\022\034\n\024row_processor_res" +
+      "ult\030\001 \002(\0142C\n\023RowProcessorService\022,\n\007Proc" +
+      "ess\022\017.ProcessRequest\032\020.ProcessResponseBH" +
+      "\n*org.apache.hadoop.hbase.protobuf.gener" +
+      "atedB\022RowProcessorProtosH\001\210\001\001\240\001\001"
     };
     com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner assigner =
       new com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner() {
@@ -1499,7 +1680,7 @@ public final class RowProcessorProtos {
           internal_static_ProcessRequest_fieldAccessorTable = new
             com.google.protobuf.GeneratedMessage.FieldAccessorTable(
               internal_static_ProcessRequest_descriptor,
-              new java.lang.String[] { "RowProcessorClassName", "RowProcessorInitializerMessageName", "RowProcessorInitializerMessage", });
+              new java.lang.String[] { "RowProcessorClassName", "RowProcessorInitializerMessageName", "RowProcessorInitializerMessage", "NonceGroup", "Nonce", });
           internal_static_ProcessResponse_descriptor =
             getDescriptor().getMessageTypes().get(1);
           internal_static_ProcessResponse_fieldAccessorTable = new

Modified: hbase/trunk/hbase-protocol/src/main/java/org/apache/hadoop/hbase/protobuf/generated/WALProtos.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-protocol/src/main/java/org/apache/hadoop/hbase/protobuf/generated/WALProtos.java?rev=1542168&r1=1542167&r2=1542168&view=diff
==============================================================================
--- hbase/trunk/hbase-protocol/src/main/java/org/apache/hadoop/hbase/protobuf/generated/WALProtos.java (original)
+++ hbase/trunk/hbase-protocol/src/main/java/org/apache/hadoop/hbase/protobuf/generated/WALProtos.java Fri Nov 15 04:36:30 2013
@@ -694,6 +694,26 @@ public final class WALProtos {
      */
     org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.UUIDOrBuilder getClusterIdsOrBuilder(
         int index);
+
+    // optional uint64 nonceGroup = 9;
+    /**
+     * <code>optional uint64 nonceGroup = 9;</code>
+     */
+    boolean hasNonceGroup();
+    /**
+     * <code>optional uint64 nonceGroup = 9;</code>
+     */
+    long getNonceGroup();
+
+    // optional uint64 nonce = 10;
+    /**
+     * <code>optional uint64 nonce = 10;</code>
+     */
+    boolean hasNonce();
+    /**
+     * <code>optional uint64 nonce = 10;</code>
+     */
+    long getNonce();
   }
   /**
    * Protobuf type {@code WALKey}
@@ -804,6 +824,16 @@ public final class WALProtos {
               clusterIds_.add(input.readMessage(org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.UUID.PARSER, extensionRegistry));
               break;
             }
+            case 72: {
+              bitField0_ |= 0x00000040;
+              nonceGroup_ = input.readUInt64();
+              break;
+            }
+            case 80: {
+              bitField0_ |= 0x00000080;
+              nonce_ = input.readUInt64();
+              break;
+            }
           }
         }
       } catch (com.google.protobuf.InvalidProtocolBufferException e) {
@@ -1078,6 +1108,38 @@ public final class WALProtos {
       return clusterIds_.get(index);
     }
 
+    // optional uint64 nonceGroup = 9;
+    public static final int NONCEGROUP_FIELD_NUMBER = 9;
+    private long nonceGroup_;
+    /**
+     * <code>optional uint64 nonceGroup = 9;</code>
+     */
+    public boolean hasNonceGroup() {
+      return ((bitField0_ & 0x00000040) == 0x00000040);
+    }
+    /**
+     * <code>optional uint64 nonceGroup = 9;</code>
+     */
+    public long getNonceGroup() {
+      return nonceGroup_;
+    }
+
+    // optional uint64 nonce = 10;
+    public static final int NONCE_FIELD_NUMBER = 10;
+    private long nonce_;
+    /**
+     * <code>optional uint64 nonce = 10;</code>
+     */
+    public boolean hasNonce() {
+      return ((bitField0_ & 0x00000080) == 0x00000080);
+    }
+    /**
+     * <code>optional uint64 nonce = 10;</code>
+     */
+    public long getNonce() {
+      return nonce_;
+    }
+
     private void initFields() {
       encodedRegionName_ = com.google.protobuf.ByteString.EMPTY;
       tableName_ = com.google.protobuf.ByteString.EMPTY;
@@ -1087,6 +1149,8 @@ public final class WALProtos {
       scopes_ = java.util.Collections.emptyList();
       followingKvCount_ = 0;
       clusterIds_ = java.util.Collections.emptyList();
+      nonceGroup_ = 0L;
+      nonce_ = 0L;
     }
     private byte memoizedIsInitialized = -1;
     public final boolean isInitialized() {
@@ -1158,6 +1222,12 @@ public final class WALProtos {
       for (int i = 0; i < clusterIds_.size(); i++) {
         output.writeMessage(8, clusterIds_.get(i));
       }
+      if (((bitField0_ & 0x00000040) == 0x00000040)) {
+        output.writeUInt64(9, nonceGroup_);
+      }
+      if (((bitField0_ & 0x00000080) == 0x00000080)) {
+        output.writeUInt64(10, nonce_);
+      }
       getUnknownFields().writeTo(output);
     }
 
@@ -1199,6 +1269,14 @@ public final class WALProtos {
         size += com.google.protobuf.CodedOutputStream
           .computeMessageSize(8, clusterIds_.get(i));
       }
+      if (((bitField0_ & 0x00000040) == 0x00000040)) {
+        size += com.google.protobuf.CodedOutputStream
+          .computeUInt64Size(9, nonceGroup_);
+      }
+      if (((bitField0_ & 0x00000080) == 0x00000080)) {
+        size += com.google.protobuf.CodedOutputStream
+          .computeUInt64Size(10, nonce_);
+      }
       size += getUnknownFields().getSerializedSize();
       memoizedSerializedSize = size;
       return size;
@@ -1256,6 +1334,16 @@ public final class WALProtos {
       }
       result = result && getClusterIdsList()
           .equals(other.getClusterIdsList());
+      result = result && (hasNonceGroup() == other.hasNonceGroup());
+      if (hasNonceGroup()) {
+        result = result && (getNonceGroup()
+            == other.getNonceGroup());
+      }
+      result = result && (hasNonce() == other.hasNonce());
+      if (hasNonce()) {
+        result = result && (getNonce()
+            == other.getNonce());
+      }
       result = result &&
           getUnknownFields().equals(other.getUnknownFields());
       return result;
@@ -1301,6 +1389,14 @@ public final class WALProtos {
         hash = (37 * hash) + CLUSTER_IDS_FIELD_NUMBER;
         hash = (53 * hash) + getClusterIdsList().hashCode();
       }
+      if (hasNonceGroup()) {
+        hash = (37 * hash) + NONCEGROUP_FIELD_NUMBER;
+        hash = (53 * hash) + hashLong(getNonceGroup());
+      }
+      if (hasNonce()) {
+        hash = (37 * hash) + NONCE_FIELD_NUMBER;
+        hash = (53 * hash) + hashLong(getNonce());
+      }
       hash = (29 * hash) + getUnknownFields().hashCode();
       memoizedHashCode = hash;
       return hash;
@@ -1445,6 +1541,10 @@ public final class WALProtos {
         } else {
           clusterIdsBuilder_.clear();
         }
+        nonceGroup_ = 0L;
+        bitField0_ = (bitField0_ & ~0x00000100);
+        nonce_ = 0L;
+        bitField0_ = (bitField0_ & ~0x00000200);
         return this;
       }
 
@@ -1519,6 +1619,14 @@ public final class WALProtos {
         } else {
           result.clusterIds_ = clusterIdsBuilder_.build();
         }
+        if (((from_bitField0_ & 0x00000100) == 0x00000100)) {
+          to_bitField0_ |= 0x00000040;
+        }
+        result.nonceGroup_ = nonceGroup_;
+        if (((from_bitField0_ & 0x00000200) == 0x00000200)) {
+          to_bitField0_ |= 0x00000080;
+        }
+        result.nonce_ = nonce_;
         result.bitField0_ = to_bitField0_;
         onBuilt();
         return result;
@@ -1605,6 +1713,12 @@ public final class WALProtos {
             }
           }
         }
+        if (other.hasNonceGroup()) {
+          setNonceGroup(other.getNonceGroup());
+        }
+        if (other.hasNonce()) {
+          setNonce(other.getNonce());
+        }
         this.mergeUnknownFields(other.getUnknownFields());
         return this;
       }
@@ -2614,6 +2728,72 @@ public final class WALProtos {
         return clusterIdsBuilder_;
       }
 
+      // optional uint64 nonceGroup = 9;
+      private long nonceGroup_ ;
+      /**
+       * <code>optional uint64 nonceGroup = 9;</code>
+       */
+      public boolean hasNonceGroup() {
+        return ((bitField0_ & 0x00000100) == 0x00000100);
+      }
+      /**
+       * <code>optional uint64 nonceGroup = 9;</code>
+       */
+      public long getNonceGroup() {
+        return nonceGroup_;
+      }
+      /**
+       * <code>optional uint64 nonceGroup = 9;</code>
+       */
+      public Builder setNonceGroup(long value) {
+        bitField0_ |= 0x00000100;
+        nonceGroup_ = value;
+        onChanged();
+        return this;
+      }
+      /**
+       * <code>optional uint64 nonceGroup = 9;</code>
+       */
+      public Builder clearNonceGroup() {
+        bitField0_ = (bitField0_ & ~0x00000100);
+        nonceGroup_ = 0L;
+        onChanged();
+        return this;
+      }
+
+      // optional uint64 nonce = 10;
+      private long nonce_ ;
+      /**
+       * <code>optional uint64 nonce = 10;</code>
+       */
+      public boolean hasNonce() {
+        return ((bitField0_ & 0x00000200) == 0x00000200);
+      }
+      /**
+       * <code>optional uint64 nonce = 10;</code>
+       */
+      public long getNonce() {
+        return nonce_;
+      }
+      /**
+       * <code>optional uint64 nonce = 10;</code>
+       */
+      public Builder setNonce(long value) {
+        bitField0_ |= 0x00000200;
+        nonce_ = value;
+        onChanged();
+        return this;
+      }
+      /**
+       * <code>optional uint64 nonce = 10;</code>
+       */
+      public Builder clearNonce() {
+        bitField0_ = (bitField0_ & ~0x00000200);
+        nonce_ = 0L;
+        onChanged();
+        return this;
+      }
+
       // @@protoc_insertion_point(builder_scope:WALKey)
     }
 
@@ -4812,22 +4992,23 @@ public final class WALProtos {
   static {
     java.lang.String[] descriptorData = {
       "\n\tWAL.proto\032\013HBase.proto\"$\n\tWALHeader\022\027\n" +
-      "\017has_compression\030\001 \001(\010\"\337\001\n\006WALKey\022\033\n\023enc" +
+      "\017has_compression\030\001 \001(\010\"\202\002\n\006WALKey\022\033\n\023enc" +
       "oded_region_name\030\001 \002(\014\022\022\n\ntable_name\030\002 \002" +
       "(\014\022\033\n\023log_sequence_number\030\003 \002(\004\022\022\n\nwrite" +
       "_time\030\004 \002(\004\022\035\n\ncluster_id\030\005 \001(\0132\005.UUIDB\002" +
       "\030\001\022\034\n\006scopes\030\006 \003(\0132\014.FamilyScope\022\032\n\022foll" +
       "owing_kv_count\030\007 \001(\r\022\032\n\013cluster_ids\030\010 \003(" +
-      "\0132\005.UUID\"=\n\013FamilyScope\022\016\n\006family\030\001 \002(\014\022" +
-      "\036\n\nscope_type\030\002 \002(\0162\n.ScopeType\"\251\001\n\024Comp" +
-      "actionDescriptor\022\022\n\ntable_name\030\001 \002(\014\022\033\n\023",
-      "encoded_region_name\030\002 \002(\014\022\023\n\013family_name" +
-      "\030\003 \002(\014\022\030\n\020compaction_input\030\004 \003(\t\022\031\n\021comp" +
-      "action_output\030\005 \003(\t\022\026\n\016store_home_dir\030\006 " +
-      "\002(\t\"\014\n\nWALTrailer*F\n\tScopeType\022\033\n\027REPLIC" +
-      "ATION_SCOPE_LOCAL\020\000\022\034\n\030REPLICATION_SCOPE" +
-      "_GLOBAL\020\001B?\n*org.apache.hadoop.hbase.pro" +
-      "tobuf.generatedB\tWALProtosH\001\210\001\000\240\001\001"
+      "\0132\005.UUID\022\022\n\nnonceGroup\030\t \001(\004\022\r\n\005nonce\030\n " +
+      "\001(\004\"=\n\013FamilyScope\022\016\n\006family\030\001 \002(\014\022\036\n\nsc" +
+      "ope_type\030\002 \002(\0162\n.ScopeType\"\251\001\n\024Compactio",
+      "nDescriptor\022\022\n\ntable_name\030\001 \002(\014\022\033\n\023encod" +
+      "ed_region_name\030\002 \002(\014\022\023\n\013family_name\030\003 \002(" +
+      "\014\022\030\n\020compaction_input\030\004 \003(\t\022\031\n\021compactio" +
+      "n_output\030\005 \003(\t\022\026\n\016store_home_dir\030\006 \002(\t\"\014" +
+      "\n\nWALTrailer*F\n\tScopeType\022\033\n\027REPLICATION" +
+      "_SCOPE_LOCAL\020\000\022\034\n\030REPLICATION_SCOPE_GLOB" +
+      "AL\020\001B?\n*org.apache.hadoop.hbase.protobuf" +
+      ".generatedB\tWALProtosH\001\210\001\000\240\001\001"
     };
     com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner assigner =
       new com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner() {
@@ -4845,7 +5026,7 @@ public final class WALProtos {
           internal_static_WALKey_fieldAccessorTable = new
             com.google.protobuf.GeneratedMessage.FieldAccessorTable(
               internal_static_WALKey_descriptor,
-              new java.lang.String[] { "EncodedRegionName", "TableName", "LogSequenceNumber", "WriteTime", "ClusterId", "Scopes", "FollowingKvCount", "ClusterIds", });
+              new java.lang.String[] { "EncodedRegionName", "TableName", "LogSequenceNumber", "WriteTime", "ClusterId", "Scopes", "FollowingKvCount", "ClusterIds", "NonceGroup", "Nonce", });
           internal_static_FamilyScope_descriptor =
             getDescriptor().getMessageTypes().get(2);
           internal_static_FamilyScope_fieldAccessorTable = new

Modified: hbase/trunk/hbase-protocol/src/main/protobuf/Client.proto
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-protocol/src/main/protobuf/Client.proto?rev=1542168&r1=1542167&r2=1542168&view=diff
==============================================================================
--- hbase/trunk/hbase-protocol/src/main/protobuf/Client.proto (original)
+++ hbase/trunk/hbase-protocol/src/main/protobuf/Client.proto Fri Nov 15 04:36:30 2013
@@ -133,6 +133,8 @@ message MutationProto {
   // 'cell' field above which is non-null when the cells are pb'd.
   optional int32 associated_cell_count = 8;
 
+  optional uint64 nonce = 9;
+
   enum Durability {
     USE_DEFAULT  = 0;
     SKIP_WAL     = 1;
@@ -182,6 +184,7 @@ message MutateRequest {
   required RegionSpecifier region = 1;
   required MutationProto mutation = 2;
   optional Condition condition = 3;
+  optional uint64 nonce_group = 4;
 }
 
 message MutateResponse {
@@ -345,6 +348,7 @@ message RegionActionResult {
  */
 message MultiRequest {
   repeated RegionAction regionAction = 1;
+  optional uint64 nonceGroup = 2;
 }
 
 message MultiResponse {

Modified: hbase/trunk/hbase-protocol/src/main/protobuf/MultiRowMutation.proto
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-protocol/src/main/protobuf/MultiRowMutation.proto?rev=1542168&r1=1542167&r2=1542168&view=diff
==============================================================================
--- hbase/trunk/hbase-protocol/src/main/protobuf/MultiRowMutation.proto (original)
+++ hbase/trunk/hbase-protocol/src/main/protobuf/MultiRowMutation.proto Fri Nov 15 04:36:30 2013
@@ -30,6 +30,8 @@ message MultiRowMutationProcessorRespons
 
 message MutateRowsRequest {
   repeated MutationProto mutation_request = 1;
+  optional uint64 nonce_group = 2;
+  optional uint64 nonce = 3;
 }
 
 message MutateRowsResponse {

Modified: hbase/trunk/hbase-protocol/src/main/protobuf/RowProcessor.proto
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-protocol/src/main/protobuf/RowProcessor.proto?rev=1542168&r1=1542167&r2=1542168&view=diff
==============================================================================
--- hbase/trunk/hbase-protocol/src/main/protobuf/RowProcessor.proto (original)
+++ hbase/trunk/hbase-protocol/src/main/protobuf/RowProcessor.proto Fri Nov 15 04:36:30 2013
@@ -30,6 +30,8 @@ message ProcessRequest {
   required string row_processor_class_name = 1;
   optional string row_processor_initializer_message_name = 2;
   optional bytes  row_processor_initializer_message = 3;
+  optional uint64 nonce_group = 4;
+  optional uint64 nonce = 5;
 }
 
 message ProcessResponse {

Modified: hbase/trunk/hbase-protocol/src/main/protobuf/WAL.proto
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-protocol/src/main/protobuf/WAL.proto?rev=1542168&r1=1542167&r2=1542168&view=diff
==============================================================================
--- hbase/trunk/hbase-protocol/src/main/protobuf/WAL.proto (original)
+++ hbase/trunk/hbase-protocol/src/main/protobuf/WAL.proto Fri Nov 15 04:36:30 2013
@@ -43,14 +43,19 @@ message WALKey {
 
   repeated FamilyScope scopes = 6;
   optional uint32 following_kv_count = 7;
+
   /*
   This field contains the list of clusters that have
   consumed the change
   */
   repeated UUID cluster_ids = 8;
+
+  optional uint64 nonceGroup = 9;
+  optional uint64 nonce = 10;
+
 /*
   optional CustomEntryType custom_entry_type = 9;
-  
+
   enum CustomEntryType {
     COMPACTION = 0;
   }

Modified: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/client/CoprocessorHConnection.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/client/CoprocessorHConnection.java?rev=1542168&r1=1542167&r2=1542168&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/client/CoprocessorHConnection.java (original)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/client/CoprocessorHConnection.java Fri Nov 15 04:36:30 2013
@@ -380,4 +380,9 @@ public class CoprocessorHConnection impl
   public boolean isDeadServer(ServerName serverName) {
     return delegate.isDeadServer(serverName);
   }
+
+  @Override
+  public NonceGenerator getNonceGenerator() {
+    return null; // don't use nonces for coprocessor connection
+  }
 }
\ No newline at end of file

Modified: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/BaseRowProcessorEndpoint.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/BaseRowProcessorEndpoint.java?rev=1542168&r1=1542167&r2=1542168&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/BaseRowProcessorEndpoint.java (original)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/BaseRowProcessorEndpoint.java Fri Nov 15 04:36:30 2013
@@ -26,6 +26,7 @@ import org.apache.hadoop.classification.
 import org.apache.hadoop.hbase.Coprocessor;
 import org.apache.hadoop.hbase.CoprocessorEnvironment;
 import org.apache.hadoop.hbase.HBaseInterfaceAudience;
+import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.protobuf.ResponseConverter;
 import org.apache.hadoop.hbase.protobuf.generated.RowProcessorProtos.ProcessRequest;
 import org.apache.hadoop.hbase.protobuf.generated.RowProcessorProtos.ProcessResponse;
@@ -68,7 +69,9 @@ extends RowProcessorService implements C
     try {
       RowProcessor<S,T> processor = constructRowProcessorFromRequest(request);
       HRegion region = env.getRegion();
-      region.processRowsWithLocks(processor);
+      long nonceGroup = request.hasNonceGroup() ? request.getNonceGroup() : HConstants.NO_NONCE;
+      long nonce = request.hasNonce() ? request.getNonce() : HConstants.NO_NONCE;
+      region.processRowsWithLocks(processor, nonceGroup, nonce);
       T result = processor.getResult();
       ProcessResponse.Builder b = ProcessResponse.newBuilder();
       b.setRowProcessorResult(result.toByteString());

Modified: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/MultiRowMutationEndpoint.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/MultiRowMutationEndpoint.java?rev=1542168&r1=1542167&r2=1542168&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/MultiRowMutationEndpoint.java (original)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/MultiRowMutationEndpoint.java Fri Nov 15 04:36:30 2013
@@ -28,6 +28,7 @@ import org.apache.hadoop.classification.
 import org.apache.hadoop.hbase.Coprocessor;
 import org.apache.hadoop.hbase.CoprocessorEnvironment;
 import org.apache.hadoop.hbase.HBaseInterfaceAudience;
+import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.HRegionInfo;
 import org.apache.hadoop.hbase.client.Mutation;
 import org.apache.hadoop.hbase.regionserver.HRegion;
@@ -80,7 +81,7 @@ public class MultiRowMutationEndpoint ex
 CoprocessorService, Coprocessor {
   private RegionCoprocessorEnvironment env;
   @Override
-  public void mutateRows(RpcController controller, MutateRowsRequest request, 
+  public void mutateRows(RpcController controller, MutateRowsRequest request,
       RpcCallback<MutateRowsResponse> done) {
     MutateRowsResponse response = MutateRowsResponse.getDefaultInstance();
     try {
@@ -110,7 +111,9 @@ CoprocessorService, Coprocessor {
         rowsToLock.add(m.getRow());
       }
       // call utility method on region
-      env.getRegion().mutateRowsWithLocks(mutations, rowsToLock);
+      long nonceGroup = request.hasNonceGroup() ? request.getNonceGroup() : HConstants.NO_NONCE;
+      long nonce = request.hasNonce() ? request.getNonce() : HConstants.NO_NONCE;
+      env.getRegion().mutateRowsWithLocks(mutations, rowsToLock, nonceGroup, nonce);
     } catch (IOException e) {
       ResponseConverter.setControllerException(controller, e);
     }

Modified: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/protobuf/ReplicationProtbufUtil.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/protobuf/ReplicationProtbufUtil.java?rev=1542168&r1=1542167&r2=1542168&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/protobuf/ReplicationProtbufUtil.java (original)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/protobuf/ReplicationProtbufUtil.java Fri Nov 15 04:36:30 2013
@@ -31,6 +31,7 @@ import java.util.UUID;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.hbase.Cell;
 import org.apache.hadoop.hbase.CellScanner;
+import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.KeyValue;
 import org.apache.hadoop.hbase.io.SizedCellScanner;
 import org.apache.hadoop.hbase.ipc.PayloadCarryingRpcController;
@@ -86,6 +87,7 @@ public class ReplicationProtbufUtil {
     HBaseProtos.UUID.Builder uuidBuilder = HBaseProtos.UUID.newBuilder();
     for (HLog.Entry entry: entries) {
       entryBuilder.clear();
+      // TODO: this duplicates a lot in HLogKey#getBuilder
       WALProtos.WALKey.Builder keyBuilder = entryBuilder.getKeyBuilder();
       HLogKey key = entry.getKey();
       keyBuilder.setEncodedRegionName(
@@ -93,6 +95,12 @@ public class ReplicationProtbufUtil {
       keyBuilder.setTableName(ZeroCopyLiteralByteString.wrap(key.getTablename().getName()));
       keyBuilder.setLogSequenceNumber(key.getLogSeqNum());
       keyBuilder.setWriteTime(key.getWriteTime());
+      if (key.getNonce() != HConstants.NO_NONCE) {
+        keyBuilder.setNonce(key.getNonce());
+      }
+      if (key.getNonceGroup() != HConstants.NO_NONCE) {
+        keyBuilder.setNonceGroup(key.getNonceGroup());
+      }
       for(UUID clusterId : key.getClusterIds()) {
         uuidBuilder.setLeastSigBits(clusterId.getLeastSignificantBits());
         uuidBuilder.setMostSigBits(clusterId.getMostSignificantBits());

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=1542168&r1=1542167&r2=1542168&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 Fri Nov 15 04:36:30 2013
@@ -68,6 +68,7 @@ import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hbase.Cell;
 import org.apache.hadoop.hbase.CellUtil;
 import org.apache.hadoop.hbase.CompoundConfiguration;
+import org.apache.hadoop.hbase.DoNotRetryIOException;
 import org.apache.hadoop.hbase.DroppedSnapshotException;
 import org.apache.hadoop.hbase.HBaseConfiguration;
 import org.apache.hadoop.hbase.HColumnDescriptor;
@@ -122,6 +123,8 @@ import org.apache.hadoop.hbase.regionser
 import org.apache.hadoop.hbase.regionserver.wal.HLog;
 import org.apache.hadoop.hbase.regionserver.wal.HLogFactory;
 import org.apache.hadoop.hbase.regionserver.wal.HLogKey;
+import org.apache.hadoop.hbase.regionserver.wal.HLogSplitter;
+import org.apache.hadoop.hbase.regionserver.wal.HLogSplitter.MutationReplay;
 import org.apache.hadoop.hbase.regionserver.wal.HLogUtil;
 import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
 import org.apache.hadoop.hbase.snapshot.SnapshotDescriptionUtils;
@@ -1897,7 +1900,7 @@ public class HRegion implements HeapSize
    * accumulating status codes and tracking the index at which processing
    * is proceeding.
    */
-  private static class BatchOperationInProgress<T> {
+  private abstract static class BatchOperationInProgress<T> {
     T[] operations;
     int nextIndexToProcess = 0;
     OperationStatus[] retCodeDetails;
@@ -1910,11 +1913,84 @@ public class HRegion implements HeapSize
       Arrays.fill(this.retCodeDetails, OperationStatus.NOT_RUN);
     }
 
+    public abstract Mutation getMutation(int index);
+    public abstract long getNonceGroup(int index);
+    public abstract long getNonce(int index);
+    /** This method is potentially expensive and should only be used for non-replay CP path. */
+    public abstract Mutation[] getMutationsForCoprocs();
+    public abstract boolean isInReplay();
+
     public boolean isDone() {
       return nextIndexToProcess == operations.length;
     }
   }
 
+  private static class MutationBatch extends BatchOperationInProgress<Mutation> {
+    private long nonceGroup;
+    private long nonce;
+    public MutationBatch(Mutation[] operations, long nonceGroup, long nonce) {
+      super(operations);
+      this.nonceGroup = nonceGroup;
+      this.nonce = nonce;
+    }
+
+    public Mutation getMutation(int index) {
+      return this.operations[index];
+    }
+
+    @Override
+    public long getNonceGroup(int index) {
+      return nonceGroup;
+    }
+
+    @Override
+    public long getNonce(int index) {
+      return nonce;
+    }
+
+    @Override
+    public Mutation[] getMutationsForCoprocs() {
+      return this.operations;
+    }
+
+    @Override
+    public boolean isInReplay() {
+      return false;
+    }
+  }
+
+  private static class ReplayBatch extends BatchOperationInProgress<HLogSplitter.MutationReplay> {
+    public ReplayBatch(MutationReplay[] operations) {
+      super(operations);
+    }
+
+    @Override
+    public Mutation getMutation(int index) {
+      return this.operations[index].mutation;
+    }
+
+    @Override
+    public long getNonceGroup(int index) {
+      return this.operations[index].nonceGroup;
+    }
+
+    @Override
+    public long getNonce(int index) {
+      return this.operations[index].nonce;
+    }
+
+    @Override
+    public Mutation[] getMutationsForCoprocs() {
+      assert false;
+      throw new RuntimeException("Should not be called for replay batch");
+    }
+
+    @Override
+    public boolean isInReplay() {
+      return true;
+    }
+  }
+
   /**
    * Perform a batch of mutations.
    * It supports only Put and Delete mutations and will ignore other types passed.
@@ -1923,8 +1999,28 @@ public class HRegion implements HeapSize
    *         OperationStatusCode and the exceptionMessage if any.
    * @throws IOException
    */
+  public OperationStatus[] batchMutate(
+      Mutation[] mutations, long nonceGroup, long nonce) throws IOException {
+    // As it stands, this is used for 3 things
+    //  * 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));
+  }
+
   public OperationStatus[] batchMutate(Mutation[] mutations) throws IOException {
-    return batchMutate(mutations, false);
+    return batchMutate(mutations, HConstants.NO_NONCE, HConstants.NO_NONCE);
+  }
+
+  /**
+   * Replay a batch of mutations.
+   * @param mutations mutations to replay.
+   * @return
+   * @throws IOException
+   */
+  public OperationStatus[] batchReplay(HLogSplitter.MutationReplay[] mutations)
+      throws IOException {
+    return batchMutate(new ReplayBatch(mutations));
   }
 
   /**
@@ -1935,21 +2031,16 @@ public class HRegion implements HeapSize
    *         OperationStatusCode and the exceptionMessage if any.
    * @throws IOException
    */
-  OperationStatus[] batchMutate(Mutation[] mutations, boolean isReplay)
-      throws IOException {
-    BatchOperationInProgress<Mutation> batchOp =
-      new BatchOperationInProgress<Mutation>(mutations);
-
+  OperationStatus[] batchMutate(BatchOperationInProgress<?> batchOp) throws IOException {
     boolean initialized = false;
-
     while (!batchOp.isDone()) {
-      if (!isReplay) {
+      if (!batchOp.isInReplay()) {
         checkReadOnly();
       }
       checkResources();
 
       long newSize;
-      if (isReplay) {
+      if (batchOp.isInReplay()) {
         startRegionOperation(Operation.REPLAY_BATCH_MUTATE);
       } else {
         startRegionOperation(Operation.BATCH_MUTATE);
@@ -1957,13 +2048,13 @@ public class HRegion implements HeapSize
 
       try {
         if (!initialized) {
-          if (!isReplay) {
+          if (!batchOp.isInReplay()) {
             this.writeRequestsCount.increment();
             doPreMutationHook(batchOp);
           }
           initialized = true;
         }
-        long addedSize = doMiniBatchMutation(batchOp, isReplay);
+        long addedSize = doMiniBatchMutation(batchOp);
         newSize = this.addAndGetGlobalMemstoreSize(addedSize);
       } finally {
         closeRegionOperation();
@@ -1976,13 +2067,13 @@ public class HRegion implements HeapSize
   }
 
 
-  private void doPreMutationHook(BatchOperationInProgress<Mutation> batchOp)
+  private void doPreMutationHook(BatchOperationInProgress<?> batchOp)
       throws IOException {
     /* Run coprocessor pre hook outside of locks to avoid deadlock */
     WALEdit walEdit = new WALEdit();
     if (coprocessorHost != null) {
       for (int i = 0 ; i < batchOp.operations.length; i++) {
-        Mutation m = batchOp.operations[i];
+        Mutation m = batchOp.getMutation(i);
         if (m instanceof Put) {
           if (coprocessorHost.prePut((Put) m, walEdit, m.getDurability())) {
             // pre hook says skip this Put
@@ -2011,9 +2102,8 @@ public class HRegion implements HeapSize
   }
 
   @SuppressWarnings("unchecked")
-  private long doMiniBatchMutation(BatchOperationInProgress<Mutation> batchOp,
-      boolean isInReplay) throws IOException {
-
+  private long doMiniBatchMutation(BatchOperationInProgress<?> batchOp) throws IOException {
+    boolean isInReplay = batchOp.isInReplay();
     // variable to note if all Put items are for the same CF -- metrics related
     boolean putsCfSetConsistent = true;
     //The set of columnFamilies first seen for Put.
@@ -2023,6 +2113,7 @@ public class HRegion implements HeapSize
     //The set of columnFamilies first seen for Delete.
     Set<byte[]> deletesCfSet = null;
 
+    long currentNonceGroup = HConstants.NO_NONCE, currentNonce = HConstants.NO_NONCE;
     WALEdit walEdit = new WALEdit(isInReplay);
     MultiVersionConsistencyControl.WriteEntry w = null;
     long txid = 0;
@@ -2046,7 +2137,7 @@ public class HRegion implements HeapSize
       int numReadyToWrite = 0;
       long now = EnvironmentEdgeManager.currentTimeMillis();
       while (lastIndexExclusive < batchOp.operations.length) {
-        Mutation mutation = batchOp.operations[lastIndexExclusive];
+        Mutation mutation = batchOp.getMutation(lastIndexExclusive);
         boolean isPutMutation = mutation instanceof Put;
 
         Map<byte[], List<Cell>> familyMap = mutation.getFamilyCellMap();
@@ -2145,7 +2236,7 @@ public class HRegion implements HeapSize
         if (batchOp.retCodeDetails[i].getOperationStatusCode()
             != OperationStatusCode.NOT_RUN) continue;
 
-        Mutation mutation = batchOp.operations[i];
+        Mutation mutation = batchOp.getMutation(i);
         if (mutation instanceof Put) {
           updateKVTimestamps(familyMaps[i].values(), byteNow);
           noOfPuts++;
@@ -2167,7 +2258,7 @@ public class HRegion implements HeapSize
       // calling the pre CP hook for batch mutation
       if (!isInReplay && coprocessorHost != null) {
         MiniBatchOperationInProgress<Mutation> miniBatchOp =
-          new MiniBatchOperationInProgress<Mutation>(batchOp.operations,
+          new MiniBatchOperationInProgress<Mutation>(batchOp.getMutationsForCoprocs(),
           batchOp.retCodeDetails, batchOp.walEditsFromCoprocessors, firstIndex, lastIndexExclusive);
         if (coprocessorHost.preBatchMutate(miniBatchOp)) return 0L;
       }
@@ -2193,6 +2284,7 @@ public class HRegion implements HeapSize
       // ------------------------------------
       // STEP 4. Build WAL edit
       // ----------------------------------
+      boolean hasWalAppends = false;
       Durability durability = Durability.USE_DEFAULT;
       for (int i = firstIndex; i < lastIndexExclusive; i++) {
         // Skip puts that were determined to be invalid during preprocessing
@@ -2202,7 +2294,7 @@ public class HRegion implements HeapSize
         }
         batchOp.retCodeDetails[i] = OperationStatus.SUCCESS;
 
-        Mutation m = batchOp.operations[i];
+        Mutation m = batchOp.getMutation(i);
         Durability tmpDur = getEffectiveDurability(m.getDurability());
         if (tmpDur.ordinal() > durability.ordinal()) {
           durability = tmpDur;
@@ -2212,6 +2304,22 @@ public class HRegion implements HeapSize
           continue;
         }
 
+        long nonceGroup = batchOp.getNonceGroup(i), nonce = batchOp.getNonce(i);
+        // In replay, the batch may contain multiple nonces. If so, write WALEdit for each.
+        // Given how nonces are originally written, these should be contiguous.
+        // txid should always increase, so having the last one is ok.
+        if (nonceGroup != currentNonceGroup || nonce != currentNonce) {
+          if (walEdit.size() > 0) {
+            assert isInReplay;
+            txid = this.log.appendNoSync(this.getRegionInfo(), htableDescriptor.getTableName(),
+                  walEdit, m.getClusterIds(), now, htableDescriptor, this.sequenceId, true,
+                  currentNonceGroup, currentNonce);
+            hasWalAppends = true;
+          }
+          currentNonceGroup = nonceGroup;
+          currentNonce = nonce;
+        }
+
         // Add WAL edits by CP
         WALEdit fromCP = batchOp.walEditsFromCoprocessors[i];
         if (fromCP != null) {
@@ -2223,12 +2331,14 @@ public class HRegion implements HeapSize
       }
 
       // -------------------------
-      // STEP 5. Append the edit to WAL. Do not sync wal.
+      // STEP 5. Append the final edit to WAL. Do not sync wal.
       // -------------------------
-      Mutation mutation = batchOp.operations[firstIndex];
+      Mutation mutation = batchOp.getMutation(firstIndex);
       if (walEdit.size() > 0) {
         txid = this.log.appendNoSync(this.getRegionInfo(), this.htableDescriptor.getTableName(),
-              walEdit, mutation.getClusterIds(), now, this.htableDescriptor, this.sequenceId);
+              walEdit, mutation.getClusterIds(), now, this.htableDescriptor, this.sequenceId,
+              true, currentNonceGroup, currentNonce);
+        hasWalAppends = true;
       }
 
       // -------------------------------
@@ -2243,14 +2353,14 @@ public class HRegion implements HeapSize
       // -------------------------
       // STEP 7. Sync wal.
       // -------------------------
-      if (walEdit.size() > 0) {
+      if (hasWalAppends) {
         syncOrDefer(txid, durability);
       }
       walSyncSuccessful = true;
       // calling the post CP hook for batch mutation
       if (!isInReplay && coprocessorHost != null) {
         MiniBatchOperationInProgress<Mutation> miniBatchOp =
-          new MiniBatchOperationInProgress<Mutation>(batchOp.operations,
+          new MiniBatchOperationInProgress<Mutation>(batchOp.getMutationsForCoprocs(),
           batchOp.retCodeDetails, batchOp.walEditsFromCoprocessors, firstIndex, lastIndexExclusive);
         coprocessorHost.postBatchMutate(miniBatchOp);
       }
@@ -2274,7 +2384,7 @@ public class HRegion implements HeapSize
               != OperationStatusCode.SUCCESS) {
             continue;
           }
-          Mutation m = batchOp.operations[i];
+          Mutation m = batchOp.getMutation(i);
           if (m instanceof Put) {
             coprocessorHost.postPut((Put) m, walEdit, m.getDurability());
           } else {
@@ -2362,10 +2472,10 @@ public class HRegion implements HeapSize
     boolean isPut = w instanceof Put;
     if (!isPut && !(w instanceof Delete))
       throw new org.apache.hadoop.hbase.DoNotRetryIOException("Action must " +
-      		"be Put or Delete");
+          "be Put or Delete");
     if (!Bytes.equals(row, w.getRow())) {
       throw new org.apache.hadoop.hbase.DoNotRetryIOException("Action's " +
-      		"getRow must match the passed row");
+          "getRow must match the passed row");
     }
 
     startRegionOperation();
@@ -2435,9 +2545,10 @@ public class HRegion implements HeapSize
     }
   }
 
-  private void doBatchMutate(Mutation mutation) throws IOException,
-      org.apache.hadoop.hbase.DoNotRetryIOException {
-    OperationStatus[] batchMutate = this.batchMutate(new Mutation[] { mutation });
+  private void doBatchMutate(Mutation mutation) throws IOException, DoNotRetryIOException {
+    // Currently this is only called for puts and deletes, so no nonces.
+    OperationStatus[] batchMutate = this.batchMutate(new Mutation[] { mutation },
+        HConstants.NO_NONCE, HConstants.NO_NONCE);
     if (batchMutate[0].getOperationStatusCode().equals(OperationStatusCode.SANITY_CHECK_FAILURE)) {
       throw new FailedSanityCheckException(batchMutate[0].getExceptionMsg());
     } else if (batchMutate[0].getOperationStatusCode().equals(OperationStatusCode.BAD_FAMILY)) {
@@ -2628,7 +2739,7 @@ public class HRegion implements HeapSize
    * called when a Put/Delete has updated memstore but subequently fails to update
    * the wal. This method is then invoked to rollback the memstore.
    */
-  private void rollbackMemstore(BatchOperationInProgress<Mutation> batchOp,
+  private void rollbackMemstore(BatchOperationInProgress<?> batchOp,
                                 Map<byte[], List<Cell>>[] familyMaps,
                                 int start, int end) {
     int kvsRolledback = 0;
@@ -2901,6 +3012,7 @@ public class HRegion implements HeapSize
       HLog.Entry entry;
       Store store = null;
       boolean reported_once = false;
+      ServerNonceManager ng = this.rsServices == null ? null : this.rsServices.getNonceManager();
 
       try {
         // How many edits seen before we check elapsed time
@@ -2916,6 +3028,10 @@ public class HRegion implements HeapSize
           HLogKey key = entry.getKey();
           WALEdit val = entry.getEdit();
 
+          if (ng != null) { // some test, or nonces disabled
+            ng.reportOperationFromWal(key.getNonceGroup(), key.getNonce(), key.getWriteTime());
+          }
+
           if (reporter != null) {
             intervalEdits += val.size();
             if (intervalEdits >= interval) {
@@ -4395,35 +4511,47 @@ public class HRegion implements HeapSize
   }
 
   public void mutateRow(RowMutations rm) throws IOException {
+    // Don't need nonces here - RowMutations only supports puts and deletes
     mutateRowsWithLocks(rm.getMutations(), Collections.singleton(rm.getRow()));
   }
 
   /**
+   * Perform atomic mutations within the region w/o nonces.
+   * See {@link #mutateRowsWithLocks(Collection, Collection, long, long)}
+   */
+  public void mutateRowsWithLocks(Collection<Mutation> mutations,
+      Collection<byte[]> rowsToLock) throws IOException {
+    mutateRowsWithLocks(mutations, rowsToLock, HConstants.NO_NONCE, HConstants.NO_NONCE);
+  }
+
+  /**
    * Perform atomic mutations within the region.
    * @param mutations The list of mutations to perform.
    * <code>mutations</code> can contain operations for multiple rows.
    * Caller has to ensure that all rows are contained in this region.
    * @param rowsToLock Rows to lock
+   * @param nonceGroup Optional nonce group of the operation (client Id)
+   * @param nonce Optional nonce of the operation (unique random id to ensure "more idempotence")
    * If multiple rows are locked care should be taken that
    * <code>rowsToLock</code> is sorted in order to avoid deadlocks.
    * @throws IOException
    */
   public void mutateRowsWithLocks(Collection<Mutation> mutations,
-      Collection<byte[]> rowsToLock) throws IOException {
-
-    MultiRowMutationProcessor proc =
-        new MultiRowMutationProcessor(mutations, rowsToLock);
-    processRowsWithLocks(proc, -1);
+      Collection<byte[]> rowsToLock, long nonceGroup, long nonce) throws IOException {
+    MultiRowMutationProcessor proc = new MultiRowMutationProcessor(mutations, rowsToLock);
+    processRowsWithLocks(proc, -1, nonceGroup, nonce);
   }
 
   /**
    * Performs atomic multiple reads and writes on a given row.
    *
    * @param processor The object defines the reads and writes to a row.
+   * @param nonceGroup Optional nonce group of the operation (client Id)
+   * @param nonce Optional nonce of the operation (unique random id to ensure "more idempotence")
    */
-  public void processRowsWithLocks(RowProcessor<?,?> processor)
+  public void processRowsWithLocks(RowProcessor<?,?> processor, long nonceGroup, long nonce)
       throws IOException {
-    processRowsWithLocks(processor, rowProcessorTimeout);
+    processRowsWithLocks(processor, rowProcessorTimeout, nonceGroup, nonce);
   }
 
   /**
@@ -4432,9 +4560,11 @@ public class HRegion implements HeapSize
    * @param processor The object defines the reads and writes to a row.
    * @param timeout The timeout of the processor.process() execution
    *                Use a negative number to switch off the time bound
+   * @param nonceGroup Optional nonce group of the operation (client Id)
+   * @param nonce Optional nonce of the operation (unique random id to ensure "more idempotence")
    */
-  public void processRowsWithLocks(RowProcessor<?,?> processor, long timeout)
-      throws IOException {
+  public void processRowsWithLocks(RowProcessor<?,?> processor, long timeout,
+      long nonceGroup, long nonce) throws IOException {
 
     for (byte[] row : processor.getRowsToLock()) {
       checkRow(row, "processRowsWithLocks");
@@ -4506,7 +4636,7 @@ public class HRegion implements HeapSize
           if (!walEdit.isEmpty()) {
             txid = this.log.appendNoSync(this.getRegionInfo(),
               this.htableDescriptor.getTableName(), walEdit, processor.getClusterIds(), now,
-              this.htableDescriptor, this.sequenceId);
+              this.htableDescriptor, this.sequenceId, true, nonceGroup, nonce);
           }
           // 8. Release region lock
           if (locked) {
@@ -4609,6 +4739,10 @@ public class HRegion implements HeapSize
     }
   }
 
+  public Result append(Append append) throws IOException {
+    return append(append, HConstants.NO_NONCE, HConstants.NO_NONCE);
+  }
+
   // TODO: There's a lot of boiler plate code identical
   // to increment... See how to better unify that.
   /**
@@ -4618,7 +4752,7 @@ public class HRegion implements HeapSize
    * @return new keyvalues after increment
    * @throws IOException
    */
-  public Result append(Append append)
+  public Result append(Append append, long nonceGroup, long nonce)
       throws IOException {
     byte[] row = append.getRow();
     checkRow(row, "append");
@@ -4749,7 +4883,8 @@ public class HRegion implements HeapSize
             // as a Put.
             txid = this.log.appendNoSync(this.getRegionInfo(),
               this.htableDescriptor.getTableName(), walEdits, new ArrayList<UUID>(),
-              EnvironmentEdgeManager.currentTimeMillis(), this.htableDescriptor, this.sequenceId);
+              EnvironmentEdgeManager.currentTimeMillis(), this.htableDescriptor, this.sequenceId,
+              true, nonceGroup, nonce);
           } else {
             recordMutationWithoutWal(append.getFamilyCellMap());
           }
@@ -4801,13 +4936,17 @@ public class HRegion implements HeapSize
     return append.isReturnResults() ? Result.create(allKVs) : null;
   }
 
+  public Result increment(Increment increment) throws IOException {
+    return increment(increment, HConstants.NO_NONCE, HConstants.NO_NONCE);
+  }
+
   /**
    * Perform one or more increment operations on a row.
    * @param increment
    * @return new keyvalues after increment
    * @throws IOException
    */
-  public Result increment(Increment increment)
+  public Result increment(Increment increment, long nonceGroup, long nonce)
   throws IOException {
     byte [] row = increment.getRow();
     checkRow(row, "increment");
@@ -4923,7 +5062,8 @@ public class HRegion implements HeapSize
             // as a Put.
             txid = this.log.appendNoSync(this.getRegionInfo(),
               this.htableDescriptor.getTableName(), walEdits, new ArrayList<UUID>(),
-              EnvironmentEdgeManager.currentTimeMillis(), this.htableDescriptor, this.sequenceId);
+              EnvironmentEdgeManager.currentTimeMillis(), this.htableDescriptor, this.sequenceId,
+              true, nonceGroup, nonce);
           } else {
             recordMutationWithoutWal(increment.getFamilyCellMap());
           }

Modified: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java?rev=1542168&r1=1542167&r2=1542168&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java (original)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java Fri Nov 15 04:36:30 2013
@@ -19,6 +19,7 @@
 package org.apache.hadoop.hbase.regionserver;
 
 import java.io.IOException;
+import java.io.InterruptedIOException;
 import java.lang.Thread.UncaughtExceptionHandler;
 import java.lang.annotation.Retention;
 import java.lang.annotation.RetentionPolicy;
@@ -93,6 +94,7 @@ import org.apache.hadoop.hbase.client.Ro
 import org.apache.hadoop.hbase.client.Scan;
 import org.apache.hadoop.hbase.coprocessor.CoprocessorHost;
 import org.apache.hadoop.hbase.exceptions.FailedSanityCheckException;
+import org.apache.hadoop.hbase.exceptions.OperationConflictException;
 import org.apache.hadoop.hbase.exceptions.OutOfOrderScannerNextException;
 import org.apache.hadoop.hbase.exceptions.RegionMovedException;
 import org.apache.hadoop.hbase.exceptions.RegionOpeningException;
@@ -441,6 +443,9 @@ public class HRegionServer implements Cl
   /** The health check chore. */
   private HealthCheckChore healthCheckChore;
 
+  /** The nonce manager chore. */
+  private Chore nonceManagerChore;
+
   /**
    * The server name the Master sees us as.  Its made from the hostname the
    * master passes us, port, and server startcode. Gets set after registration
@@ -490,6 +495,26 @@ public class HRegionServer implements Cl
   // Table level lock manager for locking for region operations
   private TableLockManager tableLockManager;
 
+  /**
+   * Nonce manager. Nonces are used to make operations like increment and append idempotent
+   * in the case where client doesn't receive the response from a successful operation and
+   * retries. We track the successful ops for some time via a nonce sent by client and handle
+   * duplicate operations (currently, by failing them; in future we might use MVCC to return
+   * result). Nonces are also recovered from WAL during, recovery; however, the caveats (from
+   * HBASE-3787) are:
+   * - WAL recovery is optimized, and under high load we won't read nearly nonce-timeout worth
+   *   of past records. If we don't read the records, we don't read and recover the nonces.
+   *   Some WALs within nonce-timeout at recovery may not even be present due to rolling/cleanup.
+   * - There's no WAL recovery during normal region move, so nonces will not be transfered.
+   * We can have separate additional "Nonce WAL". It will just contain bunch of numbers and
+   * won't be flushed on main path - because WAL itself also contains nonces, if we only flush
+   * it before memstore flush, for a given nonce we will either see it in the WAL (if it was
+   * never flushed to disk, it will be part of recovery), or we'll see it as part of the nonce
+   * log (or both occasionally, which doesn't matter). Nonce log file can be deleted after the
+   * latest nonce in it expired. It can also be recovered during move.
+   */
+  private final ServerNonceManager nonceManager;
+
   private UserProvider userProvider;
 
   /**
@@ -529,6 +554,9 @@ public class HRegionServer implements Cl
 
     this.sleeper = new Sleeper(this.msgInterval, this);
 
+    boolean isNoncesEnabled = conf.getBoolean(HConstants.HBASE_RS_NONCES_ENABLED, true);
+    this.nonceManager = isNoncesEnabled ? new ServerNonceManager(this.conf) : null;
+
     this.maxScannerResultSize = conf.getLong(
       HConstants.HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE_KEY,
       HConstants.DEFAULT_HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE);
@@ -789,6 +817,11 @@ public class HRegionServer implements Cl
     // Create the thread to clean the moved regions list
     movedRegionsCleaner = MovedRegionsCleaner.createAndStart(this);
 
+    if (this.nonceManager != null) {
+      // Create the chore that cleans up nonces.
+      nonceManagerChore = this.nonceManager.createCleanupChore(this);
+    }
+
     // Setup RPC client for master communication
     rpcClient = new RpcClient(conf, clusterId, new InetSocketAddress(
         this.isa.getAddress(), 0));
@@ -910,6 +943,9 @@ public class HRegionServer implements Cl
     if (this.healthCheckChore != null) {
       this.healthCheckChore.interrupt();
     }
+    if (this.nonceManagerChore != null) {
+      this.nonceManagerChore.interrupt();
+    }
 
     // Stop the snapshot handler, forcefully killing all running tasks
     try {
@@ -1556,8 +1592,11 @@ public class HRegionServer implements Cl
     Threads.setDaemonThreadRunning(this.periodicFlusher.getThread(), n +
         ".periodicFlusher", uncaughtExceptionHandler);
     if (this.healthCheckChore != null) {
-    Threads
-        .setDaemonThreadRunning(this.healthCheckChore.getThread(), n + ".healthChecker",
+      Threads.setDaemonThreadRunning(this.healthCheckChore.getThread(), n + ".healthChecker",
+            uncaughtExceptionHandler);
+    }
+    if (this.nonceManagerChore != null) {
+      Threads.setDaemonThreadRunning(this.nonceManagerChore.getThread(), n + ".nonceCleaner",
             uncaughtExceptionHandler);
     }
 
@@ -1811,6 +1850,9 @@ public class HRegionServer implements Cl
    * have already been called.
    */
   protected void join() {
+    if (this.nonceManagerChore != null) {
+      Threads.shutdown(this.nonceManagerChore.getThread());
+    }
     Threads.shutdown(this.compactionChecker.getThread());
     Threads.shutdown(this.periodicFlusher.getThread());
     this.cacheFlusher.join();
@@ -2826,15 +2868,19 @@ public class HRegionServer implements Cl
       if (!region.getRegionInfo().isMetaTable()) {
         cacheFlusher.reclaimMemStoreMemory();
       }
+      long nonceGroup = request.hasNonceGroup()
+          ? request.getNonceGroup() : HConstants.NO_NONCE;
       Result r = null;
       Boolean processed = null;
       MutationType type = mutation.getMutateType();
       switch (type) {
       case APPEND:
-        r = append(region, mutation, cellScanner);
+        // TODO: this doesn't actually check anything.
+        r = append(region, mutation, cellScanner, nonceGroup);
         break;
       case INCREMENT:
-        r = increment(region, mutation, cellScanner);
+        // TODO: this doesn't actually check anything.
+        r = increment(region, mutation, cellScanner, nonceGroup);
         break;
       case PUT:
         Put put = ProtobufUtil.toPut(mutation, cellScanner);
@@ -3251,6 +3297,8 @@ public class HRegionServer implements Cl
     CellScanner cellScanner = controller != null ? controller.cellScanner(): null;
     if (controller != null) controller.setCellScanner(null);
 
+    long nonceGroup = request.hasNonceGroup() ? request.getNonceGroup() : HConstants.NO_NONCE;
+
     // this will contain all the cells that we need to return. It's created later, if needed.
     List<CellScannable> cellsToReturn = null;
     MultiResponse.Builder responseBuilder = MultiResponse.newBuilder();
@@ -3279,7 +3327,7 @@ public class HRegionServer implements Cl
       } else {
         // doNonAtomicRegionMutation manages the exception internally
         cellsToReturn = doNonAtomicRegionMutation(region, regionAction, cellScanner,
-            regionActionResultBuilder, cellsToReturn);
+            regionActionResultBuilder, cellsToReturn, nonceGroup);
       }
       responseBuilder.addRegionActionResult(regionActionResultBuilder.build());
     }
@@ -3303,7 +3351,7 @@ public class HRegionServer implements Cl
    */
   private List<CellScannable> doNonAtomicRegionMutation(final HRegion region,
       final RegionAction actions, final CellScanner cellScanner,
-      final RegionActionResult.Builder builder, List<CellScannable> cellsToReturn) {
+      final RegionActionResult.Builder builder, List<CellScannable> cellsToReturn, long nonceGroup) {
     // Gather up CONTIGUOUS Puts and Deletes in this mutations List.  Idea is that rather than do
     // one at a time, we instead pass them in batch.  Be aware that the corresponding
     // ResultOrException instance that matches each Put or Delete is then added down in the
@@ -3326,10 +3374,10 @@ public class HRegionServer implements Cl
           }
           switch (type) {
           case APPEND:
-            r = append(region, action.getMutation(), cellScanner);
+            r = append(region, action.getMutation(), cellScanner, nonceGroup);
             break;
           case INCREMENT:
-            r = increment(region, action.getMutation(), cellScanner);
+            r = increment(region, action.getMutation(), cellScanner,  nonceGroup);
             break;
           case PUT:
           case DELETE:
@@ -3862,12 +3910,18 @@ public class HRegionServer implements Cl
         entries.get(0).getKey().getEncodedRegionName().toStringUtf8());
       RegionCoprocessorHost coprocessorHost = region.getCoprocessorHost();
       List<Pair<HLogKey, WALEdit>> walEntries = new ArrayList<Pair<HLogKey, WALEdit>>();
-      List<Pair<MutationType, Mutation>> mutations = new ArrayList<Pair<MutationType, Mutation>>();
+      List<HLogSplitter.MutationReplay> mutations = new ArrayList<HLogSplitter.MutationReplay>();
       for (WALEntry entry : entries) {
+        if (nonceManager != null) {
+          long nonceGroup = entry.getKey().hasNonceGroup()
+              ? entry.getKey().getNonceGroup() : HConstants.NO_NONCE;
+          long nonce = entry.getKey().hasNonce() ? entry.getKey().getNonce() : HConstants.NO_NONCE;
+          nonceManager.reportOperationFromWal(nonceGroup, nonce, entry.getKey().getWriteTime());
+        }
         Pair<HLogKey, WALEdit> walEntry = (coprocessorHost == null) ? null :
           new Pair<HLogKey, WALEdit>();
-        List<Pair<MutationType, Mutation>> edits = HLogSplitter.getMutationsFromWALEntry(entry,
-          cells, walEntry);
+        List<HLogSplitter.MutationReplay> edits =
+            HLogSplitter.getMutationsFromWALEntry(entry, cells, walEntry);
         if (coprocessorHost != null) {
           // Start coprocessor replay here. The coprocessor is for each WALEdit instead of a
           // KeyValue.
@@ -3882,7 +3936,7 @@ public class HRegionServer implements Cl
       }
 
       if (!mutations.isEmpty()) {
-        OperationStatus[] result = doBatchOp(region, mutations, true);
+        OperationStatus[] result = doReplayBatchOp(region, mutations);
         // check if it's a partial success
         for (int i = 0; result != null && i < result.length; i++) {
           if (result[i] != OperationStatus.SUCCESS) {
@@ -3987,7 +4041,7 @@ public class HRegionServer implements Cl
    * @throws IOException
    */
   protected Result append(final HRegion region,
-      final MutationProto m, final CellScanner cellScanner) throws IOException {
+      final MutationProto m, final CellScanner cellScanner, long nonceGroup) throws IOException {
     long before = EnvironmentEdgeManager.currentTimeMillis();
     Append append = ProtobufUtil.toAppend(m, cellScanner);
     Result r = null;
@@ -3995,7 +4049,14 @@ public class HRegionServer implements Cl
       r = region.getCoprocessorHost().preAppend(append);
     }
     if (r == null) {
-      r = region.append(append);
+      long nonce = startNonceOperation(m, nonceGroup);
+      boolean success = false;
+      try {
+        r = region.append(append, nonceGroup, nonce);
+        success = true;
+      } finally {
+        endNonceOperation(m, nonceGroup, success);
+      }
       if (region.getCoprocessorHost() != null) {
         region.getCoprocessorHost().postAppend(append, r);
       }
@@ -4013,8 +4074,7 @@ public class HRegionServer implements Cl
    * @throws IOException
    */
   protected Result increment(final HRegion region, final MutationProto mutation,
-      final CellScanner cells)
-  throws IOException {
+      final CellScanner cells, long nonceGroup) throws IOException {
     long before = EnvironmentEdgeManager.currentTimeMillis();
     Increment increment = ProtobufUtil.toIncrement(mutation, cells);
     Result r = null;
@@ -4022,7 +4082,14 @@ public class HRegionServer implements Cl
       r = region.getCoprocessorHost().preIncrement(increment);
     }
     if (r == null) {
-      r = region.increment(increment);
+      long nonce = startNonceOperation(mutation, nonceGroup);
+      boolean success = false;
+      try {
+        r = region.increment(increment, nonceGroup, nonce);
+        success = true;
+      } finally {
+        endNonceOperation(mutation, nonceGroup, success);
+      }
       if (region.getCoprocessorHost() != null) {
         r = region.getCoprocessorHost().postIncrement(increment, r);
       }
@@ -4032,6 +4099,49 @@ public class HRegionServer implements Cl
   }
 
   /**
+   * Starts the nonce operation for a mutation, if needed.
+   * @param mutation Mutation.
+   * @param nonceGroup Nonce group from the request.
+   * @returns Nonce used (can be NO_NONCE).
+   */
+  private long startNonceOperation(final MutationProto mutation, long nonceGroup)
+      throws IOException, OperationConflictException {
+    if (nonceManager == null || !mutation.hasNonce()) return HConstants.NO_NONCE;
+    boolean canProceed = false;
+    try {
+      canProceed = nonceManager.startOperation(nonceGroup, mutation.getNonce(), this);
+    } catch (InterruptedException ex) {
+      // Probably should not happen.
+      throw new InterruptedIOException("Nonce start operation interrupted");
+    }
+    if (!canProceed) {
+      // TODO: instead, we could convert append/increment to get w/mvcc
+      String message = "The operation with nonce {" + nonceGroup + ", " + mutation.getNonce()
+          + "} on row [" + Bytes.toString(mutation.getRow().toByteArray())
+          + "] may have already completed";
+      throw new OperationConflictException(message);
+    }
+    return mutation.getNonce();
+  }
+
+  /**
+   * Ends nonce operation for a mutation, if needed.
+   * @param mutation Mutation.
+   * @param nonceGroup Nonce group from the request. Always 0 in initial implementation.
+   * @param success Whether the operation for this nonce has succeeded.
+   */
+  private void endNonceOperation(final MutationProto mutation, long nonceGroup,
+      boolean success) {
+    if (nonceManager == null || !mutation.hasNonce()) return;
+    nonceManager.endOperation(nonceGroup, mutation.getNonce(), success);
+  }
+
+  @Override
+  public ServerNonceManager getNonceManager() {
+    return this.nonceManager;
+  }
+
+  /**
    * Execute a list of Put/Delete mutations.
    *
    * @param builder
@@ -4063,7 +4173,7 @@ public class HRegionServer implements Cl
         cacheFlusher.reclaimMemStoreMemory();
       }
 
-      OperationStatus codes[] = region.batchMutate(mArray, false);
+      OperationStatus codes[] = region.batchMutate(mArray);
       for (i = 0; i < codes.length; i++) {
         int index = mutations.get(i).getIndex();
         Exception e = null;
@@ -4119,32 +4229,31 @@ public class HRegionServer implements Cl
    * constructing MultiResponse to save a possible loop if caller doesn't need MultiResponse.
    * @param region
    * @param mutations
-   * @param isReplay
    * @return an array of OperationStatus which internally contains the OperationStatusCode and the
    *         exceptionMessage if any
    * @throws IOException
    */
-  protected OperationStatus [] doBatchOp(final HRegion region,
-      final List<Pair<MutationType, Mutation>> mutations, boolean isReplay)
-  throws IOException {
-    Mutation[] mArray = new Mutation[mutations.size()];
+  protected OperationStatus [] doReplayBatchOp(final HRegion region,
+      final List<HLogSplitter.MutationReplay> mutations) throws IOException {
+    HLogSplitter.MutationReplay[] mArray = new HLogSplitter.MutationReplay[mutations.size()];
+
     long before = EnvironmentEdgeManager.currentTimeMillis();
     boolean batchContainsPuts = false, batchContainsDelete = false;
     try {
       int i = 0;
-      for (Pair<MutationType, Mutation> m : mutations) {
-        if (m.getFirst() == MutationType.PUT) {
+      for (HLogSplitter.MutationReplay m : mutations) {
+        if (m.type == MutationType.PUT) {
           batchContainsPuts = true;
         } else {
           batchContainsDelete = true;
         }
-        mArray[i++] = m.getSecond();
+        mArray[i++] = m;
       }
       requestCount.add(mutations.size());
       if (!region.getRegionInfo().isMetaTable()) {
         cacheFlusher.reclaimMemStoreMemory();
       }
-      return region.batchMutate(mArray, isReplay);
+      return region.batchReplay(mArray);
     } finally {
       long after = EnvironmentEdgeManager.currentTimeMillis();
       if (batchContainsPuts) {

Modified: hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RegionServerServices.java
URL: http://svn.apache.org/viewvc/hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RegionServerServices.java?rev=1542168&r1=1542167&r2=1542168&view=diff
==============================================================================
--- hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RegionServerServices.java (original)
+++ hbase/trunk/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RegionServerServices.java Fri Nov 15 04:36:30 2013
@@ -115,4 +115,10 @@ public interface RegionServerServices
    * @return set of recovering regions on the hosting region server
    */
   Map<String, HRegion> getRecoveringRegions();
+
+  /**
+   * Only required for "old" log replay; if it's removed, remove this.
+   * @return The RegionServer's NonceManager
+   */
+  public ServerNonceManager getNonceManager();
 }



Mime
View raw message