Repository: GoogleCloudPlatform/cloud-bigtable-client
Pull request #2104 (2019-02-20 10:45:35)

Migrated IBigtableDataClient ListenableFuture to adapt to ApiFuture
- Migrated all async operations of `IBigtableDataClient` to adapt to ApiFuture.
- Removed `thenApply(r -> null)` where unneeded.


Patch:
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/core/IBigtableDataClient.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/core/IBigtableDataClient.java
index ecbfbad64..cede53619 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/core/IBigtableDataClient.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/core/IBigtableDataClient.java
@@ -24,11 +24,9 @@ import com.google.cloud.bigtable.data.v2.models.Row;
 import com.google.cloud.bigtable.data.v2.models.RowMutation;
 import com.google.cloud.bigtable.grpc.scanner.FlatRow;
 import com.google.cloud.bigtable.grpc.scanner.ResultScanner;
-import com.google.common.util.concurrent.ListenableFuture;
 import io.grpc.stub.StreamObserver;
 
 import java.util.List;
-import java.util.concurrent.ExecutionException;
 
 /**
  * Interface to wrap {@link com.google.cloud.bigtable.grpc.BigtableDataClient} with
@@ -47,11 +45,10 @@ public interface IBigtableDataClient {
    * Mutate a row atomically.
    *
    * @param rowMutation a {@link RowMutation} model object.
-   * @return a {@link ListenableFuture} of type {@link Void} will be set when request is
+   * @return a {@link ApiFuture} of type {@link Void} will be set when request is
    *     successful otherwise exception will be thrown.
    */
-  //TODO(rahulkql): Once it is adapted to v2.models, change the return type to ApiFuture.
-  ListenableFuture<Void> mutateRowAsync(RowMutation rowMutation);
+  ApiFuture<Void> mutateRowAsync(RowMutation rowMutation);
 
   /**
    * Perform an atomic read-modify-write operation on a row.
@@ -65,7 +62,7 @@ public interface IBigtableDataClient {
    * Perform an atomic read-modify-write operation on a row.
    *
    * @param readModifyWriteRow a {@link ReadModifyWriteRow} model object.
-   * @return a {@link ListenableFuture} of type {@link Row} will be set when request is
+   * @return a {@link ApiFuture} of type {@link Row} will be set when request is
    *     successful otherwise exception will be thrown.
    */
   ApiFuture<Row> readModifyWriteRowAsync(ReadModifyWriteRow readModifyWriteRow);
@@ -79,11 +76,10 @@ public interface IBigtableDataClient {
    * Mutate a row atomically dependent on a precondition.
    *
    * @param conditionalRowMutation a {@link ConditionalRowMutation} model object.
-   * @return a {@link ListenableFuture} of type {@link Boolean} will be set when request is
+   * @return a {@link ApiFuture} of type {@link Boolean} will be set when request is
    *     successful otherwise exception will be thrown.
    */
-  //TODO(rahulkql): Once it is adapted to v2.models, change the return type to ApiFuture.
-  ListenableFuture<Boolean> checkAndMutateRowAsync(ConditionalRowMutation conditionalRowMutation);
+  ApiFuture<Boolean> checkAndMutateRowAsync(ConditionalRowMutation conditionalRowMutation);
 
   /**
    * Mutate a row atomically dependent on a precondition.
@@ -106,10 +102,9 @@ public interface IBigtableDataClient {
    * completed.
    *
    * @param tableId a String object.
-   * @return a {@link ListenableFuture} object.
+   * @return a {@link ApiFuture} object.
    */
-  //TODO(rahulkql): Once it is adapted to v2.models, change the return type to ApiFuture.
-  ListenableFuture< List<KeyOffset>> sampleRowKeysAsync(String tableId);
+  ApiFuture< List<KeyOffset>> sampleRowKeysAsync(String tableId);
 
   /**
    * Perform a scan over {@link Row}s, in key order.
@@ -122,11 +117,11 @@ public interface IBigtableDataClient {
   /**
    * Read multiple {@link Row}s into an in-memory list, in key order.
    *
-   * @return a {@link ListenableFuture} that will finish when
+   * @return a {@link ApiFuture} that will finish when
    * all reads have completed.
    * @param request a {@link Query} object.
    */
-  ListenableFuture<List<Row>> readRowsAsync(Query request);
+  ApiFuture<List<Row>> readRowsAsync(Query request);
 
   /**
    * Returns a list of {@link FlatRow}s, in key order.
@@ -147,12 +142,11 @@ public interface IBigtableDataClient {
   /**
    * Read multiple {@link FlatRow}s into an in-memory list, in key order.
    *
-   * @return a {@link ListenableFuture} that will finish when
+   * @return a {@link ApiFuture} that will finish when
    * all reads have completed.
    * @param request a {@link Query} object.
    */
-  //TODO(rahulkql): Once it is adapted to v2.models, change the return type to ApiFuture.
-  ListenableFuture<List<FlatRow>> readFlatRowsAsync(Query request);
+  ApiFuture<List<FlatRow>> readFlatRowsAsync(Query request);
 
   /**
    * Read {@link FlatRow} asynchronously, and pass them to a stream observer to be processed.
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableDataClientWrapper.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableDataClientWrapper.java
index e31f1a51d..a1343e958 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableDataClientWrapper.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableDataClientWrapper.java
@@ -15,7 +15,9 @@
  */
 package com.google.cloud.bigtable.grpc;
 
+import com.google.api.core.ApiFunction;
 import com.google.api.core.ApiFuture;
+import com.google.api.core.ApiFutures;
 import com.google.bigtable.v2.CheckAndMutateRowRequest;
 import com.google.bigtable.v2.CheckAndMutateRowResponse;
 import com.google.bigtable.v2.MutateRowRequest;
@@ -75,16 +77,16 @@ public class BigtableDataClientWrapper implements IBigtableDataClient {
 
   /** {@inheritDoc} */
   @Override
-  public ListenableFuture<Void> mutateRowAsync(RowMutation rowMutation) {
+  public ApiFuture<Void> mutateRowAsync(RowMutation rowMutation) {
     ListenableFuture<MutateRowResponse> response =
         delegate.mutateRowAsync(rowMutation.toProto(requestContext));
-    return  Futures.transform(response, new Function<MutateRowResponse, Void>() {
+    return  ApiFutureUtil.adapt(Futures.transform(response, new Function<MutateRowResponse, Void>() {
       @Nullable
       @Override
       public Void apply(@Nullable MutateRowResponse mutateRowResponse) {
         return null;
       }
-    }, MoreExecutors.directExecutor());
+    }, MoreExecutors.directExecutor()));
   }
 
   /** {@inheritDoc} */
@@ -116,18 +118,19 @@ public class BigtableDataClientWrapper implements IBigtableDataClient {
 
   /** {@inheritDoc} */
   @Override
-  public ListenableFuture<Boolean> checkAndMutateRowAsync(
+  public ApiFuture<Boolean> checkAndMutateRowAsync(
       ConditionalRowMutation conditionalRowMutation) {
     final CheckAndMutateRowRequest request = conditionalRowMutation.toProto(requestContext);
     final ListenableFuture<CheckAndMutateRowResponse> response =
         delegate.checkAndMutateRowAsync(request);
-    return Futures.transform(response, new Function<CheckAndMutateRowResponse, Boolean>() {
+    return ApiFutureUtil.adapt(Futures.transform(response,
+        new Function<CheckAndMutateRowResponse, Boolean>() {
 
       @Override
       public Boolean apply(CheckAndMutateRowResponse checkAndMutateRowResponse) {
         return checkAndMutateRowResponse.getPredicateMatched();
       }
-    }, MoreExecutors.directExecutor());
+    }, MoreExecutors.directExecutor()));
   }
 
   /** {@inheritDoc} */
@@ -158,7 +161,7 @@ public class BigtableDataClientWrapper implements IBigtableDataClient {
 
   /** {@inheritDoc} */
   @Override
-  public ListenableFuture<List<KeyOffset>> sampleRowKeysAsync(String tableId) {
+  public ApiFuture<List<KeyOffset>> sampleRowKeysAsync(String tableId) {
     String fullTableName = NameUtil
         .formatTableName(requestContext.getProjectId(), requestContext.getInstanceId(), tableId);
     SampleRowKeysRequest requestProto =
@@ -166,7 +169,8 @@ public class BigtableDataClientWrapper implements IBigtableDataClient {
     ListenableFuture<List<SampleRowKeysResponse>> responseProto =
         delegate.sampleRowKeysAsync(requestProto);
 
-    return Futures.transform(responseProto, new Function<List<SampleRowKeysResponse>, List<KeyOffset>>() {
+    return ApiFutureUtil.adapt(Futures
+        .transform(responseProto, new Function<List<SampleRowKeysResponse>, List<KeyOffset>>() {
       @Override
       public List<KeyOffset> apply(@Nonnull List<SampleRowKeysResponse> rowKeysList) {
         if(rowKeysList == null || rowKeysList.isEmpty()){
@@ -180,7 +184,7 @@ public class BigtableDataClientWrapper implements IBigtableDataClient {
 
         return keyOffsetBuilder.build();
       }
-    }, MoreExecutors.directExecutor());
+    }, MoreExecutors.directExecutor()));
   }
 
   /** {@inheritDoc} */
@@ -218,10 +222,11 @@ public class BigtableDataClientWrapper implements IBigtableDataClient {
 
   /** {@inheritDoc} */
   @Override
-  public ListenableFuture<List<Row>> readRowsAsync(Query request) {
-    ListenableFuture<List<FlatRow>> responseProto = readFlatRowsAsync(request);
+  public ApiFuture<List<Row>> readRowsAsync(Query request) {
+    ApiFuture<List<FlatRow>> responseProto = readFlatRowsAsync(request);
 
-    return Futures.transform(responseProto, new Function<List<FlatRow>, List<Row>>() {
+    return ApiFutures.transform(responseProto,
+        new ApiFunction<List<FlatRow>, List<Row>>() {
       @Override
       public List<Row> apply(List<FlatRow> flatRowList) {
         ImmutableList.Builder<Row> rowBuilder =
@@ -249,8 +254,8 @@ public class BigtableDataClientWrapper implements IBigtableDataClient {
 
   /** {@inheritDoc} */
   @Override
-  public ListenableFuture<List<FlatRow>> readFlatRowsAsync(Query request) {
-    return delegate.readFlatRowsAsync(request.toProto(requestContext));
+  public ApiFuture<List<FlatRow>> readFlatRowsAsync(Query request) {
+    return ApiFutureUtil.adapt(delegate.readFlatRowsAsync(request.toProto(requestContext)));
   }
 
   /** {@inheritDoc} */
diff --git a/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/AbstractBigtableRegionLocator.java b/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/AbstractBigtableRegionLocator.java
index d53dcad2b..f55b2caf9 100644
--- a/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/AbstractBigtableRegionLocator.java
+++ b/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/AbstractBigtableRegionLocator.java
@@ -17,6 +17,10 @@
  */
 package com.google.cloud.bigtable.hbase;
 
+import com.google.api.core.ApiFunction;
+import com.google.api.core.ApiFuture;
+import com.google.api.core.ApiFutureCallback;
+import com.google.api.core.ApiFutures;
 import java.util.List;
 
 import javax.annotation.Nullable;
@@ -32,10 +36,7 @@ import com.google.cloud.bigtable.config.BigtableOptions;
 import com.google.cloud.bigtable.config.Logger;
 import com.google.cloud.bigtable.grpc.BigtableTableName;
 import com.google.cloud.bigtable.hbase.adapters.SampledRowKeysAdapter;
-import com.google.common.base.Function;
-import com.google.common.util.concurrent.FutureCallback;
 import com.google.common.util.concurrent.Futures;
-import com.google.common.util.concurrent.ListenableFuture;
 
 /**
  * <p> AbstractBigtbleRegionLocator class. </p>
@@ -52,7 +53,7 @@ public abstract class AbstractBigtableRegionLocator {
   private static final Logger LOG = new Logger(AbstractBigtableRegionLocator.class);
 
   protected final TableName tableName;
-  private ListenableFuture<List<HRegionLocation>> regionsFuture;
+  private ApiFuture<List<HRegionLocation>> regionsFuture;
   private final IBigtableDataClient client;
   private final SampledRowKeysAdapter adapter;
   private final BigtableTableName bigtableTableName;
@@ -75,7 +76,7 @@ public abstract class AbstractBigtableRegionLocator {
    * @param reload a boolean field.
    * @return a {@link List} object.
    */
-  protected synchronized ListenableFuture<List<HRegionLocation>> getRegionsAsync(boolean reload) {
+  protected synchronized ApiFuture<List<HRegionLocation>> getRegionsAsync(boolean reload) {
     // If we don't need to refresh and we have a recent enough version, just use that.
     if (!reload && regionsFuture != null &&
         regionsFetchTimeMillis + MAX_REGION_AGE_MILLIS > System.currentTimeMillis()) {
@@ -85,16 +86,15 @@ public abstract class AbstractBigtableRegionLocator {
     LOG.debug("Sampling rowkeys for table %s", bigtableTableName.toString());
 
     try {
-      ListenableFuture<List<KeyOffset>> future =
+      ApiFuture<List<KeyOffset>> future =
           client.sampleRowKeysAsync(bigtableTableName.getTableId());
-      this.regionsFuture = Futures
-          .transform(future, new Function<List<KeyOffset>, List<HRegionLocation>>() {
+      this.regionsFuture = ApiFutures.transform(future, new ApiFunction<List<KeyOffset>, List<HRegionLocation>>() {
             @Override
             public List<HRegionLocation> apply(@Nullable List<KeyOffset> input) {
               return adapter.adaptResponse(input);
             }
           }, MoreExecutors.directExecutor());
-      Futures.addCallback(this.regionsFuture, new FutureCallback<List<HRegionLocation>>() {
+      ApiFutures.addCallback(this.regionsFuture, new ApiFutureCallback<List<HRegionLocation>>() {
         @Override public void onSuccess(@Nullable List<HRegionLocation> result) {
         }
         @Override public void onFailure(Throwable t) {
diff --git a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAdmin.java b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAdmin.java
index 2ff1eeb92..79fb1a9ff 100644
--- a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAdmin.java
+++ b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAdmin.java
@@ -239,8 +239,7 @@ public class BigtableAdmin extends AbstractBigtableAdmin {
 
   protected CompletableFuture<Void> deleteTableAsyncInternal(TableName tableName) {
     return FutureUtils.toCompletableFuture(
-        tableAdminClientWrapper.deleteTableAsync(tableName.getNameAsString()))
-        .thenApply(r -> null);
+        tableAdminClientWrapper.deleteTableAsync(tableName.getNameAsString()));
   }
 
   @Override
@@ -347,8 +346,7 @@ public class BigtableAdmin extends AbstractBigtableAdmin {
    // rowKeyPrefix set to Empty String which truncates data from specified table.
    return FutureUtils.toCompletableFuture(
         tableAdminClientWrapper
-          .dropRowRangeAsync(tableName.getNameAsString(), ""))
-        .thenApply(r -> null);
+          .dropRowRangeAsync(tableName.getNameAsString(), ""));
   }
   /* ******* Unsupported methods *********** */
 
diff --git a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAsyncTable.java b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAsyncTable.java
index 5feed3175..9d3af1377 100644
--- a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAsyncTable.java
+++ b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/main/java/com/google/cloud/bigtable/hbase2_x/BigtableAsyncTable.java
@@ -231,8 +231,7 @@ public class BigtableAsyncTable implements AsyncTable<ScanResultConsumer> {
   @Override
   public CompletableFuture<Void> delete(Delete delete) {
     // figure out how to time this with Opencensus
-    return toCompletableFuture(clientWrapper.mutateRowAsync(hbaseAdapter.adapt(delete)))
-        .thenApply(r -> null);
+    return toCompletableFuture(clientWrapper.mutateRowAsync(hbaseAdapter.adapt(delete)));
   }
 
   /**
@@ -353,8 +352,7 @@ public class BigtableAsyncTable implements AsyncTable<ScanResultConsumer> {
    */
   @Override
   public CompletableFuture<Void> mutateRow(RowMutations rowMutations) {
-    return toCompletableFuture(clientWrapper.mutateRowAsync(hbaseAdapter.adapt(rowMutations)))
-        .thenApply(r -> null);
+    return toCompletableFuture(clientWrapper.mutateRowAsync(hbaseAdapter.adapt(rowMutations)));
   }
 
   /**
@@ -363,8 +361,7 @@ public class BigtableAsyncTable implements AsyncTable<ScanResultConsumer> {
   @Override
   public CompletableFuture<Void> put(Put put) {
     // figure out how to time this with Opencensus
-    return toCompletableFuture(clientWrapper.mutateRowAsync(hbaseAdapter.adapt(put)))
-        .thenApply(r -> null);
+    return toCompletableFuture(clientWrapper.mutateRowAsync(hbaseAdapter.adapt(put)));
   }
 
   /**


Test patch:
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/TestBigtableDataClientWrapper.java b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/TestBigtableDataClientWrapper.java
index ccb042a31..6ab62111b 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/TestBigtableDataClientWrapper.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/TestBigtableDataClientWrapper.java
@@ -155,7 +155,7 @@ public class TestBigtableDataClientWrapper {
     ListenableFuture<CheckAndMutateRowResponse> future = Futures.immediateFuture(response);
 
     when(client.checkAndMutateRowAsync(request)).thenReturn(future);
-    ListenableFuture<Boolean> actual = clientWrapper.checkAndMutateRowAsync(conditonalMutation);
+    ApiFuture<Boolean> actual = clientWrapper.checkAndMutateRowAsync(conditonalMutation);
     verify(client).checkAndMutateRowAsync(request);
     assertTrue(actual.get());
   }
@@ -172,7 +172,7 @@ public class TestBigtableDataClientWrapper {
     ListenableFuture<CheckAndMutateRowResponse> future = Futures.immediateFuture(response);
 
     when(client.checkAndMutateRowAsync(request)).thenReturn(future);
-    ListenableFuture<Boolean> actual = clientWrapper.checkAndMutateRowAsync(conditonalMutation);
+    ApiFuture<Boolean> actual = clientWrapper.checkAndMutateRowAsync(conditonalMutation);
     verify(client).checkAndMutateRowAsync(request);
     assertFalse(actual.get());
   }