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

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

@@ -143,6 +143,85 @@ 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.
+
+The mapping can be provided in one of three mutually exclusive ways:
+
+* **Inline JSON**, through `google.bigtable.replication.table_name_mapping`. This is
+  convenient when creating the peer in HBase shell:
+
+  ```
+  add_peer 'peer_id',
+    ENDPOINT_CLASSNAME => 'com.google.cloud.bigtable.hbase2_x.replication.HbaseToCloudBigtableReplicationEndpoint',
+    CONFIG => { 'google.bigtable.replication.table_name_mapping' => '{"my_namespace:my_table": "my_namespace_my_table"}' }
+  ```
+
+* **Per-table prefixed properties**, through
+  `google.bigtable.replication.table_name_mapping.<hbase_table>`. This avoids
+  JSON quote escaping in `hbase-site.xml` or shell `add_peer`:
+
+  ```xml
+  <property>
+      <name>google.bigtable.replication.table_name_mapping.my_namespace:my_table</name>
+      <value>my_namespace_my_table</value>
+  </property>
+  <property>
+      <name>google.bigtable.replication.table_name_mapping.other_namespace:events</name>
+      <value>events</value>
+  </property>
+  ```
+
+* **From a file**, through `google.bigtable.replication.table_name_mapping_file`.
+  The file contains a JSON object identical to the schema mapping file accepted
+  by the [HBase schema translator](../../bigtable-hbase-1.x-parent/bigtable-hbase-1.x-tools)
+  (`google.bigtable.schema.mapping.filepath`), so one file can be used to create
+  the tables in Cloud Bigtable and to replicate to them:
+
+  ```json
+  {
+    "my_namespace:my_table": "my_namespace_my_table",
+    "other_namespace:events": "events"
+  }
+  ```
+
+  The location is resolved through the Hadoop `FileSystem` API. Use explicit URI
+  schemes:
+  - `file:///path/to/mapping.json` for a local file present across region servers.
+  - `hdfs://nameservice1/hbase/mapping.json` for a file stored on HDFS.
+
+  If the URI scheme is omitted (e.g. `/etc/hbase/conf/mapping.json`), the path is
+  first checked on the default filesystem (`fs.defaultFS`), and if not found, falls
+  back to the local filesystem on the region server.
+
+  ```xml
+  <property>
+      <name>google.bigtable.replication.table_name_mapping_file</name>
+      <value>hdfs://nameservice1/hbase/bigtable-table-name-mapping.json</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.
+
 ### 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,9 @@
 
 package com.google.cloud.bigtable.hbase1_x.replication;
 
+import static com.google.cloud.bigtable.hbase.replication.configuration.HBaseToCloudBigtableReplicationConfiguration.TABLE_NAME_MAPPING_FILE_KEY;
+import static com.google.cloud.bigtable.hbase.replication.configuration.HBaseToCloudBigtableReplicationConfiguration.TABLE_NAME_MAPPING_KEY;
+import static com.google.cloud.bigtable.hbase.replication.configuration.HBaseToCloudBigtableReplicationConfiguration.TABLE_NAME_MAPPING_PREFIX;
 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;
@@ -25,17 +28,23 @@
 import com.google.cloud.bigtable.hbase.replication.utils.TestUtils;
 import com.google.cloud.bigtable.test.helper.BigtableEmulatorRule;
 import java.io.IOException;
+import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.UUID;
 import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hbase.Cell;
 import org.apache.hadoop.hbase.CellUtil;
 import org.apache.hadoop.hbase.HBaseTestingUtility;
 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;
@@ -78,13 +87,29 @@ public boolean replicate(ReplicateContext replicateContext) {
     }
 
     // return a WALEntry filter which accepts all rows except FILTERED_ROW_ENTRY
+    // and routes "my_ns" namespace tables only to their corresponding remapped peer.
     @Override
     public WALEntryFilter getWALEntryfilter() {
       return new ChainWALEntryFilter(
           super.getWALEntryfilter(),
           new WALEntryFilter() {
             @Override
             public Entry filter(Entry entry) {
+              if (entry == null) {
+                return null;
+              }
+              TableName entryTable = entry.getKey().getTablename();
+              boolean isMyNs = "my_ns".equals(entryTable.getNamespaceAsString());
+              boolean isDefaultPeer = "cbt".equals(ctx.getPeerId());
+              if (isDefaultPeer && isMyNs) {
+                return null;
+              }
+              if (!isDefaultPeer) {
+                String expectedTable = ctx.getConfiguration().get("test.peer.table");
+                if (expectedTable == null || !expectedTable.equals(entryTable.getNameAsString())) {
+                  return null;
+                }
+              }
               ArrayList<Cell> cells = entry.getEdit().getCells();
               int size = cells.size();
               for (int i = size - 1; i >= 0; i--) {
@@ -500,4 +525,158 @@ 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);
+  }
+
+  private ReplicationPeerConfig createRemappedPeerConfig(TableName sourceHBaseTable) {
+    ReplicationPeerConfig peerConfig = new ReplicationPeerConfig();
+    peerConfig.setReplicationEndpointImpl(TestReplicationEndpoint.class.getTypeName());
+    peerConfig.setClusterKey(hbaseTestingUtil.getClusterKey());
+    peerConfig.getConfiguration().put("google.bigtable.instance.id", "test-instance");
+    peerConfig.getConfiguration().put("google.bigtable.project.id", "test-project");
+    peerConfig
+        .getConfiguration()
+        .put("google.bigtable.emulator.endpoint.host", "localhost:" + bigtableEmulator.getPort());
+    peerConfig.getConfiguration().put("test.peer.table", sourceHBaseTable.getNameAsString());
+    return peerConfig;
+  }
+
+  @Test
+  public void testTableNameMappingInlineJson() throws Exception {
+    TableName sourceHBaseTable = TableName.valueOf("my_ns", "inline_src_" + UUID.randomUUID());
+    TableName targetCbtTable =
+        TableName.valueOf("my_ns_inline_dst_" + UUID.randomUUID().toString().substring(0, 8));
+    createRemappedTables(sourceHBaseTable, targetCbtTable);
+
+    String peerId = "cbt_inline";
+    ReplicationPeerConfig peerConfig = createRemappedPeerConfig(sourceHBaseTable);
+    String inlineJson =
+        String.format(
+            "{\"%s\": \"%s\"}",
+            sourceHBaseTable.getNameAsString(), targetCbtTable.getNameAsString());
+    peerConfig.getConfiguration().put(TABLE_NAME_MAPPING_KEY, inlineJson);
+
+    replicationAdmin.addPeer(peerId, peerConfig);
+    try (Table srcTable = hbaseConnection.getTable(sourceHBaseTable);
+        Table dstTable = cbtConnection.getTable(targetCbtTable)) {
+      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 {
+      replicationAdmin.removePeer(peerId);
+    }
+  }
+
+  @Test
+  public void testTableNameMappingFromHdfsFile() throws Exception {
+    TableName sourceHBaseTable = TableName.valueOf("my_ns", "hdfs_src_" + UUID.randomUUID());
+    TableName targetCbtTable =
+        TableName.valueOf("my_ns_hdfs_dst_" + UUID.randomUUID().toString().substring(0, 8));
+    createRemappedTables(sourceHBaseTable, targetCbtTable);
+
+    FileSystem dfs = hbaseTestingUtil.getDFSCluster().getFileSystem();
+    Path mappingPath =
+        dfs.makeQualified(new Path("/hbase/replication/mapping-" + UUID.randomUUID() + ".json"));
+    String jsonContent =
+        String.format(
+            "{\"%s\": \"%s\"}",
+            sourceHBaseTable.getNameAsString(), targetCbtTable.getNameAsString());
+    try (FSDataOutputStream out = dfs.create(mappingPath, true)) {
+      out.write(jsonContent.getBytes(StandardCharsets.UTF_8));
+    }
+
+    String peerId = "cbt_hdfs";
+    ReplicationPeerConfig peerConfig = createRemappedPeerConfig(sourceHBaseTable);
+    peerConfig.getConfiguration().put(TABLE_NAME_MAPPING_FILE_KEY, mappingPath.toString());
+
+    replicationAdmin.addPeer(peerId, peerConfig);
+    try (Table srcTable = hbaseConnection.getTable(sourceHBaseTable);
+        Table dstTable = cbtConnection.getTable(targetCbtTable)) {
+      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(200 + i));
+        srcTable.put(put);
+      }
+
+      TestUtils.assertTableEventuallyEquals(
+          srcTable, dstTable, () -> TestReplicationEndpoint.replicatedEntries.get() >= 25);
+    } finally {
+      replicationAdmin.removePeer(peerId);
+    }
+  }
+
+  @Test
+  public void testTableNameMappingPrefixProperty() throws Exception {
+    TableName sourceHBaseTable = TableName.valueOf("my_ns", "prefix_src_" + UUID.randomUUID());
+    TableName targetCbtTable =
+        TableName.valueOf("my_ns_prefix_dst_" + UUID.randomUUID().toString().substring(0, 8));
+    createRemappedTables(sourceHBaseTable, targetCbtTable);
+
+    String peerId = "cbt_prefix";
+    ReplicationPeerConfig peerConfig = createRemappedPeerConfig(sourceHBaseTable);
+    peerConfig
+        .getConfiguration()
+        .put(
+            TABLE_NAME_MAPPING_PREFIX + sourceHBaseTable.getNameAsString(),
+            targetCbtTable.getNameAsString());
+
+    replicationAdmin.addPeer(peerId, peerConfig);
+    try (Table srcTable = hbaseConnection.getTable(sourceHBaseTable);
+        Table dstTable = cbtConnection.getTable(targetCbtTable)) {
+      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(300 + i));
+        srcTable.put(put);
+      }
+
+      TestUtils.assertTableEventuallyEquals(
+          srcTable, dstTable, () -> TestReplicationEndpoint.replicatedEntries.get() >= 25);
+    } finally {
+      replicationAdmin.removePeer(peerId);
+    }
+  }
 }