feat(replication): map HBase table names to Cloud Bigtable table IDs

Author: vermas2012

hbase-migration-tools/bigtable-hbase-replication/README.md

@@ -143,6 +143,57 @@ Please refer
 to [HBaseToCloudBigtableReplicationConfiguration](bigtable-hbase-replication-core/src/main/java/com/google/cloud/bigtable/hbase/replication/configuration/HBaseToCloudBigtableReplicationConfiguration.java)
 for other properties that can be set.
 
+### Mapping HBase table names to Cloud Bigtable table IDs
+
+By default a WAL entry for HBase table `t` is written to the Cloud Bigtable
+table `t`. Cloud Bigtable does not have namespaces and `:` is not a valid
+character in a Cloud Bigtable table ID, so HBase tables in a non-default
+namespace (for example `my_namespace:my_table`) can not be replicated without
+renaming them.
+
+The endpoint accepts mappings from HBase table names to Cloud Bigtable
+table IDs. Source HBase table names are normalized to canonical form (for
+example, `default:my_table` and `my_table` refer to the same HBase table).
+Mappings must be injective (1:1): multiple HBase tables cannot map to the same
+Cloud Bigtable table ID.
+
+Every target must be a valid Cloud Bigtable table ID (`[_a-zA-Z0-9][-_.a-zA-Z0-9]*`,
+at most 50 characters); the endpoint refuses to start if the mapping is
+malformed or contains collisions, ensuring mutations are never replicated to the
+wrong table. Mapping targets are also reserved at runtime: if an unmapped HBase
+table has the same canonical name as a configured mapping target (for example,
+`ns:table_B` maps to `table_B` while an unmapped `default:table_B` is also
+replicated), `replicate()` fails the batch and halts replication for that peer
+until an explicit disambiguating mapping (such as `"table_B": "table_B_default"`)
+is added.
+
+The mapping is configured through `google.bigtable.replication.table_name_mapping`
+in `hbase-site.xml` using a JSON object of
+`{"source_hbase_table": "destination_bigtable_table"}` pairs (using the same
+JSON structure as the mapping file accepted by the
+[HBase schema translator](../../bigtable-hbase-1.x-parent/bigtable-hbase-1.x-tools)):
+
+```xml
+<property>
+    <name>google.bigtable.replication.table_name_mapping</name>
+    <value>{"my_namespace:my_table": "my_namespace_my_table", "other_namespace:events": "events"}</value>
+</property>
+```
+
+Note that `TABLE_CFS` in `add_peer` only *selects* the tables and column
+families that are replicated to a peer; it can not rename them. Use `TABLE_CFS`
+with the HBase table names (for example `TABLE_CFS => { 'my_namespace:my_table'
+=> [] }`) and the mapping above to choose the Cloud Bigtable table ID.
+
+**Operational notes:**
+* The mapping is loaded when the replication endpoint starts, so changing
+  `google.bigtable.replication.table_name_mapping` requires a rolling restart of
+  the HBase region servers.
+* After a restart, any un-replicated WAL entries still queued for the peer are
+  routed using the new mapping. Pause the replication peer or wait for
+  replication to catch up before remapping a table that is actively receiving
+  writes.
+
 ### Bidirectional Replication with Cloud Bigtable
 
 The bidirectional replication feature filters out mutations replicated to Hbase from Cloud Bigtable change streams with the [Bigtable-Hbase replicator](https://github.com/GoogleCloudPlatform/DataflowTemplates/tree/main/v2/bigtable-cdc-to-hbase) to prevent replication loops. 

hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-1.x-replication/src/test/java/com/google/cloud/bigtable/hbase1_x/replication/HbaseToCloudBigtableReplicationEndpointTest.java

@@ -16,6 +16,7 @@
 
 package com.google.cloud.bigtable.hbase1_x.replication;
 
+import static com.google.cloud.bigtable.hbase.replication.configuration.HBaseToCloudBigtableReplicationConfiguration.TABLE_NAME_MAPPING_KEY;
 import static com.google.cloud.bigtable.hbase.replication.utils.TestUtils.CF1;
 import static com.google.cloud.bigtable.hbase.replication.utils.TestUtils.FILTERED_ROW_KEY;
 import static com.google.cloud.bigtable.hbase.replication.utils.TestUtils.ROW_KEY;
@@ -36,6 +37,8 @@
 import org.apache.hadoop.hbase.HColumnDescriptor;
 import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.HTableDescriptor;
+import org.apache.hadoop.hbase.NamespaceDescriptor;
+import org.apache.hadoop.hbase.NamespaceExistException;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.Append;
 import org.apache.hadoop.hbase.client.Connection;
@@ -121,6 +124,11 @@ public Entry filter(Entry entry) {
   private Table cbtTable;
   private Table cbtTable2;
 
+  private static final TableName REMAPPED_SOURCE_TABLE =
+      TableName.valueOf("my_ns", "remapped_source_table");
+  private static final TableName REMAPPED_TARGET_TABLE =
+      TableName.valueOf("my_ns_remapped_target_table");
+
   @BeforeClass
   public static void setUpCluster() throws Exception {
     // Prepare HBase mini cluster configuration
@@ -131,6 +139,11 @@ public static void setUpCluster() throws Exception {
     conf.set("google.bigtable.project.id", "test-project");
     // This config will connect Replication endpoint to the emulator and not the prod CBT.
     conf.set("google.bigtable.emulator.endpoint.host", "localhost:" + bigtableEmulator.getPort());
+    conf.set(
+        TABLE_NAME_MAPPING_KEY,
+        String.format(
+            "{\"%s\": \"%s\"}",
+            REMAPPED_SOURCE_TABLE.getNameAsString(), REMAPPED_TARGET_TABLE.getNameAsString()));
 
     hbaseTestingUtil.startMiniCluster(2);
     replicationAdmin = new ReplicationAdmin(hbaseTestingUtil.getConfiguration());
@@ -500,4 +513,65 @@ public void testHBaseCBTTimestampTruncation() throws IOException, InterruptedExc
         hbaseCells.get(0).getTimestamp(),
         bigtableCells.get(0).getTimestamp());
   }
+
+  private void createRemappedTables(TableName hbaseSourceTable, TableName cbtTargetTable)
+      throws IOException {
+    String namespace = hbaseSourceTable.getNamespaceAsString();
+    if (!NamespaceDescriptor.DEFAULT_NAMESPACE_NAME_STR.equals(namespace)) {
+      try {
+        hbaseConnection.getAdmin().createNamespace(NamespaceDescriptor.create(namespace).build());
+      } catch (NamespaceExistException ignored) {
+        // Namespace already exists
+      }
+    }
+
+    HTableDescriptor hbaseHtd = new HTableDescriptor(hbaseSourceTable);
+    HColumnDescriptor cf1 = new HColumnDescriptor(TestUtils.CF1);
+    cf1.setMaxVersions(100);
+    cf1.setScope(HConstants.REPLICATION_SCOPE_GLOBAL);
+    hbaseHtd.addFamily(cf1);
+    HColumnDescriptor cf2 = new HColumnDescriptor(TestUtils.CF2);
+    cf2.setMaxVersions(100);
+    cf2.setScope(HConstants.REPLICATION_SCOPE_GLOBAL);
+    hbaseHtd.addFamily(cf2);
+    hbaseTestingUtil.getHBaseAdmin().createTable(hbaseHtd);
+
+    HTableDescriptor cbtHtd = new HTableDescriptor(cbtTargetTable);
+    HColumnDescriptor cbtCf1 = new HColumnDescriptor(TestUtils.CF1);
+    cbtCf1.setMaxVersions(100);
+    cbtHtd.addFamily(cbtCf1);
+    HColumnDescriptor cbtCf2 = new HColumnDescriptor(TestUtils.CF2);
+    cbtCf2.setMaxVersions(100);
+    cbtHtd.addFamily(cbtCf2);
+    cbtConnection.getAdmin().createTable(cbtHtd);
+  }
+
+  @Test
+  public void testTableNameMapping() throws Exception {
+    createRemappedTables(REMAPPED_SOURCE_TABLE, REMAPPED_TARGET_TABLE);
+
+    try (Table srcTable = hbaseConnection.getTable(REMAPPED_SOURCE_TABLE);
+        Table dstTable = cbtConnection.getTable(REMAPPED_TARGET_TABLE)) {
+      TestReplicationEndpoint.replicatedEntries.set(0);
+
+      for (int i = 0; i < 25; i++) {
+        Put put = new Put(TestUtils.getRowKey(i));
+        put.addColumn(TestUtils.CF1, TestUtils.COL_QUALIFIER, 0, TestUtils.getValue(i));
+        put.addColumn(TestUtils.CF2, TestUtils.COL_QUALIFIER, 0, TestUtils.getValue(100 + i));
+        srcTable.put(put);
+      }
+
+      TestUtils.assertTableEventuallyEquals(
+          srcTable, dstTable, () -> TestReplicationEndpoint.replicatedEntries.get() >= 25);
+    } finally {
+      if (hbaseTestingUtil.getHBaseAdmin().tableExists(REMAPPED_SOURCE_TABLE)) {
+        hbaseTestingUtil.getHBaseAdmin().disableTable(REMAPPED_SOURCE_TABLE);
+        hbaseTestingUtil.getHBaseAdmin().deleteTable(REMAPPED_SOURCE_TABLE);
+      }
+      if (cbtConnection.getAdmin().tableExists(REMAPPED_TARGET_TABLE)) {
+        cbtConnection.getAdmin().disableTable(REMAPPED_TARGET_TABLE);
+        cbtConnection.getAdmin().deleteTable(REMAPPED_TARGET_TABLE);
+      }
+    }
+  }
 }

hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-2.x-replication/src/test/java/com/google/cloud/bigtable/hbase2_x/replication/HbaseToCloudBigtableReplicationEndpointTest.java

@@ -16,6 +16,7 @@
 
 package com.google.cloud.bigtable.hbase2_x.replication;
 
+import static com.google.cloud.bigtable.hbase.replication.configuration.HBaseToCloudBigtableReplicationConfiguration.TABLE_NAME_MAPPING_KEY;
 import static com.google.cloud.bigtable.hbase.replication.utils.TestUtils.CF1;
 import static com.google.cloud.bigtable.hbase.replication.utils.TestUtils.CF2;
 import static com.google.cloud.bigtable.hbase.replication.utils.TestUtils.FILTERED_ROW_KEY;
@@ -36,6 +37,8 @@
 import org.apache.hadoop.hbase.HColumnDescriptor;
 import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.HTableDescriptor;
+import org.apache.hadoop.hbase.NamespaceDescriptor;
+import org.apache.hadoop.hbase.NamespaceExistException;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.Append;
 import org.apache.hadoop.hbase.client.Connection;
@@ -121,6 +124,11 @@ public Entry filter(Entry entry) {
   private Table cbtTable;
   private Table cbtTable2;
 
+  private static final TableName REMAPPED_SOURCE_TABLE =
+      TableName.valueOf("my_ns", "remapped_source_table");
+  private static final TableName REMAPPED_TARGET_TABLE =
+      TableName.valueOf("my_ns_remapped_target_table");
+
   @BeforeClass
   public static void setUpCluster() throws Exception {
     // Prepare HBase mini cluster configuration
@@ -131,6 +139,11 @@ public static void setUpCluster() throws Exception {
     conf.set("google.bigtable.project.id", "test-project");
     // This config will connect Replication endpoint to the emulator and not the prod CBT.
     conf.set("google.bigtable.emulator.endpoint.host", "localhost:" + bigtableEmulator.getPort());
+    conf.set(
+        TABLE_NAME_MAPPING_KEY,
+        String.format(
+            "{\"%s\": \"%s\"}",
+            REMAPPED_SOURCE_TABLE.getNameAsString(), REMAPPED_TARGET_TABLE.getNameAsString()));
 
     hbaseTestingUtil.startMiniCluster(2);
     replicationAdmin = new ReplicationAdmin(hbaseTestingUtil.getConfiguration());
@@ -494,4 +507,65 @@ public void testHBaseCBTTimestampTruncation() throws IOException, InterruptedExc
         hbaseCells.get(0).getTimestamp(),
         bigtableCells.get(0).getTimestamp());
   }
+
+  private void createRemappedTables(TableName hbaseSourceTable, TableName cbtTargetTable)
+      throws IOException {
+    String namespace = hbaseSourceTable.getNamespaceAsString();
+    if (!NamespaceDescriptor.DEFAULT_NAMESPACE_NAME_STR.equals(namespace)) {
+      try {
+        hbaseConnection.getAdmin().createNamespace(NamespaceDescriptor.create(namespace).build());
+      } catch (NamespaceExistException ignored) {
+        // Namespace already exists
+      }
+    }
+
+    HTableDescriptor hbaseHtd = new HTableDescriptor(hbaseSourceTable);
+    HColumnDescriptor cf1 = new HColumnDescriptor(TestUtils.CF1);
+    cf1.setMaxVersions(100);
+    cf1.setScope(HConstants.REPLICATION_SCOPE_GLOBAL);
+    hbaseHtd.addFamily(cf1);
+    HColumnDescriptor cf2 = new HColumnDescriptor(CF2);
+    cf2.setMaxVersions(100);
+    cf2.setScope(HConstants.REPLICATION_SCOPE_GLOBAL);
+    hbaseHtd.addFamily(cf2);
+    hbaseTestingUtil.getHBaseAdmin().createTable(hbaseHtd);
+
+    HTableDescriptor cbtHtd = new HTableDescriptor(cbtTargetTable);
+    HColumnDescriptor cbtCf1 = new HColumnDescriptor(TestUtils.CF1);
+    cbtCf1.setMaxVersions(100);
+    cbtHtd.addFamily(cbtCf1);
+    HColumnDescriptor cbtCf2 = new HColumnDescriptor(CF2);
+    cbtCf2.setMaxVersions(100);
+    cbtHtd.addFamily(cbtCf2);
+    cbtConnection.getAdmin().createTable(cbtHtd);
+  }
+
+  @Test
+  public void testTableNameMapping() throws Exception {
+    createRemappedTables(REMAPPED_SOURCE_TABLE, REMAPPED_TARGET_TABLE);
+
+    try (Table srcTable = hbaseConnection.getTable(REMAPPED_SOURCE_TABLE);
+        Table dstTable = cbtConnection.getTable(REMAPPED_TARGET_TABLE)) {
+      TestReplicationEndpoint.replicatedEntries.set(0);
+
+      for (int i = 0; i < 25; i++) {
+        Put put = new Put(TestUtils.getRowKey(i));
+        put.addColumn(TestUtils.CF1, TestUtils.COL_QUALIFIER, 0, TestUtils.getValue(i));
+        put.addColumn(CF2, TestUtils.COL_QUALIFIER, 0, TestUtils.getValue(100 + i));
+        srcTable.put(put);
+      }
+
+      TestUtils.assertTableEventuallyEquals(
+          srcTable, dstTable, () -> TestReplicationEndpoint.replicatedEntries.get() >= 25);
+    } finally {
+      if (hbaseTestingUtil.getHBaseAdmin().tableExists(REMAPPED_SOURCE_TABLE)) {
+        hbaseTestingUtil.getHBaseAdmin().disableTable(REMAPPED_SOURCE_TABLE);
+        hbaseTestingUtil.getHBaseAdmin().deleteTable(REMAPPED_SOURCE_TABLE);
+      }
+      if (cbtConnection.getAdmin().tableExists(REMAPPED_TARGET_TABLE)) {
+        cbtConnection.getAdmin().disableTable(REMAPPED_TARGET_TABLE);
+        cbtConnection.getAdmin().deleteTable(REMAPPED_TARGET_TABLE);
+      }
+    }
+  }
 }