deps: upgrade java-bigtable to 2.49.0 and fix tests (#4462)

* deps: upgrade java-bigtable to 2.49.0

Change-Id: I801be8a764c212d7922634de61b41afaa3f2c3b6

* update metrics test to skipping trailers behavior

Change-Id: Ib7f1e7d87b2ef177d6fba69f0d139f5291c3d38c

* a bit more cleanup

Change-Id: I23ef04f8ea555e2b6d2c53aa8b98a331c71960ae

* more cleanup

Change-Id: I002a750839ee0b324121c2098f93fbcd5385728b

* try to fix test flakiness

Change-Id: Iebf9d10d8194e7f0b2c85ee5e530aa84e6a3a895

Author: igorbernstein2

.kokoro/build.sh

@@ -51,7 +51,7 @@ javadoc)
     ;;
 integration)
 # clean needed when running more than one IT profile
-    mvn --no-transfer-progress clean verify -B ${INTEGRATION_TEST_ARGS} -Penable-integration-tests -DtrimStackTrace=false -Dclirr.skip=true -fae
+    mvn --no-transfer-progress clean verify -B ${INTEGRATION_TEST_ARGS} -Penable-integration-tests -DtrimStackTrace=false -Dclirr.skip=true -DskipUnitTests=true
     RETURN_CODE=$?
     ;;
 integration-migration)
@@ -60,7 +60,7 @@ integration-migration)
   RETURN_CODE=$?
   ;;
 clirr)
-    mvn --no-transfer-progress install -B -Denforcer.skip=true clirr:check
+    mvn --no-transfer-progress install -B -Denforcer.skip=true -DskipTests clirr:check
     RETURN_CODE=$?
     ;;
 *)

bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/wrappers/BigtableHBaseSettings.java

@@ -15,6 +15,7 @@
  */
 package com.google.cloud.bigtable.hbase.wrappers;
 
+import static com.google.cloud.bigtable.hbase.BigtableOptionsFactory.APP_PROFILE_ID_KEY;
 import static com.google.cloud.bigtable.hbase.BigtableOptionsFactory.INSTANCE_ID_KEY;
 import static com.google.cloud.bigtable.hbase.BigtableOptionsFactory.PROJECT_ID_KEY;
 import static com.google.common.base.Strings.isNullOrEmpty;
@@ -38,6 +39,7 @@ public abstract class BigtableHBaseSettings {
   private final Configuration configuration;
   private final String projectId;
   private final String instanceId;
+  private final String appProfileId;
   private final int ttlSecondsForBackup;
 
   @VisibleForTesting
@@ -47,7 +49,7 @@ public abstract class BigtableHBaseSettings {
       "bulk.mutation.close.timeout.milliseconds";
 
   // Must be non-negative. Set to 0 to disable timeout.
-  private final long bulkMutationCloseTimeoutMilliseconds;
+  private final long bulkMutationCloseTimeoutMilliseconds;;
 
   public static BigtableHBaseSettings create(Configuration configuration) throws IOException {
     return BigtableHBaseVeneerSettings.create(configuration);
@@ -57,6 +59,8 @@ public BigtableHBaseSettings(Configuration configuration) {
     this.configuration = new Configuration(configuration);
     this.projectId = getRequiredValue(PROJECT_ID_KEY, "Project ID");
     this.instanceId = getRequiredValue(INSTANCE_ID_KEY, "Instance ID");
+    this.appProfileId = configuration.get(APP_PROFILE_ID_KEY);
+
     this.ttlSecondsForBackup =
         configuration.getInt(
             BigtableOptionsFactory.BIGTABLE_SNAPSHOT_DEFAULT_TTL_SECS_KEY,
@@ -79,6 +83,10 @@ public String getInstanceId() {
     return instanceId;
   }
 
+  public String getAppProfileId() {
+    return appProfileId;
+  }
+
   public int getTtlSecondsForBackup() {
     return ttlSecondsForBackup;
   }

bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/wrappers/veneer/BigtableHBaseVeneerSettings.java

@@ -64,6 +64,7 @@
 import com.google.api.gax.batching.FlowControlSettings;
 import com.google.api.gax.core.FixedCredentialsProvider;
 import com.google.api.gax.core.NoCredentialsProvider;
+import com.google.api.gax.grpc.ChannelPoolSettings;
 import com.google.api.gax.grpc.InstantiatingGrpcChannelProvider;
 import com.google.api.gax.rpc.FixedHeaderProvider;
 import com.google.api.gax.rpc.ServerStreamingCallSettings;
@@ -301,6 +302,7 @@ private BigtableDataSettings buildBigtableDataSettings(ClientOperationTimeouts c
       String host = emulatorEndpoint.get().substring(0, split);
       int port = Integer.parseInt(emulatorEndpoint.get().substring(split + 1));
       dataBuilder = BigtableDataSettings.newBuilderForEmulator(host, port);
+      configureConnection(dataBuilder.stubSettings(), BIGTABLE_HOST_KEY, emulatorEndpoint.get());
     } else {
       String endpoint =
           BigtableDataSettings.newBuilder()
@@ -500,7 +502,8 @@ public ManagedChannelBuilder apply(ManagedChannelBuilder channelBuilder) {
     if (endpointKey.equals(BIGTABLE_HOST_KEY)) {
       String channelCount = configuration.get(BIGTABLE_DATA_CHANNEL_COUNT_KEY);
       if (!Strings.isNullOrEmpty(channelCount)) {
-        channelProvider.setPoolSize(Integer.parseInt(channelCount));
+        channelProvider.setChannelPoolSettings(
+            ChannelPoolSettings.staticallySized(Integer.parseInt(channelCount)));
       }
     }
 
@@ -511,14 +514,11 @@ public ManagedChannelBuilder apply(ManagedChannelBuilder channelBuilder) {
           channelProvider.getChannelConfigurator();
 
       channelProvider.setChannelConfigurator(
-          new ApiFunction<ManagedChannelBuilder, ManagedChannelBuilder>() {
-            @Override
-            public ManagedChannelBuilder apply(ManagedChannelBuilder channelBuilder) {
-              if (prevConfigurator != null) {
-                channelBuilder = prevConfigurator.apply(channelBuilder);
-              }
-              return channelBuilder.executor(null);
+          channelBuilder -> {
+            if (prevConfigurator != null) {
+              channelBuilder = prevConfigurator.apply(channelBuilder);
             }
+            return channelBuilder.executor(null);
           });
     }
 

bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/wrappers/veneer/BulkMutationVeneerApi.java

@@ -83,6 +83,7 @@ public synchronized void close() throws IOException {
     } catch (InterruptedException | ExecutionException e) {
       throw new IOException("Could not close the bulk mutation Batcher", e);
     } catch (TimeoutException e) {
+      bulkMutateBatcher.cancelOutstanding();
       throw new IOException("Cloud not close the bulk mutation Batcher, timed out in close()", e);
     }
   }

bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/wrappers/veneer/SharedDataClientWrapperFactory.java

@@ -15,17 +15,12 @@
  */
 package com.google.cloud.bigtable.hbase.wrappers.veneer;
 
-import com.google.api.gax.core.BackgroundResource;
 import com.google.api.gax.core.CredentialsProvider;
-import com.google.api.gax.core.FixedCredentialsProvider;
-import com.google.api.gax.core.FixedExecutorProvider;
+import com.google.api.gax.core.NoCredentialsProvider;
 import com.google.api.gax.rpc.ClientContext;
-import com.google.api.gax.rpc.FixedHeaderProvider;
-import com.google.api.gax.rpc.FixedTransportChannelProvider;
-import com.google.api.gax.rpc.FixedWatchdogProvider;
 import com.google.cloud.bigtable.data.v2.BigtableDataClient;
+import com.google.cloud.bigtable.data.v2.BigtableDataClientFactory;
 import com.google.cloud.bigtable.data.v2.BigtableDataSettings;
-import com.google.cloud.bigtable.data.v2.BigtableDataSettings.Builder;
 import com.google.cloud.bigtable.data.v2.stub.EnhancedBigtableStubSettings;
 import com.google.cloud.bigtable.hbase.wrappers.DataClientWrapper;
 import com.google.cloud.bigtable.metrics.BigtableClientMetrics;
@@ -41,7 +36,7 @@
  * <p>This class is meant to support channel pool caching feature.
  */
 class SharedDataClientWrapperFactory {
-  private final Map<Key, ClientContext> cachedContexts = new HashMap<>();
+  private final Map<Key, BigtableDataClientFactory> cachedContexts = new HashMap<>();
   private final Map<Key, Integer> refCounts = new HashMap<>();
 
   private final Map<Key, Integer> channelPoolSizes = new HashMap<>();
@@ -52,13 +47,14 @@ synchronized DataClientWrapper createDataClient(BigtableHBaseVeneerSettings sett
     Key key = Key.createFromSettings(settings.getDataSettings());
 
     // Get or create ClientContext that will contained the shared resources
-    ClientContext sharedCtx = cachedContexts.get(key);
+    BigtableDataClientFactory sharedCtx = cachedContexts.get(key);
+
     if (sharedCtx == null) {
-      EnhancedBigtableStubSettings stubSettings = settings.getDataSettings().getStubSettings();
-      sharedCtx = ClientContext.create(stubSettings);
+      sharedCtx = BigtableDataClientFactory.create(settings.getDataSettings());
       cachedContexts.put(key, sharedCtx);
       refCounts.put(key, 0);
-      int channelPoolSize = BigtableVeneerApi.getChannelPoolSize(stubSettings);
+      int channelPoolSize =
+          BigtableVeneerApi.getChannelPoolSize(settings.getDataSettings().getStubSettings());
       for (int i = 0; i < channelPoolSize; i++) {
         BigtableClientMetrics.counter(MetricLevel.Info, "grpc.channel.active").inc();
       }
@@ -68,25 +64,17 @@ synchronized DataClientWrapper createDataClient(BigtableHBaseVeneerSettings sett
     refCounts.put(key, refCounts.get(key) + 1);
 
     try {
-      // Patch settings to use shared resources
-      Builder builder = settings.getDataSettings().toBuilder();
-      builder
-          .stubSettings()
-          .setRefreshingChannel(false)
-          .setTransportChannelProvider(
-              FixedTransportChannelProvider.create(sharedCtx.getTransportChannel()))
-          .setCredentialsProvider(FixedCredentialsProvider.create(sharedCtx.getCredentials()))
-          .setExecutorProvider(FixedExecutorProvider.create(sharedCtx.getExecutor()))
-          .setStreamWatchdogProvider(FixedWatchdogProvider.create(sharedCtx.getStreamWatchdog()))
-          .setHeaderProvider(FixedHeaderProvider.create(sharedCtx.getHeaders()))
-          .setClock(sharedCtx.getClock());
-
-      BigtableDataSettings data = builder.build();
+      final BigtableDataClient client;
+      if (settings.getAppProfileId() == null) {
+        client = sharedCtx.createForInstance(settings.getProjectId(), settings.getInstanceId());
+      } else {
+        client =
+            sharedCtx.createForInstance(
+                settings.getProjectId(), settings.getInstanceId(), settings.getAppProfileId());
+      }
       // Create a reference counted client wrapper
       return new SharedDataClientWrapper(
-          this,
-          key,
-          new DataClientVeneerApi(BigtableDataClient.create(data), settings.getClientTimeouts()));
+          this, key, new DataClientVeneerApi(client, settings.getClientTimeouts()));
     } catch (IOException | RuntimeException e) {
       release(key);
       throw e;
@@ -101,13 +89,15 @@ synchronized void release(Key key) {
     }
 
     refCounts.remove(key);
-    ClientContext clientContext = cachedContexts.remove(key);
+    BigtableDataClientFactory clientContext = cachedContexts.remove(key);
     for (int i = 0; i < channelPoolSizes.get(key); i++) {
       BigtableClientMetrics.counter(MetricLevel.Info, "grpc.channel.active").dec();
     }
     channelPoolSizes.remove(key);
-    for (BackgroundResource resource : clientContext.getBackgroundResources()) {
-      resource.shutdown();
+    try {
+      clientContext.close();
+    } catch (Exception e) {
+      throw new RuntimeException(e);
     }
   }
 
@@ -119,23 +109,24 @@ synchronized void release(Key key) {
    * compatible with a ClientContext required by {@link BigtableDataSettings}.
    */
   static final class Key {
+    private static final NoCredentialsProvider NO_CREDENTIALS_PROVIDER_INSTANCE =
+        NoCredentialsProvider.create();
     private final String endpoint;
-    private final Map<String, String> headers;
     private final CredentialsProvider credentialsProvider;
 
     static Key createFromSettings(BigtableDataSettings settings) {
       EnhancedBigtableStubSettings stubSettings = settings.getStubSettings();
+      CredentialsProvider effectiveCredProvider = stubSettings.getCredentialsProvider();
+      // NoCredentialsProvider doesnt implement equals, but all instances are equivalent
+      if (effectiveCredProvider instanceof NoCredentialsProvider) {
+        effectiveCredProvider = NO_CREDENTIALS_PROVIDER_INSTANCE;
+      }
 
-      return new Key(
-          stubSettings.getEndpoint(),
-          stubSettings.getHeaderProvider().getHeaders(),
-          stubSettings.getCredentialsProvider());
+      return new Key(stubSettings.getEndpoint(), effectiveCredProvider);
     }
 
-    private Key(
-        String endpoint, Map<String, String> headers, CredentialsProvider credentialsProvider) {
+    private Key(String endpoint, CredentialsProvider credentialsProvider) {
       this.endpoint = endpoint;
-      this.headers = headers;
       this.credentialsProvider = credentialsProvider;
     }
 
@@ -149,13 +140,12 @@ public boolean equals(Object o) {
       }
       Key key = (Key) o;
       return Objects.equal(endpoint, key.endpoint)
-          && Objects.equal(headers, key.headers)
           && Objects.equal(credentialsProvider, key.credentialsProvider);
     }
 
     @Override
     public int hashCode() {
-      return Objects.hashCode(endpoint, headers, credentialsProvider);
+      return Objects.hashCode(endpoint, credentialsProvider);
     }
   }
 }