fix(bigtable): count mutations wrapped from protos towards the row limits

Mutation.MAX_MUTATIONS and MAX_BYTE_SIZE are enforced in addMutation, against
counters only addMutation maintains. fromProtoUnsafe(List),
fromProtoUnsafe(Iterable) and fromProto(List) add to the mutation list
directly and leave both counters at zero, so mutations wrapped from existing
protos count towards neither limit.

The mutation count is backstopped by RowMutationEntry.toProto() and
BulkMutation.add, which re-check the real list size, so it surfaces late and as
a different exception type. The byte size is not backstopped anywhere, so a
Mutation seeded from protos can exceed 200 MB with nothing client-side
objecting.

Count wrapped protos in the three factories, so the counters describe the whole
row. Deliberately not a checkState in the factories themselves: that would make
wrapping an already-over-limit proto throw where it currently does not.

Author: laughingman7743

java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/Mutation.java

@@ -86,6 +86,7 @@ public static Mutation createUnsafe() {
   public static Mutation fromProtoUnsafe(List<com.google.bigtable.v2.Mutation> protos) {
     Mutation mutation = new Mutation(true);
     mutation.mutations.addAll(protos);
+    mutation.countAllTowardsLimits(protos);
     return mutation;
   }
 
@@ -99,6 +100,7 @@ public static Mutation fromProtoUnsafe(List<com.google.bigtable.v2.Mutation> pro
   public static Mutation fromProtoUnsafe(Iterable<com.google.bigtable.v2.Mutation> protos) {
     Mutation mutation = new Mutation(true);
     mutation.mutations.addAll(protos);
+    mutation.countAllTowardsLimits(protos);
     return mutation;
   }
 
@@ -115,6 +117,7 @@ public static Mutation fromProtoUnsafe(Iterable<com.google.bigtable.v2.Mutation>
   static Mutation fromProto(List<com.google.bigtable.v2.Mutation> protos) {
     Mutation mutation = new Mutation(false);
     mutation.mutations.addAll(protos);
+    mutation.countAllTowardsLimits(protos);
     return mutation;
   }
 
@@ -333,10 +336,20 @@ private void addMutation(com.google.bigtable.v2.Mutation mutation) {
         byteSize + mutation.getSerializedSize() <= MAX_BYTE_SIZE,
         "Byte size of mutations is too large");
 
+    countTowardsLimits(mutation);
+
+    mutations.add(mutation);
+  }
+
+  private void countTowardsLimits(com.google.bigtable.v2.Mutation mutation) {
     numMutations++;
     byteSize += mutation.getSerializedSize();
+  }
 
-    mutations.add(mutation);
+  private void countAllTowardsLimits(Iterable<com.google.bigtable.v2.Mutation> protos) {
+    for (com.google.bigtable.v2.Mutation proto : protos) {
+      countTowardsLimits(proto);
+    }
   }
 
   private static ByteString wrapByteString(String str) {

java-bigtable/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/models/MutationTest.java

@@ -23,6 +23,7 @@
 import com.google.bigtable.v2.Mutation.DeleteFromRow;
 import com.google.bigtable.v2.Mutation.MergeToCell;
 import com.google.cloud.bigtable.data.v2.models.Range.TimestampRange;
+import com.google.common.collect.ImmutableList;
 import com.google.common.primitives.Longs;
 import com.google.protobuf.ByteString;
 import java.io.ByteArrayInputStream;
@@ -261,6 +262,58 @@ public void tooLargeRequest() {
     assertThat(actualError).isInstanceOf(IllegalStateException.class);
   }
 
+  @Test
+  public void tooManyMutationsCountsWrappedProtosTest() {
+    Mutation mutation =
+        Mutation.fromProtoUnsafe(
+            ImmutableList.of(
+                com.google.bigtable.v2.Mutation.newBuilder()
+                    .setDeleteFromRow(com.google.bigtable.v2.Mutation.DeleteFromRow.newBuilder())
+                    .build()));
+
+    for (int i = 0; i < Mutation.MAX_MUTATIONS - 1; i++) {
+      mutation.setCell("f", "", "");
+    }
+
+    Exception actualError = null;
+    try {
+      mutation.setCell("f", "", "");
+    } catch (Exception e) {
+      actualError = e;
+    }
+
+    assertThat(actualError).isInstanceOf(IllegalStateException.class);
+    assertThat(mutation.getMutations()).hasSize(Mutation.MAX_MUTATIONS);
+  }
+
+  @Test
+  public void tooLargeRequestCountsWrappedProtosTest() {
+    Mutation mutation =
+        Mutation.fromProtoUnsafe(
+            ImmutableList.of(
+                com.google.bigtable.v2.Mutation.newBuilder()
+                    .setSetCell(
+                        com.google.bigtable.v2.Mutation.SetCell.newBuilder()
+                            .setFamilyName("f")
+                            .setValue(ByteString.copyFrom(new byte[Mutation.MAX_BYTE_SIZE / 2])))
+                    .build(),
+                com.google.bigtable.v2.Mutation.newBuilder()
+                    .setSetCell(
+                        com.google.bigtable.v2.Mutation.SetCell.newBuilder()
+                            .setFamilyName("f")
+                            .setValue(ByteString.copyFrom(new byte[Mutation.MAX_BYTE_SIZE / 2])))
+                    .build()));
+
+    Exception actualError = null;
+    try {
+      mutation.setCell("f", "", "");
+    } catch (Exception e) {
+      actualError = e;
+    }
+
+    assertThat(actualError).isInstanceOf(IllegalStateException.class);
+  }
+
   @Test
   public void testWithLongValue() {
     Mutation mutation =