fix: bidirectional replication flakey test (#4130)

Thank you for opening a Pull Request! Before submitting your PR, there are a few things you can do to make sure it goes smoothly: - [ x] Make sure to open an issue as a [bug/issue](https://togithub.com/googleapis/java-bigtable-hbase/issues/new/choose) before writing your code! That way we can discuss the change, evaluate designs, and agree on the general idea - [ x] Ensure the tests and linter pass - [ x] Code coverage does not decrease (if any source code was changed) - [ x] Appropriate docs were updated (if necessary) Fixes #4116 If you write sample code, please follow the [samples format]( https://togithub.com/GoogleCloudPlatform/java-docs-samples/blob/main/SAMPLE_FORMAT.md).

Author: georgecma

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

@@ -37,7 +37,6 @@
 import org.apache.hadoop.hbase.client.RowMutations;
 import org.apache.hadoop.hbase.client.Table;
 import org.apache.hadoop.hbase.client.replication.ReplicationAdmin;
-import org.apache.hadoop.hbase.replication.ReplicationEndpoint.ReplicateContext;
 import org.apache.hadoop.hbase.replication.ReplicationPeerConfig;
 import org.junit.After;
 import org.junit.AfterClass;
@@ -60,12 +59,12 @@ public class HbaseToCloudBigtableBidirectionalReplicationEndpointTest {
 
  public static class TestReplicationEndpoint extends HbaseToCloudBigtableReplicationEndpoint {
 
- static AtomicInteger replicatedEntries = new AtomicInteger();
+ static AtomicInteger replicatingEntries = new AtomicInteger();
 
  @Override
  public boolean replicate(ReplicateContext replicateContext) {
  boolean result = super.replicate(replicateContext);
- replicatedEntries.getAndAdd(replicateContext.getEntries().size());
+ replicatingEntries.getAndAdd(replicateContext.getEntries().size());
  return result;
  }
  }
@@ -141,7 +140,7 @@ public void setupTestCase() throws IOException {
  hbaseTable = hbaseConnection.getTable(table1);
 
  // Reset the entry counts for TestReplicationEndpoint
- TestReplicationEndpoint.replicatedEntries.set(0);
+ TestReplicationEndpoint.replicatingEntries.set(0);
  }
 
  private void createTables(TableName tableName, int cf1Scope, int cf2Scope) throws IOException {
@@ -185,15 +184,16 @@ public void testDropsReplicatedEntry() throws IOException, InterruptedException
  // Wait for replication
  TestUtils.waitForReplication(
  () -> {
- // Only one entry should've been replicated
- return TestReplicationEndpoint.replicatedEntries.get() >= 1;
+ // Both entries should've gone through the replication process
+ // However, one would've been filtered out during this process
+ return TestReplicationEndpoint.replicatingEntries.get() >= 2;
  });
 
  // Hbase table should have both mutations
- Assert.assertTrue(hbaseTable.get(new Get(TestUtils.ROW_KEY)).size() == 1);
- Assert.assertTrue(hbaseTable.get(new Get(TestUtils.ROW_KEY_2)).size() == 1);
+ Assert.assertEquals(1, hbaseTable.get(new Get(TestUtils.ROW_KEY)).size());
+ Assert.assertEquals(1, hbaseTable.get(new Get(TestUtils.ROW_KEY_2)).size());
  // Cbt table should have only one mutation
  Assert.assertTrue(cbtTable.get(new Get(TestUtils.ROW_KEY)).isEmpty());
- Assert.assertTrue(cbtTable.get(new Get(TestUtils.ROW_KEY_2)).size() == 1);
+ Assert.assertEquals(1, cbtTable.get(new Get(TestUtils.ROW_KEY_2)).size());
  }
 }

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

@@ -38,7 +38,6 @@
 import org.apache.hadoop.hbase.client.RowMutations;
 import org.apache.hadoop.hbase.client.Table;
 import org.apache.hadoop.hbase.client.replication.ReplicationAdmin;
-import org.apache.hadoop.hbase.replication.ReplicationEndpoint.ReplicateContext;
 import org.apache.hadoop.hbase.replication.ReplicationPeerConfig;
 import org.junit.After;
 import org.junit.AfterClass;
@@ -61,12 +60,13 @@ public class HbaseToCloudBigtableBidirectionalReplicationEndpointTest {
 
  public static class TestReplicationEndpoint extends HbaseToCloudBigtableReplicationEndpoint {
 
- static AtomicInteger replicatedEntries = new AtomicInteger();
+ static AtomicInteger replicatingEntries = new AtomicInteger();
 
  @Override
  public boolean replicate(ReplicateContext replicateContext) {
  boolean result = super.replicate(replicateContext);
- replicatedEntries.getAndAdd(replicateContext.getEntries().size());
+ replicatingEntries.getAndAdd(replicateContext.getEntries().size());
+
  return result;
  }
  }
@@ -144,9 +144,7 @@ public void setupTestCase() throws IOException {
  hbaseTable = hbaseConnection.getTable(table1);
 
  // Reset the entry counts for TestReplicationEndpoint
- HbaseToCloudBigtableBidirectionalReplicationEndpointTest.TestReplicationEndpoint
- .replicatedEntries
- .set(0);
+ TestReplicationEndpoint.replicatingEntries.set(0);
  }
 
  private void createTables(TableName tableName, int cf1Scope, int cf2Scope) throws IOException {
@@ -191,18 +189,16 @@ public void testDropsReplicatedEntry() throws IOException, InterruptedException
  // Wait for replication
  TestUtils.waitForReplication(
  () -> {
- // Only one entry should've been replicated
- return HbaseToCloudBigtableBidirectionalReplicationEndpointTest.TestReplicationEndpoint
- .replicatedEntries
- .get()
- >= 1;
+ // Both entries should've gone through the replication process
+ // However, one would've been filtered out during this process
+ return TestReplicationEndpoint.replicatingEntries.get() >= 2;
  });
 
  // Hbase table should have both mutations
- Assert.assertTrue(hbaseTable.get(new Get(TestUtils.ROW_KEY)).size() == 1);
- Assert.assertTrue(hbaseTable.get(new Get(TestUtils.ROW_KEY_2)).size() == 1);
+ Assert.assertEquals(1, hbaseTable.get(new Get(TestUtils.ROW_KEY)).size());
+ Assert.assertEquals(1, hbaseTable.get(new Get(TestUtils.ROW_KEY_2)).size());
  // Cbt table should have only one mutation
  Assert.assertTrue(cbtTable.get(new Get(TestUtils.ROW_KEY)).isEmpty());
- Assert.assertTrue(cbtTable.get(new Get(TestUtils.ROW_KEY_2)).size() == 1);
+ Assert.assertEquals(1, cbtTable.get(new Get(TestUtils.ROW_KEY_2)).size());
  }
 }