From 657200357dfbce142776352455a62735c62af07f Mon Sep 17 00:00:00 2001 From: Swarali Joshi Date: Tue, 16 Jun 2026 02:36:03 +0530 Subject: [PATCH 1/2] HBASE-29555 TestRollbackSCP.testFailAndRollback fails sometimes because non empty Scheduler queue TestRollbackSCP restarts the master ProcedureExecutor in place, reusing the same executor and MasterProcedureScheduler instances to simulate a failover. While the executor is being reloaded, other still-running threads of the live mini-cluster master can push a procedure back into the shared scheduler in the window between clearing it and ProcedureExecutor.load() asserting that it is empty, so load() intermittently fails with "scheduler queue not empty". There are two such producers: - the asyncTaskExecutor callback that wakes a procedure after an async meta update (e.g. persistToMeta), and - an incoming reportRegionStateTransition RPC from a live region server, handled on an RpcServer thread, which wakes a procedure via ProcedureEvent. Fix ProcedureTestingUtility.restart() (test-only): - wait for the already shutdown asyncTaskExecutor to fully terminate before clearing the scheduler, so any pending wake-up callback has finished; and - reload (clear -> store start -> init) in a bounded retry loop, retrying only when load() fails because the scheduler is not empty. ProcedureExecutor.stop() is explicitly safe to call after a failed init(), so this is a clean redo. Co-authored-by: Cursor --- .../procedure2/ProcedureTestingUtility.java | 70 ++++++++++++++++++- 1 file changed, 67 insertions(+), 3 deletions(-) diff --git a/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/ProcedureTestingUtility.java b/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/ProcedureTestingUtility.java index 1efab6f5d3a8..09e05bd8afbb 100644 --- a/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/ProcedureTestingUtility.java +++ b/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/ProcedureTestingUtility.java @@ -28,6 +28,8 @@ import java.util.ArrayList; import java.util.Set; import java.util.concurrent.Callable; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.TimeUnit; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; @@ -125,14 +127,50 @@ public static void restart(ProcedureExecutor procExecutor, stopAction.call(); } procExecutor.join(); - procExecutor.getScheduler().clear(); + // HBASE-29555: Callbacks that wake up a procedure after an async operation (such as updating + // meta) run on the asyncTaskExecutor, not on a PEWorker thread. ProcedureExecutor.stop() shuts + // the asyncTaskExecutor down, but ProcedureExecutor.join() only waits a bounded amount of time + // for it to terminate before giving up. If such a wake up task is still pending, it can call + // scheduler.addFront() after we clear the scheduler below and after the executor has been + // restarted, leaving the scheduler queue non-empty and tripping the Preconditions check in + // ProcedureExecutor.load(). Since we reuse the same executor (and scheduler) instance across + // this simulated restart, wait for the already shutdown asyncTaskExecutor to fully terminate so + // any pending wake up task has finished before we clear the scheduler. + awaitAsyncTaskExecutorTermination(procExecutor); // nothing running... // re-start LOG.info("RESTART - Start"); - procStore.start(storeThreads); - procExecutor.init(execThreads, failOnCorrupted); + // HBASE-29555: External events (for example a region server reporting a region state transition + // over RPC) can wake a procedure and push it back into the scheduler in the small window + // between clearing the scheduler and ProcedureExecutor.load() checking that it is empty. Those + // producers run on threads we do not own here (RPC handlers etc.), so we cannot reliably stop + // them from this thread. Instead, reload in a retry loop: if load() fails only because the + // scheduler is not empty, stop/drain/clear again and retry. ProcedureExecutor.stop() is + // explicitly safe to call after a failed init(), so this is a clean redo of the reload. + final int maxRestartAttempts = 20; + for (int attempt = 1;; attempt++) { + procExecutor.getScheduler().clear(); + procStore.start(storeThreads); + try { + procExecutor.init(execThreads, failOnCorrupted); + break; + } catch (IllegalArgumentException e) { + if ( + attempt >= maxRestartAttempts || e.getMessage() == null + || !e.getMessage().contains("scheduler queue not empty") + ) { + throw e; + } + LOG.warn("RESTART - scheduler was repopulated during reload (attempt {}/{}), retrying", + attempt, maxRestartAttempts, e); + procExecutor.stop(); + procStore.stop(abort); + procExecutor.join(); + awaitAsyncTaskExecutorTermination(procExecutor); + } + } if (actionBeforeStartWorker != null) { actionBeforeStartWorker.call(); } @@ -147,6 +185,32 @@ public static void restart(ProcedureExecutor procExecutor, } } + /** + * Wait for the (already shutdown) asyncTaskExecutor of the given executor to fully terminate, so + * that any pending callback (e.g. one that would wake up a procedure by adding it back to the + * scheduler) has finished running. See HBASE-29555 and the caller in + * {@link #restart(ProcedureExecutor, boolean, boolean, Callable, Callable, Callable, boolean, boolean)}. + */ + private static void awaitAsyncTaskExecutorTermination(ProcedureExecutor procExecutor) { + ExecutorService asyncTaskExecutor = procExecutor.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 restarting;" + + " there may still be pending async tasks"); + return; + } + } + } catch (InterruptedException e) { + LOG.warn("Interrupted while waiting for asyncTaskExecutor to terminate while restarting", e); + Thread.currentThread().interrupt(); + } + } + public static void storeRestart(ProcedureStore procStore, ProcedureStore.ProcedureLoader loader) throws Exception { storeRestart(procStore, false, loader); From 37ca44b82abaa4ba13aca6e3f8b123c435e35369 Mon Sep 17 00:00:00 2001 From: Swarali Joshi Date: Tue, 30 Jun 2026 11:21:21 +0530 Subject: [PATCH 2/2] HBASE-29555 Make TestRollbackSCP reload faithfully reproduce master failover Fix the flaky TestRollbackSCP.testFailAndRollback ("IllegalArgumentException: scheduler queue not empty" from ProcedureExecutor.load()) entirely within the test, by making the in-place procedure-executor reload reproduce the two guarantees a real master failover provides instead of relying on a retry: - Gap #1: after stopping the executor, wait for its asyncTaskExecutor to fully terminate before reloading, so a pending async meta-update wake-up cannot re-populate the scheduler during load(). A real failover starts a fresh process, so no such callback survives. - Gap #2: hold a fair ReentrantReadWriteLock write lock across the reload, and have reportRegionStateTransition take the read lock via non-blocking tryLock, rejecting reports with PleaseHoldException during the reload (the region server retries). This mirrors MasterRpcServices.checkServiceStarted(), which rejects region reports until the master finishes initializing, i.e. after load(). Acquiring the write lock also waits for any in-flight report. The shared ProcedureTestingUtility.restart() is left unchanged so the other tests that rely on it keep their strict scheduler-empty invariant. Co-authored-by: Cursor --- .../procedure2/ProcedureTestingUtility.java | 70 +----------- .../master/assignment/TestRollbackSCP.java | 106 +++++++++++++++++- 2 files changed, 108 insertions(+), 68 deletions(-) diff --git a/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/ProcedureTestingUtility.java b/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/ProcedureTestingUtility.java index 09e05bd8afbb..1efab6f5d3a8 100644 --- a/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/ProcedureTestingUtility.java +++ b/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/ProcedureTestingUtility.java @@ -28,8 +28,6 @@ import java.util.ArrayList; import java.util.Set; import java.util.concurrent.Callable; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.TimeUnit; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; @@ -127,50 +125,14 @@ public static void restart(ProcedureExecutor procExecutor, stopAction.call(); } procExecutor.join(); - // HBASE-29555: Callbacks that wake up a procedure after an async operation (such as updating - // meta) run on the asyncTaskExecutor, not on a PEWorker thread. ProcedureExecutor.stop() shuts - // the asyncTaskExecutor down, but ProcedureExecutor.join() only waits a bounded amount of time - // for it to terminate before giving up. If such a wake up task is still pending, it can call - // scheduler.addFront() after we clear the scheduler below and after the executor has been - // restarted, leaving the scheduler queue non-empty and tripping the Preconditions check in - // ProcedureExecutor.load(). Since we reuse the same executor (and scheduler) instance across - // this simulated restart, wait for the already shutdown asyncTaskExecutor to fully terminate so - // any pending wake up task has finished before we clear the scheduler. - awaitAsyncTaskExecutorTermination(procExecutor); + procExecutor.getScheduler().clear(); // nothing running... // re-start LOG.info("RESTART - Start"); - // HBASE-29555: External events (for example a region server reporting a region state transition - // over RPC) can wake a procedure and push it back into the scheduler in the small window - // between clearing the scheduler and ProcedureExecutor.load() checking that it is empty. Those - // producers run on threads we do not own here (RPC handlers etc.), so we cannot reliably stop - // them from this thread. Instead, reload in a retry loop: if load() fails only because the - // scheduler is not empty, stop/drain/clear again and retry. ProcedureExecutor.stop() is - // explicitly safe to call after a failed init(), so this is a clean redo of the reload. - final int maxRestartAttempts = 20; - for (int attempt = 1;; attempt++) { - procExecutor.getScheduler().clear(); - procStore.start(storeThreads); - try { - procExecutor.init(execThreads, failOnCorrupted); - break; - } catch (IllegalArgumentException e) { - if ( - attempt >= maxRestartAttempts || e.getMessage() == null - || !e.getMessage().contains("scheduler queue not empty") - ) { - throw e; - } - LOG.warn("RESTART - scheduler was repopulated during reload (attempt {}/{}), retrying", - attempt, maxRestartAttempts, e); - procExecutor.stop(); - procStore.stop(abort); - procExecutor.join(); - awaitAsyncTaskExecutorTermination(procExecutor); - } - } + procStore.start(storeThreads); + procExecutor.init(execThreads, failOnCorrupted); if (actionBeforeStartWorker != null) { actionBeforeStartWorker.call(); } @@ -185,32 +147,6 @@ public static void restart(ProcedureExecutor procExecutor, } } - /** - * Wait for the (already shutdown) asyncTaskExecutor of the given executor to fully terminate, so - * that any pending callback (e.g. one that would wake up a procedure by adding it back to the - * scheduler) has finished running. See HBASE-29555 and the caller in - * {@link #restart(ProcedureExecutor, boolean, boolean, Callable, Callable, Callable, boolean, boolean)}. - */ - private static void awaitAsyncTaskExecutorTermination(ProcedureExecutor procExecutor) { - ExecutorService asyncTaskExecutor = procExecutor.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 restarting;" - + " there may still be pending async tasks"); - return; - } - } - } catch (InterruptedException e) { - LOG.warn("Interrupted while waiting for asyncTaskExecutor to terminate while restarting", e); - Thread.currentThread().interrupt(); - } - } - public static void storeRestart(ProcedureStore procStore, ProcedureStore.ProcedureLoader loader) throws Exception { storeRestart(procStore, false, loader); diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/assignment/TestRollbackSCP.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/assignment/TestRollbackSCP.java index 14863515392f..315cdf131534 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/assignment/TestRollbackSCP.java +++ b/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 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 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);