Add tests

Author: Jaehui-Lee

hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestBulkLoadReplication.java

@@ -61,6 +61,9 @@
 import org.apache.hadoop.hbase.coprocessor.RegionCoprocessor;
 import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
 import org.apache.hadoop.hbase.coprocessor.RegionObserver;
+import org.apache.hadoop.hbase.coprocessor.RegionServerCoprocessor;
+import org.apache.hadoop.hbase.coprocessor.RegionServerCoprocessorEnvironment;
+import org.apache.hadoop.hbase.coprocessor.RegionServerObserver;
 import org.apache.hadoop.hbase.io.hfile.HFile;
 import org.apache.hadoop.hbase.io.hfile.HFileContextBuilder;
 import org.apache.hadoop.hbase.replication.ReplicationPeerConfig;
@@ -113,6 +116,9 @@ public class TestBulkLoadReplication extends TestReplicationBase {
   private static AtomicInteger BULK_LOADS_COUNT;
   private static CountDownLatch BULK_LOAD_LATCH;
 
+  private static AtomicInteger REPLICATION_COUNT;
+  private static CountDownLatch REPLICATION_LATCH;
+
   protected static final HBaseTestingUtility UTIL3 = new HBaseTestingUtility();
   protected static final Configuration CONF3 = UTIL3.getConfiguration();
 
@@ -124,7 +130,7 @@ public class TestBulkLoadReplication extends TestReplicationBase {
   @ClassRule
   public static TemporaryFolder testFolder = new TemporaryFolder();
 
-  private static ReplicationQueueStorage queueStorage;
+  private static final List<ReplicationQueueStorage> queueStorages = new ArrayList<>();
 
   private static boolean replicationPeersAdded = false;
 
@@ -136,8 +142,11 @@ public static void setUpBeforeClass() throws Exception {
     setupConfig(UTIL3, "/3");
     TestReplicationBase.setUpBeforeClass();
     startThirdCluster();
-    queueStorage = ReplicationStorageFactory.getReplicationQueueStorage(UTIL1.getZooKeeperWatcher(),
-      UTIL1.getConfiguration());
+    for (HBaseTestingUtility util : new HBaseTestingUtility[] { UTIL1, UTIL2, UTIL3 }) {
+      ReplicationQueueStorage queueStorage = ReplicationStorageFactory
+        .getReplicationQueueStorage(util.getZooKeeperWatcher(), util.getConfiguration());
+      queueStorages.add(queueStorage);
+    }
   }
 
   private static void startThirdCluster() throws Exception {
@@ -183,6 +192,7 @@ public void setUpBase() throws Exception {
     }
 
     BULK_LOADS_COUNT = new AtomicInteger(0);
+    REPLICATION_COUNT = new AtomicInteger(0);
   }
 
   private ReplicationPeerConfig getPeerConfigForCluster(HBaseTestingUtility util) {
@@ -191,21 +201,37 @@ private ReplicationPeerConfig getPeerConfigForCluster(HBaseTestingUtility util)
   }
 
   private void setupCoprocessor(HBaseTestingUtility cluster) {
+    Configuration conf = cluster.getConfiguration();
+    String clusterKey = cluster.getClusterKey();
+    Class<TestBulkLoadReplication.BulkReplicationTestObserver> cpClass =
+      TestBulkLoadReplication.BulkReplicationTestObserver.class;
+
     cluster.getHBaseCluster().getRegions(tableName).forEach(r -> {
       try {
-        TestBulkLoadReplication.BulkReplicationTestObserver cp = r.getCoprocessorHost()
-          .findCoprocessor(TestBulkLoadReplication.BulkReplicationTestObserver.class);
+        RegionCoprocessorHost cpHost = r.getCoprocessorHost();
+        TestBulkLoadReplication.BulkReplicationTestObserver cp = cpHost.findCoprocessor(cpClass);
         if (cp == null) {
-          r.getCoprocessorHost().load(TestBulkLoadReplication.BulkReplicationTestObserver.class, 0,
-            cluster.getConfiguration());
-          cp = r.getCoprocessorHost()
-            .findCoprocessor(TestBulkLoadReplication.BulkReplicationTestObserver.class);
-          cp.clusterName = cluster.getClusterKey();
+          cpHost.load(cpClass, 0, conf);
+          cp = cpHost.findCoprocessor(cpClass);
+          cp.clusterName = clusterKey;
         }
       } catch (Exception e) {
         LOG.error(e.getMessage(), e);
       }
     });
+
+    try {
+      RegionServerCoprocessorHost cpHost =
+        cluster.getHBaseCluster().getRegionServer(0).getRegionServerCoprocessorHost();
+      TestBulkLoadReplication.BulkReplicationTestObserver cp = cpHost.findCoprocessor(cpClass);
+      if (cp == null) {
+        cpHost.load(cpClass, 0, conf);
+        cp = cpHost.findCoprocessor(cpClass);
+        cp.clusterName = clusterKey;
+      }
+    } catch (Exception e) {
+      LOG.error(e.getMessage(), e);
+    }
   }
 
   protected static void setupBulkLoadConfigsForCluster(Configuration config,
@@ -241,13 +267,21 @@ public void testBulkLoadReplicationActiveActive() throws Exception {
     // Each event gets 3 counts (the originator cluster, plus the two peers),
     // so BULK_LOADS_COUNT expected value is 3 * 3 = 9.
     assertEquals(9, BULK_LOADS_COUNT.get());
+    // Each bulk load event gets replicated twice (to two peers),
+    // so REPLICATION_COUNT expected value is 3 * 2 = 6.
+    assertEquals(6, REPLICATION_COUNT.get());
+    for (ReplicationQueueStorage queueStorage : queueStorages) {
+      assertEquals(0, queueStorage.getAllHFileRefs().size());
+    }
   }
 
   protected void assertBulkLoadConditions(TableName tableName, byte[] row, byte[] value,
     HBaseTestingUtility utility, Table... tables) throws Exception {
     BULK_LOAD_LATCH = new CountDownLatch(3);
+    REPLICATION_LATCH = new CountDownLatch(2);
     bulkLoadOnCluster(tableName, row, value, utility);
     assertTrue(BULK_LOAD_LATCH.await(1, TimeUnit.MINUTES));
+    assertTrue(REPLICATION_LATCH.await(1, TimeUnit.MINUTES));
     assertTableHasValue(tables[0], row, value);
     assertTableHasValue(tables[1], row, value);
     assertTableHasValue(tables[2], row, value);
@@ -305,10 +339,12 @@ private String createHFileForFamilies(byte[] row, byte[] value, Configuration cl
     return hFileLocation.getAbsoluteFile().getAbsolutePath();
   }
 
-  public static class BulkReplicationTestObserver implements RegionCoprocessor {
+  public static class BulkReplicationTestObserver
+    implements RegionCoprocessor, RegionServerCoprocessor {
 
     String clusterName;
     AtomicInteger bulkLoadCounts = new AtomicInteger();
+    AtomicInteger replicationCount = new AtomicInteger();
 
     @Override
     public Optional<RegionObserver> getRegionObserver() {
@@ -325,6 +361,22 @@ public void postBulkLoadHFile(ObserverContext<RegionCoprocessorEnvironment> ctx,
         }
       });
     }
+
+    @Override
+    public Optional<RegionServerObserver> getRegionServerObserver() {
+      return Optional.of(new RegionServerObserver() {
+
+        // FIXME: this is deprecated, but we need it for validating bulk load replication
+        @Override
+        public void
+          postReplicateLogEntries(final ObserverContext<RegionServerCoprocessorEnvironment> ctx) {
+          REPLICATION_LATCH.countDown();
+          REPLICATION_COUNT.incrementAndGet();
+          LOG.info("Replication succeeded. Total for {}: {}", clusterName,
+            replicationCount.addAndGet(1));
+        }
+      });
+    }
   }
 
   @Test
@@ -339,7 +391,11 @@ public void testBulkloadReplicationActiveActiveForNoRepFamily() throws Exception
     // additional wait to make sure no extra bulk load happens
     Thread.sleep(400);
     assertEquals(1, BULK_LOADS_COUNT.get());
-    assertEquals(0, queueStorage.getAllHFileRefs().size());
+    // No replication should happen for no-replication family
+    assertEquals(0, REPLICATION_COUNT.get());
+    for (ReplicationQueueStorage queueStorage : queueStorages) {
+      assertEquals(0, queueStorage.getAllHFileRefs().size());
+    }
   }
 
   private void assertBulkLoadConditionsForNoRepFamily(byte[] row, byte[] value,