feat: remove base repl endpoint from hbase 2.x (#3539)

Remove dependency on BaseReplicationEndpoint and add dependency on shaded bigtable guava

Author: sushanb

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

@@ -16,10 +16,16 @@
 
 package com.google.cloud.bigtable.hbase1_x.replication;
 
+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;
+import static com.google.cloud.bigtable.hbase.replication.utils.TestUtils.assertEquals;
+
 import com.google.cloud.bigtable.emulator.v2.BigtableEmulatorRule;
 import com.google.cloud.bigtable.hbase.BigtableConfiguration;
 import com.google.cloud.bigtable.hbase.replication.utils.TestUtils;
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.UUID;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -37,12 +43,15 @@
 import org.apache.hadoop.hbase.client.Get;
 import org.apache.hadoop.hbase.client.Increment;
 import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.Result;
 import org.apache.hadoop.hbase.client.Table;
 import org.apache.hadoop.hbase.client.replication.ReplicationAdmin;
-import org.apache.hadoop.hbase.replication.BaseReplicationEndpoint;
+import org.apache.hadoop.hbase.replication.ChainWALEntryFilter;
 import org.apache.hadoop.hbase.replication.ReplicationException;
 import org.apache.hadoop.hbase.replication.ReplicationPeerConfig;
+import org.apache.hadoop.hbase.replication.WALEntryFilter;
 import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.wal.WAL.Entry;
 import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.Before;
@@ -57,52 +66,43 @@
 @RunWith(JUnit4.class)
 public class HbaseToCloudBigtableReplicationEndpointTest {
 
-  public static class TestReplicationEndpoint extends BaseReplicationEndpoint {
+  public static class TestReplicationEndpoint extends HbaseToCloudBigtableReplicationEndpoint {
 
     static AtomicInteger replicatedEntries = new AtomicInteger();
-    static HbaseToCloudBigtableReplicationEndpoint delegate;
-
-    public TestReplicationEndpoint() {
-      delegate = new HbaseToCloudBigtableReplicationEndpoint();
-    }
-
-    @Override
-    protected void doStart() {
-      try {
-        delegate.start().get();
-      } catch (Exception e) {
-        new RuntimeException("Failed to start Replication Endpoint.", e);
-      }
-      notifyStarted();
-    }
-
-    @Override
-    protected void doStop() {
-      try {
-        delegate.stop().get();
-      } catch (Exception e) {
-        new RuntimeException("Failed to stop Replication Endpoint.", e);
-      }
-      notifyStopped();
-    }
-
-    @Override
-    public UUID getPeerUUID() {
-      return delegate.getPeerUUID();
-    }
-
-    @Override
-    public void init(Context ctx) throws IOException {
-      super.init(ctx);
-      delegate.init(ctx);
-    }
 
     @Override
     public boolean replicate(ReplicateContext replicateContext) {
-      boolean result = delegate.replicate(replicateContext);
+      boolean result = super.replicate(replicateContext);
       replicatedEntries.getAndAdd(replicateContext.getEntries().size());
       return result;
     }
+
+    // return a WALEntry filter which accepts all rows except FILTERED_ROW_ENTRY
+    @Override
+    public WALEntryFilter getWALEntryfilter() {
+      return new ChainWALEntryFilter(
+          super.getWALEntryfilter(),
+          new WALEntryFilter() {
+            @Override
+            public Entry filter(Entry entry) {
+              ArrayList<Cell> cells = entry.getEdit().getCells();
+              int size = cells.size();
+              for (int i = size - 1; i >= 0; i--) {
+                Cell cell = cells.get(i);
+                if (Bytes.equals(
+                    cell.getRowArray(),
+                    cell.getRowOffset(),
+                    cell.getRowLength(),
+                    FILTERED_ROW_KEY,
+                    0,
+                    FILTERED_ROW_KEY.length)) {
+                  cells.remove(i);
+                }
+              }
+              return entry;
+            }
+          });
+    }
   }
 
   private static final Logger LOG =
@@ -155,8 +155,8 @@ public void setupTestCase() throws IOException {
     // Create and set the empty tables
     TableName table1 = TableName.valueOf(UUID.randomUUID().toString());
     TableName table2 = TableName.valueOf(UUID.randomUUID().toString());
-    createTables(table1);
-    createTables(table2);
+    createTables(table1, HConstants.REPLICATION_SCOPE_GLOBAL, HConstants.REPLICATION_SCOPE_GLOBAL);
+    createTables(table2, HConstants.REPLICATION_SCOPE_GLOBAL, HConstants.REPLICATION_SCOPE_GLOBAL);
 
     cbtTable = cbtConnection.getTable(table1);
     cbtTable2 = cbtConnection.getTable(table2);
@@ -167,7 +167,7 @@ public void setupTestCase() throws IOException {
     TestReplicationEndpoint.replicatedEntries.set(0);
   }
 
-  private void createTables(TableName tableName) throws IOException {
+  private void createTables(TableName tableName, int cf1Scope, int cf2Scope) throws IOException {
     // Create table in HBase
     HTableDescriptor htd = hbaseTestingUtil.createTableDescriptor(tableName.getNameAsString());
     HColumnDescriptor cf1 = new HColumnDescriptor(TestUtils.CF1);
@@ -178,8 +178,8 @@ private void createTables(TableName tableName) throws IOException {
     htd.addFamily(cf2);
 
     // Enables replication to all peers, including CBT
-    cf1.setScope(HConstants.REPLICATION_SCOPE_GLOBAL);
-    cf2.setScope(HConstants.REPLICATION_SCOPE_GLOBAL);
+    cf1.setScope(cf1Scope);
+    cf2.setScope(cf2Scope);
     hbaseTestingUtil.getHBaseAdmin().createTable(htd);
 
     cbtConnection.getAdmin().createTable(htd);
@@ -407,6 +407,75 @@ public void testWriteFailureToBigtableDoesNotStallReplication()
     Assert.assertEquals(0, actualCells.get(1).getTimestamp());
   }
 
+  @Test
+  public void testMutationReplicationWithWALEntryFilter() throws IOException, InterruptedException {
+    Put put1 = new Put(FILTERED_ROW_KEY);
+    put1.addColumn(TestUtils.CF1, TestUtils.COL_QUALIFIER, 0, FILTERED_ROW_KEY);
+    put1.addColumn(TestUtils.CF2, TestUtils.COL_QUALIFIER, 0, FILTERED_ROW_KEY);
+    hbaseTable.put(put1);
+
+    Put put2 = new Put(ROW_KEY);
+    put2.addColumn(TestUtils.CF1, TestUtils.COL_QUALIFIER, 0, ROW_KEY);
+    put2.addColumn(TestUtils.CF2, TestUtils.COL_QUALIFIER, 0, ROW_KEY);
+    hbaseTable.put(put2);
+
+    TestUtils.waitForReplication(
+        () -> {
+          //  replicate Entries will be one as FILTERED KEY will not be passed.
+          return TestReplicationEndpoint.replicatedEntries.get() >= 1;
+        });
+
+    Result cbtResult = cbtTable.get(new Get(FILTERED_ROW_KEY).setMaxVersions());
+    Result hbaseResult = hbaseTable.get(new Get(FILTERED_ROW_KEY).setMaxVersions());
+    Assert.assertTrue(cbtResult.isEmpty());
+    Assert.assertFalse(hbaseResult.isEmpty());
+    Assert.assertEquals(
+        "Number of cells , actual cells: " + hbaseResult.listCells(),
+        2,
+        2,
+        hbaseResult.listCells().size());
+
+    Result cbtResult2 = cbtTable.get(new Get(ROW_KEY).setMaxVersions());
+    Result hbaseResult2 = hbaseTable.get(new Get(ROW_KEY).setMaxVersions());
+    TestUtils.assertEquals(cbtResult2, hbaseResult2);
+  }
+
+  @Test
+  public void testReplicationWithScope() throws IOException, InterruptedException {
+    TableName tableName = TableName.valueOf(UUID.randomUUID().toString());
+    createTables(
+        tableName, HConstants.REPLICATION_SCOPE_GLOBAL, HConstants.REPLICATION_SCOPE_LOCAL);
+    hbaseTable = hbaseConnection.getTable(tableName);
+    cbtTable = cbtConnection.getTable(tableName);
+
+    Put put1 = new Put(ROW_KEY);
+    put1.addColumn(TestUtils.CF1, TestUtils.COL_QUALIFIER, 0, ROW_KEY);
+    put1.addColumn(TestUtils.CF2, TestUtils.COL_QUALIFIER_2, 0, ROW_KEY);
+    hbaseTable.put(put1);
+
+    TestUtils.waitForReplication(
+        () -> {
+          return TestReplicationEndpoint.replicatedEntries.get() >= 1;
+        });
+
+    Result cbtResult = cbtTable.get(new Get(ROW_KEY).setMaxVersions());
+    Result hbaseResult = hbaseTable.get(new Get(ROW_KEY).setMaxVersions());
+
+    // make sure we have replicated cells
+    Assert.assertFalse(cbtResult.isEmpty());
+    List<Cell> hbaseCells = hbaseResult.listCells();
+    List<Cell> cbtCells = cbtResult.listCells();
+
+    Assert.assertEquals("bigtable cells", 1, cbtCells.size());
+    // make sure that only CF1 is replicated
+    for (int i = 0; i < hbaseCells.size(); i++) {
+      // make sure CF1 is only replicated
+      if (CellUtil.cloneFamily(hbaseCells.get(i)) == CF1) {
+        assertEquals(hbaseCells.get(i), cbtCells.get(0));
+      }
+    }
+  }
+
   @Test
   public void testHBaseCBTTimestampTruncation() throws IOException, InterruptedException {
     Put put = new Put(TestUtils.ROW_KEY);
@@ -424,7 +493,9 @@ public void testHBaseCBTTimestampTruncation() throws IOException, InterruptedExc
     List<Cell> hbaseCells = hbaseTable.get(new Get(TestUtils.ROW_KEY).setMaxVersions()).listCells();
     List<Cell> bigtableCells =
         cbtTable.get(new Get(TestUtils.ROW_KEY).setMaxVersions()).listCells();
-    Assert.assertEquals("bigtable cells", 1, bigtableCells.size());
+    //   timestamp will get truncated and value will be overwritten at BIGTABLE_MAX_TIMESTAMP.
+    Assert.assertEquals(
+        "bigtable cells truncated at BIGTABLE_MAX_TIMESTAMP.", 1, bigtableCells.size());
     Assert.assertNotEquals(
         "Timestamp match for row " + TestUtils.ROW_KEY,
         hbaseCells.get(0).getTimestamp(),

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

@@ -17,35 +17,127 @@
 package com.google.cloud.bigtable.hbase2_x.replication;
 // TODO reverse the package names for 1.x and 2.x to com.google.cloud.bigtable.replication.hbase1_x
 
+import static com.google.cloud.bigtable.hbase.replication.configuration.HBaseToCloudBigtableReplicationConfiguration.BIGTABLE_REPLICATION_WALENTRYFILTER_CONFIG_KEY;
+
 import com.google.bigtable.repackaged.com.google.api.core.InternalExtensionOnly;
+import com.google.bigtable.repackaged.com.google.common.util.concurrent.AbstractService;
 import com.google.cloud.bigtable.hbase.replication.CloudBigtableReplicator;
 import com.google.cloud.bigtable.hbase.replication.adapters.BigtableWALEntry;
 import com.google.cloud.bigtable.hbase2_x.replication.metrics.HBaseMetricsExporter;
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
-import org.apache.hadoop.hbase.replication.BaseReplicationEndpoint;
+import org.apache.hadoop.hbase.replication.ChainWALEntryFilter;
+import org.apache.hadoop.hbase.replication.NamespaceTableCfWALEntryFilter;
+import org.apache.hadoop.hbase.replication.ReplicationEndpoint;
+import org.apache.hadoop.hbase.replication.ReplicationPeer;
+import org.apache.hadoop.hbase.replication.ReplicationPeerConfig;
+import org.apache.hadoop.hbase.replication.ScopeWALEntryFilter;
+import org.apache.hadoop.hbase.replication.WALEntryFilter;
 import org.apache.hadoop.hbase.wal.WAL;
+import org.apache.hbase.thirdparty.com.google.common.collect.Lists;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
-// TODO(remove BaseReplicationEndpoint extension).
 @InternalExtensionOnly
-public class HbaseToCloudBigtableReplicationEndpoint extends BaseReplicationEndpoint {
+public class HbaseToCloudBigtableReplicationEndpoint extends AbstractService
+    implements ReplicationEndpoint {
 
   private final CloudBigtableReplicator cloudBigtableReplicator;
   private final HBaseMetricsExporter metricsExporter;
+  protected Context context;
+
+  private static final Logger LOG =
+      LoggerFactory.getLogger(HbaseToCloudBigtableReplicationEndpoint.class);
 
   public HbaseToCloudBigtableReplicationEndpoint() {
     cloudBigtableReplicator = new CloudBigtableReplicator();
     metricsExporter = new HBaseMetricsExporter();
   }
 
+  // User of the class should use this class after calling init.
+  // https://github.com/apache/hbase/blob/master/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java#L321
+  @Override
+  public void init(Context context) throws IOException {
+    this.context = context;
+
+    if (this.context != null) {
+      ReplicationPeer peer = this.context.getReplicationPeer();
+      if (peer != null) {
+        peer.registerPeerConfigListener(this);
+      } else {
+        LOG.warn(
+            "Not tracking replication peer config changes for Peer Id "
+                + this.context.getPeerId()
+                + " because there's no such peer");
+      }
+    }
+  }
+
+  @Override
+  public boolean canReplicateToSameCluster() {
+    return false;
+  }
+
   @Override
   public UUID getPeerUUID() {
     return cloudBigtableReplicator.getPeerUUID();
   }
 
+  // the default filter excludes the system tables and already replicated table
+  // https://github.com/apache/hbase/blob/master/hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java#L330
+  // Keeping the implementation same as HBaseReplicationEndpoint
+  @Override
+  public WALEntryFilter getWALEntryfilter() {
+    ArrayList<WALEntryFilter> filters = Lists.newArrayList();
+    WALEntryFilter scopeFilter = getScopeWALEntryFilter();
+    if (scopeFilter != null) {
+      filters.add(scopeFilter);
+    }
+    WALEntryFilter tableCfFilter = getNamespaceTableCfWALEntryFilter();
+    if (tableCfFilter != null) {
+      filters.add(tableCfFilter);
+    }
+    if (context != null && context.getPeerConfig() != null) {
+      String filterNameCSV =
+          context
+              .getPeerConfig()
+              .getConfiguration()
+              .get(BIGTABLE_REPLICATION_WALENTRYFILTER_CONFIG_KEY);
+      if (filterNameCSV != null && !filterNameCSV.isEmpty()) {
+        String[] filterNames = filterNameCSV.split(",");
+        for (String filterName : filterNames) {
+          try {
+            Class<?> clazz = Class.forName(filterName);
+            filters.add((WALEntryFilter) clazz.getDeclaredConstructor().newInstance());
+          } catch (Exception e) {
+            LOG.error("Unable to create WALEntryFilter " + filterName, e);
+          }
+        }
+      }
+    }
+    return filters.isEmpty() ? null : new ChainWALEntryFilter(filters);
+  }
+
+  /**
+   * Returns a WALEntryFilter for checking the scope. Subclasses can return null if they don't want
+   * this filter
+   */
+  private WALEntryFilter getScopeWALEntryFilter() {
+    return new ScopeWALEntryFilter();
+  }
+
+  /**
+   * Returns a WALEntryFilter for checking replication per table and CF. Subclasses can return null
+   * if they don't want this filter
+   */
+  private WALEntryFilter getNamespaceTableCfWALEntryFilter() {
+    return new NamespaceTableCfWALEntryFilter(context.getReplicationPeer());
+  }
+
   @Override
   public boolean replicate(ReplicateContext replicateContext) {
     Map<String, List<BigtableWALEntry>> walEntriesByTable = new HashMap<>();
@@ -61,6 +153,11 @@ public boolean replicate(ReplicateContext replicateContext) {
     return cloudBigtableReplicator.replicate(walEntriesByTable);
   }
 
+  @Override
+  public boolean isStarting() {
+    return state() == State.STARTING;
+  }
+
   @Override
   public void start() {
     startAsync();
@@ -73,8 +170,8 @@ public void stop() {
 
   @Override
   protected void doStart() {
-    metricsExporter.setMetricsSource(ctx.getMetrics());
-    cloudBigtableReplicator.start(ctx.getConfiguration(), metricsExporter);
+    metricsExporter.setMetricsSource(context.getMetrics());
+    cloudBigtableReplicator.start(context.getConfiguration(), metricsExporter);
     notifyStarted();
   }
 
@@ -83,4 +180,9 @@ protected void doStop() {
     cloudBigtableReplicator.stop();
     notifyStopped();
   }
+  // It is a no op for HBaseReplicationEndpoint. Keeping as same.
+  @Override
+  public void peerConfigUpdated(ReplicationPeerConfig replicationPeerConfig) {}
+  // TODO(we can implement this to enable/disable dry-run mode without deleting the peer)
+
 }