[fix](fe) Gate serving until startup initialization completes

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() {

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<Boolean> servingDuringInitialization = new ArrayList<>();
+        Answer<Void> 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<MetricRepo> metrics = Mockito.mockStatic(MetricRepo.class);
+                MockedConstruction<FollowerColumnSender> 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);