[draft](connectionContext) Fix potential memory leak by ensuring ConnectContext#cleanup() is called when ConnectionContext is no longer in use

fe/fe-core/src/main/java/org/apache/doris/cloud/CacheHotspotManagerUtils.java

@@ -266,7 +266,7 @@ public AutoCloseConnectContext(ConnectContext connectContext) {
 
         @Override
         public void close() {
-            ConnectContext.remove();
+            connectContext.cleanup();
             if (previousContext != null) {
                 previousContext.setThreadLocalInfo();
             }

fe/fe-core/src/main/java/org/apache/doris/cloud/CloudWarmUpJob.java

@@ -489,7 +489,7 @@ public synchronized void run() {
         } finally {
             if (Config.isCloudMode()) {
                 LOG.debug("remove context from job");
-                ConnectContext.remove();
+                ConnectContext.get().cleanup();
             }
         }
     }

fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java

@@ -1066,7 +1066,7 @@ public TPipelineFragmentParams plan(NereidsStreamLoadPlanner planner, TUniqueId
             return planParams;
         } finally {
             if (needCleanCtx) {
-                ConnectContext.remove();
+                ConnectContext.get().cleanup();
             }
             table.readUnlock();
         }

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

@@ -37,7 +37,7 @@ public void call() {
     @Override
     public void close() {
         connectContext.clear();
-        ConnectContext.remove();
+        connectContext.cleanup();
         if (previousContext != null) {
             previousContext.setThreadLocalInfo();
         }

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

@@ -327,6 +327,9 @@ public static ConnectContext get() {
         return MoreFieldsThread.getConnectContext();
     }
 
+    // This method is used to remove ConnectContext from current thread, but do not clean up resource used.
+    // if you finish the whole job, should call cleanup method directly. Call this remove method
+    // only in the tmp state during job.
     public static void remove() {
         MoreFieldsThread.removeConnectContext();
     }

fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java

@@ -1121,7 +1121,7 @@ public TMasterOpResult forward(TMasterOpRequest params) throws TException {
         } else {
             context.getState().setOk();
         }
-        ConnectContext.remove();
+        context.cleanup();
         clearCallback.run();
         return result;
     }
@@ -3255,7 +3255,7 @@ private TRestoreSnapshotResult restoreSnapshotImpl(TRestoreSnapshotRequest reque
             status.setStatusCode(TStatusCode.INTERNAL_ERROR);
             status.addToErrorMsgs(Strings.nullToEmpty(e.getMessage()) + ", command: " + restoreCommand);
         } finally {
-            ConnectContext.remove();
+            ConnectContext.get().cleanup();
         }
 
         return result;
@@ -4447,7 +4447,7 @@ public TFetchLoadJobResult fetchLoadJob(TFetchLoadJobRequest request) {
 
             return result;
         } finally {
-            ConnectContext.remove();
+            ctx.cleanup();
         }
     }
 

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

@@ -201,6 +201,6 @@ public void close() throws Exception {
         returnResultFromRemoteExecutor.clear();
         executor.finalizeQuery();
         ctx.clear();
-        ConnectContext.remove();
+        ctx.cleanup();
     }
 }