From 9f6a76192aeb2073b7a45172c110989c22c70e83 Mon Sep 17 00:00:00 2001 From: morrySnow Date: Mon, 21 Sep 2026 16:40:02 +0800 Subject: [PATCH] [fix](fe) Gate serving until startup initialization completes ### What problem does this PR solve? Issue Number: N/A Related PR: #68302 Problem Summary: An initial FOLLOWER or OBSERVER transition can be interrupted by UNKNOWN before metadata post-processing, daemon startup, and metrics initialization run. The replayer can still publish readiness, and the retained INIT/UNKNOWN state handles UNKNOWN without completing initialization. This can release startup services and allow local queries against an incompletely initialized FE. Gate public readiness and read eligibility on completion of the first successful MASTER/FOLLOWER/OBSERVER initialization and FE type commit. Initialization waits use metadata readiness directly to avoid a circular wait. Keep the gate open for an already initialized FE entering UNKNOWN, preserving its metadata-based read policy. Include both readiness conditions in startup wait diagnostics. Exercise real replayer updates at the interruption boundary, successful retries, all non-master initialization steps, initialized UNKNOWN reads and metadata expiry, and ignore_meta_check. Separate listener creation from thread startup so the event interleaving can be tested deterministically. ### Release note Fix FE startup transitions that could expose a follower or observer before initialization completed when an UNKNOWN notification interrupted startup. ### Check List (For Author) - Test: Unit Test - ./run-fe-ut.sh --run org.apache.doris.catalog.EnvStateListenerTest,org.apache.doris.catalog.EnvTest,org.apache.doris.qe.StmtExecutorTest - All 34 tests passed; the 5 new regression cases failed with serving gates disabled. - cd fe && mvn checkstyle:check -pl fe-core - Behavior changed: Yes. Startup services and local reads require completed initialization; initialized UNKNOWN nodes retain the existing read policy. - Does this need documentation: No --- .../java/org/apache/doris/catalog/Env.java | 42 +++-- .../doris/catalog/EnvStateListenerTest.java | 158 ++++++++++++++++++ 2 files changed, 188 insertions(+), 12 deletions(-) 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);