diff --git a/hbase-migration-tools/bigtable-hbase-replication/README.md b/hbase-migration-tools/bigtable-hbase-replication/README.md index 08936c1447..0f125aa04b 100644 --- a/hbase-migration-tools/bigtable-hbase-replication/README.md +++ b/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.`. This avoids + JSON quote escaping in `hbase-site.xml` or shell `add_peer`: + + ```xml + + google.bigtable.replication.table_name_mapping.my_namespace:my_table + my_namespace_my_table + + + google.bigtable.replication.table_name_mapping.other_namespace:events + events + + ``` + +* **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 + + google.bigtable.replication.table_name_mapping_file + hdfs://nameservice1/hbase/bigtable-table-name-mapping.json + + ``` + +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. diff --git a/hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-1.x-replication/src/test/java/com/google/cloud/bigtable/hbase1_x/replication/HbaseToCloudBigtableReplicationEndpointTest.java b/hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-1.x-replication/src/test/java/com/google/cloud/bigtable/hbase1_x/replication/HbaseToCloudBigtableReplicationEndpointTest.java index fad550fa40..9a3367c8d4 100644 --- a/hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-1.x-replication/src/test/java/com/google/cloud/bigtable/hbase1_x/replication/HbaseToCloudBigtableReplicationEndpointTest.java +++ b/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,6 +87,7 @@ 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( @@ -85,6 +95,21 @@ public WALEntryFilter 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 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); + } + } } diff --git a/hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-2.x-replication/src/test/java/com/google/cloud/bigtable/hbase2_x/replication/HbaseToCloudBigtableReplicationEndpointTest.java b/hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-2.x-replication/src/test/java/com/google/cloud/bigtable/hbase2_x/replication/HbaseToCloudBigtableReplicationEndpointTest.java index 3ec3bd728f..20514ceff2 100644 --- a/hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-2.x-replication/src/test/java/com/google/cloud/bigtable/hbase2_x/replication/HbaseToCloudBigtableReplicationEndpointTest.java +++ b/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,9 @@ package com.google.cloud.bigtable.hbase2_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.CF2; import static com.google.cloud.bigtable.hbase.replication.utils.TestUtils.FILTERED_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; @@ -69,6 +78,13 @@ public class HbaseToCloudBigtableReplicationEndpointTest { public static class TestReplicationEndpoint extends HbaseToCloudBigtableReplicationEndpoint { static AtomicInteger replicatedEntries = new AtomicInteger(); + private Context endpointContext; + + @Override + public void init(Context context) throws IOException { + super.init(context); + this.endpointContext = context; + } @Override public boolean replicate(ReplicateContext replicateContext) { @@ -78,6 +94,7 @@ 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( @@ -85,6 +102,22 @@ public WALEntryFilter 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 = + endpointContext == null || "cbt".equals(endpointContext.getPeerId()); + if (isDefaultPeer && isMyNs) { + return null; + } + if (!isDefaultPeer) { + String expectedTable = endpointContext.getConfiguration().get("test.peer.table"); + if (expectedTable == null || !expectedTable.equals(entryTable.getNameAsString())) { + return null; + } + } ArrayList cells = entry.getEdit().getCells(); int size = cells.size(); for (int i = size - 1; i >= 0; i--) { @@ -494,4 +527,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(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); + } + + 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(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(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(CF2, TestUtils.COL_QUALIFIER, 0, TestUtils.getValue(300 + i)); + srcTable.put(put); + } + + TestUtils.assertTableEventuallyEquals( + srcTable, dstTable, () -> TestReplicationEndpoint.replicatedEntries.get() >= 25); + } finally { + replicationAdmin.removePeer(peerId); + } + } } diff --git a/hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-replication-core/src/main/java/com/google/cloud/bigtable/hbase/replication/CloudBigtableReplicator.java b/hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-replication-core/src/main/java/com/google/cloud/bigtable/hbase/replication/CloudBigtableReplicator.java index f808e1d43c..71044ddb72 100644 --- a/hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-replication-core/src/main/java/com/google/cloud/bigtable/hbase/replication/CloudBigtableReplicator.java +++ b/hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-replication-core/src/main/java/com/google/cloud/bigtable/hbase/replication/CloudBigtableReplicator.java @@ -35,19 +35,21 @@ import static com.google.cloud.bigtable.hbase.replication.metrics.HBaseToCloudBigtableReplicationMetrics.BIDIRECTIONAL_REPL_INELIGIBLE_WAL_ENTRY_METRIC_KEY; import static com.google.cloud.bigtable.hbase.replication.metrics.HBaseToCloudBigtableReplicationMetrics.ONE_WAY_REPL_ELIGIBLE_MUTATIONS_METRIC_KEY; import static com.google.cloud.bigtable.hbase.replication.metrics.HBaseToCloudBigtableReplicationMetrics.ONE_WAY_REPL_ELIGIBLE_WAL_ENTRY_METRIC_KEY; -import static java.util.stream.Collectors.groupingBy; import com.google.cloud.bigtable.hbase.BigtableConfiguration; import com.google.cloud.bigtable.hbase.BigtableOptionsFactory; import com.google.cloud.bigtable.hbase.replication.adapters.BigtableWALEntry; import com.google.cloud.bigtable.hbase.replication.adapters.IncompatibleMutationAdapter; import com.google.cloud.bigtable.hbase.replication.adapters.IncompatibleMutationAdapterFactory; +import com.google.cloud.bigtable.hbase.replication.configuration.TableNameMapper; import com.google.cloud.bigtable.hbase.replication.metrics.MetricsExporter; import com.google.common.util.concurrent.ThreadFactoryBuilder; +import java.io.IOException; import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.Arrays; import java.util.HashMap; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.UUID; @@ -56,10 +58,12 @@ import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.Cell; import org.apache.hadoop.hbase.CellUtil; import org.apache.hadoop.hbase.KeyValue; +import org.apache.hadoop.hbase.TableName; import org.apache.hadoop.hbase.client.Connection; import org.apache.hadoop.hbase.util.ByteRange; import org.apache.hadoop.hbase.util.Bytes; @@ -84,6 +88,7 @@ static class SharedResources { private static SharedResources INSTANCE; private final ExecutorService executorService; + /** * Bigtable connection owned by this class and shared by all the instances of {@link * CloudBigtableReplicator} class. DO NOT CLOSE THIS CONNECTION from any instance. The state of @@ -108,7 +113,6 @@ static class SharedResources { static synchronized SharedResources getInstance(Configuration conf) { - numReferences++; if (INSTANCE == null) { /* Create a bounded cached thread pool with unbounded queue. At any point, we will only have 1 * replicate() method active per replicationEndpoint object, so we will never have too many @@ -125,7 +129,15 @@ static synchronized SharedResources getInstance(Configuration conf) { .build()); // Create a connection to Cloud Bigtable. + // Note: HBase 2.x ReplicationPeerImpl wraps peer + cluster configuration in a + // CompoundConfiguration whose internal `properties` field is null, so `new + // Configuration(conf)` + // drops all compound entries. Copy via the Configuration iterator to preserve both + // hbase-site.xml and per-peer ReplicationPeerConfig properties. Configuration configurationCopy = new Configuration(conf); + for (Map.Entry entry : conf) { + configurationCopy.set(entry.getKey(), entry.getValue()); + } String projectId = configurationCopy.get(PROJECT_KEY); String instanceId = configurationCopy.get(INSTANCE_KEY); @@ -138,12 +150,18 @@ static synchronized SharedResources getInstance(Configuration conf) { configurationCopy.set(BigtableOptionsFactory.CUSTOM_USER_AGENT_KEY, "HBaseReplication"); } - // If an App profile is provided, it will be picked automatically by the connection. - Connection connection = BigtableConfiguration.connect(configurationCopy); - LOG.info("Created a connection to CBT. projects/{}/instances/{}", projectId, instanceId); - - INSTANCE = new SharedResources(connection, executorService); + try { + // If an App profile is provided, it will be picked automatically by the connection. + Connection connection = BigtableConfiguration.connect(configurationCopy); + LOG.info("Created a connection to CBT. projects/{}/instances/{}", projectId, instanceId); + + INSTANCE = new SharedResources(connection, executorService); + } catch (RuntimeException e) { + executorService.shutdownNow(); + throw e; + } } + numReferences++; return INSTANCE; } @@ -191,6 +209,9 @@ private static synchronized void decrementReferenceCount() { private byte[] sourceCbtQualifier; private MetricsExporter metricsExporter; + /** Maps HBase table names to Cloud Bigtable table IDs. Identity unless configured. */ + private TableNameMapper tableNameMapper = TableNameMapper.identity(); + /** * Shared resources that are not tied to an object of this class. Lifecycle of resources in this * object is usually tied to the lifecycle of JVM. @@ -227,15 +248,28 @@ synchronized void start( // If enabled, gets source qualifiers from configs. if (this.bidirectionalReplicationEnabled) { this.sourceHbaseQualifier = - configuration.get(CBT_REPL_HBASE_QUALIFIER_KEY, CBT_REPL_HBASE_QUALIFIER).getBytes(); + configuration + .get(CBT_REPL_HBASE_QUALIFIER_KEY, CBT_REPL_HBASE_QUALIFIER) + .getBytes(StandardCharsets.UTF_8); this.sourceCbtQualifier = - configuration.get(CBT_REPL_CBT_QUALIFIER_KEY, CBT_REPL_CBT_QUALIFIER).getBytes(); + configuration + .get(CBT_REPL_CBT_QUALIFIER_KEY, CBT_REPL_CBT_QUALIFIER) + .getBytes(StandardCharsets.UTF_8); LOG.info( "Bidirectional direction enabled. \nCbt qualifier: {}\nHbase qualifier: {}", Bytes.toStringBinary(sourceCbtQualifier), Bytes.toStringBinary(sourceHbaseQualifier)); } + + // Load the optional HBase -> Cloud Bigtable table name mapping. A broken mapping must stop + // the endpoint from starting: replicating to the wrong table is worse than not replicating. + try { + this.tableNameMapper = TableNameMapper.fromConfiguration(configuration); + } catch (IOException e) { + throw new IllegalStateException( + "Failed to load the HBase to Cloud Bigtable table name mapping.", e); + } } if (isDryRun) { @@ -280,21 +314,78 @@ public UUID getPeerUUID() { * *

Does not throw any exceptions to the caller. In case of any errors, should return false. * - * @param walEntriesByTable Map of WALEntries keyed by table name. + * @param walEntriesByTable Map of WALEntries keyed by HBase table name. */ public boolean replicate(Map> walEntriesByTable) { long startTime = System.currentTimeMillis(); boolean succeeded = true; - List> futures = new ArrayList<>(); - - for (Map.Entry> walEntriesForTable : - walEntriesByTable.entrySet()) { - futures.addAll(replicateTable(walEntriesForTable.getKey(), walEntriesForTable.getValue())); + if (walEntriesByTable == null || walEntriesByTable.isEmpty()) { + return true; } - // Check on the result for all the batches. try { + // Coalesce WAL entries by mapped Cloud Bigtable table name. If multiple HBase source table + // names map to the same target Bigtable table, their WAL entries must be merged into a single + // list and replicated in one call to avoid concurrent out-of-order writes for the same rows. + Map> coalescedWalEntries = new LinkedHashMap<>(); + Map targetToFirstCanonicalSource = new HashMap<>(); + for (Map.Entry> walEntriesForTable : + walEntriesByTable.entrySet()) { + String hbaseTableName = walEntriesForTable.getKey(); + if (hbaseTableName == null || hbaseTableName.trim().isEmpty()) { + LOG.warn("Encountered null or blank HBase table name in WAL entries map. Skipping."); + continue; + } + List entries = walEntriesForTable.getValue(); + if (entries == null || entries.isEmpty()) { + continue; + } + String bigtableTableName = tableNameMapper.map(hbaseTableName); + if (bigtableTableName == null || bigtableTableName.trim().isEmpty()) { + LOG.warn( + "TableNameMapper mapped HBase table '{}' to null or blank. Skipping.", + hbaseTableName); + continue; + } + + String canonicalHBaseTable; + try { + canonicalHBaseTable = TableName.valueOf(hbaseTableName.trim()).getNameAsString(); + } catch (IllegalArgumentException e) { + canonicalHBaseTable = hbaseTableName.trim(); + } + + String existingSource = + targetToFirstCanonicalSource.putIfAbsent(bigtableTableName, canonicalHBaseTable); + if (existingSource != null && !existingSource.equals(canonicalHBaseTable)) { + throw new IllegalStateException( + "Multiple distinct HBase source tables ('" + + existingSource + + "' and '" + + canonicalHBaseTable + + "') map to the same Cloud Bigtable table '" + + bigtableTableName + + "'."); + } + + if (!bigtableTableName.equals(hbaseTableName) && LOG.isTraceEnabled()) { + LOG.trace( + "Replicating HBase table {} to Cloud Bigtable table {}", + hbaseTableName, + bigtableTableName); + } + coalescedWalEntries + .computeIfAbsent(bigtableTableName, k -> new ArrayList<>()) + .addAll(entries); + } + + List> futures = new ArrayList<>(); + for (Map.Entry> entry : coalescedWalEntries.entrySet()) { + futures.addAll(replicateTable(entry.getKey(), entry.getValue())); + } + + // Check on the result for all the batches. for (Future future : futures) { // replicate method should succeed only when all the entries are successfully replicated. // TODO Add to readme about setting some timeouts on CBT writes. MutateRow should not @@ -319,7 +410,6 @@ private List> replicateTable( String tableName, List walEntries) { List> futures = new ArrayList<>(); List cellsToReplicateForTable = new ArrayList<>(); - int batchSizeInBytes = 0; for (BigtableWALEntry walEntry : walEntries) { // To prevent loops in bidirectional replication, @@ -349,8 +439,9 @@ private List> replicateTable( Map> cellsToReplicateByRow = cellsToReplicateForTable.stream() .collect( - groupingBy( - k -> new SimpleByteRange(k.getRowArray(), k.getRowOffset(), k.getRowLength()))); + Collectors.groupingBy( + k -> new SimpleByteRange(k.getRowArray(), k.getRowOffset(), k.getRowLength()), + Collectors.toCollection(ArrayList::new))); // Now we have cells to replicate by rows, this list can be big and processing it on a single // thread is not efficient. As this thread will have to do proto translation and may need to @@ -358,6 +449,7 @@ private List> replicateTable( // threads) and send them on separate threads. Map> batchToReplicate = new HashMap<>(); int numCellsInBatch = 0; + int batchSizeInBytes = 0; for (Map.Entry> rowCells : cellsToReplicateByRow.entrySet()) { // If bidirectional replication is enabled, tag row cells with source info @@ -366,14 +458,38 @@ private List> replicateTable( appendSourceTagToCells(rowCells); } + int rowCellCount = rowCells.getValue().size(); + int rowSizeInBytes = getRowSize(rowCells); + + // If adding this row would exceed the batch size threshold or the maximum cell limit, + // flush the current batch first. + if (!batchToReplicate.isEmpty() + && (batchSizeInBytes + rowSizeInBytes > batchSizeThresholdInBytes + || numCellsInBatch + rowCellCount >= 100_000)) { + LOG.trace( + "Replicating a batch of " + + batchToReplicate.size() + + " rows and " + + numCellsInBatch + + " cells with heap size " + + batchSizeInBytes + + " for table: " + + tableName); + + futures.add(replicateBatch(tableName, batchToReplicate)); + batchToReplicate = new HashMap<>(); + numCellsInBatch = 0; + batchSizeInBytes = 0; + } + // TODO handle the case where a single row has >100K mutations (very rare, but should not // fail) - numCellsInBatch += rowCells.getValue().size(); - batchSizeInBytes += getRowSize(rowCells); + numCellsInBatch += rowCellCount; + batchSizeInBytes += rowSizeInBytes; batchToReplicate.put(rowCells.getKey(), rowCells.getValue()); - // TODO add tests for batch split on size and cell counts - if (batchSizeInBytes >= batchSizeThresholdInBytes || numCellsInBatch >= 100_000 - 1) { + // If the batch meets or exceeds the threshold, flush it immediately. + if (batchSizeInBytes >= batchSizeThresholdInBytes || numCellsInBatch >= 100_000) { LOG.trace( "Replicating a batch of " + batchToReplicate.size() @@ -423,13 +539,17 @@ private Future replicateBatch( /** * Checks if WAL entry is eligible for replication. If the following are both true, then entry is - * NOT eligible for replication: 1. Bidirectional replication is enabled, AND 2. Last mutation in - * WAL entry is a special mutation matching SOURCE_CBT qualifier + * NOT eligible for replication: 1. Bidirectional replication is enabled, AND 2. A mutation in the + * WAL entry is a special delete marker matching SOURCE_CBT qualifier and timestamp 0. * * @param walEntry from HBase replication WAL log. - * @return true if walEntry is from Bigtable. + * @return true if walEntry is eligible for replication to Bigtable. */ private boolean isEligibleForReplication(BigtableWALEntry walEntry) { + if (walEntry == null || walEntry.getCells() == null || walEntry.getCells().isEmpty()) { + return false; + } + // Always return eligible if bidirectional replication was not enabled int mutationCount = walEntry.getCells().size(); if (!bidirectionalReplicationEnabled) { @@ -438,27 +558,45 @@ private boolean isEligibleForReplication(BigtableWALEntry walEntry) { return true; } - Cell lastCell = walEntry.getCells().get(walEntry.getCells().size() - 1); + Cell markerCell = null; + if (sourceCbtQualifier != null) { + for (Cell cell : walEntry.getCells()) { + if (cell != null + && cell.getTimestamp() == 0L + && CellUtil.isDelete(cell) + && CellUtil.matchingQualifier(cell, sourceCbtQualifier)) { + markerCell = cell; + break; + } + } + } - if (Arrays.equals(CellUtil.cloneQualifier(lastCell), sourceCbtQualifier)) { + if (markerCell != null) { // Wrap in statement so row is not needlessly converted. if (LOG.isTraceEnabled()) { LOG.trace( "Dropping WAL entry as it came from CBT , Row key: " + Bytes.toStringBinary( - lastCell.getRowArray(), lastCell.getRowOffset(), lastCell.getRowLength())); + markerCell.getRowArray(), + markerCell.getRowOffset(), + markerCell.getRowLength())); } incrementMetric(BIDIRECTIONAL_REPL_INELIGIBLE_WAL_ENTRY_METRIC_KEY, 1); - // We decrement one because the last mutation would've been the special mutation from CBT. + // We decrement one because the special marker mutation from CBT is not a user mutation. incrementMetric(BIDIRECTIONAL_REPL_INELIGIBLE_MUTATIONS_METRIC_KEY, mutationCount - 1); return false; } else { // Wrap in statement so row is not needlessly converted. if (LOG.isTraceEnabled()) { - LOG.trace( - "Replicating WAL entry as it originated on HBase , Row key: " - + Bytes.toStringBinary( - lastCell.getRowArray(), lastCell.getRowOffset(), lastCell.getRowLength())); + for (Cell cell : walEntry.getCells()) { + if (cell != null) { + LOG.trace( + "Replicating WAL entry as it originated on HBase , Row key: " + + Bytes.toStringBinary( + cell.getRowArray(), cell.getRowOffset(), cell.getRowLength())); + break; + } + } } incrementMetric(BIDIRECTIONAL_REPL_ELIGIBLE_WAL_ENTRY_METRIC_KEY, 1); incrementMetric(BIDIRECTIONAL_REPL_ELIGIBLE_MUTATIONS_METRIC_KEY, mutationCount); diff --git a/hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-replication-core/src/main/java/com/google/cloud/bigtable/hbase/replication/configuration/HBaseToCloudBigtableReplicationConfiguration.java b/hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-replication-core/src/main/java/com/google/cloud/bigtable/hbase/replication/configuration/HBaseToCloudBigtableReplicationConfiguration.java index d44a4cef31..d6bc57d707 100644 --- a/hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-replication-core/src/main/java/com/google/cloud/bigtable/hbase/replication/configuration/HBaseToCloudBigtableReplicationConfiguration.java +++ b/hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-replication-core/src/main/java/com/google/cloud/bigtable/hbase/replication/configuration/HBaseToCloudBigtableReplicationConfiguration.java @@ -67,6 +67,7 @@ private HBaseToCloudBigtableReplicationConfiguration() {} * Bigtable at the expense of more RPCs. */ public static final String BATCH_SIZE_KEY = "google.bigtable.replication.batch_size_bytes"; + // TODO: Tune this parameter. Usually, this should be smaller than the HBase replication source // batch capacity by counts and bytes. These capacity are set by `replication.source.nb.capacity` // and `replication.source.size.capacity` config keys. @@ -90,4 +91,51 @@ private HBaseToCloudBigtableReplicationConfiguration() {} public static final boolean CBT_REPL_BIDIRECTIONAL_REPLICATION_MODE = false; public static final String CBT_REPL_HBASE_QUALIFIER = "SOURCE_HBASE"; public static final String CBT_REPL_CBT_QUALIFIER = "SOURCE_CBT"; + + /** + * Inline JSON map of HBase table name to Cloud Bigtable table ID, for example {@code + * {"my_namespace:my_table": "my_namespace_my_table"}}. WAL entries for an HBase table present in + * the map are written to the mapped Cloud Bigtable table; tables that are not present in the map + * are replicated to a Cloud Bigtable table with the same name as the HBase table (the default + * behavior). + * + *

The JSON format is identical to the schema mapping file accepted by the HBase schema + * translator ({@code google.bigtable.schema.mapping.filepath}), so the same file can be used to + * create the tables in Cloud Bigtable and to replicate to them. + * + *

This is mainly useful for HBase tables in a non-default namespace, since {@code :} is not a + * valid character in a Cloud Bigtable table ID. + * + *

Mutually exclusive with {@link #TABLE_NAME_MAPPING_FILE_KEY} and {@link + * #TABLE_NAME_MAPPING_PREFIX}. + */ + public static final String TABLE_NAME_MAPPING_KEY = + "google.bigtable.replication.table_name_mapping"; + + /** + * Location of a JSON file containing the table name mapping described in {@link + * #TABLE_NAME_MAPPING_KEY}. The location is resolved through the Hadoop {@code FileSystem} API, + * so it can be a local path ({@code file:///etc/hbase/conf/table-name-mapping.json}), an HDFS + * path ({@code hdfs://nameservice1/hbase/table-name-mapping.json}) or any other file system + * available on the region servers. The file must be readable by every region server. + * + *

Mutually exclusive with {@link #TABLE_NAME_MAPPING_KEY} and {@link + * #TABLE_NAME_MAPPING_PREFIX}. + */ + public static final String TABLE_NAME_MAPPING_FILE_KEY = + "google.bigtable.replication.table_name_mapping_file"; + + /** + * Configuration property prefix for specifying individual per-table mappings directly in {@code + * hbase-site.xml} or via HBase shell {@code add_peer} CONFIG without JSON formatting: {@code + * google.bigtable.replication.table_name_mapping.=}. + * + *

For example: {@code + * google.bigtable.replication.table_name_mapping.my_namespace:my_table=my_namespace_my_table}. + * + *

Mutually exclusive with {@link #TABLE_NAME_MAPPING_KEY} and {@link + * #TABLE_NAME_MAPPING_FILE_KEY}. + */ + public static final String TABLE_NAME_MAPPING_PREFIX = + "google.bigtable.replication.table_name_mapping."; } diff --git a/hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-replication-core/src/main/java/com/google/cloud/bigtable/hbase/replication/configuration/TableNameMapper.java b/hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-replication-core/src/main/java/com/google/cloud/bigtable/hbase/replication/configuration/TableNameMapper.java new file mode 100644 index 0000000000..4ac78c9577 --- /dev/null +++ b/hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-replication-core/src/main/java/com/google/cloud/bigtable/hbase/replication/configuration/TableNameMapper.java @@ -0,0 +1,474 @@ +/* + * Copyright 2026 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.replication.configuration; + +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 com.google.bigtable.repackaged.com.google.gson.stream.JsonReader; +import com.google.bigtable.repackaged.com.google.gson.stream.JsonToken; +import java.io.ByteArrayOutputStream; +import java.io.File; +import java.io.FileNotFoundException; +import java.io.IOException; +import java.io.StringReader; +import java.nio.charset.StandardCharsets; +import java.util.Collections; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.regex.Pattern; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FSDataInputStream; +import org.apache.hadoop.fs.FileStatus; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hbase.TableName; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Maps HBase table names to Cloud Bigtable table IDs during replication. + * + *

The mapping defines canonical HBase table names to Cloud Bigtable table IDs and can be loaded + * from one of three mutually exclusive mechanisms: + * + *

    + *
  • {@link HBaseToCloudBigtableReplicationConfiguration#TABLE_NAME_MAPPING_KEY}: inline JSON + * string (e.g. {@code {"ns:table": "mapped_table"}}). + *
  • {@link HBaseToCloudBigtableReplicationConfiguration#TABLE_NAME_MAPPING_FILE_KEY}: path to a + * JSON file resolvable via Hadoop {@code FileSystem} (e.g. {@code file:///path/to/file.json} + * or {@code hdfs://...}). + *
  • {@link HBaseToCloudBigtableReplicationConfiguration#TABLE_NAME_MAPPING_PREFIX}: individual + * properties in {@code Configuration} prefixed with {@code + * google.bigtable.replication.table_name_mapping.=}. + *
+ * + *

Source HBase table names are normalized to canonical form (for example, {@code default:table} + * resolves to {@code table}). Mappings must be injective (1:1): multiple HBase tables cannot map to + * the same Cloud Bigtable table ID. + * + *

Tables not present in the mapping keep their canonical HBase name, preserving default + * behavior. + * + *

Instances are immutable and thread safe. + */ +public final class TableNameMapper { + + private static final Logger LOG = LoggerFactory.getLogger(TableNameMapper.class); + + /** + * Cloud Bigtable table ID constraints. See + * https://cloud.google.com/bigtable/docs/reference/admin/rpc/google.bigtable.admin.v2#createtablerequest + */ + private static final Pattern BIGTABLE_TABLE_ID_PATTERN = + Pattern.compile("[_a-zA-Z0-9][-_.a-zA-Z0-9]*"); + + private static final int MAX_BIGTABLE_TABLE_ID_LENGTH = 50; + + /** + * Upper bound on the size of the mapping document. A mapping configuration is a small list of + * table names; anything larger is almost certainly a misconfiguration and must not be buffered in + * the region server heap. + */ + static final long MAX_MAPPING_FILE_SIZE_BYTES = 1024 * 1024; + + private static final TableNameMapper IDENTITY = + new TableNameMapper( + Collections.emptyMap(), Collections.emptyMap()); + + private final Map mappings; + private final Map targetToSource; + + private TableNameMapper(Map mappings, Map targetToSource) { + this.mappings = Collections.unmodifiableMap(new HashMap<>(mappings)); + this.targetToSource = Collections.unmodifiableMap(new HashMap<>(targetToSource)); + } + + /** Returns a mapper that leaves every table name unchanged. */ + public static TableNameMapper identity() { + return IDENTITY; + } + + /** + * Builds a mapper from the replication configuration. + * + * @throws IllegalArgumentException if more than one mapping mechanism is configured or if the + * mapping is malformed. + * @throws IOException if the mapping file cannot be read. + */ + public static TableNameMapper fromConfiguration(Configuration configuration) throws IOException { + if (configuration == null) { + return IDENTITY; + } + + String inlineJson = configuration.getTrimmed(TABLE_NAME_MAPPING_KEY); + String mappingFile = configuration.getTrimmed(TABLE_NAME_MAPPING_FILE_KEY); + + Map prefixMappings = new LinkedHashMap<>(); + for (Map.Entry entry : configuration) { + String key = entry.getKey(); + if (key != null && key.startsWith(TABLE_NAME_MAPPING_PREFIX)) { + String hbaseTable = key.substring(TABLE_NAME_MAPPING_PREFIX.length()).trim(); + if (hbaseTable.isEmpty()) { + throw new IllegalArgumentException( + "Configuration property '" + + key + + "' has an empty HBase table name after prefix '" + + TABLE_NAME_MAPPING_PREFIX + + "'."); + } + prefixMappings.put(hbaseTable, configuration.getTrimmed(key)); + } + } + + boolean hasInline = inlineJson != null && !inlineJson.isEmpty(); + boolean hasFile = mappingFile != null && !mappingFile.isEmpty(); + boolean hasPrefix = !prefixMappings.isEmpty(); + + int sourceCount = (hasInline ? 1 : 0) + (hasFile ? 1 : 0) + (hasPrefix ? 1 : 0); + if (sourceCount > 1) { + throw new IllegalArgumentException( + "Only one of inline JSON (" + + TABLE_NAME_MAPPING_KEY + + "), mapping file (" + + TABLE_NAME_MAPPING_FILE_KEY + + "), or prefixed properties (" + + TABLE_NAME_MAPPING_PREFIX + + ") can be set."); + } + + if (hasInline) { + return fromJson(inlineJson, TABLE_NAME_MAPPING_KEY); + } + if (hasFile) { + return fromJson(readMappingFile(configuration, mappingFile), mappingFile); + } + if (hasPrefix) { + return validateAndBuild( + prefixMappings, "prefixed properties (" + TABLE_NAME_MAPPING_PREFIX + ")"); + } + return IDENTITY; + } + + /** + * Builds a mapper from a JSON object of {@code "hbaseTableName": "bigtableTableId"} pairs. + * + * @param json the JSON document. + * @param source human readable description of where the JSON came from, used in error messages. + * @throws IllegalArgumentException if the JSON is malformed or contains an invalid table name. + */ + public static TableNameMapper fromJson(String json, String source) { + Map rawMappings = parseJsonMappings(json, source); + return validateAndBuild(rawMappings, source); + } + + @SuppressWarnings("deprecation") + private static Map parseJsonMappings(String json, String source) { + if (json == null || json.trim().isEmpty()) { + throw new IllegalArgumentException("Table name mapping from " + source + " is empty."); + } + if (json.length() > MAX_MAPPING_FILE_SIZE_BYTES + || json.getBytes(StandardCharsets.UTF_8).length > MAX_MAPPING_FILE_SIZE_BYTES) { + throw new IllegalArgumentException( + "Table name mapping from " + + source + + " exceeds maximum allowed size of " + + MAX_MAPPING_FILE_SIZE_BYTES + + " bytes."); + } + + Map rawMappings = new LinkedHashMap<>(); + try (JsonReader reader = new JsonReader(new StringReader(json))) { + reader.setLenient(false); + JsonToken firstToken = reader.peek(); + if (firstToken != JsonToken.BEGIN_OBJECT) { + throw new IllegalArgumentException( + "Table name mapping from " + + source + + " is not a valid JSON object: expected BEGIN_OBJECT, found " + + firstToken); + } + reader.beginObject(); + while (reader.hasNext()) { + String hbaseTableName = reader.nextName(); + JsonToken valueToken = reader.peek(); + if (valueToken != JsonToken.STRING) { + throw new IllegalArgumentException( + "Table name mapping from " + + source + + " has invalid non-string value for HBase table '" + + hbaseTableName + + "': expected STRING, found " + + valueToken); + } + String bigtableTableId = reader.nextString(); + if (rawMappings.put(hbaseTableName, bigtableTableId) != null) { + throw new IllegalArgumentException( + "Table name mapping from " + + source + + " contains duplicate key: '" + + hbaseTableName + + "'."); + } + } + reader.endObject(); + if (reader.peek() != JsonToken.END_DOCUMENT) { + throw new IllegalArgumentException( + "Table name mapping from " + source + " has trailing characters after JSON object."); + } + } catch (IllegalArgumentException e) { + throw e; + } catch (Exception e) { + throw new IllegalArgumentException( + "Table name mapping from " + source + " is not a valid JSON object.", e); + } + return rawMappings; + } + + private static TableNameMapper validateAndBuild(Map rawMappings, String source) { + if (rawMappings == null || rawMappings.isEmpty()) { + return IDENTITY; + } + + Map validated = new HashMap<>(rawMappings.size()); + Map targetToSource = new HashMap<>(rawMappings.size()); + + for (Map.Entry entry : rawMappings.entrySet()) { + String rawHBaseTable = entry.getKey(); + String bigtableTableId = entry.getValue(); + + String canonicalHBaseTable = canonicalizeHBaseTableName(rawHBaseTable, source); + + if (validated.containsKey(canonicalHBaseTable)) { + throw new IllegalArgumentException( + "Table name mapping from " + + source + + " contains duplicate or conflicting mapping for HBase table '" + + canonicalHBaseTable + + "' (source key: '" + + rawHBaseTable + + "')."); + } + + validateBigtableTableId(canonicalHBaseTable, bigtableTableId, source); + + if (targetToSource.containsKey(bigtableTableId)) { + String existingSource = targetToSource.get(bigtableTableId); + throw new IllegalArgumentException( + "Table name mapping from " + + source + + " maps multiple HBase tables to the same Cloud Bigtable table '" + + bigtableTableId + + "': '" + + existingSource + + "' and '" + + canonicalHBaseTable + + "'. N:1 table mappings are not supported."); + } + + validated.put(canonicalHBaseTable, bigtableTableId); + targetToSource.put(bigtableTableId, canonicalHBaseTable); + } + + LOG.info("Loaded {} table name mapping(s) from {}: {}", validated.size(), source, validated); + return new TableNameMapper(validated, targetToSource); + } + + private static String canonicalizeHBaseTableName(String rawHBaseTable, String source) { + if (rawHBaseTable == null) { + throw new IllegalArgumentException( + "Table name mapping from " + source + " contains an empty HBase table name."); + } + String trimmed = rawHBaseTable.trim(); + if (trimmed.isEmpty()) { + throw new IllegalArgumentException( + "Table name mapping from " + source + " contains an empty HBase table name."); + } + try { + return TableName.valueOf(trimmed).getNameAsString(); + } catch (IllegalArgumentException e) { + throw new IllegalArgumentException( + "Table name mapping from " + + source + + " contains an invalid HBase table name: '" + + rawHBaseTable + + "'", + e); + } + } + + private static void validateBigtableTableId( + String hbaseTableName, String bigtableTableId, String source) { + if (bigtableTableId == null + || bigtableTableId.isEmpty() + || bigtableTableId.length() > MAX_BIGTABLE_TABLE_ID_LENGTH + || !BIGTABLE_TABLE_ID_PATTERN.matcher(bigtableTableId).matches()) { + throw new IllegalArgumentException( + "Table name mapping from " + + source + + " maps HBase table '" + + hbaseTableName + + "' to '" + + bigtableTableId + + "', which is not a valid Cloud Bigtable table ID. Table IDs must match " + + BIGTABLE_TABLE_ID_PATTERN.pattern() + + " and be at most " + + MAX_BIGTABLE_TABLE_ID_LENGTH + + " characters."); + } + } + + private static String readMappingFile(Configuration configuration, String location) + throws IOException { + Configuration fsConf = new Configuration(configuration); + for (Map.Entry entry : configuration) { + fsConf.set(entry.getKey(), entry.getValue()); + } + Path path = new Path(location); + FileSystem fs = path.getFileSystem(fsConf); + + boolean existsOnDefaultFs = false; + IOException defaultFsException = null; + try { + existsOnDefaultFs = fs.exists(path); + } catch (IOException e) { + defaultFsException = e; + LOG.debug("Failed to check existence on default filesystem {}", fs.getUri(), e); + } + + if (!existsOnDefaultFs) { + if (path.toUri().getScheme() == null) { + FileSystem localFs = FileSystem.getLocal(fsConf); + Path localPath = new Path(new File(location).getAbsolutePath()); + if (localFs.exists(localPath)) { + fs = localFs; + path = localPath; + } else { + FileNotFoundException fnfe = + new FileNotFoundException( + "Table name mapping file does not exist: " + + location + + " (resolved against FileSystem URI: " + + fs.getUri() + + "; if referencing a local file on region servers, prefix with 'file://' or ensure the file exists; if referencing HDFS, prefix with 'hdfs://')."); + if (defaultFsException != null) { + fnfe.initCause(defaultFsException); + } + throw fnfe; + } + } else { + FileNotFoundException fnfe = + new FileNotFoundException( + "Table name mapping file does not exist: " + + location + + " (resolved against FileSystem URI: " + + fs.getUri() + + "; prefix with 'file://' for local files or 'hdfs://' for HDFS)."); + if (defaultFsException != null) { + fnfe.initCause(defaultFsException); + } + throw fnfe; + } + } + + FileStatus status = fs.getFileStatus(path); + if (status.isDirectory()) { + throw new IOException("Table name mapping file " + location + " is a directory."); + } + if (status.getLen() > MAX_MAPPING_FILE_SIZE_BYTES) { + throw new IOException( + "Table name mapping file " + + location + + " is " + + status.getLen() + + " bytes; the maximum supported size is " + + MAX_MAPPING_FILE_SIZE_BYTES + + " bytes."); + } + + ByteArrayOutputStream out = new ByteArrayOutputStream(); + byte[] buffer = new byte[4096]; + long totalRead = 0; + try (FSDataInputStream in = fs.open(path)) { + int bytesRead; + while ((bytesRead = in.read(buffer)) != -1) { + totalRead += bytesRead; + if (totalRead > MAX_MAPPING_FILE_SIZE_BYTES) { + throw new IOException( + "Table name mapping file " + + location + + " exceeds maximum allowed size of " + + MAX_MAPPING_FILE_SIZE_BYTES + + " bytes."); + } + out.write(buffer, 0, bytesRead); + } + } + return new String(out.toByteArray(), StandardCharsets.UTF_8); + } + + /** + * Returns the Cloud Bigtable table ID that WAL entries for {@code hbaseTableName} should be + * written to. Returns {@code hbaseTableName} (or its canonical form) unchanged if it is not + * present in the mapping. + */ + public String map(String hbaseTableName) { + if (hbaseTableName == null) { + return null; + } + String canonical; + try { + canonical = TableName.valueOf(hbaseTableName.trim()).getNameAsString(); + } catch (IllegalArgumentException e) { + canonical = hbaseTableName.trim(); + } + String mapped = mappings.get(canonical); + if (mapped != null) { + return mapped; + } + String conflictingSource = targetToSource.get(canonical); + if (conflictingSource != null) { + throw new IllegalArgumentException( + "Unmapped HBase table '" + + canonical + + "' collides with target Cloud Bigtable table '" + + canonical + + "' already mapped from HBase table '" + + conflictingSource + + "'. Map '" + + canonical + + "' explicitly to a distinct Cloud Bigtable table ID."); + } + return canonical; + } + + /** Returns true if this mapper does not rename any table. */ + public boolean isIdentity() { + return mappings.isEmpty(); + } + + /** Returns an unmodifiable view of the configured mappings. */ + public Map getMappings() { + return mappings; + } + + @Override + public String toString() { + return "TableNameMapper" + mappings; + } +} diff --git a/hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-replication-core/src/test/java/com/google/cloud/bigtable/hbase/replication/CloudBigtableReplicatorTest.java b/hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-replication-core/src/test/java/com/google/cloud/bigtable/hbase/replication/CloudBigtableReplicatorTest.java index 7c3a4ed821..7347751e07 100644 --- a/hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-replication-core/src/test/java/com/google/cloud/bigtable/hbase/replication/CloudBigtableReplicatorTest.java +++ b/hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-replication-core/src/test/java/com/google/cloud/bigtable/hbase/replication/CloudBigtableReplicatorTest.java @@ -20,6 +20,8 @@ import static com.google.cloud.bigtable.hbase.replication.configuration.HBaseToCloudBigtableReplicationConfiguration.CBT_REPL_CBT_QUALIFIER_KEY; import static com.google.cloud.bigtable.hbase.replication.configuration.HBaseToCloudBigtableReplicationConfiguration.CBT_REPL_HBASE_QUALIFIER; import static com.google.cloud.bigtable.hbase.replication.configuration.HBaseToCloudBigtableReplicationConfiguration.CBT_REPL_HBASE_QUALIFIER_KEY; +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.metrics.HBaseToCloudBigtableReplicationMetrics.BIDIRECTIONAL_REPL_ELIGIBLE_MUTATIONS_METRIC_KEY; import static com.google.cloud.bigtable.hbase.replication.metrics.HBaseToCloudBigtableReplicationMetrics.BIDIRECTIONAL_REPL_ELIGIBLE_WAL_ENTRY_METRIC_KEY; import static com.google.cloud.bigtable.hbase.replication.metrics.HBaseToCloudBigtableReplicationMetrics.BIDIRECTIONAL_REPL_INELIGIBLE_MUTATIONS_METRIC_KEY; @@ -41,6 +43,7 @@ import static org.apache.hadoop.hbase.KeyValue.Type.DeleteFamilyVersion; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.verifyNoInteractions; @@ -53,8 +56,11 @@ import com.google.cloud.bigtable.hbase.replication.adapters.IncompatibleMutationAdapter; import com.google.cloud.bigtable.hbase.replication.metrics.MetricsExporter; import java.io.IOException; +import java.nio.charset.StandardCharsets; import java.util.Arrays; +import java.util.Collections; import java.util.HashMap; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.concurrent.Callable; @@ -815,4 +821,926 @@ public void testBidirectionalReplicationMultipleMutations() throws IOException { verifyNoMoreInteractions(mockMetricExporter, mockExecutorService); verifyNoInteractions(mockConnection); } + + @Test + public void testReplicateAppliesTableNameMapping() throws IOException { + // Map the namespaced HBase table to a flat Cloud Bigtable table ID, leave the other one alone. + String namespacedTable = "my_namespace:" + TABLE_NAME_STRING; + String mappedTable = "my_namespace_" + TABLE_NAME_STRING; + Configuration mappingConf = new Configuration(false); + mappingConf.set( + TABLE_NAME_MAPPING_KEY, "{\"" + namespacedTable + "\": \"" + mappedTable + "\"}"); + + CloudBigtableReplicator replicator = new CloudBigtableReplicator(); + replicator.start(sharedResources, incompatibleMutationAdapter, 2000, false, mappingConf, null); + + Cell cell1 = new KeyValue(getRowKey(1), CF1, null, TIMESTAMP, getValue(1)); + BigtableWALEntry walEntry1 = + new BigtableWALEntry(TIMESTAMP, Arrays.asList(cell1), namespacedTable); + Cell cell2 = new KeyValue(getRowKey(2), CF1, null, TIMESTAMP, getValue(2)); + BigtableWALEntry walEntry2 = + new BigtableWALEntry(TIMESTAMP, Arrays.asList(cell2), TABLE_NAME_STRING_2); + + Map> walsToReplicate = new HashMap<>(); + walsToReplicate.put(namespacedTable, Arrays.asList(walEntry1)); + walsToReplicate.put(TABLE_NAME_STRING_2, Arrays.asList(walEntry2)); + + Map> expectedMappedBatch = new HashMap<>(); + expectedMappedBatch.put(new SimpleByteRange(getRowKey(1)), walEntry1.getCells()); + Map> expectedUnmappedBatch = new HashMap<>(); + expectedUnmappedBatch.put(new SimpleByteRange(getRowKey(2)), walEntry2.getCells()); + + when(mockExecutorService.submit((Callable) any())) + .thenReturn(CompletableFuture.completedFuture(true)); + + assertTrue(replicator.replicate(walsToReplicate)); + + // The namespaced table is written to the mapped Cloud Bigtable table... + verify(mockExecutorService) + .submit(new CloudBigtableReplicationTask(mappedTable, mockConnection, expectedMappedBatch)); + // ...and the unmapped table keeps its HBase name. + verify(mockExecutorService) + .submit( + new CloudBigtableReplicationTask( + TABLE_NAME_STRING_2, mockConnection, expectedUnmappedBatch)); + verifyNoMoreInteractions(mockExecutorService); + verifyNoInteractions(mockConnection); + } + + @Test + public void testStartFailsOnInvalidTableNameMapping() { + Configuration mappingConf = new Configuration(false); + // ':' is not a valid character in a Cloud Bigtable table ID. + mappingConf.set(TABLE_NAME_MAPPING_KEY, "{\"ns:table\": \"ns:table\"}"); + + CloudBigtableReplicator replicator = new CloudBigtableReplicator(); + expectThrows( + IllegalArgumentException.class, + () -> + replicator.start( + sharedResources, incompatibleMutationAdapter, 2000, false, mappingConf, null)); + } + + @Test + public void testReplicateCoalescesSourceTablesMappingToSameBigtableTable() throws IOException { + String mappedTable = "mapped_target"; + Configuration mappingConf = new Configuration(false); + mappingConf.set(TABLE_NAME_MAPPING_KEY, "{\"table1\": \"" + mappedTable + "\"}"); + + CloudBigtableReplicator replicator = new CloudBigtableReplicator(); + replicator.start(sharedResources, incompatibleMutationAdapter, 2000, false, mappingConf, null); + + Cell cell1 = new KeyValue(getRowKey(1), CF1, null, TIMESTAMP, getValue(1)); + BigtableWALEntry walEntry1 = new BigtableWALEntry(TIMESTAMP, Arrays.asList(cell1), "table1"); + Cell cell2 = new KeyValue(getRowKey(1), CF1, null, TIMESTAMP + 1, getValue(2)); + BigtableWALEntry walEntry2 = + new BigtableWALEntry(TIMESTAMP + 1, Arrays.asList(cell2), "default:table1"); + + Map> walsToReplicate = new LinkedHashMap<>(); + walsToReplicate.put("table1", Arrays.asList(walEntry1)); + walsToReplicate.put("default:table1", Arrays.asList(walEntry2)); + + // Both entries should be merged into a single batch and grouped under the same row key. + Map> expectedBatch = new HashMap<>(); + expectedBatch.put(new SimpleByteRange(getRowKey(1)), Arrays.asList(cell1, cell2)); + + when(mockExecutorService.submit((Callable) any())) + .thenReturn(CompletableFuture.completedFuture(true)); + + assertTrue(replicator.replicate(walsToReplicate)); + + // Verify only ONE task was submitted for mappedTable, containing both mutations. + verify(mockExecutorService) + .submit(new CloudBigtableReplicationTask(mappedTable, mockConnection, expectedBatch)); + verifyNoMoreInteractions(mockExecutorService); + verifyNoInteractions(mockConnection); + } + + @Test + public void testReplicateCoalescesUnmappedDefaultNamespaceAliases() throws IOException { + CloudBigtableReplicator replicator = new CloudBigtableReplicator(); + replicator.start(sharedResources, incompatibleMutationAdapter, 2000, false, conf, null); + + Cell cell1 = new KeyValue(getRowKey(1), CF1, null, TIMESTAMP, getValue(1)); + BigtableWALEntry walEntry1 = + new BigtableWALEntry(TIMESTAMP, Arrays.asList(cell1), TABLE_NAME_STRING); + Cell cell2 = new KeyValue(getRowKey(1), CF1, null, TIMESTAMP + 1, getValue(2)); + BigtableWALEntry walEntry2 = + new BigtableWALEntry(TIMESTAMP + 1, Arrays.asList(cell2), "default:" + TABLE_NAME_STRING); + + Map> walsToReplicate = new LinkedHashMap<>(); + walsToReplicate.put(TABLE_NAME_STRING, Arrays.asList(walEntry1)); + walsToReplicate.put("default:" + TABLE_NAME_STRING, Arrays.asList(walEntry2)); + + Map> expectedBatch = new HashMap<>(); + expectedBatch.put(new SimpleByteRange(getRowKey(1)), Arrays.asList(cell1, cell2)); + + when(mockExecutorService.submit((Callable) any())) + .thenReturn(CompletableFuture.completedFuture(true)); + + assertTrue(replicator.replicate(walsToReplicate)); + + verify(mockExecutorService) + .submit(new CloudBigtableReplicationTask(TABLE_NAME_STRING, mockConnection, expectedBatch)); + verifyNoMoreInteractions(mockExecutorService); + verifyNoInteractions(mockConnection); + } + + @Test + public void testStartFailsOnMissingTableNameMappingFile() { + Configuration mappingConf = new Configuration(false); + mappingConf.set(TABLE_NAME_MAPPING_FILE_KEY, "file:///definitely/not/a/real/mapping.json"); + + CloudBigtableReplicator replicator = new CloudBigtableReplicator(); + expectThrows( + IllegalStateException.class, + () -> + replicator.start( + sharedResources, incompatibleMutationAdapter, 2000, false, mappingConf, null)); + } + + @Test + public void testReplicateHandlesNullAndBlankHBaseTableNamesGracefully() throws IOException { + CloudBigtableReplicator replicator = new CloudBigtableReplicator(); + replicator.start(sharedResources, incompatibleMutationAdapter, 2000, false, conf, null); + + Cell cell = new KeyValue(getRowKey(1), CF1, null, TIMESTAMP, getValue(1)); + BigtableWALEntry walEntry = + new BigtableWALEntry(TIMESTAMP, Arrays.asList(cell), TABLE_NAME_STRING); + + Map> walsToReplicate = new LinkedHashMap<>(); + walsToReplicate.put(null, Arrays.asList(walEntry)); + walsToReplicate.put(" ", Arrays.asList(walEntry)); + walsToReplicate.put(TABLE_NAME_STRING, Arrays.asList(walEntry)); + + Map> expectedBatch = new HashMap<>(); + expectedBatch.put(new SimpleByteRange(getRowKey(1)), Arrays.asList(cell)); + + when(mockExecutorService.submit((Callable) any())) + .thenReturn(CompletableFuture.completedFuture(true)); + + // Must not throw NullPointerException or IllegalArgumentException + assertTrue(replicator.replicate(walsToReplicate)); + + // Only the valid table entry is replicated + verify(mockExecutorService) + .submit(new CloudBigtableReplicationTask(TABLE_NAME_STRING, mockConnection, expectedBatch)); + verifyNoMoreInteractions(mockExecutorService); + verifyNoInteractions(mockConnection); + } + + @Test + public void testReplicateCatchesUnexpectedExceptionAndReturnsFalse() { + CloudBigtableReplicator replicator = new CloudBigtableReplicator(); + replicator.start(sharedResources, incompatibleMutationAdapter, 2000, false, conf, null); + + Cell cell = new KeyValue(getRowKey(1), CF1, null, TIMESTAMP, getValue(1)); + BigtableWALEntry walEntry = + new BigtableWALEntry(TIMESTAMP, Arrays.asList(cell), TABLE_NAME_STRING); + + Map> walsToReplicate = new HashMap<>(); + walsToReplicate.put(TABLE_NAME_STRING, Arrays.asList(walEntry)); + + when(mockExecutorService.submit((Callable) any())) + .thenThrow(new RuntimeException("Simulated unexpected thread pool failure")); + + // Must not propagate RuntimeException to caller, must return false + assertFalse(replicator.replicate(walsToReplicate)); + } + + // --- Tests for Finding 1: Empty WALEntry / WALEdit poison pill fix --- + + @Test + public void testBidirectionalReplicationSkipsEmptyWalEntryWithoutFailing() { + CloudBigtableReplicator replicator = new CloudBigtableReplicator(); + Configuration replicationConf = new Configuration(false); + replicationConf.setBoolean(CBT_REPL_BIDIRECTIONAL_REPLICATION_MODE_KEY, true); + replicator.start( + sharedResources, + incompatibleMutationAdapter, + 2000, + false, + replicationConf, + mockMetricExporter); + + BigtableWALEntry emptyEntry = + new BigtableWALEntry(TIMESTAMP, Collections.emptyList(), TABLE_NAME_STRING); + + Map> walsToReplicate = new HashMap<>(); + walsToReplicate.put(TABLE_NAME_STRING, Arrays.asList(emptyEntry)); + + // Must return true without throwing IndexOutOfBoundsException + assertTrue(replicator.replicate(walsToReplicate)); + + // Metric counters called during constructor only + verify(mockMetricExporter).incCounters(INCOMPATIBLE_MUTATION_METRIC_KEY, 0); + verify(mockMetricExporter).incCounters(DROPPED_INCOMPATIBLE_MUTATION_METRIC_KEY, 0); + verify(mockMetricExporter).incCounters(INCOMPATIBLE_MUTATION_DELETES_METRICS_KEY, 0); + verify(mockMetricExporter).incCounters(INCOMPATIBLE_MUTATION_TIMESTAMP_OVERFLOW_METRIC_KEY, 0); + verify(mockMetricExporter).incCounters(PUTS_IN_FUTURE_METRIC_KEY, 0); + + // No replication tasks submitted + verifyNoMoreInteractions(mockMetricExporter, mockExecutorService); + verifyNoInteractions(mockConnection); + } + + @Test + public void testBidirectionalReplicationSkipsNullCellsWalEntryWithoutFailing() { + CloudBigtableReplicator replicator = new CloudBigtableReplicator(); + Configuration replicationConf = new Configuration(false); + replicationConf.setBoolean(CBT_REPL_BIDIRECTIONAL_REPLICATION_MODE_KEY, true); + replicator.start( + sharedResources, + incompatibleMutationAdapter, + 2000, + false, + replicationConf, + mockMetricExporter); + + BigtableWALEntry nullCellsEntry = new BigtableWALEntry(TIMESTAMP, null, TABLE_NAME_STRING); + + Map> walsToReplicate = new HashMap<>(); + walsToReplicate.put(TABLE_NAME_STRING, Arrays.asList(nullCellsEntry)); + + // Must return true without throwing NullPointerException + assertTrue(replicator.replicate(walsToReplicate)); + + verify(mockMetricExporter).incCounters(INCOMPATIBLE_MUTATION_METRIC_KEY, 0); + verify(mockMetricExporter).incCounters(DROPPED_INCOMPATIBLE_MUTATION_METRIC_KEY, 0); + verify(mockMetricExporter).incCounters(INCOMPATIBLE_MUTATION_DELETES_METRICS_KEY, 0); + verify(mockMetricExporter).incCounters(INCOMPATIBLE_MUTATION_TIMESTAMP_OVERFLOW_METRIC_KEY, 0); + verify(mockMetricExporter).incCounters(PUTS_IN_FUTURE_METRIC_KEY, 0); + + verifyNoMoreInteractions(mockMetricExporter, mockExecutorService); + verifyNoInteractions(mockConnection); + } + + @Test + public void testBidirectionalReplicationSkipsNullWalEntryWithoutFailing() { + CloudBigtableReplicator replicator = new CloudBigtableReplicator(); + Configuration replicationConf = new Configuration(false); + replicationConf.setBoolean(CBT_REPL_BIDIRECTIONAL_REPLICATION_MODE_KEY, true); + replicator.start( + sharedResources, + incompatibleMutationAdapter, + 2000, + false, + replicationConf, + mockMetricExporter); + + Map> walsToReplicate = new HashMap<>(); + walsToReplicate.put(TABLE_NAME_STRING, Collections.singletonList((BigtableWALEntry) null)); + + assertTrue(replicator.replicate(walsToReplicate)); + + verify(mockMetricExporter).incCounters(INCOMPATIBLE_MUTATION_METRIC_KEY, 0); + verify(mockMetricExporter).incCounters(DROPPED_INCOMPATIBLE_MUTATION_METRIC_KEY, 0); + verify(mockMetricExporter).incCounters(INCOMPATIBLE_MUTATION_DELETES_METRICS_KEY, 0); + verify(mockMetricExporter).incCounters(INCOMPATIBLE_MUTATION_TIMESTAMP_OVERFLOW_METRIC_KEY, 0); + verify(mockMetricExporter).incCounters(PUTS_IN_FUTURE_METRIC_KEY, 0); + + verifyNoMoreInteractions(mockMetricExporter, mockExecutorService); + verifyNoInteractions(mockConnection); + } + + @Test + public void testOneWayReplicationSkipsEmptyWalEntryWithoutIncrementingMetrics() { + CloudBigtableReplicator replicator = new CloudBigtableReplicator(); + replicator.start( + sharedResources, incompatibleMutationAdapter, 2000, false, conf, mockMetricExporter); + + BigtableWALEntry emptyEntry = + new BigtableWALEntry(TIMESTAMP, Collections.emptyList(), TABLE_NAME_STRING); + + Map> walsToReplicate = new HashMap<>(); + walsToReplicate.put(TABLE_NAME_STRING, Arrays.asList(emptyEntry)); + + assertTrue(replicator.replicate(walsToReplicate)); + + // Verify ONE_WAY_REPL_ELIGIBLE_WAL_ENTRY_METRIC_KEY was NOT incremented + verify(mockMetricExporter).incCounters(INCOMPATIBLE_MUTATION_METRIC_KEY, 0); + verify(mockMetricExporter).incCounters(DROPPED_INCOMPATIBLE_MUTATION_METRIC_KEY, 0); + verify(mockMetricExporter).incCounters(INCOMPATIBLE_MUTATION_DELETES_METRICS_KEY, 0); + verify(mockMetricExporter).incCounters(INCOMPATIBLE_MUTATION_TIMESTAMP_OVERFLOW_METRIC_KEY, 0); + verify(mockMetricExporter).incCounters(PUTS_IN_FUTURE_METRIC_KEY, 0); + + verifyNoMoreInteractions(mockMetricExporter, mockExecutorService); + verifyNoInteractions(mockConnection); + } + + @Test + public void testReplicateFiltersEmptyWalEntriesAndReplicatesValidOnes() throws IOException { + CloudBigtableReplicator replicator = new CloudBigtableReplicator(); + Configuration replicationConf = new Configuration(false); + replicationConf.setBoolean(CBT_REPL_BIDIRECTIONAL_REPLICATION_MODE_KEY, true); + replicator.start( + sharedResources, + incompatibleMutationAdapter, + 2000, + false, + replicationConf, + mockMetricExporter); + + BigtableWALEntry emptyEntry = + new BigtableWALEntry(TIMESTAMP, Collections.emptyList(), TABLE_NAME_STRING); + Cell validCell = new KeyValue(getRowKey(1), CF1, null, TIMESTAMP, getValue(1)); + BigtableWALEntry validEntry = + new BigtableWALEntry(TIMESTAMP, Arrays.asList(validCell), TABLE_NAME_STRING); + + Map> walsToReplicate = new HashMap<>(); + walsToReplicate.put(TABLE_NAME_STRING, Arrays.asList(emptyEntry, validEntry)); + + Cell sourceMarker = + new KeyValue( + getRowKey(1), + CF1, + CBT_REPL_HBASE_QUALIFIER.getBytes(StandardCharsets.UTF_8), + 0L, + KeyValue.Type.Delete); + Map> expectedBatch = new HashMap<>(); + expectedBatch.put(new SimpleByteRange(getRowKey(1)), Arrays.asList(validCell, sourceMarker)); + + when(mockExecutorService.submit((Callable) any())) + .thenReturn(CompletableFuture.completedFuture(true)); + + assertTrue(replicator.replicate(walsToReplicate)); + + verify(mockExecutorService) + .submit(new CloudBigtableReplicationTask(TABLE_NAME_STRING, mockConnection, expectedBatch)); + } + + // --- Tests for Finding 2: Identity fallback collisions & cross-batch safety --- + + @Test + public void testReplicateFailsOnUnmappedIdentityCollisionWithMappedTarget() { + Configuration mappingConf = new Configuration(false); + // Explicitly map ns:table_B to table_B + mappingConf.set(TABLE_NAME_MAPPING_KEY, "{\"ns:table_B\": \"table_B\"}"); + + CloudBigtableReplicator replicator = new CloudBigtableReplicator(); + replicator.start(sharedResources, incompatibleMutationAdapter, 2000, false, mappingConf, null); + + Cell cell = new KeyValue(getRowKey(1), CF1, null, TIMESTAMP, getValue(1)); + // An unmapped default-namespace table named table_B + BigtableWALEntry unmappedEntry = + new BigtableWALEntry(TIMESTAMP, Arrays.asList(cell), "table_B"); + + Map> walsToReplicate = new HashMap<>(); + walsToReplicate.put("table_B", Arrays.asList(unmappedEntry)); + + // TableNameMapper throws IllegalArgumentException on unmapped collision, + // which replicate() must catch, log, and return false. + assertFalse(replicator.replicate(walsToReplicate)); + verifyNoInteractions(mockExecutorService); + } + + @Test + public void testReplicateFailsWhenBatchContainsBothMappedAndUnmappedCollidingTables() { + Configuration mappingConf = new Configuration(false); + // Explicitly map ns:table_B to table_B + mappingConf.set(TABLE_NAME_MAPPING_KEY, "{\"ns:table_B\": \"table_B\"}"); + + CloudBigtableReplicator replicator = new CloudBigtableReplicator(); + replicator.start(sharedResources, incompatibleMutationAdapter, 2000, false, mappingConf, null); + + Cell cell1 = new KeyValue(getRowKey(1), CF1, null, TIMESTAMP, getValue(1)); + BigtableWALEntry mappedEntry = + new BigtableWALEntry(TIMESTAMP, Arrays.asList(cell1), "ns:table_B"); + + Cell cell2 = new KeyValue(getRowKey(2), CF1, null, TIMESTAMP, getValue(2)); + BigtableWALEntry unmappedEntry = + new BigtableWALEntry(TIMESTAMP, Arrays.asList(cell2), "table_B"); + + Map> walsToReplicate = new LinkedHashMap<>(); + walsToReplicate.put("ns:table_B", Arrays.asList(mappedEntry)); + walsToReplicate.put("table_B", Arrays.asList(unmappedEntry)); + + assertFalse(replicator.replicate(walsToReplicate)); + verifyNoInteractions(mockExecutorService); + } + + // --- Tests for Finding 3: UTF-8 qualifiers for bidirectional markers --- + + @Test + public void testBidirectionalReplicationSupportsUtf8Qualifiers() throws IOException { + String customCbtQualifier = "_source_cbt_日本語"; + String customHbaseQualifier = "_source_hbase_日本語"; + + Configuration replicationConf = new Configuration(false); + replicationConf.setBoolean(CBT_REPL_BIDIRECTIONAL_REPLICATION_MODE_KEY, true); + replicationConf.set(CBT_REPL_CBT_QUALIFIER_KEY, customCbtQualifier); + replicationConf.set(CBT_REPL_HBASE_QUALIFIER_KEY, customHbaseQualifier); + + CloudBigtableReplicator replicator = new CloudBigtableReplicator(); + replicator.start( + sharedResources, + incompatibleMutationAdapter, + 2000, + false, + replicationConf, + mockMetricExporter); + + // 1. CBT marker with UTF-8 qualifier should be dropped + Cell cbtMarkerCell = + new KeyValue( + getRowKey(1), + CF1, + customCbtQualifier.getBytes(StandardCharsets.UTF_8), + 0L, + KeyValue.Type.Delete); + BigtableWALEntry cbtEntry = + new BigtableWALEntry(TIMESTAMP, Arrays.asList(cbtMarkerCell), TABLE_NAME_STRING); + + assertTrue( + replicator.replicate(Collections.singletonMap(TABLE_NAME_STRING, Arrays.asList(cbtEntry)))); + verify(mockMetricExporter).incCounters(BIDIRECTIONAL_REPL_INELIGIBLE_WAL_ENTRY_METRIC_KEY, 1); + verifyNoInteractions(mockExecutorService); + + // 2. HBase mutation should be tagged with UTF-8 HBase qualifier + Cell userCell = new KeyValue(getRowKey(2), CF1, null, TIMESTAMP, getValue(2)); + BigtableWALEntry hbaseEntry = + new BigtableWALEntry(TIMESTAMP, Arrays.asList(userCell), TABLE_NAME_STRING); + + Cell expectedMarker = + new KeyValue( + getRowKey(2), + CF1, + customHbaseQualifier.getBytes(StandardCharsets.UTF_8), + 0L, + KeyValue.Type.Delete); + Map> expectedBatch = new HashMap<>(); + expectedBatch.put(new SimpleByteRange(getRowKey(2)), Arrays.asList(userCell, expectedMarker)); + + when(mockExecutorService.submit((Callable) any())) + .thenReturn(CompletableFuture.completedFuture(true)); + + assertTrue( + replicator.replicate( + Collections.singletonMap(TABLE_NAME_STRING, Arrays.asList(hbaseEntry)))); + + verify(mockExecutorService) + .submit(new CloudBigtableReplicationTask(TABLE_NAME_STRING, mockConnection, expectedBatch)); + } + + // --- Tests for Round 3 Finding 1: Robust Bidirectional Marker Detection --- + + @Test + public void testBidirectionalReplicationDetectsMarkerAtFirstPositionDropsEntry() + throws IOException { + CloudBigtableReplicator replicator = new CloudBigtableReplicator(); + Configuration replicationConf = new Configuration(false); + replicationConf.setBoolean(CBT_REPL_BIDIRECTIONAL_REPLICATION_MODE_KEY, true); + replicator.start( + sharedResources, + incompatibleMutationAdapter, + 2000, + false, + replicationConf, + mockMetricExporter); + + // CBT source marker delete placed as the FIRST cell (index 0), followed by a normal cell + Cell cbtMarkerCell = + new KeyValue( + getRowKey(1), CF1, CBT_REPL_CBT_QUALIFIER.getBytes(), 0L, KeyValue.Type.Delete); + Cell userCell = new KeyValue(getRowKey(1), CF1, COL_QUALIFIER, TIMESTAMP, getValue(1)); + + BigtableWALEntry inputWAL = + new BigtableWALEntry(TIMESTAMP, Arrays.asList(cbtMarkerCell, userCell), TABLE_NAME_STRING); + Map> walsToReplicate = new HashMap<>(); + walsToReplicate.put(TABLE_NAME_STRING, Arrays.asList(inputWAL)); + + assertTrue(replicator.replicate(walsToReplicate)); + + verify(mockMetricExporter).incCounters(INCOMPATIBLE_MUTATION_METRIC_KEY, 0); + verify(mockMetricExporter).incCounters(DROPPED_INCOMPATIBLE_MUTATION_METRIC_KEY, 0); + verify(mockMetricExporter).incCounters(INCOMPATIBLE_MUTATION_DELETES_METRICS_KEY, 0); + verify(mockMetricExporter).incCounters(INCOMPATIBLE_MUTATION_TIMESTAMP_OVERFLOW_METRIC_KEY, 0); + verify(mockMetricExporter).incCounters(PUTS_IN_FUTURE_METRIC_KEY, 0); + // Expect the WAL entry to be dropped because marker is detected at first position + verify(mockMetricExporter).incCounters(BIDIRECTIONAL_REPL_INELIGIBLE_WAL_ENTRY_METRIC_KEY, 1); + verify(mockMetricExporter).incCounters(BIDIRECTIONAL_REPL_INELIGIBLE_MUTATIONS_METRIC_KEY, 1); + verifyNoInteractions(mockExecutorService); + } + + @Test + public void testBidirectionalReplicationUserPutWithMarkerQualifierIsNotDropped() + throws IOException { + CloudBigtableReplicator replicator = new CloudBigtableReplicator(); + Configuration replicationConf = new Configuration(false); + replicationConf.setBoolean(CBT_REPL_BIDIRECTIONAL_REPLICATION_MODE_KEY, true); + replicator.start( + sharedResources, + incompatibleMutationAdapter, + 2000, + false, + replicationConf, + mockMetricExporter); + + // Normal user Put cell with qualifier equal to CBT_REPL_CBT_QUALIFIER and non-zero timestamp + Cell userPutCell = + new KeyValue(getRowKey(1), CF1, CBT_REPL_CBT_QUALIFIER.getBytes(), TIMESTAMP, getValue(1)); + + BigtableWALEntry inputWAL = + new BigtableWALEntry(TIMESTAMP, Arrays.asList(userPutCell), TABLE_NAME_STRING); + Map> walsToReplicate = new HashMap<>(); + walsToReplicate.put(TABLE_NAME_STRING, Arrays.asList(inputWAL)); + + // When replicated, HBase source marker is appended to the row cells + Cell hbaseSourceTaggedCell = + new KeyValue( + getRowKey(1), CF1, CBT_REPL_HBASE_QUALIFIER.getBytes(), 0L, KeyValue.Type.Delete); + BigtableWALEntry expectedWAL = + new BigtableWALEntry( + TIMESTAMP, Arrays.asList(userPutCell, hbaseSourceTaggedCell), TABLE_NAME_STRING); + Map> expectedBatchOfWal = new HashMap<>(); + expectedBatchOfWal.put(new SimpleByteRange(getRowKey(1)), expectedWAL.getCells()); + + when(mockExecutorService.submit((Callable) any())) + .thenReturn(CompletableFuture.completedFuture(true)); + + assertTrue(replicator.replicate(walsToReplicate)); + + verify(mockMetricExporter).incCounters(INCOMPATIBLE_MUTATION_METRIC_KEY, 0); + verify(mockMetricExporter).incCounters(DROPPED_INCOMPATIBLE_MUTATION_METRIC_KEY, 0); + verify(mockMetricExporter).incCounters(INCOMPATIBLE_MUTATION_DELETES_METRICS_KEY, 0); + verify(mockMetricExporter).incCounters(INCOMPATIBLE_MUTATION_TIMESTAMP_OVERFLOW_METRIC_KEY, 0); + verify(mockMetricExporter).incCounters(PUTS_IN_FUTURE_METRIC_KEY, 0); + // Expect WAL entry NOT to be dropped; eligible metrics should be recorded + verify(mockMetricExporter).incCounters(BIDIRECTIONAL_REPL_ELIGIBLE_WAL_ENTRY_METRIC_KEY, 1); + verify(mockMetricExporter).incCounters(BIDIRECTIONAL_REPL_ELIGIBLE_MUTATIONS_METRIC_KEY, 1); + + verify(mockExecutorService) + .submit( + new CloudBigtableReplicationTask( + TABLE_NAME_STRING, mockConnection, expectedBatchOfWal)); + verifyNoMoreInteractions(mockMetricExporter, mockExecutorService); + verifyNoInteractions(mockConnection); + } + + @Test + public void testBidirectionalReplicationDetectsMarkerInMiddlePositionDropsEntry() + throws IOException { + CloudBigtableReplicator replicator = new CloudBigtableReplicator(); + Configuration replicationConf = new Configuration(false); + replicationConf.setBoolean(CBT_REPL_BIDIRECTIONAL_REPLICATION_MODE_KEY, true); + replicator.start( + sharedResources, + incompatibleMutationAdapter, + 2000, + false, + replicationConf, + mockMetricExporter); + + // Cells sorted such that CBT marker delete is in the MIDDLE between user cells + Cell userCell1 = + new KeyValue( + getRowKey(1), CF1, "col_a".getBytes(StandardCharsets.UTF_8), TIMESTAMP, getValue(1)); + Cell cbtMarkerCell = + new KeyValue( + getRowKey(1), CF1, CBT_REPL_CBT_QUALIFIER.getBytes(), 0L, KeyValue.Type.Delete); + Cell userCell2 = + new KeyValue( + getRowKey(1), CF1, "col_z".getBytes(StandardCharsets.UTF_8), TIMESTAMP, getValue(2)); + + BigtableWALEntry inputWAL = + new BigtableWALEntry( + TIMESTAMP, Arrays.asList(userCell1, cbtMarkerCell, userCell2), TABLE_NAME_STRING); + Map> walsToReplicate = new HashMap<>(); + walsToReplicate.put(TABLE_NAME_STRING, Arrays.asList(inputWAL)); + + assertTrue(replicator.replicate(walsToReplicate)); + + verify(mockMetricExporter).incCounters(INCOMPATIBLE_MUTATION_METRIC_KEY, 0); + verify(mockMetricExporter).incCounters(DROPPED_INCOMPATIBLE_MUTATION_METRIC_KEY, 0); + verify(mockMetricExporter).incCounters(INCOMPATIBLE_MUTATION_DELETES_METRICS_KEY, 0); + verify(mockMetricExporter).incCounters(INCOMPATIBLE_MUTATION_TIMESTAMP_OVERFLOW_METRIC_KEY, 0); + verify(mockMetricExporter).incCounters(PUTS_IN_FUTURE_METRIC_KEY, 0); + // Entry dropped, 2 user mutations counted as ineligible (3 total - 1 marker) + verify(mockMetricExporter).incCounters(BIDIRECTIONAL_REPL_INELIGIBLE_WAL_ENTRY_METRIC_KEY, 1); + verify(mockMetricExporter).incCounters(BIDIRECTIONAL_REPL_INELIGIBLE_MUTATIONS_METRIC_KEY, 2); + verifyNoInteractions(mockExecutorService); + } + + @Test + public void testBidirectionalReplicationUserDeleteWithMarkerQualifierIsNotDropped() + throws IOException { + CloudBigtableReplicator replicator = new CloudBigtableReplicator(); + Configuration replicationConf = new Configuration(false); + replicationConf.setBoolean(CBT_REPL_BIDIRECTIONAL_REPLICATION_MODE_KEY, true); + replicator.start( + sharedResources, + incompatibleMutationAdapter, + 2000, + false, + replicationConf, + mockMetricExporter); + + // Normal user Delete cell with qualifier equal to CBT_REPL_CBT_QUALIFIER and non-zero timestamp + Cell userDeleteCell = + new KeyValue( + getRowKey(1), CF1, CBT_REPL_CBT_QUALIFIER.getBytes(), TIMESTAMP, KeyValue.Type.Delete); + + BigtableWALEntry inputWAL = + new BigtableWALEntry(TIMESTAMP, Arrays.asList(userDeleteCell), TABLE_NAME_STRING); + Map> walsToReplicate = new HashMap<>(); + walsToReplicate.put(TABLE_NAME_STRING, Arrays.asList(inputWAL)); + + Cell hbaseSourceTaggedCell = + new KeyValue( + getRowKey(1), CF1, CBT_REPL_HBASE_QUALIFIER.getBytes(), 0L, KeyValue.Type.Delete); + BigtableWALEntry expectedWAL = + new BigtableWALEntry( + TIMESTAMP, Arrays.asList(userDeleteCell, hbaseSourceTaggedCell), TABLE_NAME_STRING); + Map> expectedBatchOfWal = new HashMap<>(); + expectedBatchOfWal.put(new SimpleByteRange(getRowKey(1)), expectedWAL.getCells()); + + when(mockExecutorService.submit((Callable) any())) + .thenReturn(CompletableFuture.completedFuture(true)); + + assertTrue(replicator.replicate(walsToReplicate)); + + verify(mockMetricExporter).incCounters(INCOMPATIBLE_MUTATION_METRIC_KEY, 0); + verify(mockMetricExporter).incCounters(DROPPED_INCOMPATIBLE_MUTATION_METRIC_KEY, 0); + verify(mockMetricExporter).incCounters(INCOMPATIBLE_MUTATION_DELETES_METRICS_KEY, 0); + verify(mockMetricExporter).incCounters(INCOMPATIBLE_MUTATION_TIMESTAMP_OVERFLOW_METRIC_KEY, 0); + verify(mockMetricExporter).incCounters(PUTS_IN_FUTURE_METRIC_KEY, 0); + verify(mockMetricExporter).incCounters(BIDIRECTIONAL_REPL_ELIGIBLE_WAL_ENTRY_METRIC_KEY, 1); + verify(mockMetricExporter).incCounters(BIDIRECTIONAL_REPL_ELIGIBLE_MUTATIONS_METRIC_KEY, 1); + + verify(mockExecutorService) + .submit( + new CloudBigtableReplicationTask( + TABLE_NAME_STRING, mockConnection, expectedBatchOfWal)); + verifyNoMoreInteractions(mockMetricExporter, mockExecutorService); + verifyNoInteractions(mockConnection); + } + + @Test + public void testBidirectionalReplicationUserPutWithMarkerQualifierAtTimestampZeroIsNotDropped() + throws IOException { + CloudBigtableReplicator replicator = new CloudBigtableReplicator(); + Configuration replicationConf = new Configuration(false); + replicationConf.setBoolean(CBT_REPL_BIDIRECTIONAL_REPLICATION_MODE_KEY, true); + replicator.start( + sharedResources, + incompatibleMutationAdapter, + 2000, + false, + replicationConf, + mockMetricExporter); + + // Normal user Put cell with qualifier equal to CBT_REPL_CBT_QUALIFIER and timestamp 0L (Put, + // not Delete) + Cell userPutCell = + new KeyValue(getRowKey(1), CF1, CBT_REPL_CBT_QUALIFIER.getBytes(), 0L, KeyValue.Type.Put); + + BigtableWALEntry inputWAL = + new BigtableWALEntry(TIMESTAMP, Arrays.asList(userPutCell), TABLE_NAME_STRING); + Map> walsToReplicate = new HashMap<>(); + walsToReplicate.put(TABLE_NAME_STRING, Arrays.asList(inputWAL)); + + Cell hbaseSourceTaggedCell = + new KeyValue( + getRowKey(1), CF1, CBT_REPL_HBASE_QUALIFIER.getBytes(), 0L, KeyValue.Type.Delete); + BigtableWALEntry expectedWAL = + new BigtableWALEntry( + TIMESTAMP, Arrays.asList(userPutCell, hbaseSourceTaggedCell), TABLE_NAME_STRING); + Map> expectedBatchOfWal = new HashMap<>(); + expectedBatchOfWal.put(new SimpleByteRange(getRowKey(1)), expectedWAL.getCells()); + + when(mockExecutorService.submit((Callable) any())) + .thenReturn(CompletableFuture.completedFuture(true)); + + assertTrue(replicator.replicate(walsToReplicate)); + + verify(mockMetricExporter).incCounters(INCOMPATIBLE_MUTATION_METRIC_KEY, 0); + verify(mockMetricExporter).incCounters(DROPPED_INCOMPATIBLE_MUTATION_METRIC_KEY, 0); + verify(mockMetricExporter).incCounters(INCOMPATIBLE_MUTATION_DELETES_METRICS_KEY, 0); + verify(mockMetricExporter).incCounters(INCOMPATIBLE_MUTATION_TIMESTAMP_OVERFLOW_METRIC_KEY, 0); + verify(mockMetricExporter).incCounters(PUTS_IN_FUTURE_METRIC_KEY, 0); + verify(mockMetricExporter).incCounters(BIDIRECTIONAL_REPL_ELIGIBLE_WAL_ENTRY_METRIC_KEY, 1); + verify(mockMetricExporter).incCounters(BIDIRECTIONAL_REPL_ELIGIBLE_MUTATIONS_METRIC_KEY, 1); + + verify(mockExecutorService) + .submit( + new CloudBigtableReplicationTask( + TABLE_NAME_STRING, mockConnection, expectedBatchOfWal)); + verifyNoMoreInteractions(mockMetricExporter, mockExecutorService); + verifyNoInteractions(mockConnection); + } + + @Test + public void testBidirectionalReplicationDetectsDeleteColumnMarkerDropsEntry() throws IOException { + CloudBigtableReplicator replicator = new CloudBigtableReplicator(); + Configuration replicationConf = new Configuration(false); + replicationConf.setBoolean(CBT_REPL_BIDIRECTIONAL_REPLICATION_MODE_KEY, true); + replicator.start( + sharedResources, + incompatibleMutationAdapter, + 2000, + false, + replicationConf, + mockMetricExporter); + + // CBT source marker delete using DeleteColumn type at timestamp 0L + Cell cbtMarkerCell = + new KeyValue( + getRowKey(1), CF1, CBT_REPL_CBT_QUALIFIER.getBytes(), 0L, KeyValue.Type.DeleteColumn); + Cell userCell = new KeyValue(getRowKey(1), CF1, COL_QUALIFIER, TIMESTAMP, getValue(1)); + + BigtableWALEntry inputWAL = + new BigtableWALEntry(TIMESTAMP, Arrays.asList(cbtMarkerCell, userCell), TABLE_NAME_STRING); + Map> walsToReplicate = new HashMap<>(); + walsToReplicate.put(TABLE_NAME_STRING, Arrays.asList(inputWAL)); + + assertTrue(replicator.replicate(walsToReplicate)); + + verify(mockMetricExporter).incCounters(INCOMPATIBLE_MUTATION_METRIC_KEY, 0); + verify(mockMetricExporter).incCounters(DROPPED_INCOMPATIBLE_MUTATION_METRIC_KEY, 0); + verify(mockMetricExporter).incCounters(INCOMPATIBLE_MUTATION_DELETES_METRICS_KEY, 0); + verify(mockMetricExporter).incCounters(INCOMPATIBLE_MUTATION_TIMESTAMP_OVERFLOW_METRIC_KEY, 0); + verify(mockMetricExporter).incCounters(PUTS_IN_FUTURE_METRIC_KEY, 0); + verify(mockMetricExporter).incCounters(BIDIRECTIONAL_REPL_INELIGIBLE_WAL_ENTRY_METRIC_KEY, 1); + verify(mockMetricExporter).incCounters(BIDIRECTIONAL_REPL_INELIGIBLE_MUTATIONS_METRIC_KEY, 1); + verifyNoInteractions(mockExecutorService); + } + + // --- Tests for Finding 4: Pre-flush batch threshold check --- + + @Test + public void testReplicatePreFlushesBatchBeforeAddingRowExceedingSizeThreshold() + throws IOException { + Cell cell1 = new KeyValue(getRowKey(1), CF1, null, TIMESTAMP, getValue(1)); + int singleRowSize = (int) org.apache.hadoop.hbase.CellUtil.estimatedHeapSizeOf(cell1); + + // Set threshold so that 1 row fits, but 2 rows exceed it + int threshold = (int) (singleRowSize * 1.5); + + CloudBigtableReplicator replicator = new CloudBigtableReplicator(); + replicator.start(sharedResources, incompatibleMutationAdapter, threshold, false, null, null); + + Cell cell2 = new KeyValue(getRowKey(2), CF1, null, TIMESTAMP, getValue(2)); + BigtableWALEntry walEntry1 = + new BigtableWALEntry(TIMESTAMP, Arrays.asList(cell1), TABLE_NAME_STRING); + BigtableWALEntry walEntry2 = + new BigtableWALEntry(TIMESTAMP, Arrays.asList(cell2), TABLE_NAME_STRING); + + Map> expectedBatch1 = new HashMap<>(); + expectedBatch1.put(new SimpleByteRange(getRowKey(1)), walEntry1.getCells()); + Map> expectedBatch2 = new HashMap<>(); + expectedBatch2.put(new SimpleByteRange(getRowKey(2)), walEntry2.getCells()); + + Map> walsToReplicate = new LinkedHashMap<>(); + walsToReplicate.put(TABLE_NAME_STRING, Arrays.asList(walEntry1, walEntry2)); + + when(mockExecutorService.submit((Callable) any())) + .thenReturn(CompletableFuture.completedFuture(true)); + + assertTrue(replicator.replicate(walsToReplicate)); + + // Verify that the batch was pre-flushed before adding row 2, producing 2 distinct batches + verify(mockExecutorService) + .submit( + new CloudBigtableReplicationTask(TABLE_NAME_STRING, mockConnection, expectedBatch1)); + verify(mockExecutorService) + .submit( + new CloudBigtableReplicationTask(TABLE_NAME_STRING, mockConnection, expectedBatch2)); + } + + @Test + public void testReplicatePreFlushesBatchBeforeAddingRowExceedingCellCountThreshold() + throws IOException { + CloudBigtableReplicator replicator = new CloudBigtableReplicator(); + // High byte threshold so only cell count threshold (100,000) triggers + replicator.start( + sharedResources, incompatibleMutationAdapter, Integer.MAX_VALUE, false, null, null); + + // Row 1 with 60,000 cells + Cell[] row1Cells = new Cell[60_000]; + for (int i = 0; i < 60_000; i++) { + row1Cells[i] = + new KeyValue( + getRowKey(1), + CF1, + ("col" + i).getBytes(StandardCharsets.UTF_8), + TIMESTAMP, + getValue(1)); + } + BigtableWALEntry walEntry1 = + new BigtableWALEntry(TIMESTAMP, Arrays.asList(row1Cells), TABLE_NAME_STRING); + + // Row 2 with 50,000 cells (60,000 + 50,000 = 110,000 >= 100,000) + Cell[] row2Cells = new Cell[50_000]; + for (int i = 0; i < 50_000; i++) { + row2Cells[i] = + new KeyValue( + getRowKey(2), + CF1, + ("col" + i).getBytes(StandardCharsets.UTF_8), + TIMESTAMP, + getValue(2)); + } + BigtableWALEntry walEntry2 = + new BigtableWALEntry(TIMESTAMP, Arrays.asList(row2Cells), TABLE_NAME_STRING); + + Map> expectedBatch1 = new HashMap<>(); + expectedBatch1.put(new SimpleByteRange(getRowKey(1)), walEntry1.getCells()); + Map> expectedBatch2 = new HashMap<>(); + expectedBatch2.put(new SimpleByteRange(getRowKey(2)), walEntry2.getCells()); + + Map> walsToReplicate = new LinkedHashMap<>(); + walsToReplicate.put(TABLE_NAME_STRING, Arrays.asList(walEntry1, walEntry2)); + + when(mockExecutorService.submit((Callable) any())) + .thenReturn(CompletableFuture.completedFuture(true)); + + assertTrue(replicator.replicate(walsToReplicate)); + + // Verify row 1 was pre-flushed before row 2 was added, resulting in 2 separate tasks + verify(mockExecutorService) + .submit( + new CloudBigtableReplicationTask(TABLE_NAME_STRING, mockConnection, expectedBatch1)); + verify(mockExecutorService) + .submit( + new CloudBigtableReplicationTask(TABLE_NAME_STRING, mockConnection, expectedBatch2)); + } + + @Test + public void testReplicatePreFlushesAndPostFlushesOversizedRow() throws IOException { + Cell cell1 = new KeyValue(getRowKey(1), CF1, null, TIMESTAMP, getValue(1)); + int singleCellSize = (int) org.apache.hadoop.hbase.CellUtil.estimatedHeapSizeOf(cell1); + + // Threshold is 2 * singleCellSize + int threshold = singleCellSize * 2; + + CloudBigtableReplicator replicator = new CloudBigtableReplicator(); + replicator.start(sharedResources, incompatibleMutationAdapter, threshold, false, null, null); + + // Row 1: 1 cell (fits) + BigtableWALEntry walEntry1 = + new BigtableWALEntry(TIMESTAMP, Arrays.asList(cell1), TABLE_NAME_STRING); + + // Row 2: 3 cells (3 * singleCellSize > threshold, oversized row) + Cell cell2a = + new KeyValue( + getRowKey(2), CF1, "c1".getBytes(StandardCharsets.UTF_8), TIMESTAMP, getValue(2)); + Cell cell2b = + new KeyValue( + getRowKey(2), CF1, "c2".getBytes(StandardCharsets.UTF_8), TIMESTAMP, getValue(2)); + Cell cell2c = + new KeyValue( + getRowKey(2), CF1, "c3".getBytes(StandardCharsets.UTF_8), TIMESTAMP, getValue(2)); + BigtableWALEntry walEntry2 = + new BigtableWALEntry(TIMESTAMP, Arrays.asList(cell2a, cell2b, cell2c), TABLE_NAME_STRING); + + // Row 3: 1 cell (fits) + Cell cell3 = new KeyValue(getRowKey(3), CF1, null, TIMESTAMP, getValue(3)); + BigtableWALEntry walEntry3 = + new BigtableWALEntry(TIMESTAMP, Arrays.asList(cell3), TABLE_NAME_STRING); + + Map> expectedBatch1 = new HashMap<>(); + expectedBatch1.put(new SimpleByteRange(getRowKey(1)), walEntry1.getCells()); + Map> expectedBatch2 = new HashMap<>(); + expectedBatch2.put(new SimpleByteRange(getRowKey(2)), walEntry2.getCells()); + Map> expectedBatch3 = new HashMap<>(); + expectedBatch3.put(new SimpleByteRange(getRowKey(3)), walEntry3.getCells()); + + Map> walsToReplicate = new LinkedHashMap<>(); + walsToReplicate.put(TABLE_NAME_STRING, Arrays.asList(walEntry1, walEntry2, walEntry3)); + + when(mockExecutorService.submit((Callable) any())) + .thenReturn(CompletableFuture.completedFuture(true)); + + assertTrue(replicator.replicate(walsToReplicate)); + + // Verify 3 distinct batches: Row 1 pre-flushed before Row 2, Row 2 post-flushed immediately, + // Row 3 flushed at end. + verify(mockExecutorService) + .submit( + new CloudBigtableReplicationTask(TABLE_NAME_STRING, mockConnection, expectedBatch1)); + verify(mockExecutorService) + .submit( + new CloudBigtableReplicationTask(TABLE_NAME_STRING, mockConnection, expectedBatch2)); + verify(mockExecutorService) + .submit( + new CloudBigtableReplicationTask(TABLE_NAME_STRING, mockConnection, expectedBatch3)); + } + + /** JUnit 4.12 (pulled in transitively by hbase-client) does not have assertThrows. */ + private static void expectThrows(Class expected, ThrowingRunnable runnable) { + try { + runnable.run(); + } catch (Throwable t) { + if (expected.isInstance(t)) { + return; + } + throw new AssertionError( + "Expected " + expected.getName() + " but got " + t.getClass().getName(), t); + } + fail("Expected " + expected.getName() + " to be thrown."); + } + + private interface ThrowingRunnable { + void run() throws Throwable; + } } diff --git a/hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-replication-core/src/test/java/com/google/cloud/bigtable/hbase/replication/configuration/TableNameMapperTest.java b/hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-replication-core/src/test/java/com/google/cloud/bigtable/hbase/replication/configuration/TableNameMapperTest.java new file mode 100644 index 0000000000..b2999beea1 --- /dev/null +++ b/hbase-migration-tools/bigtable-hbase-replication/bigtable-hbase-replication-core/src/test/java/com/google/cloud/bigtable/hbase/replication/configuration/TableNameMapperTest.java @@ -0,0 +1,517 @@ +/* + * Copyright 2026 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.replication.configuration; + +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 org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.io.File; +import java.io.FileNotFoundException; +import java.io.FileOutputStream; +import java.io.IOException; +import java.io.OutputStream; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import org.apache.hadoop.conf.Configuration; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +@RunWith(JUnit4.class) +public class TableNameMapperTest { + + private static final String MAPPING_JSON = + "{\"smartbrain:oraperi_new\": \"smartbrain_oraperi_new\", \"plain\": \"renamed-plain\"}"; + + @Rule public final TemporaryFolder tempFolder = new TemporaryFolder(); + + private File writeFile(String name, byte[] contents) throws IOException { + File file = tempFolder.newFile(name); + try (OutputStream out = new FileOutputStream(file)) { + out.write(contents); + } + return file; + } + + @Test + public void identityWhenNothingConfigured() throws IOException { + TableNameMapper mapper = TableNameMapper.fromConfiguration(new Configuration(false)); + assertTrue(mapper.isIdentity()); + assertEquals("ns:table", mapper.map("ns:table")); + assertEquals("table", mapper.map("table")); + } + + @Test + public void identityWhenConfigurationIsNull() throws IOException { + assertTrue(TableNameMapper.fromConfiguration(null).isIdentity()); + } + + @Test + public void identityWhenKeysAreBlank() throws IOException { + Configuration conf = new Configuration(false); + conf.set(TABLE_NAME_MAPPING_KEY, " "); + conf.set(TABLE_NAME_MAPPING_FILE_KEY, ""); + assertTrue(TableNameMapper.fromConfiguration(conf).isIdentity()); + } + + @Test + public void loadsInlineJson() throws IOException { + Configuration conf = new Configuration(false); + conf.set(TABLE_NAME_MAPPING_KEY, MAPPING_JSON); + + TableNameMapper mapper = TableNameMapper.fromConfiguration(conf); + + assertFalse(mapper.isIdentity()); + assertEquals("smartbrain_oraperi_new", mapper.map("smartbrain:oraperi_new")); + assertEquals("renamed-plain", mapper.map("plain")); + // Unmapped tables keep their HBase name. + assertEquals("other:table", mapper.map("other:table")); + assertEquals(2, mapper.getMappings().size()); + } + + @Test + public void loadsJsonFromLocalFile() throws IOException { + File file = writeFile("mapping.json", MAPPING_JSON.getBytes(StandardCharsets.UTF_8)); + Configuration conf = new Configuration(false); + conf.set(TABLE_NAME_MAPPING_FILE_KEY, file.toURI().toString()); + + TableNameMapper mapper = TableNameMapper.fromConfiguration(conf); + + assertEquals("smartbrain_oraperi_new", mapper.map("smartbrain:oraperi_new")); + assertEquals("renamed-plain", mapper.map("plain")); + } + + @Test + public void loadsJsonFromLocalFileWithoutScheme() throws IOException { + File file = writeFile("mapping.json", MAPPING_JSON.getBytes(StandardCharsets.UTF_8)); + Configuration conf = new Configuration(false); + conf.set(TABLE_NAME_MAPPING_FILE_KEY, file.getAbsolutePath()); + + TableNameMapper mapper = TableNameMapper.fromConfiguration(conf); + + assertEquals("smartbrain_oraperi_new", mapper.map("smartbrain:oraperi_new")); + } + + @Test + public void rejectsBothInlineAndFile() throws IOException { + File file = writeFile("mapping.json", MAPPING_JSON.getBytes(StandardCharsets.UTF_8)); + Configuration conf = new Configuration(false); + conf.set(TABLE_NAME_MAPPING_KEY, MAPPING_JSON); + conf.set(TABLE_NAME_MAPPING_FILE_KEY, file.getAbsolutePath()); + + expectThrows(IllegalArgumentException.class, () -> TableNameMapper.fromConfiguration(conf)); + } + + @Test + public void rejectsMissingFile() { + Configuration conf = new Configuration(false); + conf.set( + TABLE_NAME_MAPPING_FILE_KEY, + new File(tempFolder.getRoot(), "does-not-exist.json").getAbsolutePath()); + + expectThrows(IOException.class, () -> TableNameMapper.fromConfiguration(conf)); + } + + @Test + public void rejectsDirectory() throws IOException { + File dir = tempFolder.newFolder("mapping-dir"); + Configuration conf = new Configuration(false); + conf.set(TABLE_NAME_MAPPING_FILE_KEY, dir.getAbsolutePath()); + + expectThrows(IOException.class, () -> TableNameMapper.fromConfiguration(conf)); + } + + @Test + public void rejectsOversizedFile() throws IOException { + byte[] contents = new byte[(int) TableNameMapper.MAX_MAPPING_FILE_SIZE_BYTES + 1]; + Arrays.fill(contents, (byte) ' '); + File file = writeFile("huge.json", contents); + Configuration conf = new Configuration(false); + conf.set(TABLE_NAME_MAPPING_FILE_KEY, file.getAbsolutePath()); + + expectThrows(IOException.class, () -> TableNameMapper.fromConfiguration(conf)); + } + + @Test + public void rejectsMalformedJson() { + expectThrows( + IllegalArgumentException.class, () -> TableNameMapper.fromJson("{not json", "test")); + expectThrows( + IllegalArgumentException.class, () -> TableNameMapper.fromJson("[\"a\", \"b\"]", "test")); + expectThrows(IllegalArgumentException.class, () -> TableNameMapper.fromJson("", "test")); + expectThrows(IllegalArgumentException.class, () -> TableNameMapper.fromJson("null", "test")); + } + + @Test + public void acceptsEmptyObject() { + assertTrue(TableNameMapper.fromJson("{}", "test").isIdentity()); + } + + @Test + public void rejectsInvalidBigtableTableIds() { + // Namespace separator is not allowed in a Bigtable table ID. + expectThrows( + IllegalArgumentException.class, + () -> TableNameMapper.fromJson("{\"ns:t\": \"ns:t\"}", "test")); + // Empty / null target. + expectThrows( + IllegalArgumentException.class, () -> TableNameMapper.fromJson("{\"ns:t\": \"\"}", "test")); + expectThrows( + IllegalArgumentException.class, () -> TableNameMapper.fromJson("{\"ns:t\": null}", "test")); + // Leading '-' is not allowed. + expectThrows( + IllegalArgumentException.class, + () -> TableNameMapper.fromJson("{\"ns:t\": \"-bad\"}", "test")); + // Whitespace and slashes are not allowed. + expectThrows( + IllegalArgumentException.class, + () -> TableNameMapper.fromJson("{\"ns:t\": \"bad id\"}", "test")); + expectThrows( + IllegalArgumentException.class, + () -> TableNameMapper.fromJson("{\"ns:t\": \"a/b\"}", "test")); + // Too long (51 characters). + StringBuilder tooLong = new StringBuilder(); + for (int i = 0; i < 51; i++) { + tooLong.append('a'); + } + expectThrows( + IllegalArgumentException.class, + () -> TableNameMapper.fromJson("{\"ns:t\": \"" + tooLong + "\"}", "test")); + } + + @Test + public void rejectsEmptyHbaseTableName() { + expectThrows( + IllegalArgumentException.class, () -> TableNameMapper.fromJson("{\"\": \"t\"}", "test")); + expectThrows( + IllegalArgumentException.class, () -> TableNameMapper.fromJson("{\" \": \"t\"}", "test")); + } + + @Test + public void acceptsValidBigtableTableIds() { + TableNameMapper mapper = + TableNameMapper.fromJson( + "{\"a\": \"_leading_underscore\", \"b\": \"with.dots-and-dashes_1\", \"c\": \"9\"}", + "test"); + assertEquals("_leading_underscore", mapper.map("a")); + assertEquals("with.dots-and-dashes_1", mapper.map("b")); + assertEquals("9", mapper.map("c")); + } + + // --- Tests for Finding 1(a): Canonicalization & 1:1 Injectivity --- + + @Test + public void normalizesDefaultNamespaceToCanonicalHBaseTable() { + TableNameMapper mapper = + TableNameMapper.fromJson("{\"default:my_table\": \"mapped_table\"}", "test"); + // Both default:my_table and my_table resolve to mapped_table. + assertEquals("mapped_table", mapper.map("default:my_table")); + assertEquals("mapped_table", mapper.map("my_table")); + assertEquals("mapped_table", mapper.map(" default:my_table ")); + } + + @Test + public void rejectsConflictingMappingsForSameCanonicalHBaseTable() { + expectThrows( + IllegalArgumentException.class, + () -> + TableNameMapper.fromJson( + "{\"my_table\": \"bt1\", \"default:my_table\": \"bt2\"}", "test")); + } + + @Test + public void rejectsDuplicateMappingForSameCanonicalHBaseTable() { + expectThrows( + IllegalArgumentException.class, + () -> + TableNameMapper.fromJson( + "{\"my_table\": \"bt1\", \"default:my_table\": \"bt1\"}", "test")); + } + + @Test + public void rejectsNToOneMappingAcrossDistinctHBaseTables() { + // Distinct canonical HBase tables mapping to same Bigtable table ID. + expectThrows( + IllegalArgumentException.class, + () -> + TableNameMapper.fromJson( + "{\"ns1:table\": \"shared_bt\", \"ns2:table\": \"shared_bt\"}", "test")); + + expectThrows( + IllegalArgumentException.class, + () -> + TableNameMapper.fromJson( + "{\"table1\": \"shared_bt\", \"default:table2\": \"shared_bt\"}", "test")); + } + + @Test + public void rejectsUnmappedIdentityFallbackCollidingWithMappedTarget() { + TableNameMapper mapper = TableNameMapper.fromJson("{\"ns:table_B\": \"table_B\"}", "test"); + + // The explicitly mapped table maps as expected. + assertEquals("table_B", mapper.map("ns:table_B")); + + // An unmapped table literally named "table_B" collides with the mapped target "table_B". + try { + mapper.map("table_B"); + fail("Expected IllegalArgumentException on collision with mapped target"); + } catch (IllegalArgumentException e) { + assertTrue( + "Error message must mention collision and conflicting source: " + e.getMessage(), + e.getMessage() + .contains( + "Unmapped HBase table 'table_B' collides with target Cloud Bigtable table" + + " 'table_B' already mapped from HBase table 'ns:table_B'")); + } + + // Default namespace alias of table_B also collides. + try { + mapper.map("default:table_B"); + fail("Expected IllegalArgumentException on collision with mapped target"); + } catch (IllegalArgumentException e) { + assertTrue( + "Error message must mention collision: " + e.getMessage(), + e.getMessage() + .contains( + "Unmapped HBase table 'table_B' collides with target Cloud Bigtable table" + + " 'table_B' already mapped from HBase table 'ns:table_B'")); + } + + // An unmapped table with a distinct name does NOT collide. + assertEquals("table_C", mapper.map("table_C")); + } + + @Test + public void allowsDisambiguatedMappingsForCollidingNames() { + TableNameMapper mapper = + TableNameMapper.fromJson( + "{\"ns:table_B\": \"table_B\", \"table_B\": \"table_B_default\"}", "test"); + + assertEquals("table_B", mapper.map("ns:table_B")); + assertEquals("table_B_default", mapper.map("table_B")); + assertEquals("table_B_default", mapper.map("default:table_B")); + } + + @Test + public void allowsExplicitIdentityMappingWithoutCollision() { + TableNameMapper mapper = TableNameMapper.fromJson("{\"table_B\": \"table_B\"}", "test"); + + assertEquals("table_B", mapper.map("table_B")); + assertEquals("table_B", mapper.map("default:table_B")); + } + + // --- Tests for Finding 3: Local FS Fallback & Clear Error Messages --- + + @Test + public void loadsLocalFileWhenSchemeOmittedAndMissingOnDefaultFs() throws IOException { + File file = writeFile("mapping.json", MAPPING_JSON.getBytes(StandardCharsets.UTF_8)); + Configuration conf = new Configuration(false); + // Simulate a cluster where defaultFS points to a remote/nonexistent location. + conf.set("fs.defaultFS", "file:///non/existent/default/fs/mount"); + conf.set(TABLE_NAME_MAPPING_FILE_KEY, file.getAbsolutePath()); + + TableNameMapper mapper = TableNameMapper.fromConfiguration(conf); + assertEquals("smartbrain_oraperi_new", mapper.map("smartbrain:oraperi_new")); + assertEquals("renamed-plain", mapper.map("plain")); + } + + @Test + public void informativeFileNotFoundExceptionMessage() { + Configuration conf = new Configuration(false); + conf.set(TABLE_NAME_MAPPING_FILE_KEY, "/etc/hbase/conf/nonexistent-mapping.json"); + + try { + TableNameMapper.fromConfiguration(conf); + fail("Expected FileNotFoundException"); + } catch (FileNotFoundException e) { + assertTrue( + "Message should explain resolved FileSystem URI and scheme options: " + e.getMessage(), + e.getMessage().contains("resolved against FileSystem URI") + && e.getMessage().contains("file://") + && e.getMessage().contains("hdfs://")); + } catch (IOException e) { + fail("Expected FileNotFoundException but caught " + e.getClass().getName()); + } + } + + // --- Tests for Finding 4: Enforce Size Cap on Inline JSON --- + + @Test + public void rejectsOversizedInlineJson() { + char[] chars = new char[(int) TableNameMapper.MAX_MAPPING_FILE_SIZE_BYTES + 10]; + Arrays.fill(chars, ' '); + String hugeJson = "{\"ns:t\": \"val\"" + new String(chars) + "}"; + + expectThrows( + IllegalArgumentException.class, () -> TableNameMapper.fromJson(hugeJson, "inline")); + } + + // --- Tests for Finding 5: Strict JSON Parsing --- + + @Test + public void rejectsNumberValueInJson() { + expectThrows( + IllegalArgumentException.class, () -> TableNameMapper.fromJson("{\"ns:t\": 123}", "test")); + } + + @Test + public void rejectsBooleanValueInJson() { + expectThrows( + IllegalArgumentException.class, () -> TableNameMapper.fromJson("{\"ns:t\": true}", "test")); + } + + @Test + public void rejectsNullValueInJson() { + expectThrows( + IllegalArgumentException.class, () -> TableNameMapper.fromJson("{\"ns:t\": null}", "test")); + } + + @Test + public void rejectsArrayValueInJson() { + expectThrows( + IllegalArgumentException.class, + () -> TableNameMapper.fromJson("{\"ns:t\": [\"target\"]}", "test")); + } + + @Test + public void rejectsNestedObjectValueInJson() { + expectThrows( + IllegalArgumentException.class, + () -> TableNameMapper.fromJson("{\"ns:t\": {\"inner\": \"val\"}}", "test")); + } + + @Test + public void rejectsDuplicateRawKeyInJson() { + expectThrows( + IllegalArgumentException.class, + () -> TableNameMapper.fromJson("{\"ns:t\": \"a\", \"ns:t\": \"b\"}", "test")); + } + + @Test + public void rejectsTrailingGarbageAfterJsonObject() { + expectThrows( + IllegalArgumentException.class, + () -> TableNameMapper.fromJson("{\"ns:t\": \"a\"} trailing", "test")); + } + + // --- Tests for Finding 6: Prefix Configuration Support --- + + @Test + public void loadsPrefixConfiguration() throws IOException { + Configuration conf = new Configuration(false); + conf.set(TABLE_NAME_MAPPING_PREFIX + "smartbrain:oraperi_new", "smartbrain_oraperi_new"); + conf.set(TABLE_NAME_MAPPING_PREFIX + "plain", "renamed-plain"); + + TableNameMapper mapper = TableNameMapper.fromConfiguration(conf); + + assertFalse(mapper.isIdentity()); + assertEquals("smartbrain_oraperi_new", mapper.map("smartbrain:oraperi_new")); + assertEquals("renamed-plain", mapper.map("plain")); + assertEquals("other:table", mapper.map("other:table")); + } + + @Test + public void prefixConfigurationNormalizesDefaultNamespace() throws IOException { + Configuration conf = new Configuration(false); + conf.set(TABLE_NAME_MAPPING_PREFIX + "default:plain", "renamed-plain"); + + TableNameMapper mapper = TableNameMapper.fromConfiguration(conf); + + assertEquals("renamed-plain", mapper.map("plain")); + assertEquals("renamed-plain", mapper.map("default:plain")); + } + + @Test + public void rejectsMutualExclusivityInlineAndPrefix() { + Configuration conf = new Configuration(false); + conf.set(TABLE_NAME_MAPPING_KEY, "{\"ns:t\": \"mapped\"}"); + conf.set(TABLE_NAME_MAPPING_PREFIX + "other", "mapped_other"); + + expectThrows(IllegalArgumentException.class, () -> TableNameMapper.fromConfiguration(conf)); + } + + @Test + public void rejectsMutualExclusivityFileAndPrefix() throws IOException { + File file = writeFile("mapping.json", MAPPING_JSON.getBytes(StandardCharsets.UTF_8)); + Configuration conf = new Configuration(false); + conf.set(TABLE_NAME_MAPPING_FILE_KEY, file.getAbsolutePath()); + conf.set(TABLE_NAME_MAPPING_PREFIX + "other", "mapped_other"); + + expectThrows(IllegalArgumentException.class, () -> TableNameMapper.fromConfiguration(conf)); + } + + @Test + public void rejectsPrefixWithInvalidTargetTableId() { + Configuration conf = new Configuration(false); + conf.set(TABLE_NAME_MAPPING_PREFIX + "ns:table", "-invalid-prefix"); + + expectThrows(IllegalArgumentException.class, () -> TableNameMapper.fromConfiguration(conf)); + } + + @Test + public void rejectsPrefixNToOneCollisions() { + Configuration conf = new Configuration(false); + conf.set(TABLE_NAME_MAPPING_PREFIX + "ns1:t", "shared_target"); + conf.set(TABLE_NAME_MAPPING_PREFIX + "ns2:t", "shared_target"); + + expectThrows(IllegalArgumentException.class, () -> TableNameMapper.fromConfiguration(conf)); + } + + @Test + public void rejectsPrefixConflictingDefaultNamespace() { + Configuration conf = new Configuration(false); + conf.set(TABLE_NAME_MAPPING_PREFIX + "plain", "target1"); + conf.set(TABLE_NAME_MAPPING_PREFIX + "default:plain", "target2"); + + expectThrows(IllegalArgumentException.class, () -> TableNameMapper.fromConfiguration(conf)); + } + + @Test + public void rejectsPrefixWithEmptyHBaseTableName() { + Configuration conf1 = new Configuration(false); + conf1.set(TABLE_NAME_MAPPING_PREFIX, "some_target"); + expectThrows(IllegalArgumentException.class, () -> TableNameMapper.fromConfiguration(conf1)); + + Configuration conf2 = new Configuration(false); + conf2.set(TABLE_NAME_MAPPING_PREFIX + " ", "some_target"); + expectThrows(IllegalArgumentException.class, () -> TableNameMapper.fromConfiguration(conf2)); + } + + /** JUnit 4.12 (pulled in transitively by hbase-client) does not have assertThrows. */ + private static void expectThrows(Class expected, ThrowingRunnable runnable) { + try { + runnable.run(); + } catch (Throwable t) { + if (expected.isInstance(t)) { + return; + } + throw new AssertionError( + "Expected " + expected.getName() + " but got " + t.getClass().getName(), t); + } + fail("Expected " + expected.getName() + " to be thrown."); + } + + private interface ThrowingRunnable { + void run() throws Throwable; + } +}