[fix](fe) Redirect forward-to-master statements to the real master after failover

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

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

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