Repository: googleapis/java-bigtable-hbase
Pull request #3031 (2021-06-23 20:38:09)

feat: allow DirectPath by default + update integration tests
* DirectPath is now opt-out by setting google.bigtable.direct.path.allowed to false
* Integration tests can now verify that they are running over the desired mode by specifying "bigtable.connection-mode" system property
* Integration tests update their user agent to signal membership in allow/deny lists
* DirectPathFallbackIT now supports IPv4


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 39bbf4ef7..340cc405e 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,12 @@
 # Changelog
 
+### [1.21.1](https://www.github.com/googleapis/java-bigtable-hbase/compare/v1.21.0...v1.21.1) (2021-06-29)
+
+
+### Bug Fixes
+
+* use binary search to get region location ([#3040](https://www.github.com/googleapis/java-bigtable-hbase/issues/3040)) ([7b8663f](https://www.github.com/googleapis/java-bigtable-hbase/commit/7b8663f31e805280ac594aaa98576c7d466eb1a2))
+
 ## [1.21.0](https://www.github.com/googleapis/java-bigtable-hbase/compare/v1.20.1...v1.21.0) (2021-06-22)
 
 
diff --git a/bigtable-client-core-parent/bigtable-client-core/pom.xml b/bigtable-client-core-parent/bigtable-client-core/pom.xml
index 7ef2d5e5c..950e891ca 100644
--- a/bigtable-client-core-parent/bigtable-client-core/pom.xml
+++ b/bigtable-client-core-parent/bigtable-client-core/pom.xml
@@ -20,7 +20,7 @@ limitations under the License.
   <parent>
     <groupId>com.google.cloud.bigtable</groupId>
     <artifactId>bigtable-client-core-parent</artifactId>
-    <version>1.21.1-SNAPSHOT</version> <!-- {x-version-update:bigtable-client-core-parent:current} -->
+    <version>1.21.1</version> <!-- {x-version-update:bigtable-client-core-parent:current} -->
   </parent>
 
   <artifactId>bigtable-client-core</artifactId>
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/config/BigtableOptions.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/config/BigtableOptions.java
index a36542542..98bdc9391 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/config/BigtableOptions.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/config/BigtableOptions.java
@@ -62,11 +62,6 @@ public class BigtableOptions implements Serializable, Cloneable {
   @InternalApi("For internal usage only")
   public static final String BIGTABLE_DATA_HOST_DEFAULT = "bigtable.googleapis.com";
 
-  // Temporary DirectPath config
-  private static final String DIRECT_PATH_ENV_VAR = "GOOGLE_CLOUD_ENABLE_DIRECT_PATH";
-  private static final String BIGTABLE_DIRECTPATH_DATA_HOST_DEFAULT =
-      "directpath-bigtable.googleapis.com";
-
   /** For internal use only - public for technical reasons. */
   @InternalApi("For internal usage only")
   public static final String BIGTABLE_BATCH_DATA_HOST_DEFAULT = "batch-bigtable.googleapis.com";
@@ -104,27 +99,6 @@ public class BigtableOptions implements Serializable, Cloneable {
     return builder().build();
   }
 
-  /**
-   * Checks if DirectPath is enabled via an environment variable.
-   *
-   * <p>For internal use only - public for technical reasons.
-   */
-  @InternalApi("For internal use only")
-  public static boolean isDirectPathEnabled() {
-    String whiteList = MoreObjects.firstNonNull(System.getenv(DIRECT_PATH_ENV_VAR), "").trim();
-
-    if (whiteList.isEmpty()) {
-      return false;
-    }
-
-    for (String entry : whiteList.split(",")) {
-      if (BIGTABLE_DATA_HOST_DEFAULT.contains(entry)) {
-        return true;
-      }
-    }
-    return false;
-  }
-
   /** Create a new instance of the {@link Builder}. */
   public static Builder builder() {
     return new Builder();
@@ -141,12 +115,10 @@ public class BigtableOptions implements Serializable, Cloneable {
       options.appProfileId = BIGTABLE_APP_PROFILE_DEFAULT;
 
       // Optional configuration for hosts - useful for the Bigtable team, more than anything else.
-      options.dataHost =
-          isDirectPathEnabled()
-              ? BIGTABLE_DIRECTPATH_DATA_HOST_DEFAULT
-              : BIGTABLE_DATA_HOST_DEFAULT;
+      options.dataHost = BIGTABLE_DATA_HOST_DEFAULT;
       options.adminHost = BIGTABLE_ADMIN_HOST_DEFAULT;
       options.port = BIGTABLE_PORT_DEFAULT;
+      options.directPathAllowed = true;
 
       options.dataChannelCount = BIGTABLE_DATA_CHANNEL_COUNT_DEFAULT;
       options.usePlaintextNegotiation = false;
@@ -212,6 +184,17 @@ public class BigtableOptions implements Serializable, Cloneable {
       return this;
     }
 
+    public Builder disableDirectPath() {
+      options.directPathAllowed = false;
+      return this;
+    }
+
+    /** For internal use only - public for technical reasons. */
+    @InternalApi("Visible for test only")
+    public ChannelConfigurator getChannelConfigurator() {
+      return this.options.channelConfigurator;
+    }
+
     /** For internal use only - public for technical reasons. */
     @InternalApi("Visible for test only")
     public Builder setChannelConfigurator(ChannelConfigurator configurator) {
@@ -372,6 +355,7 @@ public class BigtableOptions implements Serializable, Cloneable {
   private String adminHost;
   private String dataHost;
   private int port;
+  private boolean directPathAllowed;
   private String projectId;
   private String instanceId;
   private String appProfileId = BIGTABLE_APP_PROFILE_DEFAULT;
@@ -448,6 +432,11 @@ public class BigtableOptions implements Serializable, Cloneable {
     return port;
   }
 
+  @BetaApi("surface for DirectPath is still unstable and may change in the future")
+  public boolean isDirectPathAllowed() {
+    return directPathAllowed;
+  }
+
   /**
    * Get the credential this object was constructed with. May be null.
    *
@@ -562,6 +551,7 @@ public class BigtableOptions implements Serializable, Cloneable {
         && (dataChannelCount == other.dataChannelCount)
         && (usePlaintextNegotiation == other.usePlaintextNegotiation)
         && (useCachedDataPool == other.useCachedDataPool)
+        && (directPathAllowed == other.directPathAllowed)
         && Objects.equals(adminHost, other.adminHost)
         && Objects.equals(dataHost, other.dataHost)
         && Objects.equals(projectId, other.projectId)
@@ -585,6 +575,7 @@ public class BigtableOptions implements Serializable, Cloneable {
         .omitNullValues()
         .add("dataHost", dataHost)
         .add("adminHost", adminHost)
+        .add("directPathEnabled", directPathAllowed)
         .add("port", port)
         .add("projectId", projectId)
         .add("instanceId", instanceId)
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableClusterUtilities.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableClusterUtilities.java
index fbfeceee3..b4e133b69 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableClusterUtilities.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableClusterUtilities.java
@@ -154,7 +154,8 @@ public class BigtableClusterUtilities implements AutoCloseable {
 
     ClientInterceptor[] interceptors =
         BigtableSession.createAdminApiInterceptors(options).toArray(new ClientInterceptor[0]);
-    channel = BigtableSession.createNettyChannel(options.getAdminHost(), options, interceptors);
+    channel =
+        BigtableSession.createNettyChannel(options.getAdminHost(), options, false, interceptors);
     client = new BigtableInstanceGrpcClient(channel);
   }
 
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableSession.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableSession.java
index 40c16793b..8c3be6a0a 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableSession.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/BigtableSession.java
@@ -58,6 +58,7 @@ import com.google.cloud.bigtable.grpc.io.Watchdog;
 import com.google.cloud.bigtable.grpc.io.WatchdogInterceptor;
 import com.google.cloud.bigtable.metrics.BigtableClientMetrics;
 import com.google.cloud.bigtable.metrics.BigtableClientMetrics.MetricLevel;
+import com.google.cloud.bigtable.util.DirectPathUtil;
 import com.google.cloud.bigtable.util.ReferenceCountedHashMap;
 import com.google.cloud.bigtable.util.ReferenceCountedHashMap.Callable;
 import com.google.cloud.bigtable.util.ThreadUtil;
@@ -134,6 +135,9 @@ public class BigtableSession implements Closeable {
   // Use this conservative values for timeout (10s)
   static final long CHANNEL_KEEP_ALIVE_TIMEOUT_SECONDS = 10;
 
+  static final long DIRECT_PATH_KEEP_ALIVE_TIME_SECONDS = 3600;
+  static final long DIRECT_PATH_KEEP_ALIVE_TIMEOUT_SECONDS = 20;
+
   @VisibleForTesting
   static final String PROJECT_ID_EMPTY_OR_NULL = "ProjectId must not be empty or null.";
 
@@ -210,7 +214,6 @@ public class BigtableSession implements Closeable {
 
   private final BigtableOptions options;
   private final List<ManagedChannel> managedChannels;
-  @Deprecated private final List<ClientInterceptor> dataChannelInterceptors;
 
   private final BigtableDataClient dataClient;
   private final RequestContext dataRequestContext;
@@ -311,31 +314,34 @@ public class BigtableSession implements Closeable {
       this.dataClient = null;
       this.throttlingDataClient = null;
       this.dataRequestContext = null;
-      this.dataChannelInterceptors = null;
     } else {
+      boolean useDirectPath =
+          options.isDirectPathAllowed()
+              && DirectPathUtil.shouldAttemptDirectPath(
+                  options.getDataHost(), options.getPort(), options.getCredentialOptions());
+
       // Get a raw data channel pool - depending on the settings, this channel can either be
       // cached/shared or it can specific to this session. If it's specific to this session,
       // it will be added to managedChannels and cleaned up when this session is closed.
       ManagedChannel rawDataChannelPool;
       if (options.useCachedChannel()) {
         synchronized (BigtableSession.class) {
-          String key = String.format("%s:%d", options.getDataHost(), options.getPort());
+          String key =
+              String.format("%s:%s:%d", useDirectPath, options.getDataHost(), options.getPort());
           rawDataChannelPool = cachedDataChannelPools.get(key);
           if (rawDataChannelPool == null) {
-            rawDataChannelPool = createRawDataChannelPool(options);
+            rawDataChannelPool = createRawDataChannelPool(options, useDirectPath);
             cachedDataChannelPools.put(key, rawDataChannelPool);
           }
         }
       } else {
-        rawDataChannelPool = createRawDataChannelPool(options);
+        rawDataChannelPool = createRawDataChannelPool(options, useDirectPath);
         managedChannels.add(rawDataChannelPool);
       }
 
-      // TODO: stop saving the data channel interceptors as instance variables, this is here only to
-      // support deprecated methods
-      dataChannelInterceptors = createDataApiInterceptors(options);
       Channel dataChannel =
-          ClientInterceptors.intercept(rawDataChannelPool, dataChannelInterceptors);
+          ClientInterceptors.intercept(
+              rawDataChannelPool, createDataApiInterceptors(options, useDirectPath));
 
       this.dataRequestContext =
           RequestContext.create(
@@ -366,7 +372,7 @@ public class BigtableSession implements Closeable {
           new BigtableDataGrpcClient(asyncDataChannel, sharedPools.getRetryExecutor(), options);
       throttlingDataClient.setDeadlineGeneratorFactory(callOptionsFactory);
 
-      ManagedChannel rawAdminChannel = createNettyChannel(options.getAdminHost(), options);
+      ManagedChannel rawAdminChannel = createNettyChannel(options.getAdminHost(), options, false);
       managedChannels.add(rawAdminChannel);
 
       Channel adminChannel =
@@ -414,8 +420,8 @@ public class BigtableSession implements Closeable {
     return interceptors.build();
   }
 
-  private List<ClientInterceptor> createDataApiInterceptors(BigtableOptions options)
-      throws IOException {
+  private List<ClientInterceptor> createDataApiInterceptors(
+      BigtableOptions options, boolean useDirectPath) throws IOException {
     ImmutableList.Builder<ClientInterceptor> interceptors = ImmutableList.builder();
 
     // TODO: instanceName should never be null
@@ -428,7 +434,7 @@ public class BigtableSession implements Closeable {
 
     interceptors.add(setupWatchdog());
 
-    if (!BigtableOptions.isDirectPathEnabled()) {
+    if (!useDirectPath) {
       ClientInterceptor authInterceptor = createAuthInterceptor(options);
       if (authInterceptor != null) {
         interceptors.add(authInterceptor);
@@ -486,13 +492,13 @@ public class BigtableSession implements Closeable {
 
   // <editor-fold desc="Channel management">
 
-  private static ChannelPool createRawDataChannelPool(final BigtableOptions options)
-      throws IOException {
+  private static ChannelPool createRawDataChannelPool(
+      final BigtableOptions options, final boolean useDirectPath) throws IOException {
     ChannelPool.ChannelFactory channelFactory =
         new ChannelPool.ChannelFactory() {
           @Override
           public ManagedChannel create() throws IOException {
-            return createNettyChannel(options.getDataHost(), options);
+            return createNettyChannel(options.getDataHost(), options, useDirectPath);
           }
         };
     return new ChannelPool(channelFactory, options.getChannelCount());
@@ -500,10 +506,11 @@ public class BigtableSession implements Closeable {
   /** For internal use only - public for technical reasons. */
   @InternalApi("For internal usage only")
   public static ManagedChannel createNettyChannel(
-      String host, BigtableOptions options, ClientInterceptor... interceptors) throws SSLException {
-
-    // DirectPath is only supported for data currently
-    boolean isDirectPath = BigtableOptions.isDirectPathEnabled() && !host.contains("admin");
+      String host,
+      BigtableOptions options,
+      boolean useDirectPath,
+      ClientInterceptor... interceptors)
+      throws SSLException {
 
     LOG.info("Creating new channel for %s", host);
     if (LOG.getLog().isTraceEnabled()) {
@@ -511,11 +518,7 @@ public class BigtableSession implements Closeable {
     }
 
     ManagedChannelBuilder<?> builder;
-    if (isDirectPath) {
-      LOG.warn(
-          "Connecting to Bigtable using DirectPath."
-              + " This is currently an experimental feature and should not be used in production.");
-
+    if (useDirectPath) {
       builder = ComputeEngineChannelBuilder.forAddress(host, options.getPort());
       // When channel pooling is enabled, force the pick_first grpclb strategy.
       // This is necessary to avoid the multiplicative effect of creating channel pool with
@@ -535,13 +538,19 @@ public class BigtableSession implements Closeable {
       ImmutableMap<String, Object> loadBalancingConfig =
           ImmutableMap.<String, Object>of("loadBalancingConfig", ImmutableList.of(grpcLbPolicy));
 
-      builder.defaultServiceConfig(loadBalancingConfig);
+      builder
+          .defaultServiceConfig(loadBalancingConfig)
+          .keepAliveTime(DIRECT_PATH_KEEP_ALIVE_TIME_SECONDS, TimeUnit.SECONDS)
+          .keepAliveTimeout(DIRECT_PATH_KEEP_ALIVE_TIMEOUT_SECONDS, TimeUnit.SECONDS);
     } else {
       builder = ManagedChannelBuilder.forAddress(host, options.getPort());
 
       if (options.usePlaintextNegotiation()) {
         builder.usePlaintext();
       }
+      builder
+          .keepAliveTime(CHANNEL_KEEP_ALIVE_TIME_SECONDS, TimeUnit.SECONDS)
+          .keepAliveTimeout(CHANNEL_KEEP_ALIVE_TIMEOUT_SECONDS, TimeUnit.SECONDS);
     }
 
     if (options.getChannelConfigurator() != null) {
@@ -549,10 +558,9 @@ public class BigtableSession implements Closeable {
     }
 
     return builder
+        .disableServiceConfigLookUp()
         .idleTimeout(Long.MAX_VALUE, TimeUnit.SECONDS)
         .maxInboundMessageSize(MAX_MESSAGE_SIZE)
-        .keepAliveTime(CHANNEL_KEEP_ALIVE_TIME_SECONDS, TimeUnit.SECONDS)
-        .keepAliveTimeout(CHANNEL_KEEP_ALIVE_TIMEOUT_SECONDS, TimeUnit.SECONDS)
         // Default behavior Do not use keepalive without any outstanding rpc calls as it can add a
         // bunch of load.
         .userAgent(BigtableVersionInfo.CORE_USER_AGENT + "," + options.getUserAgent())
@@ -705,7 +713,7 @@ public class BigtableSession implements Closeable {
   public static BigtableInstanceClient createInstanceClient(BigtableOptions options)
       throws IOException, GeneralSecurityException {
 
-    ManagedChannel rawAdminChannel = createNettyChannel(options.getAdminHost(), options);
+    ManagedChannel rawAdminChannel = createNettyChannel(options.getAdminHost(), options, false);
     Channel adminChannel =
         ClientInterceptors.intercept(rawAdminChannel, createAdminApiInterceptors(options));
     return new BigtableInstanceGrpcClient(adminChannel);
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/util/DirectPathUtil.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/util/DirectPathUtil.java
new file mode 100644
index 000000000..20eeb86ac
--- /dev/null
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/util/DirectPathUtil.java
@@ -0,0 +1,88 @@
+/*
+ * Copyright 2021 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.google.cloud.bigtable.util;
+
+import com.google.api.core.InternalApi;
+import com.google.auth.Credentials;
+import com.google.auth.oauth2.ComputeEngineCredentials;
+import com.google.cloud.bigtable.config.BigtableOptions;
+import com.google.cloud.bigtable.config.CredentialFactory;
+import com.google.cloud.bigtable.config.CredentialOptions;
+import com.google.cloud.bigtable.config.CredentialOptions.CredentialType;
+import com.google.cloud.bigtable.config.Logger;
+import com.google.common.base.Supplier;
+import com.google.common.base.Suppliers;
+import com.google.common.io.Files;
+import java.io.File;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.security.GeneralSecurityException;
+
+@InternalApi("For internal use only")
+public class DirectPathUtil {
+  private static final Logger LOG = new Logger(DirectPathUtil.class);
+
+  private static final String GCE_PRODUCTION_NAME_PRIOR_2016 = "Google";
+  private static final String GCE_PRODUCTION_NAME_AFTER_2016 = "Google Compute Engine";
+
+  private static final Supplier<Boolean> isOnComputeEngine =
+      Suppliers.memoize(
+          new Supplier<Boolean>() {
+            @Override
+            public Boolean get() {
+              String osName = System.getProperty("os.name");
+              if ("Linux".equals(osName)) {
+                try {
+                  String result =
+                      Files.asCharSource(
+                              new File("/sys/class/dmi/id/product_name"), StandardCharsets.UTF_8)
+                          .readFirstLine();
+
+                  return result != null
+                      && (result.contains(GCE_PRODUCTION_NAME_PRIOR_2016)
+                          || result.contains(GCE_PRODUCTION_NAME_AFTER_2016));
+                } catch (IOException e) {
+                  return false;
+                }
+              }
+              return false;
+            }
+          });
+
+  private DirectPathUtil() {}
+
+  public static boolean shouldAttemptDirectPath(
+      String endpoint, int port, CredentialOptions creds) {
+
+    return BigtableOptions.BIGTABLE_DATA_HOST_DEFAULT.equals(endpoint)
+        && BigtableOptions.BIGTABLE_PORT_DEFAULT == port
+        && isOnComputeEngine.get()
+        && areCredsDirectPathCompatible(creds);
+  }
+
+  private static boolean areCredsDirectPathCompatible(CredentialOptions credentialOptions) {
+    if (credentialOptions.getCredentialType() != CredentialType.DefaultCredentials) {
+      return false;
+    }
+    Credentials credentials = null;
+    try {
+      credentials = CredentialFactory.getCredentials(credentialOptions);
+    } catch (IOException | GeneralSecurityException e) {
+      LOG.warn("Failed to probe credentials, assuming they are not DirectPath compatible", e);
+    }
+    return credentials instanceof ComputeEngineCredentials;
+  }
+}
diff --git a/bigtable-client-core-parent/bigtable-hbase/pom.xml b/bigtable-client-core-parent/bigtable-hbase/pom.xml
index 122ebcdf2..361c11f2e 100644
--- a/bigtable-client-core-parent/bigtable-hbase/pom.xml
+++ b/bigtable-client-core-parent/bigtable-hbase/pom.xml
@@ -20,7 +20,7 @@ limitations under the License.
   <parent>
     <groupId>com.google.cloud.bigtable</groupId>
     <artifactId>bigtable-client-core-parent</artifactId>
-    <version>1.21.1-SNAPSHOT</version> <!-- {x-version-update:bigtable-client-core-parent:current} -->
+    <version>1.21.1</version> <!-- {x-version-update:bigtable-client-core-parent:current} -->
   </parent>
 
   <artifactId>bigtable-hbase</artifactId>
diff --git a/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/BigtableOptionsFactory.java b/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/BigtableOptionsFactory.java
index 703f1d959..45ffce726 100644
--- a/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/BigtableOptionsFactory.java
+++ b/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/BigtableOptionsFactory.java
@@ -26,20 +26,25 @@ import static com.google.cloud.bigtable.config.CallOptionsConfig.USE_TIMEOUT_DEF
 import static com.google.common.base.Strings.isNullOrEmpty;
 
 import com.google.api.core.BetaApi;
+import com.google.api.core.InternalApi;
 import com.google.api.core.InternalExtensionOnly;
 import com.google.auth.Credentials;
 import com.google.cloud.bigtable.config.BigtableOptions;
+import com.google.cloud.bigtable.config.BigtableOptions.ChannelConfigurator;
 import com.google.cloud.bigtable.config.BulkOptions;
 import com.google.cloud.bigtable.config.CallOptionsConfig;
 import com.google.cloud.bigtable.config.CredentialOptions;
 import com.google.cloud.bigtable.config.Logger;
 import com.google.cloud.bigtable.config.RetryOptions;
+import com.google.cloud.bigtable.hbase.util.IpVerificationInterceptor;
 import com.google.common.base.Optional;
 import com.google.common.base.Preconditions;
+import io.grpc.ManagedChannelBuilder;
 import io.grpc.Status;
 import java.io.FileInputStream;
 import java.io.FileNotFoundException;
 import java.io.IOException;
+import java.util.regex.Pattern;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.util.VersionInfo;
 
@@ -59,6 +64,8 @@ public class BigtableOptionsFactory {
   /** Constant <code>BIGTABLE_HOST_KEY="google.bigtable.endpoint.host"</code> */
   public static final String BIGTABLE_HOST_KEY = "google.bigtable.endpoint.host";
 
+  @BetaApi("The surface for DirectPath is currently unstable and may change in the future")
+  public static final String BIGTABLE_DIRECT_PATH_ALLOWED = "google.bigtable.direct.path.allowed";
   /**
    * Constant <code>BIGTABLE_HOST_KEY="google.bigtable.emulator.endpoint.host"</code>. Values should
    * be of format: `host:port`
@@ -331,6 +338,9 @@ public class BigtableOptionsFactory {
   @BetaApi("The API for setting tracing cookie is not yet stable and may change in the future")
   public static final String BIGTABLE_TRACING_COOKIE = "google.bigtable.tracing.cookie.header";
 
+  @InternalApi("for testing only")
+  public static final String BIGTABLE_TEST_DATA_IP_REGEX = "google.bigtable.test.data.ip.regex";
+
   /**
    * fromConfiguration.
    *
@@ -368,6 +378,10 @@ public class BigtableOptionsFactory {
       bigtableOptionsBuilder.setPort(Integer.parseInt(portOverrideStr));
     }
 
+    if (!configuration.getBoolean(BIGTABLE_DIRECT_PATH_ALLOWED, true)) {
+      bigtableOptionsBuilder.disableDirectPath();
+    }
+
     String usePlaintextStr = configuration.get(BIGTABLE_USE_PLAINTEXT_NEGOTIATION);
     if (usePlaintextStr != null) {
       bigtableOptionsBuilder.setUsePlaintextNegotiation(Boolean.parseBoolean(usePlaintextStr));
@@ -396,6 +410,26 @@ public class BigtableOptionsFactory {
     if (tracingCookie != null) {
       bigtableOptionsBuilder.setTracingCookie(tracingCookie);
     }
+
+    final String s = configuration.get(BIGTABLE_TEST_DATA_IP_REGEX);
+    if (s != null) {
+      final ChannelConfigurator oldConfigurator = bigtableOptionsBuilder.getChannelConfigurator();
+      bigtableOptionsBuilder.setChannelConfigurator(
+          new ChannelConfigurator() {
+            @Override
+            public ManagedChannelBuilder configureChannel(
+                ManagedChannelBuilder builder, String host) {
+              if (oldConfigurator != null) {
+                builder = oldConfigurator.configureChannel(builder, host);
+              }
+              if (!host.contains("admin")) {
+                builder = builder.intercept(new IpVerificationInterceptor(Pattern.compile(s)));
+              }
+              return builder;
+            }
+          });
+    }
+
     return bigtableOptionsBuilder.build();
   }
 
diff --git a/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/BigtableRegionLocator.java b/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/BigtableRegionLocator.java
index 6fb0e22e1..85518a2ed 100644
--- a/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/BigtableRegionLocator.java
+++ b/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/BigtableRegionLocator.java
@@ -22,6 +22,7 @@ import com.google.cloud.bigtable.core.IBigtableDataClient;
 import java.io.IOException;
 import java.util.List;
 import java.util.concurrent.ExecutionException;
+import org.apache.hadoop.hbase.HRegionInfo;
 import org.apache.hadoop.hbase.HRegionLocation;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.RegionLocator;
@@ -61,11 +62,30 @@ public abstract class BigtableRegionLocator extends AbstractBigtableRegionLocato
   /** {@inheritDoc} */
   @Override
   public HRegionLocation getRegionLocation(byte[] row, boolean reload) throws IOException {
-    for (HRegionLocation region : getRegions(reload)) {
-      if (region.getRegionInfo().containsRow(row)) {
-        return region;
+    List<HRegionLocation> regions = getRegions(reload);
+    return findRegion(regions, row);
+  }
+
+  private HRegionLocation findRegion(List<HRegionLocation> regions, byte[] row) throws IOException {
+    int low = 0;
+    int high = regions.size() - 1;
+
+    while (low <= high) {
+      int mid = (low + high) >>> 1;
+      HRegionLocation regionLocation = regions.get(mid);
+      HRegionInfo regionInfo = regionLocation.getRegionInfo();
+
+      // This isn't the last region (endKey != "") and row key is greater than the current bound
+      if (regionInfo.getEndKey().length > 0 && Bytes.compareTo(row, regionInfo.getEndKey()) >= 0) {
+        low = mid + 1;
+      } else if (Bytes.compareTo(row, regionInfo.getStartKey()) < 0) {
+        // no need to check empty key because it will compare naturally
+        high = mid - 1;
+      } else {
+        return regionLocation;
       }
     }
+    // Should never happen because the regions are contiguous
     throw new IOException("Region not found for row: " + Bytes.toStringBinary(row));
   }
 
diff --git a/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/util/IpVerificationInterceptor.java b/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/util/IpVerificationInterceptor.java
new file mode 100644
index 000000000..17b69da4c
--- /dev/null
+++ b/bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/util/IpVerificationInterceptor.java
@@ -0,0 +1,79 @@
+/*
+ * Copyright 2021 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.google.cloud.bigtable.hbase.util;
+
+import com.google.api.core.InternalApi;
+import com.google.common.base.Preconditions;
+import io.grpc.CallOptions;
+import io.grpc.CallOptions.Key;
+import io.grpc.Channel;
+import io.grpc.ClientCall;
+import io.grpc.ClientInterceptor;
+import io.grpc.ForwardingClientCall.SimpleForwardingClientCall;
+import io.grpc.ForwardingClientCallListener.SimpleForwardingClientCallListener;
+import io.grpc.Grpc;
+import io.grpc.Metadata;
+import io.grpc.MethodDescriptor;
+import java.net.SocketAddress;
+import java.util.regex.Pattern;
+
+@InternalApi("for testing only")
+public class IpVerificationInterceptor implements ClientInterceptor {
+  public static CallOptions.Key<Boolean> SKIP_IP_VERIFICATION =
+      Key.createWithDefault("SKIP_IP_VERIFICATION", false);
+  private final Pattern regex;
+
+  public IpVerificationInterceptor(Pattern regex) {
+    this.regex = regex;
+  }
+
+  @Override
+  public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
+      MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
+    final ClientCall<ReqT, RespT> clientCall = next.newCall(method, callOptions);
+
+    if (callOptions.getOption(SKIP_IP_VERIFICATION)) {
+      return clientCall;
+    }
+
+    return new SimpleForwardingClientCall<ReqT, RespT>(clientCall) {
+      @Override
+      public void start(Listener<RespT> responseListener, Metadata headers) {
+        super.start(
+            new SimpleForwardingClientCallListener<RespT>(responseListener) {
+              @Override
+              public void onHeaders(Metadata headers) {
+                // Check peer IP after connection is established.
+                SocketAddress remoteAddr =
+                    clientCall.getAttributes().get(Grpc.TRANSPORT_ATTR_REMOTE_ADDR);
+
+                Preconditions.checkNotNull(
+                    remoteAddr, "Call was missing TRANSPORT_ATTR_REMOTE_ADDR");
+
+                Preconditions.checkState(
+                    regex.matcher(remoteAddr.toString()).matches(),
+                    "Remote address %s didn't match the pattern %s",
+                    remoteAddr,
+                    regex);
+
+                super.onHeaders(headers);
+              }
+            },
+            headers);
+      }
+    };
+  }
+}
diff --git a/bigtable-client-core-parent/pom.xml b/bigtable-client-core-parent/pom.xml
index 83266a00e..aaffcb6a4 100644
--- a/bigtable-client-core-parent/pom.xml
+++ b/bigtable-client-core-parent/pom.xml
@@ -20,7 +20,7 @@ limitations under the License.
   <parent>
     <groupId>com.google.cloud.bigtable</groupId>
     <artifactId>bigtable-client-parent</artifactId>
-    <version>1.21.1-SNAPSHOT</version> <!-- {x-version-update:bigtable-client-parent:current} -->
+    <version>1.21.1</version> <!-- {x-version-update:bigtable-client-parent:current} -->
   </parent>
 
   <artifactId>bigtable-client-core-parent</artifactId>
diff --git a/bigtable-dataflow-parent/bigtable-beam-import/pom.xml b/bigtable-dataflow-parent/bigtable-beam-import/pom.xml
index 7b0de215e..ab4fce428 100644
--- a/bigtable-dataflow-parent/bigtable-beam-import/pom.xml
+++ b/bigtable-dataflow-parent/bigtable-beam-import/pom.xml
@@ -18,7 +18,7 @@ limitations under the License.
   <parent>
     <groupId>com.google.cloud.bigtable</groupId>
     <artifactId>bigtable-dataflow-parent</artifactId>
-    <version>1.21.1-SNAPSHOT</version> <!-- {x-version-update:bigtable-dataflow-parent:current} -->
+    <version>1.21.1</version> <!-- {x-version-update:bigtable-dataflow-parent:current} -->
   </parent>
   <modelVersion>4.0.0</modelVersion>
 
diff --git a/bigtable-dataflow-parent/bigtable-hbase-beam/pom.xml b/bigtable-dataflow-parent/bigtable-hbase-beam/pom.xml
index 25b3c47c9..4ff633b06 100644
--- a/bigtable-dataflow-parent/bigtable-hbase-beam/pom.xml
+++ b/bigtable-dataflow-parent/bigtable-hbase-beam/pom.xml
@@ -16,7 +16,7 @@ limitations under the License.
   <parent>
     <groupId>com.google.cloud.bigtable</groupId>
     <artifactId>bigtable-dataflow-parent</artifactId>
-    <version>1.21.1-SNAPSHOT</version> <!-- {x-version-update:bigtable-dataflow-parent:current} -->
+    <version>1.21.1</version> <!-- {x-version-update:bigtable-dataflow-parent:current} -->
   </parent>
 
   <artifactId>bigtable-hbase-beam</artifactId>
diff --git a/bigtable-dataflow-parent/pom.xml b/bigtable-dataflow-parent/pom.xml
index 6c109a433..24d156441 100644
--- a/bigtable-dataflow-parent/pom.xml
+++ b/bigtable-dataflow-parent/pom.xml
@@ -20,7 +20,7 @@ limitations under the License.
   <parent>
     <groupId>com.google.cloud.bigtable</groupId>
     <artifactId>bigtable-client-parent</artifactId>
-    <version>1.21.1-SNAPSHOT</version> <!-- {x-version-update:bigtable-dataflow-parent:current} -->
+    <version>1.21.1</version> <!-- {x-version-update:bigtable-dataflow-parent:current} -->
   </parent>
 
   <artifactId>bigtable-dataflow-parent</artifactId>
diff --git a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-benchmarks/pom.xml b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-benchmarks/pom.xml
index 3d2b49fbc..03018c4d2 100644
--- a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-benchmarks/pom.xml
+++ b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-benchmarks/pom.xml
@@ -3,7 +3,7 @@
   <parent>
     <artifactId>bigtable-hbase-1.x-parent</artifactId>
     <groupId>com.google.cloud.bigtable</groupId>
-    <version>1.21.1-SNAPSHOT</version> <!-- {x-version-update:bigtable-hbase-1x-parent:current} -->
+    <version>1.21.1</version> <!-- {x-version-update:bigtable-hbase-1x-parent:current} -->
   </parent>
   <modelVersion>4.0.0</modelVersion>
 
diff --git a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-hadoop/pom.xml b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-hadoop/pom.xml
index ad4ce785b..a96b5a40a 100644
--- a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-hadoop/pom.xml
+++ b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-hadoop/pom.xml
@@ -18,7 +18,7 @@ limitations under the License.
   <parent>
     <artifactId>bigtable-hbase-1.x-parent</artifactId>
     <groupId>com.google.cloud.bigtable</groupId>
-    <version>1.21.1-SNAPSHOT</version> <!-- {x-version-update:bigtable-hbase-1x-parent:current} -->
+    <version>1.21.1</version> <!-- {x-version-update:bigtable-hbase-1x-parent:current} -->
   </parent>
   <modelVersion>4.0.0</modelVersion>
 
diff --git a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-mapreduce/pom.xml b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-mapreduce/pom.xml
index 518e9d49a..a6a1ab302 100644
--- a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-mapreduce/pom.xml
+++ b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-mapreduce/pom.xml
@@ -20,7 +20,7 @@ limitations under the License.
   <parent>
     <groupId>com.google.cloud.bigtable</groupId>
     <artifactId>bigtable-hbase-1.x-parent</artifactId>
-    <version>1.21.1-SNAPSHOT</version> <!-- {x-version-update:bigtable-hbase-1x-parent:current} -->
+    <version>1.21.1</version> <!-- {x-version-update:bigtable-hbase-1x-parent:current} -->
   </parent>
 
   <artifactId>bigtable-hbase-1.x-mapreduce</artifactId>
diff --git a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-shaded/pom.xml b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-shaded/pom.xml
index 07d6d7a3b..bbe4855d7 100644
--- a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-shaded/pom.xml
+++ b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-shaded/pom.xml
@@ -18,7 +18,7 @@ limitations under the License.
   <parent>
     <artifactId>bigtable-hbase-1.x-parent</artifactId>
     <groupId>com.google.cloud.bigtable</groupId>
-    <version>1.21.1-SNAPSHOT</version> <!-- {x-version-update:bigtable-hbase-1x-parent:current} -->
+    <version>1.21.1</version> <!-- {x-version-update:bigtable-hbase-1x-parent:current} -->
   </parent>
   <modelVersion>4.0.0</modelVersion>
 
diff --git a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-tools/pom.xml b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-tools/pom.xml
index 7b13743d2..16f6323d8 100644
--- a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-tools/pom.xml
+++ b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-tools/pom.xml
@@ -5,7 +5,7 @@
   <parent>
     <artifactId>bigtable-hbase-1.x-parent</artifactId>
     <groupId>com.google.cloud.bigtable</groupId>
-    <version>1.21.1-SNAPSHOT</version> <!-- {x-version-update:bigtable-hbase-1x-parent:current} -->
+    <version>1.21.1</version> <!-- {x-version-update:bigtable-hbase-1x-parent:current} -->
   </parent>
   <modelVersion>4.0.0</modelVersion>
 
diff --git a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x/pom.xml b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x/pom.xml
index 1763623e4..790195b94 100644
--- a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x/pom.xml
+++ b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x/pom.xml
@@ -20,7 +20,7 @@ limitations under the License.
   <parent>
     <groupId>com.google.cloud.bigtable</groupId>
     <artifactId>bigtable-hbase-1.x-parent</artifactId>
-    <version>1.21.1-SNAPSHOT</version> <!-- {x-version-update:bigtable-hbase-1x-parent:current} -->
+    <version>1.21.1</version> <!-- {x-version-update:bigtable-hbase-1x-parent:current} -->
   </parent>
 
   <artifactId>bigtable-hbase-1.x</artifactId>
diff --git a/bigtable-hbase-1.x-parent/pom.xml b/bigtable-hbase-1.x-parent/pom.xml
index d339d19a3..4f61e254b 100644
--- a/bigtable-hbase-1.x-parent/pom.xml
+++ b/bigtable-hbase-1.x-parent/pom.xml
@@ -20,7 +20,7 @@ limitations under the License.
   <parent>
     <groupId>com.google.cloud.bigtable</groupId>
     <artifactId>bigtable-client-parent</artifactId>
-    <version>1.21.1-SNAPSHOT</version> <!-- {x-version-update:bigtable-hbase-1x-parent:current} -->
+    <version>1.21.1</version> <!-- {x-version-update:bigtable-hbase-1x-parent:current} -->
   </parent>
 
   <artifactId>bigtable-hbase-1.x-parent</artifactId>
diff --git a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-hadoop/pom.xml b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-hadoop/pom.xml
index 9af6f2af9..c42282030 100644
--- a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-hadoop/pom.xml
+++ b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-hadoop/pom.xml
@@ -18,7 +18,7 @@ limitations under the License.
   <parent>
     <artifactId>bigtable-hbase-2.x-parent</artifactId>
     <groupId>com.google.cloud.bigtable</groupId>
-    <version>1.21.1-SNAPSHOT</version> <!-- {x-version-update:bigtable-hbase-2x-parent:current} -->
+    <version>1.21.1</version> <!-- {x-version-update:bigtable-hbase-2x-parent:current} -->
   </parent>
   <modelVersion>4.0.0</modelVersion>
 
diff --git a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-shaded/pom.xml b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-shaded/pom.xml
index 18ab89f23..eda64aaf1 100644
--- a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-shaded/pom.xml
+++ b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-shaded/pom.xml
@@ -18,7 +18,7 @@ limitations under the License.
   <parent>
     <artifactId>bigtable-hbase-2.x-parent</artifactId>
     <groupId>com.google.cloud.bigtable</groupId>
-    <version>1.21.1-SNAPSHOT</version> <!-- {x-version-update:bigtable-hbase-2x-parent:current} -->
+    <version>1.21.1</version> <!-- {x-version-update:bigtable-hbase-2x-parent:current} -->
   </parent>
   <modelVersion>4.0.0</modelVersion>
 
diff --git a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/pom.xml b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/pom.xml
index 8fd310d46..fa1e24624 100644
--- a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/pom.xml
+++ b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/pom.xml
@@ -20,7 +20,7 @@ limitations under the License.
   <parent>
     <groupId>com.google.cloud.bigtable</groupId>
     <artifactId>bigtable-hbase-2.x-parent</artifactId>
-    <version>1.21.1-SNAPSHOT</version> <!-- {x-version-update:bigtable-hbase-2x-parent:current} -->
+    <version>1.21.1</version> <!-- {x-version-update:bigtable-hbase-2x-parent:current} -->
   </parent>
 
   <properties>
@@ -149,6 +149,14 @@ limitations under the License.
       <version>${mockito.version}</version>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>com.google.protobuf</groupId>
+      <artifactId>protobuf-java</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>com.google.api.grpc</groupId>
+      <artifactId>grpc-google-cloud-bigtable-v2</artifactId>
+    </dependency>
   </dependencies>
   <build>
     <plugins>
diff --git a/bigtable-hbase-2.x-parent/pom.xml b/bigtable-hbase-2.x-parent/pom.xml
index aea75f257..882787eb8 100644
--- a/bigtable-hbase-2.x-parent/pom.xml
+++ b/bigtable-hbase-2.x-parent/pom.xml
@@ -20,7 +20,7 @@ limitations under the License.
   <parent>
     <groupId>com.google.cloud.bigtable</groupId>
     <artifactId>bigtable-client-parent</artifactId>
-    <version>1.21.1-SNAPSHOT</version> <!-- {x-version-update:bigtable-client-parent:current} -->
+    <version>1.21.1</version> <!-- {x-version-update:bigtable-client-parent:current} -->
   </parent>
 
   <artifactId>bigtable-hbase-2.x-parent</artifactId>
diff --git a/pom.xml b/pom.xml
index babb6d674..564a1dc1e 100644
--- a/pom.xml
+++ b/pom.xml
@@ -19,7 +19,7 @@ limitations under the License.
 
   <groupId>com.google.cloud.bigtable</groupId>
   <artifactId>bigtable-client-parent</artifactId>
-  <version>1.21.1-SNAPSHOT</version> <!-- {x-version-update:bigtable-client-parent:current} -->
+  <version>1.21.1</version> <!-- {x-version-update:bigtable-client-parent:current} -->
   <packaging>pom</packaging>
   <name>${project.groupId}:${project.artifactId}</name>
   <url>https://cloud.google.com/bigtable/</url>
diff --git a/versions.txt b/versions.txt
index f88550c6c..b1ad900c2 100644
--- a/versions.txt
+++ b/versions.txt
@@ -1,9 +1,9 @@
 # Format:
 # module:released-version:current-version
 
-bigtable-client-parent:1.21.0:1.21.1-SNAPSHOT
-bigtable-client-core-parent:1.21.0:1.21.1-SNAPSHOT
-bigtable-hbase-1x-parent:1.21.0:1.21.1-SNAPSHOT
-bigtable-hbase-2x-parent:1.21.0:1.21.1-SNAPSHOT
-bigtable-dataflow-parent:1.21.0:1.21.1-SNAPSHOT
-bigtable-test:1.21.0:1.21.1-SNAPSHOT
+bigtable-client-parent:1.21.1:1.21.1
+bigtable-client-core-parent:1.21.1:1.21.1
+bigtable-hbase-1x-parent:1.21.1:1.21.1
+bigtable-hbase-2x-parent:1.21.1:1.21.1
+bigtable-dataflow-parent:1.21.1:1.21.1
+bigtable-test:1.21.1:1.21.1


Test patch:
diff --git a/bigtable-client-core-parent/bigtable-hbase-integration-tests-common/pom.xml b/bigtable-client-core-parent/bigtable-hbase-integration-tests-common/pom.xml
index 7f0cb14ac..13fc46abd 100644
--- a/bigtable-client-core-parent/bigtable-hbase-integration-tests-common/pom.xml
+++ b/bigtable-client-core-parent/bigtable-hbase-integration-tests-common/pom.xml
@@ -19,7 +19,7 @@ limitations under the License.
   <parent>
     <groupId>com.google.cloud.bigtable</groupId>
     <artifactId>bigtable-client-core-parent</artifactId>
-    <version>1.21.1-SNAPSHOT</version> <!-- {x-version-update:bigtable-client-core-parent:current} -->
+    <version>1.21.1</version> <!-- {x-version-update:bigtable-client-core-parent:current} -->
   </parent>
 
   <artifactId>bigtable-hbase-integration-tests-common</artifactId>
diff --git a/bigtable-client-core-parent/bigtable-hbase-integration-tests-common/src/test/java/com/google/cloud/bigtable/hbase/test_env/BigtableEnv.java b/bigtable-client-core-parent/bigtable-hbase-integration-tests-common/src/test/java/com/google/cloud/bigtable/hbase/test_env/BigtableEnv.java
index fe6a36927..22e9c62c8 100644
--- a/bigtable-client-core-parent/bigtable-hbase-integration-tests-common/src/test/java/com/google/cloud/bigtable/hbase/test_env/BigtableEnv.java
+++ b/bigtable-client-core-parent/bigtable-hbase-integration-tests-common/src/test/java/com/google/cloud/bigtable/hbase/test_env/BigtableEnv.java
@@ -81,6 +81,29 @@ class BigtableEnv extends SharedTestEnv {
       }
     }
 
+    // Configure DirectPath settings:
+    // - when a required mode is specified, ensure that ip address match the mode
+    // - specify a user agent that will trigger deny/allow lists on the serverside
+    switch (getConnectionMode()) {
+      case DEFAULT:
+        // nothing special
+        break;
+      case REQUIRE_CFE:
+        configuration.set("google.bigtable.test.data.ip.regex", "^(?!2001:4860:8040|34\\.126).*");
+        configuration.set("google.bigtable.custom.user.agent", "bigtable-directpath-disable");
+        break;
+      case REQUIRE_DIRECT_PATH:
+        configuration.set("google.bigtable.test.data.ip.regex", "^(?:2001:4860:8040|34\\.126).*");
+        configuration.set("google.bigtable.custom.user.agent", "bigtable-directpath-enable");
+        break;
+      case REQUIRE_DIRECT_PATH_IPV4:
+        configuration.set("google.bigtable.test.data.ip.regex", "^(?:34\\.126).*");
+        configuration.set("google.bigtable.custom.user.agent", "bigtable-directpath-enable");
+        break;
+      default:
+        throw new IllegalStateException("Unknown ConnectionMode: " + getConnectionMode());
+    }
+
     // Garbage collect tables that previous runs failed to clean up
     ListeningExecutorService executor = MoreExecutors.listeningDecorator(getExecutor());
     try (Connection connection = ConnectionFactory.createConnection(configuration);
diff --git a/bigtable-client-core-parent/bigtable-hbase-integration-tests-common/src/test/java/com/google/cloud/bigtable/hbase/test_env/ConnectionMode.java b/bigtable-client-core-parent/bigtable-hbase-integration-tests-common/src/test/java/com/google/cloud/bigtable/hbase/test_env/ConnectionMode.java
new file mode 100644
index 000000000..a65e3c0ec
--- /dev/null
+++ b/bigtable-client-core-parent/bigtable-hbase-integration-tests-common/src/test/java/com/google/cloud/bigtable/hbase/test_env/ConnectionMode.java
@@ -0,0 +1,26 @@
+/*
+ * Copyright 2021 Google Inc. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.google.cloud.bigtable.hbase.test_env;
+
+/**
+ * ConnectionMode to enforce for the tests. This will affect remote ip interceptors and use agent
+ */
+public enum ConnectionMode {
+  DEFAULT,
+  REQUIRE_CFE,
+  REQUIRE_DIRECT_PATH,
+  REQUIRE_DIRECT_PATH_IPV4
+}
diff --git a/bigtable-client-core-parent/bigtable-hbase-integration-tests-common/src/test/java/com/google/cloud/bigtable/hbase/test_env/SharedTestEnv.java b/bigtable-client-core-parent/bigtable-hbase-integration-tests-common/src/test/java/com/google/cloud/bigtable/hbase/test_env/SharedTestEnv.java
index 568f11478..a3ef8a042 100644
--- a/bigtable-client-core-parent/bigtable-hbase-integration-tests-common/src/test/java/com/google/cloud/bigtable/hbase/test_env/SharedTestEnv.java
+++ b/bigtable-client-core-parent/bigtable-hbase-integration-tests-common/src/test/java/com/google/cloud/bigtable/hbase/test_env/SharedTestEnv.java
@@ -15,6 +15,7 @@
  */
 package com.google.cloud.bigtable.hbase.test_env;
 
+import com.google.common.base.MoreObjects;
 import com.google.common.base.Preconditions;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import java.io.IOException;
@@ -87,4 +88,11 @@ abstract class SharedTestEnv {
   public Configuration getConfiguration() {
     return configuration;
   }
+
+  public ConnectionMode getConnectionMode() {
+    String modeStr =
+        MoreObjects.firstNonNull(
+            System.getProperty("bigtable.connection-mode"), ConnectionMode.DEFAULT.name());
+    return ConnectionMode.valueOf(modeStr);
+  }
 }
diff --git a/bigtable-client-core-parent/bigtable-hbase-integration-tests-common/src/test/java/com/google/cloud/bigtable/hbase/test_env/SharedTestEnvRule.java b/bigtable-client-core-parent/bigtable-hbase-integration-tests-common/src/test/java/com/google/cloud/bigtable/hbase/test_env/SharedTestEnvRule.java
index 083df1159..40a201116 100644
--- a/bigtable-client-core-parent/bigtable-hbase-integration-tests-common/src/test/java/com/google/cloud/bigtable/hbase/test_env/SharedTestEnvRule.java
+++ b/bigtable-client-core-parent/bigtable-hbase-integration-tests-common/src/test/java/com/google/cloud/bigtable/hbase/test_env/SharedTestEnvRule.java
@@ -149,4 +149,8 @@ public class SharedTestEnvRule extends ExternalResource {
   public ExecutorService getExecutor() {
     return sharedTestEnv.getExecutor();
   }
+
+  public ConnectionMode getConnectionMode() {
+    return sharedTestEnv.getConnectionMode();
+  }
 }
diff --git a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-integration-tests/pom.xml b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-integration-tests/pom.xml
index ab2ca7b9d..fd03316f6 100644
--- a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-integration-tests/pom.xml
+++ b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-integration-tests/pom.xml
@@ -20,7 +20,7 @@ limitations under the License.
   <parent>
     <groupId>com.google.cloud.bigtable</groupId>
     <artifactId>bigtable-hbase-1.x-parent</artifactId>
-    <version>1.21.1-SNAPSHOT</version> <!-- {x-version-update:bigtable-hbase-1x-parent:current} -->
+    <version>1.21.1</version> <!-- {x-version-update:bigtable-hbase-1x-parent:current} -->
   </parent>
 
   <artifactId>bigtable-hbase-1.x-integration-tests</artifactId>
diff --git a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/DirectPathFallbackIT.java b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/DirectPathFallbackIT.java
index c21abfc2a..de6c954e0 100644
--- a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/DirectPathFallbackIT.java
+++ b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x-integration-tests/src/test/java/com/google/cloud/bigtable/hbase/DirectPathFallbackIT.java
@@ -20,8 +20,17 @@ import com.google.bigtable.v2.RowSet;
 import com.google.cloud.bigtable.config.BigtableOptions;
 import com.google.cloud.bigtable.config.BigtableOptions.ChannelConfigurator;
 import com.google.cloud.bigtable.grpc.BigtableSession;
+import com.google.cloud.bigtable.hbase.test_env.ConnectionMode;
+import com.google.cloud.bigtable.hbase.util.IpVerificationInterceptor;
+import com.google.common.base.Stopwatch;
+import com.google.common.collect.ImmutableSet;
 import com.google.protobuf.ByteString;
+import io.grpc.CallOptions;
+import io.grpc.Channel;
+import io.grpc.ClientCall;
+import io.grpc.ClientInterceptor;
 import io.grpc.ManagedChannelBuilder;
+import io.grpc.MethodDescriptor;
 import io.grpc.alts.ComputeEngineChannelBuilder;
 import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder;
 import io.grpc.netty.shaded.io.netty.channel.ChannelDuplexHandler;
@@ -34,10 +43,12 @@ import io.grpc.netty.shaded.io.netty.channel.socket.nio.NioSocketChannel;
 import io.grpc.netty.shaded.io.netty.util.ReferenceCountUtil;
 import java.io.IOException;
 import java.lang.reflect.Field;
-import java.net.Inet6Address;
+import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
+import java.util.Set;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import org.junit.After;
@@ -57,9 +68,13 @@ public class DirectPathFallbackIT extends AbstractTest {
   private static final int MIN_COMPLETE_READ_CALLS = 40;
   private static final int NUM_RPCS_TO_SEND = 20;
 
-  private AtomicBoolean blackholeIPv6 = new AtomicBoolean();
+  // IP address prefixes allocated for DirectPath backends.
+  private static final String DP_IPV6_PREFIX = "2001:4860:8040";
+  private static final String DP_IPV4_PREFIX = "34.126";
+
+  private AtomicBoolean blackholeDpAddr = new AtomicBoolean();
   private AtomicInteger numBlocked = new AtomicInteger();
-  private AtomicInteger numIPv6Read = new AtomicInteger();
+  private AtomicInteger numDpAddrRead = new AtomicInteger();
 
   private ChannelFactory<NioSocketChannel> channelFactory;
   private EventLoopGroup eventLoopGroup;
@@ -73,9 +88,12 @@ public class DirectPathFallbackIT extends AbstractTest {
 
   @Before
   public void setup() throws IOException {
-    if (!BigtableOptions.isDirectPathEnabled()) {
+    Set<ConnectionMode> validModes =
+        ImmutableSet.of(
+            ConnectionMode.REQUIRE_DIRECT_PATH, ConnectionMode.REQUIRE_DIRECT_PATH_IPV4);
+    if (!validModes.contains(sharedTestEnv.getConnectionMode())) {
       throw new AssumptionViolatedException(
-          "DirectPath integration tests can only run against DirectPathEnv");
+          "DirectPathFallbackIT can only return when explicitly requested");
     }
 
     BigtableOptions.Builder bigtableOptions =
@@ -83,16 +101,32 @@ public class DirectPathFallbackIT extends AbstractTest {
             .toBuilder()
             .setDataChannelCount(1);
 
+    final ChannelConfigurator oldConfigurator = bigtableOptions.getChannelConfigurator();
     bigtableOptions.setChannelConfigurator(
         new ChannelConfigurator() {
           @Override
           public ManagedChannelBuilder configureChannel(
               ManagedChannelBuilder builder, String host) {
+            if (oldConfigurator != null) {
+              builder = oldConfigurator.configureChannel(builder, host);
+            }
             // TODO: remove this when admin api supports DirectPath
             if (host.contains("admin")) {
               return builder;
             }
             injectNettyChannelHandler(builder);
+            // Since we are forcing fallback, disable ip verification
+            builder.intercept(
+                new ClientInterceptor() {
+                  @Override
+                  public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
+                      MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
+                    return next.newCall(
+                        method,
+                        callOptions.withOption(
+                            IpVerificationInterceptor.SKIP_IP_VERIFICATION, true));
+                  }
+                });
             // Fail fast when blackhole is active
             builder.keepAliveTime(1, TimeUnit.SECONDS);
             builder.keepAliveTimeout(1, TimeUnit.SECONDS);
@@ -114,7 +148,47 @@ public class DirectPathFallbackIT extends AbstractTest {
   }
 
   @Test
-  public void testFallback() throws InterruptedException {
+  public void testFallback() throws InterruptedException, TimeoutException {
+    // Precondition: wait for DirectPath to connect
+    Assert.assertTrue("Failed to observe RPCs over DirectPath", exerciseDirectPath());
+
+    // Enable the blackhole, which will prevent communication with grpclb and thus DirectPath.
+    blackholeDpAddr.set(true);
+
+    // Send a request, which should be routed over IPv4 and CFE.
+    instrumentedSession
+        .getDataClient()
+        .readFlatRowsList(
+            ReadRowsRequest.newBuilder()
+                .setTableName(
+                    String.format(
+                        "projects/%s/instances/%s/tables/%s",
+                        sharedTestEnv.getConfiguration().get(BigtableOptionsFactory.PROJECT_ID_KEY),
+                        sharedTestEnv
+                            .getConfiguration()
+                            .get(BigtableOptionsFactory.INSTANCE_ID_KEY),
+                        sharedTestEnv.getDefaultTableName().toString()))
+                .setRows(RowSet.newBuilder().addRowKeys(ByteString.copyFromUtf8("nonexistent-row")))
+                .setRowsLimit(1)
+                .build());
+
+    // Verify that the above check was meaningful, by verifying that the blackhole actually dropped
+    // packets.
+    Assert.assertTrue("Failed to detect any IPv6 traffic in blackhole", numBlocked.get() > 0);
+
+    // Make sure that the client will start reading from IPv6 again by sending new requests and
+    // checking the injected IPv6 counter has been updated.
+    blackholeDpAddr.set(false);
+
+    Assert.assertTrue("Failed to upgrade back to DirectPath", exerciseDirectPath());
+  }
+
+  private boolean exerciseDirectPath() throws InterruptedException, TimeoutException {
+    Stopwatch stopwatch = Stopwatch.createStarted();
+    numDpAddrRead.set(0);
+
+    boolean seenEnough = false;
+
     ReadRowsRequest request =
         ReadRowsRequest.newBuilder()
             .setTableName(
@@ -126,34 +200,16 @@ public class DirectPathFallbackIT extends AbstractTest {
             .setRows(RowSet.newBuilder().addRowKeys(ByteString.copyFromUtf8("nonexistent-row")))
             .setRowsLimit(1)
             .build();
-    // Verify that we can send a request
-    instrumentedSession.getDataClient().readFlatRowsList(request);
-
-    // Enable the blackhole, which will prevent communication via IPv6 and thus DirectPath.
-    blackholeIPv6.set(true);
-
-    // Send a request, which should be routed over IPv4 and CFE.
-    instrumentedSession.getDataClient().readFlatRowsList(request);
-
-    // Verify that the above check was meaningful, by verifying that the blackhole actually dropped
-    // packets.
-    Assert.assertTrue("Failed to detect any IPv6 traffic in blackhole", numBlocked.get() > 0);
 
-    // Make sure that the client will start reading from IPv6 again by sending new requests and
-    // checking the injected IPv6 counter has been updated.
-    blackholeIPv6.set(false);
-    numIPv6Read.set(0);
-
-    while (numIPv6Read.get() < MIN_COMPLETE_READ_CALLS) {
+    while (!seenEnough && stopwatch.elapsed(TimeUnit.MINUTES) < 2) {
       for (int i = 0; i < NUM_RPCS_TO_SEND; i++) {
+        // Verify that we can send a request
         instrumentedSession.getDataClient().readFlatRowsList(request);
       }
-
       Thread.sleep(100);
+      seenEnough = numDpAddrRead.get() >= MIN_COMPLETE_READ_CALLS;
     }
-
-    Assert.assertTrue(
-        "Failed to upgrade back to DirectPath", numIPv6Read.get() > MIN_COMPLETE_READ_CALLS);
+    return seenEnough;
   }
 
   /**
@@ -175,6 +231,17 @@ public class DirectPathFallbackIT extends AbstractTest {
       NettyChannelBuilder nettyChannelBuilder = (NettyChannelBuilder) delegateChannelBuilder;
       nettyChannelBuilder.channelFactory(channelFactory);
       nettyChannelBuilder.eventLoopGroup(eventLoopGroup);
+
+      channelBuilder.intercept(
+          new ClientInterceptor() {
+            @Override
+            public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
+                MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, Channel next) {
+              return next.newCall(
+                  method,
+                  callOptions.withOption(IpVerificationInterceptor.SKIP_IP_VERIFICATION, true));
+            }
+          });
     } catch (NoSuchFieldException | IllegalAccessException e) {
       throw new RuntimeException("Failed to inject the netty ChannelHandler", e);
     }
@@ -196,7 +263,7 @@ public class DirectPathFallbackIT extends AbstractTest {
    * make IPv6 packets disappear
    */
   private class MyChannelHandler extends ChannelDuplexHandler {
-    private boolean isIPv6;
+    private boolean isDpAddr;
 
     @Override
     public void connect(
@@ -206,11 +273,13 @@ public class DirectPathFallbackIT extends AbstractTest {
         ChannelPromise promise)
         throws Exception {
 
-      this.isIPv6 =
-          (remoteAddress instanceof InetSocketAddress)
-              && ((InetSocketAddress) remoteAddress).getAddress() instanceof Inet6Address;
+      if (remoteAddress instanceof InetSocketAddress) {
+        InetAddress inetAddress = ((InetSocketAddress) remoteAddress).getAddress();
+        String addr = inetAddress.getHostAddress();
+        isDpAddr = addr.startsWith(DP_IPV6_PREFIX) || addr.startsWith(DP_IPV4_PREFIX);
+      }
 
-      if (!(isIPv6 && blackholeIPv6.get())) {
+      if (!(isDpAddr && blackholeDpAddr.get())) {
         super.connect(ctx, remoteAddress, localAddress, promise);
       } else {
         // Fail the connection fast
@@ -220,7 +289,7 @@ public class DirectPathFallbackIT extends AbstractTest {
 
     @Override
     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
-      boolean dropCall = isIPv6 && blackholeIPv6.get();
+      boolean dropCall = isDpAddr && blackholeDpAddr.get();
 
       if (dropCall) {
         // Don't notify the next handler and increment counter
@@ -233,14 +302,14 @@ public class DirectPathFallbackIT extends AbstractTest {
 
     @Override
     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
-      boolean dropCall = isIPv6 && blackholeIPv6.get();
+      boolean dropCall = isDpAddr && blackholeDpAddr.get();
 
       if (dropCall) {
         // Don't notify the next handler and increment counter
         numBlocked.incrementAndGet();
       } else {
-        if (isIPv6) {
-          numIPv6Read.incrementAndGet();
+        if (isDpAddr) {
+          numDpAddrRead.incrementAndGet();
         }
         super.channelReadComplete(ctx);
       }
diff --git a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x/src/test/java/com/google/cloud/bigtable/hbase1_x/TestBigtableConnection.java b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x/src/test/java/com/google/cloud/bigtable/hbase1_x/TestBigtableConnection.java
index 8f4e50bad..5a31f033e 100644
--- a/bigtable-hbase-1.x-parent/bigtable-hbase-1.x/src/test/java/com/google/cloud/bigtable/hbase1_x/TestBigtableConnection.java
+++ b/bigtable-hbase-1.x-parent/bigtable-hbase-1.x/src/test/java/com/google/cloud/bigtable/hbase1_x/TestBigtableConnection.java
@@ -37,6 +37,7 @@ import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.HRegionInfo;
+import org.apache.hadoop.hbase.HRegionLocation;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.Connection;
 import org.apache.hadoop.hbase.util.Bytes;
@@ -119,6 +120,15 @@ public class TestBigtableConnection {
     assertEquals("rowKey", Bytes.toString(regionInfoList.get(0).getEndKey()));
   }
 
+  @Test
+  public void testGetRegionLocation() throws IOException {
+    HRegionLocation regionLocation =
+        connection.getRegionLocator(TABLE_NAME).getRegionLocation("rowKey".getBytes());
+
+    assertEquals("rowKey", Bytes.toString(regionLocation.getRegionInfo().getStartKey()));
+    assertEquals("", Bytes.toString(regionLocation.getRegionInfo().getEndKey()));
+  }
+
   private static class FakeDataService extends BigtableGrpc.BigtableImplBase {
     final BlockingQueue<Object> requests = Queues.newLinkedBlockingDeque();
 
diff --git a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-integration-tests/pom.xml b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-integration-tests/pom.xml
index 50df7c28c..680127b93 100644
--- a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-integration-tests/pom.xml
+++ b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x-integration-tests/pom.xml
@@ -20,7 +20,7 @@ limitations under the License.
   <parent>
     <groupId>com.google.cloud.bigtable</groupId>
     <artifactId>bigtable-hbase-2.x-parent</artifactId>
-    <version>1.21.1-SNAPSHOT</version> <!-- {x-version-update:bigtable-hbase-2x-parent:current} -->
+    <version>1.21.1</version> <!-- {x-version-update:bigtable-hbase-2x-parent:current} -->
   </parent>
 
   <artifactId>bigtable-hbase-2.x-integration-tests</artifactId>
diff --git a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/test/java/com/google/cloud/bigtable/hbase/TestBigtableConnection.java b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/test/java/com/google/cloud/bigtable/hbase/TestBigtableConnection.java
index a4c814576..ed5e1e645 100644
--- a/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/test/java/com/google/cloud/bigtable/hbase/TestBigtableConnection.java
+++ b/bigtable-hbase-2.x-parent/bigtable-hbase-2.x/src/test/java/com/google/cloud/bigtable/hbase/TestBigtableConnection.java
@@ -15,14 +15,30 @@
  */
 package com.google.cloud.bigtable.hbase;
 
+import static org.junit.Assert.assertEquals;
+
+import com.google.bigtable.v2.BigtableGrpc;
+import com.google.bigtable.v2.SampleRowKeysRequest;
+import com.google.bigtable.v2.SampleRowKeysResponse;
 import com.google.cloud.bigtable.hbase2_x.BigtableConnection;
+import com.google.common.collect.Queues;
+import com.google.protobuf.ByteString;
+import io.grpc.Server;
+import io.grpc.ServerBuilder;
+import io.grpc.stub.StreamObserver;
 import java.io.IOException;
+import java.net.ServerSocket;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.TimeUnit;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HRegionLocation;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.Admin;
 import org.apache.hadoop.hbase.client.BufferedMutator;
 import org.apache.hadoop.hbase.client.Hbck;
+import org.apache.hadoop.hbase.client.RegionLocator;
 import org.apache.hadoop.hbase.client.Table;
+import org.apache.hadoop.hbase.util.Bytes;
 import org.junit.Assert;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -108,4 +124,75 @@ public class TestBigtableConnection {
       Assert.assertEquals("UnsupportedHbck", hbck1.toString());
     }
   }
+
+  @Test
+  public void testGetRegionLocation() throws IOException {
+    FakeDataService fakeDataService = new FakeDataService();
+    int dataPort;
+    try (ServerSocket s = new ServerSocket(0)) {
+      dataPort = s.getLocalPort();
+    }
+    Server server = ServerBuilder.forPort(dataPort).addService(fakeDataService).build();
+    server.start();
+
+    Configuration configuration = new Configuration(false);
+    configuration.set(BigtableOptionsFactory.PROJECT_ID_KEY, "project_id");
+    configuration.set(BigtableOptionsFactory.INSTANCE_ID_KEY, "instance_id");
+    configuration.set(BigtableOptionsFactory.BIGTABLE_NULL_CREDENTIAL_ENABLE_KEY, "true");
+    configuration.set(BigtableOptionsFactory.BIGTABLE_DATA_CHANNEL_COUNT_KEY, "1");
+    configuration.set(BigtableOptionsFactory.BIGTABLE_EMULATOR_HOST_KEY, "localhost:" + dataPort);
+    try (BigtableConnection connection = new BigtableConnection(configuration)) {
+      RegionLocator regionLocator = connection.getRegionLocator(TableName.valueOf("table_id"));
+
+      // SampleRowKeys returns a, b, c ... z
+      HRegionLocation regionLocation = regionLocator.getRegionLocation("1".getBytes());
+      assertEquals("", Bytes.toString(regionLocation.getRegion().getStartKey()));
+      assertEquals("a", Bytes.toString(regionLocation.getRegion().getEndKey()));
+
+      regionLocation = regionLocator.getRegionLocation("a".getBytes());
+      assertEquals("a", Bytes.toString(regionLocation.getRegion().getStartKey()));
+      assertEquals("b", Bytes.toString(regionLocation.getRegion().getEndKey()));
+
+      regionLocation = regionLocator.getRegionLocation("bbb".getBytes());
+      assertEquals("b", Bytes.toString(regionLocation.getRegion().getStartKey()));
+      assertEquals("c", Bytes.toString(regionLocation.getRegion().getEndKey()));
+
+      regionLocation = regionLocator.getRegionLocation("d".getBytes());
+      assertEquals("d", Bytes.toString(regionLocation.getRegion().getStartKey()));
+      assertEquals("e", Bytes.toString(regionLocation.getRegion().getEndKey()));
+
+      regionLocation = regionLocator.getRegionLocation("z".getBytes());
+      assertEquals("z", Bytes.toString(regionLocation.getRegion().getStartKey()));
+      assertEquals("", Bytes.toString(regionLocation.getRegion().getEndKey()));
+
+      regionLocation = regionLocator.getRegionLocation("zzz".getBytes());
+      assertEquals("z", Bytes.toString(regionLocation.getRegion().getStartKey()));
+      assertEquals("", Bytes.toString(regionLocation.getRegion().getEndKey()));
+    }
+  }
+
+  private static class FakeDataService extends BigtableGrpc.BigtableImplBase {
+    final BlockingQueue<Object> requests = Queues.newLinkedBlockingDeque();
+
+    @SuppressWarnings("unchecked")
+    <T> T popLastRequest() throws InterruptedException {
+      return (T) requests.poll(1, TimeUnit.SECONDS);
+    }
+
+    @Override
+    public void sampleRowKeys(
+        SampleRowKeysRequest request, StreamObserver<SampleRowKeysResponse> responseObserver) {
+      requests.add(request);
+      long offset = 1000L;
+      for (char i = 'a'; i <= 'z'; i++) {
+        responseObserver.onNext(
+            SampleRowKeysResponse.newBuilder()
+                .setRowKey(ByteString.copyFromUtf8(String.valueOf(i)))
+                .setOffsetBytes(offset)
+                .build());
+        offset += 1000;
+      }
+      responseObserver.onCompleted();
+    }
+  }
 }
diff --git a/bigtable-test/bigtable-emulator-maven-plugin/pom.xml b/bigtable-test/bigtable-emulator-maven-plugin/pom.xml
index bbbc2ab67..50fab32ab 100644
--- a/bigtable-test/bigtable-emulator-maven-plugin/pom.xml
+++ b/bigtable-test/bigtable-emulator-maven-plugin/pom.xml
@@ -20,7 +20,7 @@ limitations under the License.
   <parent>
     <artifactId>bigtable-test</artifactId>
     <groupId>com.google.cloud.bigtable</groupId>
-    <version>1.21.1-SNAPSHOT</version> <!-- {x-version-update:bigtable-test:current} -->
+    <version>1.21.1</version> <!-- {x-version-update:bigtable-test:current} -->
   </parent>
 
 
diff --git a/bigtable-test/pom.xml b/bigtable-test/pom.xml
index 9a5cc6c97..6d73dc3c5 100644
--- a/bigtable-test/pom.xml
+++ b/bigtable-test/pom.xml
@@ -20,7 +20,7 @@ limitations under the License.
   <parent>
     <artifactId>bigtable-client-parent</artifactId>
     <groupId>com.google.cloud.bigtable</groupId>
-    <version>1.21.1-SNAPSHOT</version> <!-- {x-version-update:bigtable-client-parent:current} -->
+    <version>1.21.1</version> <!-- {x-version-update:bigtable-client-parent:current} -->
   </parent>
 
   <artifactId>bigtable-test</artifactId>