diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java index 5f20e4feb39302..93ed130d7c0bbf 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java +++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java @@ -432,12 +432,14 @@ public class Env { protected boolean isFirstTimeStartUp = false; protected boolean isElectable; - // set to true after finished replay all meta and ready to serve - // set to false when catalog is not ready. + // Metadata readiness, updated by the replayer independently of startup initialization. private AtomicBoolean isReady = new AtomicBoolean(false); + // Published after the first successful MASTER/FOLLOWER/OBSERVER initialization and FE type commit. + // Keep this true across UNKNOWN transitions so initialized nodes retain their existing read policy. + private volatile boolean startupInitialized = false; // set to true after http server start private AtomicBoolean httpReady = new AtomicBoolean(false); - // set to true if FE can offer READ service. + // Metadata read eligibility; serving reads also requires startupInitialized. // canRead can be true even if isReady is false. // for example: OBSERVER transfer to UNKNOWN, then isReady will be set to false, but canRead can still be true private AtomicBoolean canRead = new AtomicBoolean(false); @@ -1304,13 +1306,18 @@ public void waitForReady() throws InterruptedException { Thread.sleep(100); if (counter++ % 100 == 0) { String reason = editLog == null ? "editlog is null" : editLog.getNotReadyReason(); - LOG.info("wait catalog to be ready. feType:{} isReady:{}, counter:{} reason: {}", - feType, isReady.get(), counter, reason); + LOG.info("wait catalog to be ready. feType:{} metadataReady:{} startupInitialized:{}, " + + "counter:{} reason: {}", + feType, isMetadataReady(), startupInitialized, counter, reason); } } } public boolean isReady() { + return startupInitialized && isMetadataReady(); + } + + private boolean isMetadataReady() { return isReady.get(); } @@ -1953,7 +1960,8 @@ void advanceNextId() { */ public boolean postProcessAfterMetadataReplayed(boolean waitCatalogReady) { if (waitCatalogReady) { - while (!isReady()) { + // Startup initialization itself must not wait for the serving gate that it will open. + while (!isMetadataReady()) { // Avoid endless waiting if the state has changed. // // Consider the following situation: @@ -2134,7 +2142,7 @@ private boolean transferToNonMaster(FrontendNodeType newType) { replayer.start(); } - // 'isReady' will be set to true in 'setCanRead()' method + // The replayer publishes metadata readiness before startup initialization completes. if (!postProcessAfterMetadataReplayed(true)) { // A newer BDB state is already waiting in typeTransferQueue. Abort this stale transition so the // state listener can process the newer state instead of waiting indefinitely for this node to @@ -2210,7 +2218,7 @@ private void initLowerCaseTableNames() { // After the cluster initialization is complete, 'lower_case_table_names' can not be modified during the cluster // restart or upgrade. private void checkLowerCaseTableNames() { - while (!isReady()) { + while (!isMetadataReady()) { // Waiting for lower_case_table_names to initialize value from image or editlog. try { LOG.info("Waiting for \'lower_case_table_names\' initialization."); @@ -3264,7 +3272,12 @@ public void notifyNewFETypeTransfer(FrontendNodeType newType) { } public void startStateListener() { - listener = new Daemon("stateListener", STATE_CHANGE_CHECK_INTERVAL_MS) { + listener = createStateListener(); + listener.start(); + } + + Daemon createStateListener() { + Daemon stateListener = new Daemon("stateListener", STATE_CHANGE_CHECK_INTERVAL_MS) { @Override protected synchronized void runOneCycle() { @@ -3378,13 +3391,18 @@ protected synchronized void runOneCycle() { continue; } feType = newType; + // INIT -> UNKNOWN is a completed no-op, not a completed startup initialization. + if (newType == FrontendNodeType.MASTER || newType == FrontendNodeType.FOLLOWER + || newType == FrontendNodeType.OBSERVER) { + startupInitialized = true; + } LOG.info("finished to transfer FE type to {}", feType); } } // end runOneCycle }; - listener.setMetaContext(metaContext); - listener.start(); + stateListener.setMetaContext(metaContext); + return stateListener; } public synchronized boolean replayJournal(long toJournalId) { @@ -5535,7 +5553,7 @@ public void setMaster(MasterInfo info) { } public boolean canRead() { - return this.canRead.get(); + return startupInitialized && canRead.get(); } public boolean isElectable() { diff --git a/fe/fe-core/src/test/java/org/apache/doris/catalog/EnvStateListenerTest.java b/fe/fe-core/src/test/java/org/apache/doris/catalog/EnvStateListenerTest.java index 0a41817f74b073..cf069df3abfeec 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/catalog/EnvStateListenerTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/catalog/EnvStateListenerTest.java @@ -17,18 +17,36 @@ package org.apache.doris.catalog; +import org.apache.doris.common.Config; import org.apache.doris.common.util.Daemon; +import org.apache.doris.datasource.CatalogMgr; import org.apache.doris.ha.FrontendNodeType; +import org.apache.doris.metric.MetricRepo; +import org.apache.doris.mysql.privilege.Auth; +import org.apache.doris.statistics.analysis.AnalysisManager; +import org.apache.doris.statistics.analysis.FollowerColumnSender; +import org.apache.doris.statistics.cache.StatisticsCache; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; +import org.mockito.MockedConstruction; +import org.mockito.MockedStatic; import org.mockito.Mockito; +import org.mockito.stubbing.Answer; import java.lang.reflect.Field; +import java.lang.reflect.Method; +import java.util.ArrayList; +import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +@Timeout(30) public class EnvStateListenerTest { @Test public void testInterruptedNonMasterTransitionDoesNotCommitFeType() throws Exception { @@ -63,9 +81,149 @@ public void testInterruptedNonMasterTransitionDoesNotCommitFeType() throws Excep Assertions.assertEquals(FrontendNodeType.INIT, env.getFeType()); } finally { stateListener.exit(); + env.notifyNewFETypeTransfer(env.getFeType()); + stateListener.join(5000); + Assertions.assertFalse(stateListener.isAlive()); } } + @ParameterizedTest + @CsvSource({"INIT, FOLLOWER", "INIT, OBSERVER", "UNKNOWN, FOLLOWER", "UNKNOWN, OBSERVER"}) + public void testUnknownInterruptionKeepsStartupGateClosedUntilRetryCompletes( + FrontendNodeType initialType, FrontendNodeType targetType) throws Exception { + Env env = Mockito.spy(new Env(false)); + setField(env, "feType", initialType); + Mockito.doReturn(false).when(env).replayJournal(-1); + env.createReplayer(); + Daemon replayer = (Daemon) getField(env, "replayer"); + Daemon stateListener = env.createStateListener(); + + Auth auth = Mockito.mock(Auth.class); + setField(env, "auth", auth); + CatalogMgr catalogMgr = Mockito.spy(env.getCatalogMgr()); + setField(env, "catalogMgr", catalogMgr); + AnalysisManager analysisManager = Mockito.mock(AnalysisManager.class); + StatisticsCache statisticsCache = Mockito.mock(StatisticsCache.class); + Mockito.when(analysisManager.getStatisticsCache()).thenReturn(statisticsCache); + setField(env, "analysisManager", analysisManager); + + List servingDuringInitialization = new ArrayList<>(); + Answer recordServingState = invocation -> { + servingDuringInitialization.add(env.isReady() || env.canRead()); + return null; + }; + Mockito.doAnswer(recordServingState).when(env).startNonMasterDaemonThreads(); + Mockito.doAnswer(recordServingState).when(statisticsCache).preHeat(); + + AtomicInteger transitionAttempts = new AtomicInteger(); + Mockito.doAnswer(invocation -> { + if (transitionAttempts.incrementAndGet() == 1) { + boolean completed = (boolean) invocation.callRealMethod(); + // Run the real replayer readiness update after UNKNOWN interrupts the wait, before + // the listener handles UNKNOWN. This deterministically reproduces the unsafe interleaving. + replayFreshMetadata(env, replayer); + return completed; + } + // transferToNonMaster resets metadata readiness on every attempt. Let the replayer + // catch up again before the retry executes the real metadata post-processing. + replayFreshMetadata(env, replayer); + return invocation.callRealMethod(); + }).when(env).postProcessAfterMetadataReplayed(true); + + try (MockedStatic metrics = Mockito.mockStatic(MetricRepo.class); + MockedConstruction senders = Mockito.mockConstruction( + FollowerColumnSender.class, + (sender, context) -> Mockito.doAnswer(recordServingState).when(sender).start())) { + metrics.when(MetricRepo::init).thenAnswer(recordServingState); + + env.notifyNewFETypeTransfer(targetType); + env.notifyNewFETypeTransfer(FrontendNodeType.UNKNOWN); + // An equal event ends runOneCycle. For INIT, first commit UNKNOWN; for an initial + // UNKNOWN, the interruption event itself is already equal to the retained FE type. + if (initialType == FrontendNodeType.INIT) { + env.notifyNewFETypeTransfer(FrontendNodeType.UNKNOWN); + } + runOneCycle(stateListener); + + Assertions.assertEquals(FrontendNodeType.UNKNOWN, env.getFeType()); + Assertions.assertEquals(1, transitionAttempts.get()); + Assertions.assertTrue(((AtomicBoolean) getField(env, "isReady")).get()); + Assertions.assertTrue(((AtomicBoolean) getField(env, "canRead")).get()); + Assertions.assertFalse(env.isReady()); + Assertions.assertFalse(env.canRead()); + Mockito.verify(auth, Mockito.never()).rectifyPrivs(); + Mockito.verify(catalogMgr, Mockito.never()).registerCatalogRefreshListener(env); + Mockito.verify(env, Mockito.never()).startNonMasterDaemonThreads(); + Mockito.verify(statisticsCache, Mockito.never()).preHeat(); + metrics.verify(MetricRepo::init, Mockito.never()); + Assertions.assertTrue(senders.constructed().isEmpty()); + + replayFreshMetadata(env, replayer); + Assertions.assertFalse(env.isReady()); + Assertions.assertFalse(env.canRead()); + + env.notifyNewFETypeTransfer(targetType); + env.notifyNewFETypeTransfer(targetType); + runOneCycle(stateListener); + + Assertions.assertEquals(targetType, env.getFeType()); + Assertions.assertEquals(2, transitionAttempts.get()); + Mockito.verify(auth).rectifyPrivs(); + Mockito.verify(catalogMgr).registerCatalogRefreshListener(env); + Mockito.verify(env).startNonMasterDaemonThreads(); + metrics.verify(MetricRepo::init); + Mockito.verify(statisticsCache).preHeat(); + Assertions.assertEquals(1, senders.constructed().size()); + Mockito.verify(senders.constructed().get(0)).start(); + Assertions.assertEquals(List.of(false, false, false, false), servingDuringInitialization); + Assertions.assertTrue(env.isReady()); + Assertions.assertTrue(env.canRead()); + env.waitForReady(); + + // Once startup has completed, UNKNOWN retains the existing read policy until metadata expires. + env.notifyNewFETypeTransfer(FrontendNodeType.UNKNOWN); + env.notifyNewFETypeTransfer(FrontendNodeType.UNKNOWN); + runOneCycle(stateListener); + Assertions.assertFalse(env.isReady()); + Assertions.assertTrue(env.canRead()); + replayFreshMetadata(env, replayer); + Assertions.assertTrue(env.isReady()); + Assertions.assertTrue(env.canRead()); + env.setSynchronizedTime(0); + runOneCycle(replayer); + Assertions.assertFalse(env.isReady()); + Assertions.assertFalse(env.canRead()); + } + } + + @Test + public void testIgnoreMetaCheckDoesNotBypassStartupGate() throws Exception { + Env env = Mockito.spy(new Env(false)); + Mockito.doReturn(false).when(env).replayJournal(-1); + env.createReplayer(); + boolean originalIgnoreMetaCheck = Config.ignore_meta_check; + try { + Config.ignore_meta_check = true; + runOneCycle((Daemon) getField(env, "replayer")); + Assertions.assertTrue(((AtomicBoolean) getField(env, "canRead")).get()); + Assertions.assertFalse(env.isReady()); + Assertions.assertFalse(env.canRead()); + } finally { + Config.ignore_meta_check = originalIgnoreMetaCheck; + } + } + + private static void replayFreshMetadata(Env env, Daemon replayer) throws ReflectiveOperationException { + env.setSynchronizedTime(System.currentTimeMillis()); + runOneCycle(replayer); + } + + private static void runOneCycle(Daemon daemon) throws ReflectiveOperationException { + Method method = daemon.getClass().getDeclaredMethod("runOneCycle"); + method.setAccessible(true); + method.invoke(daemon); + } + private static Object getField(Env env, String fieldName) throws ReflectiveOperationException { Field field = Env.class.getDeclaredField(fieldName); field.setAccessible(true);