From 164a1503c7891dea9821b658e16890782f9f0c83 Mon Sep 17 00:00:00 2001 From: morrySnow Date: Mon, 21 Sep 2026 16:16:39 +0800 Subject: [PATCH] [fix](fetype) Commit FE type only after transition completes (#68302) ### What problem does this PR solve? Problem Summary: A non-master FE transition can be interrupted when a newer BDB state is queued while the listener waits for metadata readiness. The listener previously still committed the target FE type even though non-master initialization, including MetricRepo.init(), had not run. A repeated FOLLOWER or OBSERVER notification was then treated as redundant, leaving the FE in an incompletely initialized state. This change makes transferToNonMaster() report whether the transition completed and updates feType only after all initialization finishes. An interrupted transition retains the previous committed state so the queued event is evaluated against the state that was actually initialized and can retry the transition. ### Release note Fix FE state transitions that could leave a follower or observer serving queries before non-master initialization completed. --- .../java/org/apache/doris/catalog/Env.java | 34 ++++++-- .../doris/catalog/EnvStateListenerTest.java | 80 +++++++++++++++++++ 2 files changed, 106 insertions(+), 8 deletions(-) create mode 100644 fe/fe-core/src/test/java/org/apache/doris/catalog/EnvStateListenerTest.java 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 265fe780bd74e0..fad5bcdfa365ad 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 @@ -2058,7 +2058,7 @@ protected void startNonMasterDaemonThreads() { splitSourceManager.start(); } - private void transferToNonMaster(FrontendNodeType newType) { + private boolean transferToNonMaster(FrontendNodeType newType) { isReady.set(false); try { @@ -2068,7 +2068,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 @@ -2080,8 +2080,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(); @@ -2098,11 +2101,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; } } @@ -3125,6 +3130,8 @@ protected synchronized void runOneCycle() { return; } + boolean transferCompleted = true; + /* * INIT -> MASTER: transferToMaster * INIT -> FOLLOWER/OBSERVER: transferToNonMaster @@ -3142,7 +3149,7 @@ protected synchronized void runOneCycle() { } case FOLLOWER: case OBSERVER: { - transferToNonMaster(newType); + transferCompleted = transferToNonMaster(newType); break; } case UNKNOWN: @@ -3160,7 +3167,7 @@ protected synchronized void runOneCycle() { } case FOLLOWER: case OBSERVER: { - transferToNonMaster(newType); + transferCompleted = transferToNonMaster(newType); break; } default: @@ -3175,7 +3182,7 @@ protected synchronized void runOneCycle() { break; } case UNKNOWN: { - transferToNonMaster(newType); + transferCompleted = transferToNonMaster(newType); break; } default: @@ -3186,7 +3193,7 @@ protected synchronized void runOneCycle() { case OBSERVER: { switch (newType) { case UNKNOWN: { - transferToNonMaster(newType); + transferCompleted = transferToNonMaster(newType); break; } default: @@ -3206,6 +3213,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); } 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 new file mode 100644 index 00000000000000..0a41817f74b073 --- /dev/null +++ b/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); + } +}