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); + } +}