diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java b/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java
index 8274ec45ab9354..fcf2ca3d94a784 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java
@@ -861,6 +861,9 @@ public PlSqlOperation getPlSqlOperation() {
         return plSqlOperation;
     }
 
+    /**
+     * This method is idempotent.
+     */
     protected void closeChannel() {
         if (mysqlChannel != null) {
             mysqlChannel.close();
@@ -1027,6 +1030,10 @@ public void kill(boolean killConnection) {
         }
         // Now, cancel running query.
         cancelQuery(new Status(TStatusCode.CANCELLED, "cancel query by user from " + getRemoteHostPortString()));
+        // Clean up after cancelQuery to avoid needing session variables etc. inside cancelQuery
+        if (killConnection) {
+            cleanup();
+        }
     }
 
     // kill operation with no protect by timeout.
@@ -1050,6 +1057,10 @@ private void killByTimeout(boolean killConnection) {
             executorRef.cancel(new Status(TStatusCode.TIMEOUT,
                     "query is timeout, killed by timeout checker"));
         }
+        // Clean up after cancelQuery to avoid needing session variables etc. inside cancelQuery
+        if (killConnection) {
+            cleanup();
+        }
     }
 
     public void cancelQuery(Status cancelReason) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/service/arrowflight/sessions/FlightSqlConnectContext.java b/fe/fe-core/src/main/java/org/apache/doris/service/arrowflight/sessions/FlightSqlConnectContext.java
index 35293273f1e0aa..00d8a63554eb4b 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/service/arrowflight/sessions/FlightSqlConnectContext.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/service/arrowflight/sessions/FlightSqlConnectContext.java
@@ -80,6 +80,10 @@ public void kill(boolean killConnection) {
         }
         // Now, cancel running query.
         cancelQuery(new Status(TStatusCode.CANCELLED, "arrow flight query killed by user"));
+        // Clean up after cancelQuery to avoid needing session variables etc. inside cancelQuery
+        if (killConnection) {
+            cleanup();
+        }
     }
 
     @Override
diff --git a/regression-test/suites/temp_table_p0/test_temp_table_with_conn_timeout.groovy b/regression-test/suites/temp_table_p0/test_temp_table_with_conn_timeout.groovy
new file mode 100644
index 00000000000000..d78774160a03f0
--- /dev/null
+++ b/regression-test/suites/temp_table_p0/test_temp_table_with_conn_timeout.groovy
@@ -0,0 +1,45 @@
+// 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.
+
+suite("test_temp_table_with_conn_timeout", "p0") {
+    String db = context.config.getDbNameByFile(context.file)
+    def tableName = "t_test_temp_table_with_conn_timeout"
+    String tempTableFullName
+    sql "select 1" // ensure db is created
+    connect(context.config.jdbcUser, context.config.jdbcPassword, context.config.jdbcUrl) {
+        sql"use ${db}"
+        sql """create temporary table ${tableName}(id int) properties("replication_num" = "1") """
+        def show_result = sql_return_maparray("show data")
+
+        show_result.each {  row ->
+            if (row.TableName.contains(tableName)) {
+                tempTableFullName = row.TableName
+            }
+        }
+        assert tempTableFullName != null
+
+        // set session variable for a short connection timeout
+        sql "set interactive_timeout=5"
+        sql "set wait_timeout=5"
+
+        sleep(10*1000)
+    }
+
+    // temp table should not exist after session exit
+    def tables = sql_return_maparray("show data")
+    assert tables.find { it.TableName == tempTableFullName } == null
+}