diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/FEOpExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/qe/FEOpExecutor.java
index f0e31c4aa5ffe3..37f51998c530ea 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/FEOpExecutor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/FEOpExecutor.java
@@ -163,6 +163,41 @@ protected TMasterOpResult forward(TMasterOpRequest params) throws Exception {
         }
     }
 
+    /**
+     * Whether this executor should redirect to the real master when the target FE
+     * rejects the request with NOT_MASTER. Only master-directed executors enable this;
+     * generic FE-to-FE calls (config propagation, cross-FE query kill) target a
+     * specific FE on purpose and must NOT be redirected.
+     */
+    protected boolean supportNotMasterRedirect() {
+        return false;
+    }
+
+    /**
+     * Send the request to the given address, ignoring NOT_MASTER semantics. Used for
+     * the single redirect retry against a re-discovered master.
+     */
+    protected TMasterOpResult forwardTo(TMasterOpRequest params, TNetworkAddress target) throws Exception {
+        FrontendService.Client client;
+        try {
+            client = ClientPool.frontendPool.borrowObject(target, thriftTimeoutMs);
+        } catch (Exception e) {
+            throw new Exception("Failed to get client for " + target + ".", e);
+        }
+        boolean isReturnToPool = false;
+        try {
+            TMasterOpResult result = client.forward(params);
+            isReturnToPool = true;
+            return result;
+        } finally {
+            if (isReturnToPool) {
+                ClientPool.frontendPool.returnObject(target, client);
+            } else {
+                ClientPool.frontendPool.invalidateObject(target, client);
+            }
+        }
+    }
+
     protected TMasterOpRequest buildStmtForwardParams() throws AnalysisException {
         TMasterOpRequest params = new TMasterOpRequest();
         // node ident
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/MasterOpExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/qe/MasterOpExecutor.java
index e4669d318fee26..d616603c83edec 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/MasterOpExecutor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/MasterOpExecutor.java
@@ -19,18 +19,24 @@
 
 import org.apache.doris.analysis.RedirectStatus;
 import org.apache.doris.catalog.Env;
+import org.apache.doris.common.Config;
 import org.apache.doris.common.DdlException;
 import org.apache.doris.common.LoadException;
+import org.apache.doris.ha.FrontendNodeType;
 import org.apache.doris.resource.BackendSelection;
 import org.apache.doris.resource.BackendSelectionManager;
+import org.apache.doris.system.Frontend;
 import org.apache.doris.thrift.TGroupCommitInfo;
 import org.apache.doris.thrift.TMasterOpRequest;
 import org.apache.doris.thrift.TMasterOpResult;
 import org.apache.doris.thrift.TNetworkAddress;
 
+import com.google.common.base.Strings;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
+import java.net.InetSocketAddress;
+
 /**
  * MasterOpExecutor is used to send request to Master FE.
  * It is inherited from FEOpExecutor. The difference is that MasterOpExecutor may need to wait the journal being
@@ -57,9 +63,23 @@ public MasterOpExecutor(ConnectContext ctx) {
         this(null, ctx, RedirectStatus.FORWARD_WITH_SYNC, true);
     }
 
+    @Override
+    public boolean supportNotMasterRedirect() {
+        return true;
+    }
+
     @Override
     public void execute() throws Exception {
-        super.execute();
+        TMasterOpRequest params = buildStmtForwardParams();
+        result = forward(params);
+        if (isNotMasterResult(result)) {
+            // The FE we forwarded to is not the master any more (our masterInfo is stale,
+            // typically after a master failover while journal replay is lagging).
+            // The statement was NOT executed there, so it is safe to re-discover the real
+            // master and retry once, even for non-idempotent statements.
+            result = redirectAndRetry(params);
+        }
+        processForwardResult(result);
         waitOnReplaying();
     }
 
@@ -69,7 +89,174 @@ public void cancel() throws Exception {
         waitOnReplaying();
     }
 
+    private void processForwardResult(TMasterOpResult result) throws Exception {
+        if (ctx.isTxnModel()) {
+            if (result.isSetTxnLoadInfo()) {
+                ctx.getTxnEntry().setTxnLoadInfoInObserver(result.getTxnLoadInfo());
+            } else {
+                ctx.setTxnEntry(null);
+                LOG.info("set txn entry to null");
+            }
+        }
+        if (result.isSetAffectedRows()) {
+            ctx.updateReturnRows((int) result.getAffectedRows());
+        }
+    }
+
+    private boolean isNotMasterResult(TMasterOpResult result) {
+        return isNotMasterResultForTest(result);
+    }
+
+    static boolean isNotMasterResultForTest(TMasterOpResult result) {
+        return result != null && result.isSetNotMaster() && result.isNotMaster();
+    }
+
+    private TNetworkAddress validateHint(TNetworkAddress hint) {
+        return validateHintForTest(hint);
+    }
+
+    /**
+     * Handle a NOT_MASTER rejection: validate the hint carried by the rejecting FE,
+     * or discover the current master on our own, then retry the request once against it.
+     *
+     * The hint is best-effort and must not be trusted blindly: a degraded old master
+     * may still keep masterInfo = itself, so a hint pointing back to the failed target
+     * (or to this node) is rejected and we fall back to our own discovery.
+     */
+    private TMasterOpResult redirectAndRetry(TMasterOpRequest params) throws Exception {
+        TNetworkAddress newMaster = validateHint(result.getMasterAddress());
+        if (newMaster == null) {
+            newMaster = discoverMasterByLeader();
+        }
+        if (newMaster == null) {
+            newMaster = discoverMasterByProbe();
+        }
+        if (newMaster == null) {
+            LOG.warn("forward target {} is not master and no new master could be discovered", feAddr);
+            throw new MasterRedirectException(
+                    "forward to master FE " + feAddr + " failed: target is not master any more"
+                            + " and no new master could be discovered. You may need to check FE's status");
+        }
+        LOG.warn("forward target {} is not master any more, retry against the new master {}",
+                feAddr, newMaster);
+        TMasterOpResult retryResult = forwardTo(params, newMaster);
+        if (isNotMasterResult(retryResult)) {
+            // single retry only: never loop on redirects
+            throw new MasterRedirectException(
+                    "forward to master FE " + newMaster + " also rejected as not-master");
+        }
+        feAddr = newMaster;
+        return retryResult;
+    }
+
+    /**
+     * Thrown when a forward target rejects the request as NOT_MASTER and no usable
+     * new master can be discovered (or the retry target also rejects it).
+     */
+    public static class MasterRedirectException extends RuntimeException {
+        public MasterRedirectException(String msg) {
+            super(msg);
+        }
+    }
+
+    /**
+     * A usable hint must be non-empty, and must not point back to the failed target or to
+     * this node (both are possible for a degraded old master whose masterInfo = itself).
+     */
+    TNetworkAddress validateHintForTest(TNetworkAddress hint) {
+        if (hint == null || Strings.isNullOrEmpty(hint.hostname) || hint.port <= 0) {
+            return null;
+        }
+        if (hint.hostname.equals(feAddr.getHostname()) && hint.port == feAddr.getPort()) {
+            return null;
+        }
+        String selfHost = Env.getCurrentEnv().getSelfNode().getHost();
+        if (hint.hostname.equals(selfHost) && hint.port == Config.rpc_port) {
+            return null;
+        }
+        return hint;
+    }
+
+    /**
+     * Discover the current master by asking the bdbje group directly (independent of
+     * journal replay), then map the leader's (host, editLogPort) to its rpc port via the
+     * local frontend list. May fail when the bdbje channel itself is partitioned.
+     */
+    private TNetworkAddress discoverMasterByLeader() {
+        try {
+            InetSocketAddress leader = ctx.getEnv().getHaProtocol().getLeader();
+            if (leader == null) {
+                return null;
+            }
+            for (Frontend fe : ctx.getEnv().getFrontends(null)) {
+                if (fe.getRole() == FrontendNodeType.FOLLOWER
+                        && fe.getHost().equals(leader.getHostString())
+                        && fe.getEditLogPort() == leader.getPort()) {
+                    return new TNetworkAddress(fe.getHost(), fe.getRpcPort());
+                }
+            }
+        } catch (Exception e) {
+            LOG.warn("failed to discover master by bdbje leader: {}", e.getMessage());
+        }
+        return null;
+    }
+
+    /**
+     * Last-resort discovery, tolerant of a partitioned bdbje channel: probe the thrift
+     * endpoints of alive followers (excluding the failed target and this node) with a
+     * lightweight isMasterProbe request, bounded to at most a few probes.
+     */
+    private TNetworkAddress discoverMasterByProbe() {
+        int probed = 0;
+        final int maxProbes = 2;
+        String selfHost = Env.getCurrentEnv().getSelfNode().getHost();
+        for (Frontend fe : ctx.getEnv().getFrontends(FrontendNodeType.FOLLOWER)) {
+            if (probed >= maxProbes) {
+                break;
+            }
+            if (!fe.isAlive()) {
+                continue;
+            }
+            TNetworkAddress candidate = new TNetworkAddress(fe.getHost(), fe.getRpcPort());
+            if (candidate.hostname.equals(feAddr.getHostname()) && candidate.port == feAddr.getPort()) {
+                continue;
+            }
+            if (fe.getHost().equals(selfHost)) {
+                continue;
+            }
+            probed++;
+            try {
+                TMasterOpResult probeResult = forwardTo(buildMasterProbeParams(), candidate);
+                if (probeResult.isSetNotMaster() && !probeResult.isNotMaster()) {
+                    return candidate;
+                }
+            } catch (Exception e) {
+                LOG.warn("master probe to {} failed: {}", candidate, e.getMessage());
+            }
+        }
+        return null;
+    }
+
+    private TMasterOpRequest buildMasterProbeParams() {
+        TMasterOpRequest params = new TMasterOpRequest();
+        params.setClientNodeHost(Env.getCurrentEnv().getSelfNode().getHost());
+        params.setClientNodePort(Env.getCurrentEnv().getSelfNode().getPort());
+        params.setIsMasterProbe(true);
+        params.setDb(ctx.getDatabase());
+        params.setUser(ctx.getQualifiedUser());
+        // just make the protocol happy
+        params.setSql("");
+        return params;
+    }
+
     private void waitOnReplaying() throws DdlException {
+        if (isNotMasterResult(result)) {
+            // A NOT_MASTER rejection carries no valid journal-sync target; waiting on it
+            // (typically journal id 0 or a stale id from the rejecting FE) would hang the
+            // client for the whole journal-wait timeout. Surface the error immediately.
+            LOG.info("forward result is a NOT_MASTER rejection, skip journal replay wait");
+            return;
+        }
         LOG.info("forwarding to master get result max journal id: {}", result.maxJournalId);
         ctx.getEnv().getJournalObservable().waitOn(result.maxJournalId, journalWaitTimeoutMs);
     }
diff --git a/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java b/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
index ea4c8c18a069f1..c8407616b63bb5 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
@@ -68,6 +68,7 @@
 import org.apache.doris.common.Config;
 import org.apache.doris.common.DdlException;
 import org.apache.doris.common.DuplicatedRequestException;
+import org.apache.doris.common.ErrorCode;
 import org.apache.doris.common.FeConstants;
 import org.apache.doris.common.InternalErrorCode;
 import org.apache.doris.common.LabelAlreadyUsedException;
@@ -1152,6 +1153,15 @@ public TFetchResourceResult fetchResource() throws TException {
     @Override
     public TMasterOpResult forward(TMasterOpRequest params) throws TException {
         Frontend requester = validateForwardRequester(params);
+        // If this node is not the master any more (e.g. after a master failover while the
+        // sender's journal replay is still lagging), reject the forwarded statement before
+        // executing it. Otherwise the statement would run deep into StmtExecutor and fail
+        // with "Master FE is not ready", which the sender cannot recover from.
+        if (!params.isSetIsMasterProbe() || !params.isIsMasterProbe()) {
+            if (!Env.getCurrentEnv().isMaster()) {
+                return buildNotMasterResult();
+            }
+        }
         TMasterOpResult shortcut = handleForwardShortcut(params);
         if (shortcut != null) {
             return shortcut;
@@ -1168,6 +1178,35 @@ public TMasterOpResult forward(TMasterOpRequest params) throws TException {
         }
     }
 
+    /**
+     * Build a NOT_MASTER rejection for a forwarded statement. The statement is NOT executed,
+     * so the sender may safely retry it (even non-idempotent statements) against the real
+     * master discovered via the hint below or by its own discovery.
+     */
+    private TMasterOpResult buildNotMasterResult() {
+        TMasterOpResult result = new TMasterOpResult();
+        // No journal-sync target for a rejection; the sender must skip its journal wait.
+        result.setMaxJournalId(0L);
+        result.setPacket(new byte[0]);
+        result.setNotMaster(true);
+        result.setStatus(QueryState.MysqlStateType.ERR.name());
+        result.setStatusCode(ErrorCode.ERR_UNKNOWN_ERROR.getCode());
+        result.setErrMessage(NOT_MASTER_REJECT_MSG.replace("{}", Env.getCurrentEnv().getSelfNode().getHost()));
+        // Hint the sender with the master this node knows about, if any. This is best-effort:
+        // the hint may be stale or even point to this node itself, so the sender must
+        // validate it (not equal to the failed target / itself) before use.
+        if (Env.getCurrentEnv().isReady() && !Strings.isNullOrEmpty(Env.getCurrentEnv().getMasterHost())) {
+            result.setMasterAddress(new TNetworkAddress(Env.getCurrentEnv().getMasterHost(),
+                    Env.getCurrentEnv().getMasterRpcPort()));
+        }
+        LOG.warn("reject forwarded statement because current node is not master. known master: {}",
+                result.getMasterAddress());
+        return result;
+    }
+
+    private static final String NOT_MASTER_REJECT_MSG =
+            "Current FE({}) is not the master any more, please retry against the current master.";
+
     private Frontend validateForwardRequester(TMasterOpRequest params) throws TException {
         Frontend fe = Env.getCurrentEnv().checkFeExist(params.getClientNodeHost(), params.getClientNodePort());
         if (fe != null) {
@@ -1178,6 +1217,9 @@ private Frontend validateForwardRequester(TMasterOpRequest params) throws TExcep
     }
 
     private TMasterOpResult handleForwardShortcut(TMasterOpRequest params) throws TException {
+        if (params.isSetIsMasterProbe() && params.isIsMasterProbe()) {
+            return handleMasterProbe();
+        }
         if (params.isSyncJournalOnly()) {
             return createForwardResultWithJournalSync();
         }
@@ -1202,6 +1244,29 @@ private TMasterOpResult createForwardResultWithJournalSync() {
         return result;
     }
 
+    /**
+     * Lightweight probe: answer whether this node is the master, without executing anything.
+     * Used by a sender re-discovering the real master after a NOT_MASTER rejection.
+     */
+    private TMasterOpResult handleMasterProbe() {
+        TMasterOpResult result = new TMasterOpResult();
+        result.setMaxJournalId(0L);
+        result.setPacket(new byte[0]);
+        if (Env.getCurrentEnv().isMaster()) {
+            result.setNotMaster(false);
+            // confirm self as master: hint points to this node
+            result.setMasterAddress(new TNetworkAddress(Env.getCurrentEnv().getSelfNode().getHost(),
+                    Config.rpc_port));
+        } else {
+            result.setNotMaster(true);
+            if (Env.getCurrentEnv().isReady() && !Strings.isNullOrEmpty(Env.getCurrentEnv().getMasterHost())) {
+                result.setMasterAddress(new TNetworkAddress(Env.getCurrentEnv().getMasterHost(),
+                        Env.getCurrentEnv().getMasterRpcPort()));
+            }
+        }
+        return result;
+    }
+
     private TMasterOpResult createForwardResultWithoutJournalSync() {
         TMasterOpResult result = new TMasterOpResult();
         // Group commit shortcuts update master memory without producing a journal id. Say so explicitly
diff --git a/fe/fe-core/src/test/java/org/apache/doris/qe/MasterOpExecutorNotMasterTest.java b/fe/fe-core/src/test/java/org/apache/doris/qe/MasterOpExecutorNotMasterTest.java
new file mode 100644
index 00000000000000..94f6e4b269e1ef
--- /dev/null
+++ b/fe/fe-core/src/test/java/org/apache/doris/qe/MasterOpExecutorNotMasterTest.java
@@ -0,0 +1,103 @@
+// 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.qe;
+
+import org.apache.doris.analysis.RedirectStatus;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.Config;
+import org.apache.doris.datasource.InternalCatalog;
+import org.apache.doris.system.SystemInfoService;
+import org.apache.doris.thrift.TMasterOpResult;
+import org.apache.doris.thrift.TNetworkAddress;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+public class MasterOpExecutorNotMasterTest {
+
+    // The static Env mock must stay open for the whole test method, because
+    // validateHintForTest() reads Env.getCurrentEnv() outside the constructor.
+    private static MasterOpExecutor newExecutor(String masterHost, int masterPort, MockedStatic<Env> mockedEnv) {
+        Env env = Mockito.mock(Env.class);
+        Mockito.when(env.getSelfNode()).thenReturn(new SystemInfoService.HostInfo("127.0.0.1", 9010));
+        Mockito.when(env.getMasterHost()).thenReturn(masterHost);
+        Mockito.when(env.getMasterRpcPort()).thenReturn(masterPort);
+        InternalCatalog catalog = Mockito.mock(InternalCatalog.class);
+        Mockito.when(catalog.getName()).thenReturn("internal");
+        Mockito.when(env.getInternalCatalog()).thenReturn(catalog);
+        mockedEnv.when(Env::getCurrentEnv).thenReturn(env);
+        ConnectContext ctx = new ConnectContext();
+        ctx.setEnv(env);
+        ctx.setSessionVariable(VariableMgr.newSessionVariable());
+        ctx.getSessionVariable().setQueryTimeoutS(10);
+        return new MasterOpExecutor(null, ctx, RedirectStatus.FORWARD_WITH_SYNC, true);
+    }
+
+    // A hint equal to the failed target must be rejected (a degraded old master whose
+    // masterInfo = itself would otherwise create a retry loop).
+    @Test
+    public void testHintPointingToFailedTargetRejected() {
+        try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class)) {
+            MasterOpExecutor executor = newExecutor("10.0.0.1", 9020, mockedEnv);
+            Assert.assertNull(executor.validateHintForTest(new TNetworkAddress("10.0.0.1", 9020)));
+        }
+    }
+
+    // A hint pointing to this node must be rejected as well.
+    @Test
+    public void testHintPointingToSelfRejected() {
+        try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class)) {
+            MasterOpExecutor executor = newExecutor("10.0.0.1", 9020, mockedEnv);
+            Assert.assertNull(executor.validateHintForTest(new TNetworkAddress("127.0.0.1", Config.rpc_port)));
+        }
+    }
+
+    // A valid hint pointing elsewhere is accepted.
+    @Test
+    public void testValidHintAccepted() {
+        try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class)) {
+            MasterOpExecutor executor = newExecutor("10.0.0.1", 9020, mockedEnv);
+            TNetworkAddress hint = new TNetworkAddress("10.0.0.2", 9020);
+            Assert.assertEquals(hint, executor.validateHintForTest(hint));
+        }
+    }
+
+    // Empty/invalid hints are rejected.
+    @Test
+    public void testInvalidHintRejected() {
+        try (MockedStatic<Env> mockedEnv = Mockito.mockStatic(Env.class)) {
+            MasterOpExecutor executor = newExecutor("10.0.0.1", 9020, mockedEnv);
+            Assert.assertNull(executor.validateHintForTest(null));
+            Assert.assertNull(executor.validateHintForTest(new TNetworkAddress("", 9020)));
+            Assert.assertNull(executor.validateHintForTest(new TNetworkAddress("10.0.0.2", 0)));
+        }
+    }
+
+    // NOT_MASTER detection.
+    @Test
+    public void testIsNotMasterResult() {
+        Assert.assertFalse(MasterOpExecutor.isNotMasterResultForTest(null));
+        TMasterOpResult normal = new TMasterOpResult();
+        Assert.assertFalse(MasterOpExecutor.isNotMasterResultForTest(normal));
+        TMasterOpResult notMaster = new TMasterOpResult();
+        notMaster.setNotMaster(true);
+        Assert.assertTrue(MasterOpExecutor.isNotMasterResultForTest(notMaster));
+    }
+}
diff --git a/gensrc/thrift/FrontendService.thrift b/gensrc/thrift/FrontendService.thrift
index 9f197d691f8b75..a74042d77ef7be 100644
--- a/gensrc/thrift/FrontendService.thrift
+++ b/gensrc/thrift/FrontendService.thrift
@@ -441,6 +441,11 @@ struct TMasterOpRequest {
     1005: optional string delegated_credential_token
     1006: optional i64 delegated_credential_expires_at_millis
     1007: optional string delegated_credential_session_id
+
+    // if set to true, this request is a lightweight probe asking whether the receiving
+    // FE is the master. Used by a sender to re-discover the real master after a
+    // NOT_MASTER rejection. No statement is executed.
+    37: optional bool isMasterProbe
 }
 
 struct TColumnDefinition {
@@ -474,6 +479,13 @@ struct TMasterOpResult {
     11: optional i64 affectedRows;
     // Lets the forwarding FE wait for the final statistics of external write fragments.
     12: optional list<i64> auditStatisticsBackendIds;
+    // Set when the receiving FE is not the master, so the sender can refresh its stale
+    // masterInfo and retry against the real master instead of failing with
+    // "Master FE is not ready".
+    14: optional bool notMaster;
+    // The master address known by the rejecting FE (may be stale or even point to the
+    // rejecting FE itself; the sender must validate it before use).
+    15: optional Types.TNetworkAddress masterAddress;
 }
 
 // Certificate-based authentication info forwarded from BE to FE