[To dev/1.3] Avoiding error log by removing ThreadLocal for Mqtt Client

(cherry picked from commit 8a8a7be)

Author: JackieTien97

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/mqtt/MPPPublishHandler.java

@@ -88,6 +88,7 @@ public void onConnect(InterceptConnectMessage msg) {
           ZoneId.systemDefault().toString(),
           TSProtocolVersion.IOTDB_SERVICE_PROTOCOL_V3,
           ClientVersion.V_1_0);
+      sessionManager.registerSessionForMqtt(session);
       clientIdToSessionMap.put(msg.getClientID(), session);
     }
   }
@@ -96,6 +97,7 @@ public void onConnect(InterceptConnectMessage msg) {
   public void onDisconnect(InterceptDisconnectMessage msg) {
     MqttClientSession session = clientIdToSessionMap.remove(msg.getClientID());
     if (null != session) {
+      sessionManager.removeCurrSessionForMqtt(session);
       sessionManager.closeSession(session, Coordinator.getInstance()::cleanupQueryExecution);
     }
   }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/session/SessionManager.java

@@ -316,6 +316,12 @@ public void removeCurrSession() {
     currSessionIdleTime.remove();
   }
 
+  public void removeCurrSessionForMqtt(MqttClientSession mqttClientSession) {
+    if (mqttClientSession != null) {
+      sessions.remove(mqttClientSession);
+    }
+  }
+
   /**
    * this method can be only used in client-thread model. Do not use this method in message-thread
    * model based service.
@@ -333,6 +339,14 @@ public boolean registerSession(IClientSession session) {
     return true;
   }
 
+  /**
+   * this method can be only used in mqtt model. Do not use this method in client-thread model based
+   * service.
+   */
+  public void registerSessionForMqtt(IClientSession session) {
+    sessions.put(session, placeHolder);
+  }
+
   /** must be called after registerSession()) will mark the session login. */
   public void supplySession(
       IClientSession session,