Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 26 additions & 8 deletions fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java
Original file line number Diff line number Diff line change
Expand Up @@ -2058,7 +2058,7 @@ protected void startNonMasterDaemonThreads() {
splitSourceManager.start();
}

private void transferToNonMaster(FrontendNodeType newType) {
private boolean transferToNonMaster(FrontendNodeType newType) {
isReady.set(false);

try {
Expand All @@ -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
Expand All @@ -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();
Expand All @@ -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;
}
}

Expand Down Expand Up @@ -3125,6 +3130,8 @@ protected synchronized void runOneCycle() {
return;
}

boolean transferCompleted = true;

/*
* INIT -> MASTER: transferToMaster
* INIT -> FOLLOWER/OBSERVER: transferToNonMaster
Expand All @@ -3142,7 +3149,7 @@ protected synchronized void runOneCycle() {
}
case FOLLOWER:
case OBSERVER: {
transferToNonMaster(newType);
transferCompleted = transferToNonMaster(newType);
break;
}
case UNKNOWN:
Expand All @@ -3160,7 +3167,7 @@ protected synchronized void runOneCycle() {
}
case FOLLOWER:
case OBSERVER: {
transferToNonMaster(newType);
transferCompleted = transferToNonMaster(newType);
break;
}
default:
Expand All @@ -3175,7 +3182,7 @@ protected synchronized void runOneCycle() {
break;
}
case UNKNOWN: {
transferToNonMaster(newType);
transferCompleted = transferToNonMaster(newType);
break;
}
default:
Expand All @@ -3186,7 +3193,7 @@ protected synchronized void runOneCycle() {
case OBSERVER: {
switch (newType) {
case UNKNOWN: {
transferToNonMaster(newType);
transferCompleted = transferToNonMaster(newType);
break;
}
default:
Expand All @@ -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);
}
Expand Down
Original file line number Diff line number Diff line change
@@ -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);
}
}
Loading