fix: allow creating a managed channel (#3824)

* fix: allow creating a managed channel

* small fixes

Author: mutianf

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

@@ -306,4 +306,12 @@ public class BigtableOptionsFactory {
   /** Tracing cookie to send in the header with the requests */
   @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";
+
+  /**
+   * If this option is set to true, log a warning when creating a managed connection instead of
+   * throwing illegal argument exception.
+   */
+  @BetaApi("This API is not yet stable and may change in the future")
+  public static final String MANAGED_CONNECTION_WARNING =
+      "google.bigtable.managed.connection.warning";
 }

bigtable-client-core-parent/bigtable-hbase/src/main/java/org/apache/hadoop/hbase/client/AbstractBigtableConnection.java

@@ -19,6 +19,7 @@
 import com.google.api.core.InternalApi;
 import com.google.cloud.bigtable.hbase.BigtableBufferedMutator;
 import com.google.cloud.bigtable.hbase.BigtableHBaseVersion;
+import com.google.cloud.bigtable.hbase.BigtableOptionsFactory;
 import com.google.cloud.bigtable.hbase.BigtableRegionLocator;
 import com.google.cloud.bigtable.hbase.adapters.Adapters;
 import com.google.cloud.bigtable.hbase.adapters.HBaseRequestAdapter;
@@ -107,7 +108,12 @@ public AbstractBigtableConnection(Configuration conf) throws IOException {
   protected AbstractBigtableConnection(
       Configuration conf, boolean managed, ExecutorService pool, User user) throws IOException {
     if (managed) {
-      throw new IllegalArgumentException("Bigtable does not support managed connections.");
+      if (conf.getBoolean(BigtableOptionsFactory.MANAGED_CONNECTION_WARNING, false)) {
+        LOG.warn(
+            "Bigtable does not support managed connections. This connection will end up leaking.");
+      } else {
+        throw new IllegalArgumentException("Bigtable does not support managed connections.");
+      }
     }
 
     try {

bigtable-client-core-parent/bigtable-hbase/src/test/java/org/apache/hadoop/hbase/client/TestAbstractBigtableConnection.java

@@ -20,6 +20,7 @@
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
 import static org.mockito.Mockito.when;
 
 import com.google.bigtable.v2.BigtableGrpc;
@@ -50,13 +51,15 @@
 import java.util.List;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.HRegionInfo;
 import org.apache.hadoop.hbase.HRegionLocation;
 import org.apache.hadoop.hbase.ServerName;
 import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.security.User;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.util.VersionInfo;
 import org.junit.After;
@@ -222,12 +225,35 @@ public void testHeaders() throws IOException {
     }
   }
 
+  @Test
+  public void testManagedConnectionOverride() throws IOException {
+    Configuration configuration = new Configuration(false);
+    configuration.set(BigtableOptionsFactory.PROJECT_ID_KEY, PROJECT_ID);
+    configuration.set(BigtableOptionsFactory.INSTANCE_ID_KEY, INSTANCE_ID);
+    configuration.setBoolean(BigtableOptionsFactory.BIGTABLE_NULL_CREDENTIAL_ENABLE_KEY, true);
+    configuration.setInt(BigtableOptionsFactory.BIGTABLE_DATA_CHANNEL_COUNT_KEY, 1);
+    configuration.set(
+        BigtableOptionsFactory.BIGTABLE_EMULATOR_HOST_KEY, HOST_NAME + ":" + server.getPort());
+    configuration.setBoolean(BigtableOptionsFactory.MANAGED_CONNECTION_WARNING, true);
+    try (Connection newConnection =
+        new TestBigtableConnectionImpl(
+            configuration, true, Executors.newSingleThreadExecutor(), null)) {
+    } catch (IllegalArgumentException e) {
+      fail("Should not throw IllegalArgumentException");
+    }
+  }
+
   private class TestBigtableConnectionImpl extends AbstractBigtableConnection {
 
     TestBigtableConnectionImpl(Configuration conf) throws IOException {
       super(conf);
     }
 
+    TestBigtableConnectionImpl(Configuration conf, boolean managed, ExecutorService pool, User user)
+        throws IOException {
+      super(conf, managed, pool, user);
+    }
+
     @Override
     protected SampledRowKeysAdapter createSampledRowKeysAdapter(
         TableName tableName, ServerName serverName) {