Pipe: Fixed the issue that frequent logins on the receiving end caused indicator leaks (apache#16076)

* Pipe: Fixed the issue that frequent logins on the receiving end caused indicator leaks

* fix

* fix

* spotless

Author: luoluoyuyu

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java

@@ -908,15 +908,28 @@ private TSStatus executeStatementWithPermissionCheckAndRetryOnDataTypeMismatch(
 
   @Override
   protected TSStatus login() {
-    final BasicOpenSessionResp openSessionResp =
-        SESSION_MANAGER.login(
-            SESSION_MANAGER.getCurrSession(),
-            username,
-            password,
-            ZoneId.systemDefault().toString(),
-            SessionManager.CURRENT_RPC_VERSION,
-            IoTDBConstant.ClientVersion.V_1_0);
-    return RpcUtils.getStatus(openSessionResp.getCode(), openSessionResp.getMessage());
+    final IClientSession session = SESSION_MANAGER.getCurrSession();
+
+    if (session != null && !session.isLogin()) {
+      final BasicOpenSessionResp openSessionResp =
+          SESSION_MANAGER.login(
+              session,
+              username,
+              password,
+              ZoneId.systemDefault().toString(),
+              SessionManager.CURRENT_RPC_VERSION,
+              IoTDBConstant.ClientVersion.V_1_0);
+      return RpcUtils.getStatus(openSessionResp.getCode(), openSessionResp.getMessage());
+    }
+
+    Long timeToExpire = SESSION_MANAGER.checkPasswordExpiration(username, password);
+    if (timeToExpire != null && timeToExpire <= System.currentTimeMillis()) {
+      return RpcUtils.getStatus(
+          TSStatusCode.ILLEGAL_PASSWORD.getStatusCode(),
+          "Password has expired, please use \"ALTER USER\" to change to a new one");
+    }
+
+    return AuthorityChecker.checkUser(username, password);
   }
 
   private TSStatus executeStatementForTableModel(

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

@@ -139,7 +139,7 @@ public BasicOpenSessionResp login(
    * @return the timestamp when the password will expire. Long.MAX if the password never expires.
    *     Null if the password history cannot be found.
    */
-  private Long checkPasswordExpiration(String username, String password) {
+  public Long checkPasswordExpiration(String username, String password) {
     // check password expiration
     long passwordExpirationDays =
         CommonDescriptor.getInstance().getConfig().getPasswordExpirationDays();