From 9d2fe962dbaddc8555361fe4e45a608c81e5cdd2 Mon Sep 17 00:00:00 2001 From: Rico Date: Tue, 15 Nov 2022 10:03:04 +0800 Subject: [PATCH] inner-1956:fix MultiNodeSelectHandler hang (#3461) Signed-off-by: dcy10000 --- .../mysql/nio/handler/MultiNodeHandler.java | 1 + .../nio/handler/MultiNodeQueryHandler.java | 1 + .../nio/handler/MultiNodeSelectHandler.java | 7 +++++-- .../MySQLBackAuthService.java | 19 +++++++++++++++++++ 4 files changed, 26 insertions(+), 2 deletions(-) diff --git a/src/main/java/com/actiontech/dble/backend/mysql/nio/handler/MultiNodeHandler.java b/src/main/java/com/actiontech/dble/backend/mysql/nio/handler/MultiNodeHandler.java index 789606967..c7d8f0531 100644 --- a/src/main/java/com/actiontech/dble/backend/mysql/nio/handler/MultiNodeHandler.java +++ b/src/main/java/com/actiontech/dble/backend/mysql/nio/handler/MultiNodeHandler.java @@ -33,6 +33,7 @@ public abstract class MultiNodeHandler implements ResponseHandler { protected Set unResponseRrns = Sets.newConcurrentHashSet(); protected int errorConnsCnt = 0; protected boolean firstResponsed = false; + protected boolean complexQuery = false; public MultiNodeHandler(NonBlockingSession session) { if (session == null) { diff --git a/src/main/java/com/actiontech/dble/backend/mysql/nio/handler/MultiNodeQueryHandler.java b/src/main/java/com/actiontech/dble/backend/mysql/nio/handler/MultiNodeQueryHandler.java index 21ef65603..def160df3 100644 --- a/src/main/java/com/actiontech/dble/backend/mysql/nio/handler/MultiNodeQueryHandler.java +++ b/src/main/java/com/actiontech/dble/backend/mysql/nio/handler/MultiNodeQueryHandler.java @@ -181,6 +181,7 @@ public class MultiNodeQueryHandler extends MultiNodeHandler implements LoadDataR } conn.getBackendService().setResponseHandler(this); conn.getBackendService().setSession(session); + conn.getBackendService().setComplexQuery(complexQuery); conn.getBackendService().executeMultiNode(node, session.getShardingService(), sessionAutocommit && !session.getShardingService().isTxStart() && !node.isModifySQL()); } diff --git a/src/main/java/com/actiontech/dble/backend/mysql/nio/handler/MultiNodeSelectHandler.java b/src/main/java/com/actiontech/dble/backend/mysql/nio/handler/MultiNodeSelectHandler.java index e460f7f70..6f170bb1b 100644 --- a/src/main/java/com/actiontech/dble/backend/mysql/nio/handler/MultiNodeSelectHandler.java +++ b/src/main/java/com/actiontech/dble/backend/mysql/nio/handler/MultiNodeSelectHandler.java @@ -50,6 +50,7 @@ public class MultiNodeSelectHandler extends MultiNodeQueryHandler { public MultiNodeSelectHandler(RouteResultset rrs, NonBlockingSession session) { super(rrs, session, false); + this.complexQuery = true; this.queueSize = SystemConfig.getInstance().getMergeQueueSize(); this.queues = new ConcurrentHashMap<>(); if (CollectionUtil.isEmpty(rrs.getSelectCols())) { @@ -212,7 +213,7 @@ public class MultiNodeSelectHandler extends MultiNodeQueryHandler { } while (!heap.isEmpty()) { if (isFail()) - return; + break; HeapItem top = heap.peak(); if (top.isNullItem()) { heap.poll(); @@ -252,7 +253,9 @@ public class MultiNodeSelectHandler extends MultiNodeQueryHandler { iterator.remove(); } doSqlStat(); - nextHandler.rowEofResponse(null, false, null); + if (!isFail()) { + nextHandler.rowEofResponse(null, false, null); + } } catch (MySQLOutPutException e) { String msg = e.getLocalizedMessage(); LOGGER.info(msg, e); diff --git a/src/main/java/com/actiontech/dble/services/mysqlauthenticate/MySQLBackAuthService.java b/src/main/java/com/actiontech/dble/services/mysqlauthenticate/MySQLBackAuthService.java index 44bee50ad..171cf9fb0 100644 --- a/src/main/java/com/actiontech/dble/services/mysqlauthenticate/MySQLBackAuthService.java +++ b/src/main/java/com/actiontech/dble/services/mysqlauthenticate/MySQLBackAuthService.java @@ -1,6 +1,7 @@ package com.actiontech.dble.services.mysqlauthenticate; import com.actiontech.dble.DbleServer; +import com.actiontech.dble.backend.heartbeat.HeartbeatSQLJob; import com.actiontech.dble.backend.mysql.CharsetUtil; import com.actiontech.dble.backend.mysql.nio.handler.ResponseHandler; import com.actiontech.dble.backend.pool.PooledConnectionListener; @@ -21,6 +22,7 @@ import org.slf4j.LoggerFactory; import java.io.IOException; import java.security.NoSuchAlgorithmException; +import java.util.concurrent.Executor; import static com.actiontech.dble.config.ErrorCode.ER_ACCESS_DENIED_ERROR; @@ -40,9 +42,13 @@ public class MySQLBackAuthService extends BackendService implements AuthService private volatile boolean authSwitchMore; private volatile PluginName pluginName; private volatile long serverCapabilities; + private volatile boolean highPriority = false; public MySQLBackAuthService(BackendConnection connection, String user, String schema, String passwd, PooledConnectionListener listener, ResponseHandler handler) { super(connection); + if (handler instanceof HeartbeatSQLJob) { + highPriority = true; + } this.user = user; this.schema = schema; this.passwd = passwd; @@ -53,6 +59,9 @@ public class MySQLBackAuthService extends BackendService implements AuthService // only for com_change_user public MySQLBackAuthService(BackendConnection connection, String user, String passwd, ResponseHandler handler) { super(connection); + if (handler instanceof HeartbeatSQLJob) { + highPriority = true; + } this.user = user; this.passwd = passwd; this.handler = handler; @@ -290,4 +299,14 @@ public class MySQLBackAuthService extends BackendService implements AuthService public boolean haveNotReceivedMessage() { throw new UnsupportedOperationException(); } + + + @Override + protected Executor getExecutor() { + if (highPriority) { + return DbleServer.getInstance().getComplexQueryExecutor(); + } else { + return DbleServer.getInstance().getBackendBusinessExecutor(); + } + } }