HBASE-29555 TestRollbackSCP.testFailAndRollback fails sometimes because non empty Scheduler queue

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

@@ -25,9 +25,13 @@
 
 import java.io.IOException;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.HBaseTestingUtil;
+import org.apache.hadoop.hbase.PleaseHoldException;
 import org.apache.hadoop.hbase.ServerName;
 import org.apache.hadoop.hbase.StartTestingClusterOption;
 import org.apache.hadoop.hbase.TableName;
@@ -55,8 +59,12 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import org.apache.hadoop.hbase.shaded.protobuf.generated.ProcedureProtos.ProcedureState;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.RegionServerStatusProtos.ReportRegionStateTransitionRequest;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.RegionServerStatusProtos.ReportRegionStateTransitionResponse;
 
 /**
  * SCP does not support rollback actually, here we just want to simulate that when there is a code
@@ -67,6 +75,8 @@
 @Tag(LargeTests.TAG)
 public class TestRollbackSCP {
 
+  private static final Logger LOG = LoggerFactory.getLogger(TestRollbackSCP.class);
+
   private static final HBaseTestingUtil UTIL = new HBaseTestingUtil();
 
   private static final TableName TABLE_NAME = TableName.valueOf("test");
@@ -75,12 +85,45 @@ public class TestRollbackSCP {
 
   private static final AtomicBoolean INJECTED = new AtomicBoolean(false);
 
+  // HBASE-29555: guards the in-place procedure-executor reload against region servers concurrently
+  // reporting region state transitions. See restartMasterProcedureExecutorLikeFailover(). Fair so
+  // that, while the reload is waiting for / holding the write lock, new reports are rejected rather
+  // than barging in. A region state report takes the read lock; the reload takes the write lock.
+  private static final ReentrantReadWriteLock RELOAD_LOCK = new ReentrantReadWriteLock(true);
+
   private static final class AssignmentManagerForTest extends AssignmentManager {
 
     public AssignmentManagerForTest(MasterServices master, MasterRegion masterRegion) {
       super(master, masterRegion);
     }
 
+    @Override
+    public ReportRegionStateTransitionResponse reportRegionStateTransition(
+      ReportRegionStateTransitionRequest req) throws PleaseHoldException {
+      // HBASE-29555 (Gap #2): in production a starting/recovering master rejects region state
+      // reports (via HMaster.checkServiceStarted) until it has finished initializing, which happens
+      // only after the procedure executor has loaded. The test instead reloads the executor in place
+      // under a live master, so a report can wake a procedure and re-populate the scheduler in the
+      // window between clearing it and ProcedureExecutor.load() asserting it is empty. Mirror
+      // production: while the reload holds the write lock, reject reports with PleaseHoldException
+      // (the region server retries). tryLock (non-blocking) is used so a report never blocks the
+      // reload; acquiring the write lock in the reload still waits for any already in-flight report.
+      boolean locked = false;
+      try {
+        locked = RELOAD_LOCK.readLock().tryLock(0, TimeUnit.SECONDS);
+      } catch (InterruptedException e) {
+        Thread.currentThread().interrupt();
+      }
+      if (!locked) {
+        throw new PleaseHoldException("master procedure executor is reloading");
+      }
+      try {
+        return super.reportRegionStateTransition(req);
+      } finally {
+        RELOAD_LOCK.readLock().unlock();
+      }
+    }
+
     @Override
     CompletableFuture<Void> persistToMeta(RegionStateNode regionNode) {
       TransitRegionStateProcedure proc = regionNode.getProcedure();
@@ -160,6 +203,67 @@ public void describeTo(Description description) {
     };
   }
 
+  /**
+   * Wait for the (already shutdown) asyncTaskExecutor of the given executor to fully terminate.
+   * HBASE-29555 (Gap #1): callbacks that wake a procedure after an async operation (e.g.
+   * AssignmentManager#persistToMeta run on the executor's asyncTaskExecutor, not on a
+   * PEWorker. In production a master restart is a fresh process, so no such callback can survive
+   * into the reloaded executor. Here we reuse the same executor in-place, so a pending callback can
+   * call scheduler.addFront during the reload. ProcedureExecutor shuts the
+   * asyncTaskExecutor down; waiting for it to terminate guarantees any pending callback has run
+   * before we clear the scheduler, reproducing the "no async work survives the restart" guarantee.
+   */
+  private static void awaitAsyncTaskExecutorTermination(ProcedureExecutor<?> procExec) {
+    ExecutorService asyncTaskExecutor = procExec.getAsyncTaskExecutor();
+    if (asyncTaskExecutor == null) {
+      return;
+    }
+    long deadline = System.currentTimeMillis() + 60000;
+    try {
+      while (!asyncTaskExecutor.awaitTermination(1, TimeUnit.SECONDS)) {
+        if (System.currentTimeMillis() > deadline) {
+          LOG.warn("asyncTaskExecutor did not terminate in time while reloading the executor");
+          return;
+        }
+      }
+    } catch (InterruptedException e) {
+      LOG.warn("Interrupted while waiting for asyncTaskExecutor to terminate while reloading", e);
+      Thread.currentThread().interrupt();
+    }
+  }
+
+  /**
+   * Reload the master procedure executor in place (delegating the actual stop/clear/reload to
+   * MasterProcedureTestingUtility.restartMasterProcedureExecutor), but closing the two gaps
+   * that make the in-place reload diverge from a real master failover (HBASE-29555), so that the
+   * test reproduces production rather than relying on a retry:
+   * Gap #1 (no async work survives a real failover): stop the executor and wait for its
+   * asyncTaskExecutor to fully terminate before the reload, so a pending meta-update wake-up cannot
+   * re-populate the scheduler during load(). A real failover gets this for free because the
+   * old process (and its asyncTaskExecutor) is gone.
+   * Gap #2 (no region report during load): hold the reload write lock for the whole
+   * reload so that reportRegionStateTransition is rejected (and any in-flight report is
+   * waited for) -- exactly like a real master rejects reports via checkServiceStarted until
+   * it finishes initializing, which is after load()
+   */
+  private void restartMasterProcedureExecutorLikeFailover(
+    ProcedureExecutor<MasterProcedureEnv> procExec) throws Exception {
+    // Gap #2: block region reports for the duration of the reload. Acquiring the write lock waits for
+    // any in-flight report to finish first.
+    RELOAD_LOCK.writeLock().lock();
+    try {
+      // Gap #1: stop the executor (which shuts down the asyncTaskExecutor) and wait for the
+      // asyncTaskExecutor to fully terminate, so any pending wake-up has run before the reload
+      // clears the scheduler. The subsequent restartMasterProcedureExecutor will clear/reload, and
+      // its stop()/join() are safe to call again.
+      procExec.stop();
+      awaitAsyncTaskExecutorTermination(procExec);
+      MasterProcedureTestingUtility.restartMasterProcedureExecutor(procExec);
+    } finally {
+      RELOAD_LOCK.writeLock().unlock();
+    }
+  }
+
   @Test
   public void testFailAndRollback() throws Exception {
     HRegionServer rsWithMeta = UTIL.getRSForFirstRegionInTable(TableName.META_TABLE_NAME);
@@ -172,7 +276,7 @@ public void testFailAndRollback() throws Exception {
     UTIL.waitFor(30000, () -> !procExec.isRunning());
     // make sure that finally we could successfully rollback the procedure
     while (scp.getState() != ProcedureState.FAILED || !procExec.isRunning()) {
-      MasterProcedureTestingUtility.restartMasterProcedureExecutor(procExec);
+      restartMasterProcedureExecutorLikeFailover(procExec);
       ProcedureTestingUtility.waitProcedure(procExec, scp);
     }
     assertEquals(scp.getState(), ProcedureState.FAILED);