Batcher error accounting is incorrect when the RPC fails

[igorbernstein2]

MutateRowsAttemptCallable when it encounters either RPC or entry errors will resolve its future with a synthetic MutateRowsException. The synthetic error has an INTERNAL status but it marked as retriable. This is done so that the parent RetryableFuture retries the remaining entries. However, once there are no more retriable entries, the MutateRowsException will bubble up to BatcherImpl.Batch#onBatchFailure, which will feed into BatcherStats, which will record the RPC error as INTERNAL

[igorbernstein2]

Here are 2 tests for EnhancedBigtableStub that show the 2 problems

@Test
  public void testBatchMutationsPartialFailure() throws InterruptedException {
    Batcher<RowMutationEntry, Void> batcher = enhancedBigtableStub.newMutateRowsBatcher("table1",
        GrpcCallContext.createDefault());

    batcher.add(RowMutationEntry.create("key0").deleteRow());
    batcher.add(RowMutationEntry.create("key1").deleteRow());

    Mockito.doAnswer(invocationOnMock -> {
      StreamObserver<MutateRowsResponse> observer = invocationOnMock.getArgument(1);
      observer.onNext(MutateRowsResponse.newBuilder()
              .addEntries(
              Entry.newBuilder()
                  .setIndex(0)
                  .setStatus(Status.newBuilder().setCode(Code.OK_VALUE))
                  .build())
              .addEntries(
                  Entry.newBuilder()
                      .setIndex(1)
                      .setStatus(Status.newBuilder().setCode(Code.PERMISSION_DENIED_VALUE))
                      .build())
          .build());
      observer.onCompleted();
      return null;
    }).when(fakeDataService).mutateRows(Mockito.any(MutateRowsRequest.class), Mockito.any(StreamObserver.class));

    BatchingException batchingException = Assert.assertThrows(BatchingException.class,
        () -> batcher.close());

    // Overall RPC status was ok, so it should be a partial failure
    assertThat(batchingException).hasMessageThat().contains("1 partial failure");
    // The MutateRows exception's status should not bubble up
    assertThat(batchingException).hasMessageThat().doesNotContain("INTERNAL");
    // Neither should the overall exception
    assertThat(batchingException).hasMessageThat().doesNotContain("ApiException");
  }

  @Test
  public void testBatchMutationRPCErrorCode() throws InterruptedException {
    Batcher<RowMutationEntry, Void> batcher = enhancedBigtableStub.newMutateRowsBatcher("table1",
        GrpcCallContext.createDefault());

    Mockito.doAnswer(invocationOnMock -> {
      StreamObserver<MutateRowsResponse> observer = invocationOnMock.getArgument(1);
      observer.onError(io.grpc.Status.PERMISSION_DENIED.asException());
      return null;
    }).when(fakeDataService).mutateRows(Mockito.any(MutateRowsRequest.class), Mockito.any(StreamObserver.class));

    batcher.add(RowMutationEntry.create("key0").deleteRow());

    BatchingException batchingException = Assert.assertThrows(BatchingException.class,
        () -> batcher.close());

    // Overall RPC status was PERMISSION_DENIED, so the error message should be something like Batching finished with 1 batches failed to apply due to: 1 ApiException(1 PERMISSION_DENIED)
    assertThat(batchingException).hasMessageThat().contains("PERMISSION_DENIED");
  }

[igorbernstein2]

In addition to this, Bigtable validates all of the entries for the entire RPC before processing each entry. This leads to a bad UX when a batch as an invalid entry (ie. no entries). In this case all of the entries will get the error that the entries are empty