[fix](fetype) Commit FE type only after transition completes

fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java

@@ -2114,7 +2114,7 @@ protected void startNonMasterDaemonThreads() {
         splitSourceManager.start();
     }
 
-    private void transferToNonMaster(FrontendNodeType newType) {
+    private boolean transferToNonMaster(FrontendNodeType newType) {
         isReady.set(false);
 
         try {
@@ -2124,7 +2124,7 @@ private void transferToNonMaster(FrontendNodeType newType) {
                 // not set canRead here, leave canRead as what is was.
                 // if meta out of date, canRead will be set to false in replayer thread.
                 metaReplayState.setTransferToUnknown();
-                return;
+                return true;
             }
 
             // transfer from INIT/UNKNOWN to OBSERVER/FOLLOWER
@@ -2136,8 +2136,11 @@ private void transferToNonMaster(FrontendNodeType newType) {
 
             // 'isReady' will be set to true in 'setCanRead()' method
             if (!postProcessAfterMetadataReplayed(true)) {
-                // the state has changed, exit early.
-                return;
+                // 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
+                // become ready as a non-master. The caller must not publish newType to feType in this case:
+                // none of the non-master initialization below, including MetricRepo.init(), has completed yet.
+                return false;
             }
 
             checkLowerCaseTableNames();
@@ -2154,11 +2157,13 @@ private void transferToNonMaster(FrontendNodeType newType) {
                 followerColumnSender = new FollowerColumnSender();
                 followerColumnSender.start();
             }
+            return true;
         } catch (Throwable e) {
             // When failed to transfer to non-master, we need to exit the process.
             // Otherwise, the process will be in an unknown state.
             LOG.error("failed to transfer to non-master.", e);
             System.exit(-1);
+            return false;
         }
     }
 
@@ -3278,6 +3283,8 @@ protected synchronized void runOneCycle() {
                         return;
                     }
 
+                    boolean transferCompleted = true;
+
                     /*
                      * INIT -> MASTER: transferToMaster
                      * INIT -> FOLLOWER/OBSERVER: transferToNonMaster
@@ -3295,7 +3302,7 @@ protected synchronized void runOneCycle() {
                                 }
                                 case FOLLOWER:
                                 case OBSERVER: {
-                                    transferToNonMaster(newType);
+                                    transferCompleted = transferToNonMaster(newType);
                                     break;
                                 }
                                 case UNKNOWN:
@@ -3313,7 +3320,7 @@ protected synchronized void runOneCycle() {
                                 }
                                 case FOLLOWER:
                                 case OBSERVER: {
-                                    transferToNonMaster(newType);
+                                    transferCompleted = transferToNonMaster(newType);
                                     break;
                                 }
                                 default:
@@ -3328,7 +3335,7 @@ protected synchronized void runOneCycle() {
                                     break;
                                 }
                                 case UNKNOWN: {
-                                    transferToNonMaster(newType);
+                                    transferCompleted = transferToNonMaster(newType);
                                     break;
                                 }
                                 default:
@@ -3339,7 +3346,7 @@ protected synchronized void runOneCycle() {
                         case OBSERVER: {
                             switch (newType) {
                                 case UNKNOWN: {
-                                    transferToNonMaster(newType);
+                                    transferCompleted = transferToNonMaster(newType);
                                     break;
                                 }
                                 default:
@@ -3359,6 +3366,17 @@ protected synchronized void runOneCycle() {
                             break;
                     } // end switch formerFeType
 
+                    if (!transferCompleted) {
+                        // feType represents the last fully initialized FE state, not merely the latest state
+                        // reported by BDB. A non-master transition can be interrupted when a newer BDB state is
+                        // queued while it waits for metadata to become ready. Committing newType after that early
+                        // return would make a repeated FOLLOWER/OBSERVER event look redundant and skip the
+                        // incomplete initialization permanently. Keep the previous committed state so the queued
+                        // event is evaluated against the state that was actually initialized and can retry the
+                        // transition or take a different path.
+                        LOG.info("skip committing incomplete FE type transfer from {} to {}", feType, newType);
+                        continue;
+                    }
                     feType = newType;
                     LOG.info("finished to transfer FE type to {}", feType);
                 }

fe/fe-core/src/test/java/org/apache/doris/catalog/EnvStateListenerTest.java

@@ -0,0 +1,80 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.catalog;
+
+import org.apache.doris.common.util.Daemon;
+import org.apache.doris.ha.FrontendNodeType;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.lang.reflect.Field;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class EnvStateListenerTest {
+    @Test
+    public void testInterruptedNonMasterTransitionDoesNotCommitFeType() throws Exception {
+        Env env = Mockito.spy(new Env(false));
+        setField(env, "replayer", Mockito.mock(Daemon.class));
+
+        CountDownLatch firstTransitionInterrupted = new CountDownLatch(1);
+        CountDownLatch repeatedTransitionAttempted = new CountDownLatch(1);
+        AtomicInteger transitionAttempts = new AtomicInteger();
+        Mockito.doAnswer(invocation -> {
+            if (transitionAttempts.incrementAndGet() == 1) {
+                firstTransitionInterrupted.countDown();
+            } else {
+                // Let runOneCycle return after the repeated FOLLOWER transition is interrupted. Without this
+                // event, the state listener would correctly keep waiting for another state after the assertion.
+                env.notifyNewFETypeTransfer(FrontendNodeType.INIT);
+                repeatedTransitionAttempted.countDown();
+            }
+            return false;
+        }).when(env).postProcessAfterMetadataReplayed(true);
+
+        env.startStateListener();
+        Daemon stateListener = (Daemon) getField(env, "listener");
+        try {
+            env.notifyNewFETypeTransfer(FrontendNodeType.FOLLOWER);
+            Assertions.assertTrue(firstTransitionInterrupted.await(5, TimeUnit.SECONDS));
+
+            // The first transition was interrupted before non-master initialization completed. A repeated
+            // FOLLOWER event must retry the transition instead of being discarded as an already completed state.
+            env.notifyNewFETypeTransfer(FrontendNodeType.FOLLOWER);
+            Assertions.assertTrue(repeatedTransitionAttempted.await(5, TimeUnit.SECONDS));
+            Assertions.assertEquals(FrontendNodeType.INIT, env.getFeType());
+        } finally {
+            stateListener.exit();
+        }
+    }
+
+    private static Object getField(Env env, String fieldName) throws ReflectiveOperationException {
+        Field field = Env.class.getDeclaredField(fieldName);
+        field.setAccessible(true);
+        return field.get(env);
+    }
+
+    private static void setField(Env env, String fieldName, Object value) throws ReflectiveOperationException {
+        Field field = Env.class.getDeclaredField(fieldName);
+        field.setAccessible(true);
+        field.set(env, value);
+    }
+}