@Test public void testReadModifyWriteAsyncWithOneRow() throws ExecutionException, InterruptedException { String family_1 = "col-family-1"; String family_2 = "col-family-2"; String family_3 = "col-family-3"; ByteString qualifier_1 = ByteString.copyFromUtf8("test-qualifier_1-1"); ByteString qualifier_2 = ByteString.copyFromUtf8("test-qualifier_1-2"); ByteString qualifier_3 = ByteString.copyFromUtf8("test-qualifier_1-2"); ByteString value_1 = ByteString.copyFromUtf8("test-values-1"); ByteString value_2 = ByteString.copyFromUtf8("test-values-2"); ByteString value_3 = ByteString.copyFromUtf8("test-values-3"); ByteString value_4 = ByteString.copyFromUtf8("test-values-4"); ByteString value_5 = ByteString.copyFromUtf8("test-values-5"); ByteString value_6 = ByteString.copyFromUtf8("test-values-6"); Row row = Row.newBuilder() .setKey(ROW_KEY) .addFamilies( Family.newBuilder() .setName(family_1) .addColumns( Column.newBuilder() .setQualifier(qualifier_1) .addCells( Cell.newBuilder() .setTimestampMicros(11_111L) .setValue(value_1) .addLabels("label-1")) .addCells( Cell.newBuilder() .setTimestampMicros(22_222L) .setValue(value_2) .addLabels("label-2"))) .addColumns( Column.newBuilder() .setQualifier(qualifier_2) .addCells( Cell.newBuilder() .setTimestampMicros(11_111L) .setValue(value_3) .addLabels("label-3") .addLabels("label-4")) .addCells( Cell.newBuilder().setTimestampMicros(22_222L).setValue(value_4))) .build()) .addFamilies( Family.newBuilder() .setName(family_2) .addColumns( Column.newBuilder() .setQualifier(qualifier_1) .addCells( Cell.newBuilder().setTimestampMicros(11_111L).setValue(value_5)))) .addFamilies( Family.newBuilder() .setName(family_3) .addColumns( Column.newBuilder() .setQualifier(qualifier_3) .addCells(Cell.newBuilder().setValue(value_6).addLabels("label-6")))) .build(); ReadModifyWriteRow readModify = ReadModifyWriteRow.create(TABLE_ID, "test-key"); ReadModifyWriteRowRequest request = readModify.toProto(REQUEST_CONTEXT); ReadModifyWriteRowResponse response = ReadModifyWriteRowResponse.newBuilder().setRow(row).build(); ListenableFuture<ReadModifyWriteRowResponse> listenableResponse = Futures.immediateFuture(response); when(delegate.readModifyWriteRowAsync(request)).thenReturn(listenableResponse); Result result = dataClientWrapper.readModifyWriteRowAsync(readModify).get(); assertEquals(6, result.rawCells().length); List<org.apache.hadoop.hbase.Cell> cells1 = result.getColumnCells(family_1.getBytes(), qualifier_1.toByteArray()); assertEquals(2, cells1.size()); assertEquals(11L, cells1.get(0).getTimestamp()); assertArrayEquals(value_1.toByteArray(), CellUtil.cloneValue(cells1.get(0))); assertEquals(Collections.singletonList("label-1"), ((RowCell) cells1.get(0)).getLabels()); assertEquals(22L, cells1.get(1).getTimestamp()); assertArrayEquals(value_2.toByteArray(), CellUtil.cloneValue(cells1.get(1))); assertEquals(Collections.singletonList("label-2"), ((RowCell) cells1.get(1)).getLabels()); List<org.apache.hadoop.hbase.Cell> cells2 = result.getColumnCells(family_1.getBytes(), qualifier_2.toByteArray()); assertEquals(2, cells2.size()); assertEquals(11L, cells2.get(0).getTimestamp()); assertArrayEquals(value_3.toByteArray(), CellUtil.cloneValue(cells2.get(0))); assertEquals(ImmutableList.of("label-3", "label-4"), ((RowCell) cells2.get(0)).getLabels()); assertEquals(22L, cells2.get(1).getTimestamp()); assertArrayEquals(value_4.toByteArray(), CellUtil.cloneValue(cells2.get(1))); List<org.apache.hadoop.hbase.Cell> cells3 = result.getColumnCells(family_2.getBytes(), qualifier_1.toByteArray()); assertEquals(1, cells3.size()); assertArrayEquals(value_5.toByteArray(), CellUtil.cloneValue(cells3.get(0))); List<org.apache.hadoop.hbase.Cell> cells4 = result.getColumnCells(family_3.getBytes(), qualifier_3.toByteArray()); assertEquals(1, cells4.size()); assertArrayEquals(value_6.toByteArray(), CellUtil.cloneValue(cells4.get(0))); assertEquals(Collections.singletonList("label-6"), ((RowCell) cells4.get(0)).getLabels()); }
@Override public ApiFuture<Result> readModifyWriteRowAsync(ReadModifyWriteRow readModifyWriteRow) { ListenableFuture<ReadModifyWriteRowResponse> response = delegate.readModifyWriteRowAsync(readModifyWriteRow.toProto(requestContext)); return ApiFutureUtil.transformAndAdapt( response, new Function<ReadModifyWriteRowResponse, Result>() { @Override public Result apply(ReadModifyWriteRowResponse response) { return transformRowToResult(response.getRow()); } }); }
DataClientClassicApi implements DataClientWrapper { @Override public ApiFuture<Result> readModifyWriteRowAsync(ReadModifyWriteRow readModifyWriteRow) { ListenableFuture<ReadModifyWriteRowResponse> response = delegate.readModifyWriteRowAsync(readModifyWriteRow.toProto(requestContext)); return ApiFutureUtil.transformAndAdapt( response, new Function<ReadModifyWriteRowResponse, Result>() { @Override public Result apply(ReadModifyWriteRowResponse response) { return transformRowToResult(response.getRow()); } }); } }
DataClientClassicApi implements DataClientWrapper { @Override public ApiFuture<Result> readModifyWriteRowAsync(ReadModifyWriteRow readModifyWriteRow) { ListenableFuture<ReadModifyWriteRowResponse> response = delegate.readModifyWriteRowAsync(readModifyWriteRow.toProto(requestContext)); return ApiFutureUtil.transformAndAdapt( response, new Function<ReadModifyWriteRowResponse, Result>() { @Override public Result apply(ReadModifyWriteRowResponse response) { return transformRowToResult(response.getRow()); } }); } DataClientClassicApi(BigtableSession session, RequestContext requestContext); }
DataClientClassicApi implements DataClientWrapper { @Override public ApiFuture<Result> readModifyWriteRowAsync(ReadModifyWriteRow readModifyWriteRow) { ListenableFuture<ReadModifyWriteRowResponse> response = delegate.readModifyWriteRowAsync(readModifyWriteRow.toProto(requestContext)); return ApiFutureUtil.transformAndAdapt( response, new Function<ReadModifyWriteRowResponse, Result>() { @Override public Result apply(ReadModifyWriteRowResponse response) { return transformRowToResult(response.getRow()); } }); } DataClientClassicApi(BigtableSession session, RequestContext requestContext); @Override BulkMutationWrapper createBulkMutation(String tableId); @Override BulkReadWrapper createBulkRead(String tableId); @Override ApiFuture<Void> mutateRowAsync(RowMutation rowMutation); @Override ApiFuture<Result> readModifyWriteRowAsync(ReadModifyWriteRow readModifyWriteRow); @Override ApiFuture<Boolean> checkAndMutateRowAsync(ConditionalRowMutation conditionalRowMutation); @Override ApiFuture<List<KeyOffset>> sampleRowKeysAsync(String tableId); @Override ApiFuture<Result> readRowAsync(String tableId, ByteString rowKey, Filters.Filter filter); @Override org.apache.hadoop.hbase.client.ResultScanner readRows(Query request); @Override ApiFuture<List<Result>> readRowsAsync(Query request); @Override void readRowsAsync(Query request, final StreamObserver<Result> observer); @Override void close(); }
DataClientClassicApi implements DataClientWrapper { @Override public ApiFuture<Result> readModifyWriteRowAsync(ReadModifyWriteRow readModifyWriteRow) { ListenableFuture<ReadModifyWriteRowResponse> response = delegate.readModifyWriteRowAsync(readModifyWriteRow.toProto(requestContext)); return ApiFutureUtil.transformAndAdapt( response, new Function<ReadModifyWriteRowResponse, Result>() { @Override public Result apply(ReadModifyWriteRowResponse response) { return transformRowToResult(response.getRow()); } }); } DataClientClassicApi(BigtableSession session, RequestContext requestContext); @Override BulkMutationWrapper createBulkMutation(String tableId); @Override BulkReadWrapper createBulkRead(String tableId); @Override ApiFuture<Void> mutateRowAsync(RowMutation rowMutation); @Override ApiFuture<Result> readModifyWriteRowAsync(ReadModifyWriteRow readModifyWriteRow); @Override ApiFuture<Boolean> checkAndMutateRowAsync(ConditionalRowMutation conditionalRowMutation); @Override ApiFuture<List<KeyOffset>> sampleRowKeysAsync(String tableId); @Override ApiFuture<Result> readRowAsync(String tableId, ByteString rowKey, Filters.Filter filter); @Override org.apache.hadoop.hbase.client.ResultScanner readRows(Query request); @Override ApiFuture<List<Result>> readRowsAsync(Query request); @Override void readRowsAsync(Query request, final StreamObserver<Result> observer); @Override void close();  }