BulkMutation { public ListenableFuture<MutateRowResponse> add(MutateRowRequest request) { return add(convert(request)); } @InternalApi("For internal usage only")  BulkMutation(
      BigtableTableName tableName,
      BigtableDataClient client,
      ScheduledExecutorService retryExecutorService,
      BulkOptions bulkOptions); @InternalApi("For internal usage only")  BulkMutation(
      BigtableTableName tableName,
      BigtableDataClient client,
      OperationAccountant operationAccountant,
      ScheduledExecutorService retryExecutorService,
      BulkOptions bulkOptions); ListenableFuture<MutateRowResponse> add(MutateRowRequest request); synchronized ListenableFuture<MutateRowResponse> add(MutateRowsRequest.Entry entry); void flush(); synchronized void sendUnsent(); boolean isFlushed(); @InternalApi("For internal usage only")
  static final long MAX_RPC_WAIT_TIME_NANOS; @InternalApi("For internal usage only")
  static final long MAX_NUMBER_OF_MUTATIONS; }
@Test public void testIsStale() { underTest.add(createEntry()); underTest.currentBatch.lastRpcSentTimeNanos = time.get(); Assert.assertFalse(underTest.currentBatch.isStale()); time.addAndGet(BulkMutation.MAX_RPC_WAIT_TIME_NANOS); Assert.assertTrue(underTest.currentBatch.isStale()); }
@Test public void testAdd() { MutateRowsRequest.Entry entry = createRequestEntry(); underTest.add(entry); underTest.sendUnsent(); MutateRowsRequest expected = MutateRowsRequest.newBuilder() .setTableName(TABLE_NAME.toString()) .addEntries(entry) .build(); verify(client, times(1)).mutateRowsAsync(eq(expected)); }
@Test public void testCallableSuccess() throws Exception { ListenableFuture<MutateRowResponse> rowFuture = underTest.add(createRequestEntry()); setResponse(Status.OK); MutateRowResponse result = rowFuture.get(10, TimeUnit.MILLISECONDS); Assert.assertTrue(rowFuture.isDone()); Assert.assertEquals(MutateRowResponse.getDefaultInstance(), result); Assert.assertFalse(operationAccountant.hasInflightOperations()); }
@Test public void testCallableFail() throws Exception { ListenableFuture<MutateRowResponse> rowFuture = underTest.add(createRequestEntry()); Assert.assertFalse(rowFuture.isDone()); setResponse(Status.NOT_FOUND); Assert.assertTrue(rowFuture.isDone()); try { rowFuture.get(); } catch (ExecutionException e) { Assert.assertEquals(Status.NOT_FOUND.getCode(), Status.fromThrowable(e).getCode()); Assert.assertFalse(operationAccountant.hasInflightOperations()); } }
@Test public void testCallableTooFewStatuses() throws Exception { ListenableFuture<MutateRowResponse> rowFuture1 = underTest.add(createRequestEntry()); ListenableFuture<MutateRowResponse> rowFuture2 = underTest.add(createRequestEntry()); Batch batch = underTest.currentBatch; Assert.assertFalse(rowFuture1.isDone()); Assert.assertFalse(rowFuture2.isDone()); Assert.assertEquals(2, batch.getRequestCount()); setResponse(Status.OK); Assert.assertTrue(rowFuture1.isDone()); Assert.assertTrue(rowFuture2.isDone()); Assert.assertEquals(MutateRowResponse.getDefaultInstance(), rowFuture1.get()); rowFuture1.get(); try { rowFuture2.get(); Assert.fail("Expected exception"); } catch (Exception e) { Assert.assertEquals(Status.Code.INTERNAL, Status.fromThrowable(e).getCode()); } }
@Test public void testRunOutOfTime() throws Exception { ListenableFuture<MutateRowResponse> rowFuture = underTest.add(createRequestEntry()); setResponse(Status.DEADLINE_EXCEEDED); Assert.assertTrue(rowFuture.isDone()); try { rowFuture.get(); Assert.fail("Expected exception"); } catch (ExecutionException e) { Assert.assertEquals(Status.DEADLINE_EXCEEDED.getCode(), Status.fromThrowable(e).getCode()); } Assert.assertFalse(operationAccountant.hasInflightOperations()); }
@Test public void testCallableStale() throws Exception { ListenableFuture<MutateRowResponse> rowFuture = underTest.add(createRequestEntry()); setResponse(Status.OK); MutateRowResponse result = rowFuture.get(10, TimeUnit.MILLISECONDS); Assert.assertTrue(rowFuture.isDone()); Assert.assertEquals(MutateRowResponse.getDefaultInstance(), result); Assert.assertFalse(operationAccountant.hasInflightOperations()); }
@Test public void testRequestTimer() { underTest.add(createEntry()); Assert.assertFalse(underTest.currentBatch.wasSent()); underTest.currentBatch.lastRpcSentTimeNanos = time.get(); Assert.assertFalse(underTest.currentBatch.isStale()); time.addAndGet(BulkMutation.MAX_RPC_WAIT_TIME_NANOS - 1); Assert.assertFalse(underTest.currentBatch.isStale()); time.addAndGet(1); Assert.assertTrue(underTest.currentBatch.isStale()); }
@Test public void testAutoflushDisabled() { underTest.add(createRequestEntry()); verify(retryExecutorService, never()) .schedule(any(Runnable.class), anyLong(), any(TimeUnit.class)); }
@SuppressWarnings("unchecked") @Test public void testAutoflush() throws Exception { underTest = new BulkMutation( TABLE_NAME, client, operationAccountant, retryExecutorService, BulkOptions.builder().setAutoflushMs(1000L).build()); ArgumentCaptor<Runnable> autoflusher = ArgumentCaptor.forClass(Runnable.class); when(retryExecutorService.schedule(autoflusher.capture(), anyLong(), any(TimeUnit.class))) .thenReturn(mockScheduledFuture); underTest.add(createRequestEntry()); verify(retryExecutorService, times(1)) .schedule(autoflusher.capture(), anyLong(), any(TimeUnit.class)); verify(client, never()).mutateRowsAsync(any(MutateRowsRequest.class)); autoflusher.getValue().run(); verify(client, times(1)).mutateRowsAsync(any(MutateRowsRequest.class)); }
@Test public void testLotsOfMutations() throws Exception { MutateRowsRequest.Entry smallRequest = createRequestEntry(); MutateRowsRequest.Entry.Builder bigRequest = createRequestEntry().toBuilder(); bigRequest.addAllMutations(Collections.nCopies(20_000, bigRequest.getMutations(0))); MutateRowsRequest.Entry lastRequest = bigRequest.clone().setRowKey(ByteString.copyFrom("SomeOtherKey".getBytes())).build(); underTest.add(smallRequest); underTest.add(bigRequest.build()); underTest.add(bigRequest.build()); underTest.add(lastRequest); Assert.assertTrue(underTest.currentBatch.builder.getEntriesList().contains(smallRequest)); underTest.add(lastRequest); Assert.assertFalse(underTest.currentBatch.builder.getEntriesList().contains(smallRequest)); Assert.assertTrue(underTest.currentBatch.builder.getEntriesList().contains(lastRequest)); }
@Test(expected = IllegalArgumentException.class) public void testTooManyMutations() { MutateRowsRequest.Entry.Builder bigRequest = createRequestEntry().toBuilder(); bigRequest.addAllMutations( Collections.nCopies( (int) BulkMutation.MAX_NUMBER_OF_MUTATIONS, bigRequest.getMutations(0))); underTest.add(bigRequest.build()); }