[fix](arrow-flight) Bound idle Arrow Flight SQL sessions separately from wait_timeout

fe/fe-common/src/main/java/org/apache/doris/common/Config.java

@@ -2644,6 +2644,15 @@ public class Config extends ConfigBase {
     @ConfField(description = "Maximum number of connections for the Arrow Flight Server per FE.")
     public static int arrow_flight_max_connections = 4096;
 
+    @ConfField(mutable = true, masterOnly = false, description = "Idle timeout for Arrow Flight SQL sessions, "
+            + "in seconds. A Flight session that has been sleeping longer than this is killed by the "
+            + "connection timeout checker, exactly as a MySQL connection is killed after wait_timeout, but "
+            + "with its own, shorter bound: a Flight query keeps its coordinator and its workload-group "
+            + "queue slot alive until the session's next query or its close, so an abandoned session holds "
+            + "a slot for as long as it lives. The effective idle bound is min(wait_timeout, this value). "
+            + "0 disables the Flight-specific bound (wait_timeout alone applies).")
+    public static int arrow_flight_session_idle_timeout_second = 3600;
+
     @ConfField(mutable = true, masterOnly = true, description = "In auto bucketing, the number of buckets is "
             + "estimated based on the partition size. For storage "
             + "and computing integration, a partition size of 5GB " + "is estimated as one bucket, but for cloud, a "

fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java

@@ -1268,8 +1268,8 @@ public void kill(boolean killConnection) {
     private void killByTimeout(boolean killConnection) {
         if (killConnection) {
             LOG.warn("kill wait timeout connection, connection type: {}, connectionId: {}, remote: {}, "
-                            + "wait timeout: {}",
-                    getConnectType(), connectionId, getRemoteHostPortString(), sessionVariable.getWaitTimeoutS());
+                            + "idle timeout: {}",
+                    getConnectType(), connectionId, getRemoteHostPortString(), getIdleTimeoutS());
             killConnection();
         }
         // Now, cancel running query.
@@ -1292,6 +1292,15 @@ public void cancelQuery(Status cancelReason) {
         }
     }
 
+    /**
+     * How long this connection may sleep (COM_SLEEP) before the timeout checker kills it.
+     * The MySQL protocol bound is the session's wait_timeout; protocol-specific contexts may
+     * tighten it (never widen it) — see FlightSqlConnectContext.
+     */
+    public long getIdleTimeoutS() {
+        return sessionVariable.getWaitTimeoutS();
+    }
+
     public void checkTimeout(long now) {
         if (startTime <= 0) {
             return;
@@ -1301,7 +1310,7 @@ public void checkTimeout(long now) {
         boolean killFlag = false;
         boolean killConnection = false;
         if (command == MysqlCommand.COM_SLEEP) {
-            if (delta > sessionVariable.getWaitTimeoutS() * 1000L) {
+            if (delta > getIdleTimeoutS() * 1000L) {
                 // Need kill this connection.
                 killFlag = true;
                 killConnection = true;

fe/fe-core/src/main/java/org/apache/doris/service/arrowflight/sessions/FlightSqlConnectContext.java

@@ -17,6 +17,7 @@
 
 package org.apache.doris.service.arrowflight.sessions;
 
+import org.apache.doris.common.Config;
 import org.apache.doris.common.Status;
 import org.apache.doris.mysql.MysqlChannel;
 import org.apache.doris.qe.ConnectContext;
@@ -50,6 +51,28 @@ public FlightSqlChannel getFlightSqlChannel() {
         return flightSqlChannel;
     }
 
+    /**
+     * Flight sessions get their own idle bound (Config.arrow_flight_session_idle_timeout_second),
+     * tighter than the MySQL wait_timeout: an abandoned Flight session keeps its last query's
+     * coordinator — and that query's workload-group queue slot — alive until it is closed, so a
+     * client that opens a session per query and never closes it would otherwise pin a slot for
+     * the whole wait_timeout (8h by default).
+     *
+     * A Flight session is COM_SLEEP while the client drains the result from the BE (DoGet), and
+     * its idle clock runs from the query's START — so the bound is floored at the query's own
+     * execution timeout: a long result drain is never killed before query_timeout would kill the
+     * query itself. Effective bound = min(wait_timeout, max(config, exec timeout)); 0 disables it.
+     */
+    @Override
+    public long getIdleTimeoutS() {
+        long waitTimeoutS = super.getIdleTimeoutS();
+        int flightIdleS = Config.arrow_flight_session_idle_timeout_second;
+        if (flightIdleS <= 0) {
+            return waitTimeoutS;
+        }
+        return Math.min(waitTimeoutS, Math.max((long) flightIdleS, (long) getExecTimeoutS()));
+    }
+
     @Override
     public MysqlChannel getMysqlChannel() {
         throw new RuntimeException("getMysqlChannel not in mysql connection");

fe/fe-core/src/test/java/org/apache/doris/service/arrowflight/sessions/FlightSqlSessionIdleTimeoutTest.java

@@ -0,0 +1,86 @@
+// 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.service.arrowflight.sessions;
+
+import org.apache.doris.common.Config;
+import org.apache.doris.common.FeConstants;
+import org.apache.doris.qe.ConnectContext;
+
+import org.junit.Assert;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+/**
+ * The Flight-only idle bound: a Flight session's idle timeout is
+ * min(wait_timeout, arrow_flight_session_idle_timeout_second); it never widens wait_timeout,
+ * 0 disables it, and a MySQL-protocol context is untouched.
+ */
+public class FlightSqlSessionIdleTimeoutTest {
+
+    @BeforeAll
+    public static void setUp() {
+        // ConnectContext.init() registers the session with Env unless running as a unit test.
+        FeConstants.runningUnitTest = true;
+    }
+
+    @Test
+    public void testFlightIdleBoundOnlyTightensWaitTimeout() {
+        int saved = Config.arrow_flight_session_idle_timeout_second;
+        try {
+            FlightSqlConnectContext ctx = new FlightSqlConnectContext("test-peer-identity");
+            long waitTimeoutS = ctx.getSessionVariable().getWaitTimeoutS();
+            Assert.assertTrue(waitTimeoutS > 0);
+            // the bound is floored at the query's exec timeout; pin it low so the bound is visible
+            ctx.getSessionVariable().setQueryTimeoutS(5);
+
+            // tighter than wait_timeout and above the exec timeout -> the Flight bound wins
+            Config.arrow_flight_session_idle_timeout_second = 7;
+            Assert.assertEquals(7L, ctx.getIdleTimeoutS());
+
+            // below the exec timeout -> the exec timeout floors it (a result drain is never
+            // killed before query_timeout would kill the query)
+            Config.arrow_flight_session_idle_timeout_second = 3;
+            Assert.assertEquals(5L, ctx.getIdleTimeoutS());
+            ctx.getSessionVariable().setQueryTimeoutS(20);
+            Assert.assertEquals(20L, ctx.getIdleTimeoutS());
+            ctx.getSessionVariable().setQueryTimeoutS(5);
+
+            // looser than wait_timeout -> wait_timeout still applies (the bound never widens it)
+            Config.arrow_flight_session_idle_timeout_second = (int) Math.min(Integer.MAX_VALUE, waitTimeoutS + 1000);
+            Assert.assertEquals(waitTimeoutS, ctx.getIdleTimeoutS());
+
+            // 0 disables the Flight-specific bound
+            Config.arrow_flight_session_idle_timeout_second = 0;
+            Assert.assertEquals(waitTimeoutS, ctx.getIdleTimeoutS());
+        } finally {
+            Config.arrow_flight_session_idle_timeout_second = saved;
+        }
+    }
+
+    @Test
+    public void testMysqlContextKeepsWaitTimeout() {
+        int saved = Config.arrow_flight_session_idle_timeout_second;
+        try {
+            Config.arrow_flight_session_idle_timeout_second = 7;
+            ConnectContext ctx = new ConnectContext();
+            Assert.assertEquals(ctx.getSessionVariable().getWaitTimeoutS(), ctx.getIdleTimeoutS());
+        } finally {
+            Config.arrow_flight_session_idle_timeout_second = saved;
+        }
+    }
+}