HBASE-30073 Test fixes for some flappers and a reproducible error (#8057) (#8082) (#8092)

Signed-off by: Charles Connell <cconnell@apache.org>

Author: haridsv

hbase-common/src/test/java/org/apache/hadoop/hbase/HBaseJupiterExtension.java

@@ -84,7 +84,7 @@ public class HBaseJupiterExtension implements InvocationInterceptor, BeforeAllCa
 
   private static final Map<String, Duration> TAG_TO_TIMEOUT =
     ImmutableMap.of(SmallTests.TAG, Duration.ofMinutes(3), MediumTests.TAG, Duration.ofMinutes(6),
-      LargeTests.TAG, Duration.ofMinutes(13), IntegrationTests.TAG, Duration.ZERO);
+      LargeTests.TAG, Duration.ofMinutes(20), IntegrationTests.TAG, Duration.ZERO);
 
   private static final String EXECUTOR = "executor";
 

hbase-compression/pom.xml

@@ -46,6 +46,11 @@
       <artifactId>hbase-resource-bundle</artifactId>
       <optional>true</optional>
     </dependency>
+    <dependency>
+      <groupId>org.junit.jupiter</groupId>
+      <artifactId>junit-jupiter-api</artifactId>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <build>

hbase-server/src/test/java/org/apache/hadoop/hbase/HBaseTestingUtility.java

@@ -489,7 +489,13 @@ private void createSubDirAndSystemProperty(String propertyName, Path parent, Str
 
     String sysValue = System.getProperty(propertyName);
 
-    if (sysValue != null) {
+    // Check if directory sharing should be disabled for this test.
+    // Tests that run with high parallelism and don't need shared directories can set this
+    // to avoid race conditions where one test's tearDown() deletes directories another test
+    // is still using.
+    boolean disableSharing = conf.getBoolean("hbase.test.disable-directory-sharing", false);
+
+    if (sysValue != null && !disableSharing) {
       // There is already a value set. So we do nothing but hope
       // that there will be no conflicts
       LOG.info("System.getProperty(\"" + propertyName + "\") already set to: " + sysValue
@@ -502,7 +508,7 @@ private void createSubDirAndSystemProperty(String propertyName, Path parent, Str
       }
       conf.set(propertyName, sysValue);
     } else {
-      // Ok, it's not set, so we create it as a subdirectory
+      // Ok, it's not set (or sharing is disabled), so we create it as a subdirectory
       createSubDir(propertyName, parent, subDirName);
       System.setProperty(propertyName, conf.get(propertyName));
     }

hbase-server/src/test/java/org/apache/hadoop/hbase/TestZooKeeper.java

@@ -77,7 +77,6 @@ public static void setUpBeforeClass() throws Exception {
     conf.setInt(HConstants.ZK_SESSION_TIMEOUT, 1000);
     conf.setClass(HConstants.HBASE_MASTER_LOADBALANCER_CLASS, MockLoadBalancer.class,
       LoadBalancer.class);
-    TEST_UTIL.startMiniDFSCluster(2);
   }
 
   @AfterAll

hbase-server/src/test/java/org/apache/hadoop/hbase/client/AbstractTestAsyncTableScan.java

@@ -46,6 +46,7 @@
 import java.util.stream.Stream;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.ConnectionRule;
+import org.apache.hadoop.hbase.HBaseConfiguration;
 import org.apache.hadoop.hbase.HBaseTestingUtility;
 import org.apache.hadoop.hbase.MatcherPredicate;
 import org.apache.hadoop.hbase.MiniClusterRule;
@@ -73,8 +74,21 @@
 public abstract class AbstractTestAsyncTableScan {
 
   protected static final OpenTelemetryClassRule OTEL_CLASS_RULE = OpenTelemetryClassRule.create();
-  protected static final MiniClusterRule MINI_CLUSTER_RULE = MiniClusterRule.newBuilder()
-    .setMiniClusterOption(StartMiniClusterOption.builder().numWorkers(3).build()).build();
+
+  private static Configuration createConfiguration() {
+    // Use HBaseConfiguration.create() instead of new Configuration() to properly load
+    // hbase-default.xml which contains required default values (e.g. for log cleaner plugins)
+    Configuration conf = HBaseConfiguration.create();
+    // Disable directory sharing to prevent race conditions when tests run in parallel.
+    // Each test instance gets its own isolated directories to avoid one test's tearDown()
+    // deleting directories another parallel test is still using.
+    conf.setBoolean("hbase.test.disable-directory-sharing", true);
+    return conf;
+  }
+
+  protected static final MiniClusterRule MINI_CLUSTER_RULE =
+    MiniClusterRule.newBuilder().setConfiguration(createConfiguration())
+      .setMiniClusterOption(StartMiniClusterOption.builder().numWorkers(3).build()).build();
 
   protected static final ConnectionRule CONN_RULE =
     ConnectionRule.createAsyncConnectionRule(MINI_CLUSTER_RULE::createAsyncConnection);

hbase-server/src/test/java/org/apache/hadoop/hbase/master/assignment/TestRollbackSCP.java

@@ -136,6 +136,11 @@ public static void tearDownAfterClass() throws IOException {
   @Before
   public void setUp() throws IOException {
     UTIL.ensureSomeNonStoppedRegionServersAvailable(2);
+    // Surefire reruns failed tests in the same JVM without re-running @BeforeClass. Reset injection
+    // state so compareAndSet in persistToMeta can succeed again and kill-before-store flags clear.
+    INJECTED.set(false);
+    ProcedureTestingUtility.setKillAndToggleBeforeStoreUpdateInRollback(
+      UTIL.getMiniHBaseCluster().getMaster().getMasterProcedureExecutor(), false);
   }
 
   private ServerCrashProcedure getSCPForServer(ServerName serverName) throws IOException {

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

@@ -259,6 +259,10 @@ public void setup() throws IOException {
     TEST_UTIL = HBaseTestingUtility.createLocalHTU();
     FILESYSTEM = TEST_UTIL.getTestFileSystem();
     CONF = TEST_UTIL.getConfiguration();
+    // Disable directory sharing to prevent race conditions when tests run in parallel.
+    // Each test instance gets its own isolated directories to avoid one test's tearDown()
+    // deleting directories another parallel test is still using.
+    CONF.setBoolean("hbase.test.disable-directory-sharing", true);
     NettyAsyncFSWALConfigHelper.setEventLoopConfig(CONF, GROUP, NioSocketChannel.class);
     dir = TEST_UTIL.getDataTestDir("TestHRegion").toString();
     method = name.getMethodName();
@@ -1352,18 +1356,23 @@ public void testGetWhileRegionClose() throws IOException {
         threads[i].start();
       }
     } finally {
+      done.set(true);
+      for (GetTillDoneOrException t : threads) {
+        if (t != null) {
+          try {
+            t.join(5000);
+          } catch (InterruptedException e) {
+            e.printStackTrace();
+          }
+        }
+      }
       if (this.region != null) {
         HBaseTestingUtility.closeRegionAndWAL(this.region);
         this.region = null;
       }
     }
-    done.set(true);
+    // Check for errors after threads have been stopped
     for (GetTillDoneOrException t : threads) {
-      try {
-        t.join();
-      } catch (InterruptedException e) {
-        e.printStackTrace();
-      }
       if (t.e != null) {
         LOG.info("Exception=" + t.e);
         assertFalse("Found a NPE in " + t.getName(), t.e instanceof NullPointerException);
@@ -1391,7 +1400,7 @@ class GetTillDoneOrException extends Thread {
 
     @Override
     public void run() {
-      while (!this.done.get()) {
+      while (!this.done.get() && !Thread.currentThread().isInterrupted()) {
         try {
           assertTrue(region.get(g).size() > 0);
           this.count.incrementAndGet();
@@ -4517,7 +4526,7 @@ public void checkNoError() {
     @Override
     public void run() {
       done = false;
-      while (!done) {
+      while (!done && !Thread.currentThread().isInterrupted()) {
         synchronized (this) {
           try {
             wait();
@@ -4730,7 +4739,7 @@ public void checkNoError() {
     @Override
     public void run() {
       done = false;
-      while (!done) {
+      while (!done && !Thread.currentThread().isInterrupted()) {
         try {
           for (int r = 0; r < numRows; r++) {
             byte[] row = Bytes.toBytes("row" + r);
@@ -5264,7 +5273,7 @@ public void testParallelIncrementWithMemStoreFlush() throws Exception {
     Runnable flusher = new Runnable() {
       @Override
       public void run() {
-        while (!incrementDone.get()) {
+        while (!incrementDone.get() && !Thread.currentThread().isInterrupted()) {
           try {
             region.flush(true);
           } catch (Exception e) {
@@ -5280,13 +5289,38 @@ public void run() {
     long expected = (long) threadNum * incCounter;
     Thread[] incrementers = new Thread[threadNum];
     Thread flushThread = new Thread(flusher);
+    flushThread.setName("FlushThread-" + method);
     for (int i = 0; i < threadNum; i++) {
       incrementers[i] = new Thread(new Incrementer(this.region, incCounter));
       incrementers[i].start();
     }
     flushThread.start();
-    for (int i = 0; i < threadNum; i++) {
-      incrementers[i].join();
+    try {
+      for (int i = 0; i < threadNum; i++) {
+        incrementers[i].join();
+      }
+
+      incrementDone.set(true);
+      flushThread.join();
+
+      Get get = new Get(Incrementer.incRow);
+      get.addColumn(Incrementer.family, Incrementer.qualifier);
+      get.readVersions(1);
+      Result res = this.region.get(get);
+      List<Cell> kvs = res.getColumnCells(Incrementer.family, Incrementer.qualifier);
+
+      // we just got the latest version
+      assertEquals(1, kvs.size());
+      Cell kv = kvs.get(0);
+      assertEquals(expected, Bytes.toLong(kv.getValueArray(), kv.getValueOffset()));
+    } finally {
+      // Ensure flush thread is stopped even if test fails or times out
+      incrementDone.set(true);
+      flushThread.interrupt();
+      flushThread.join(5000); // Wait up to 5 seconds for thread to stop
+      if (flushThread.isAlive()) {
+        LOG.warn("Flush thread did not stop within timeout for test " + method);
+      }
     }
 
     incrementDone.set(true);
@@ -5349,7 +5383,7 @@ public void testParallelAppendWithMemStoreFlush() throws Exception {
     Runnable flusher = new Runnable() {
       @Override
       public void run() {
-        while (!appendDone.get()) {
+        while (!appendDone.get() && !Thread.currentThread().isInterrupted()) {
           try {
             region.flush(true);
           } catch (Exception e) {
@@ -5369,13 +5403,41 @@ public void run() {
     }
     Thread[] appenders = new Thread[threadNum];
     Thread flushThread = new Thread(flusher);
+    flushThread.setName("FlushThread-" + method);
     for (int i = 0; i < threadNum; i++) {
       appenders[i] = new Thread(new Appender(this.region, appendCounter));
       appenders[i].start();
     }
     flushThread.start();
-    for (int i = 0; i < threadNum; i++) {
-      appenders[i].join();
+    try {
+      for (int i = 0; i < threadNum; i++) {
+        appenders[i].join();
+      }
+
+      appendDone.set(true);
+      flushThread.join();
+
+      Get get = new Get(Appender.appendRow);
+      get.addColumn(Appender.family, Appender.qualifier);
+      get.readVersions(1);
+      Result res = this.region.get(get);
+      List<Cell> kvs = res.getColumnCells(Appender.family, Appender.qualifier);
+
+      // we just got the latest version
+      assertEquals(1, kvs.size());
+      Cell kv = kvs.get(0);
+      byte[] appendResult = new byte[kv.getValueLength()];
+      System.arraycopy(kv.getValueArray(), kv.getValueOffset(), appendResult, 0,
+        kv.getValueLength());
+      assertArrayEquals(expected, appendResult);
+    } finally {
+      // Ensure flush thread is stopped even if test fails or times out
+      appendDone.set(true);
+      flushThread.interrupt();
+      flushThread.join(5000); // Wait up to 5 seconds for thread to stop
+      if (flushThread.isAlive()) {
+        LOG.warn("Flush thread did not stop within timeout for test " + method);
+      }
     }
 
     appendDone.set(true);
@@ -7337,7 +7399,7 @@ public void testMutateRowInParallel() throws Exception {
     // Writer thread
     Thread writerThread = new Thread(() -> {
       try {
-        while (true) {
+        while (!Thread.currentThread().isInterrupted()) {
           // If all the reader threads finish, then stop the writer thread
           if (latch.await(0, TimeUnit.MILLISECONDS)) {
             return;
@@ -7362,15 +7424,19 @@ public void testMutateRowInParallel() throws Exception {
             .addColumn(fam1, q3, tsIncrement + 1, Bytes.toBytes(1L))
             .addColumn(fam1, q4, tsAppend + 1, Bytes.toBytes("a")) });
         }
+      } catch (InterruptedException e) {
+        // Test interrupted, exit gracefully
+        Thread.currentThread().interrupt();
       } catch (Exception e) {
         assertionError.set(new AssertionError(e));
       }
     });
+    writerThread.setName("WriterThread-" + method);
     writerThread.start();
 
     // Reader threads
     for (int i = 0; i < numReaderThreads; i++) {
-      new Thread(() -> {
+      Thread readerThread = new Thread(() -> {
         try {
           for (int j = 0; j < 10000; j++) {
             // Verify the values
@@ -7399,13 +7465,24 @@ public void testMutateRowInParallel() throws Exception {
         }
 
         latch.countDown();
-      }).start();
+      });
+      readerThread.setName("ReaderThread-" + i + "-" + method);
+      readerThread.start();
     }
 
-    writerThread.join();
+    try {
+      writerThread.join();
 
-    if (assertionError.get() != null) {
-      throw assertionError.get();
+      if (assertionError.get() != null) {
+        throw assertionError.get();
+      }
+    } finally {
+      // Ensure writer thread is stopped on test timeout
+      writerThread.interrupt();
+      writerThread.join(5000);
+      if (writerThread.isAlive()) {
+        LOG.warn("Writer thread did not stop within timeout for test " + method);
+      }
     }
   }
 

hbase-server/src/test/java/org/apache/hadoop/hbase/replication/TestReplicationBase.java

@@ -183,6 +183,10 @@ protected static void loadData(String prefix, byte[] row, byte[] familyName) thr
   protected static void setupConfig(HBaseTestingUtility util, String znodeParent) {
     Configuration conf = util.getConfiguration();
     conf.set(HConstants.ZOOKEEPER_ZNODE_PARENT, znodeParent);
+    // Disable directory sharing to prevent race conditions when tests run in parallel.
+    // Each test instance gets its own isolated directories to avoid one test's tearDown()
+    // deleting directories another parallel test is still using.
+    conf.setBoolean("hbase.test.disable-directory-sharing", true);
     // We don't want too many edits per batch sent to the ReplicationEndpoint to trigger
     // sufficient number of events. But we don't want to go too low because
     // HBaseInterClusterReplicationEndpoint partitions entries into batches and we want