Fix the identity used by the asynchronous LOAD of the pipe receiver (#19029)

* Fix the identity used by the asynchronous LOAD of the pipe receiver

The load attributes of a TsFile received by the pipe receiver were built from the
default user instead of from the sender of the transfer. The receiver now passes
the authenticated sender to the active loader, and a seal request whose user can
not be determined is rejected. The type mismatch converter inherits the identity
of the load request. TsFiles placed directly in the active listening directory are
still loaded with the default user.

Tests: IoTDBPipeReceiverAsyncLoadAuthIT and IoTDBDataNodeReceiverTest.

* Fix the user resolution in the LOAD pipelines

Several places substituted the default user when a load carried no identity.
LoadUtil, ActiveLoadTsFileLoader and the type mismatch converter now require the
user of the load and fail otherwise, while a TsFile placed directly in the active
listening directory still falls back to the default user. The worker session of
the active loader is created after the per-file identity is known, and the thrift
pipe receiver returns NOT_LOGIN instead of dereferencing a missing session.

Tests: LoadUtilTest and IoTDBPipeReceiverAsyncLoadAuthIT.

* Fix the identity resolution of the pipe receiver

The V2 handshake took the user id from the handshake parameters, and the receiver
fell back to default credentials when the handshake carried none. The receiver now
resolves the identity from the handshake user name with the local user manager or
the permission manager, creates the audit entity only after a successful login,
and clears it when the handshake fails or the connection exits. Two null pointer
exceptions on the LOAD paths are fixed as well.

Tests: IoTDBFileReceiverTest and IoTDBDataNodeReceiverAuditEntityTest.

* Fix the identity binding of a pipe transfer connection

A connection that was already authenticated kept the identity of its first
transfer while the audit entity followed the new handshake. A handshake that does
not match the user of the connection is now rejected with NO_PERMISSION, and the
identity of a transfer is resolved from the verified user name, with the address
of the connection preferred over the hostname of the handshake.

Tests: IoTDBFileReceiverTest.

* Harden the user resolution of the pipe receiver and the load pipeline

- Use "{}" placeholders in the new LOG_ templates and keep the en and zh
  constants in parity.
- Fail closed when the session of a pipe connection is not logged in: an
  unauthenticated session still carries the id of the super user.
- Resolve the user id of a load from its user name instead of trusting the
  default id of the session, in the tree model schema verifier and in the
  TsFile data type converter.
- Allow a local TsFile that is dropped into a sub directory of the active
  listening directory to be loaded with the default user, while a file that
  carries attributes but no user is still rejected.

* Make the load user identity requirement configurable

The pipe receiver and the LOAD pipeline now fail when the user that
executes a load can not be resolved. Keep the legacy default-user
fallback by default and gate the strict behavior behind two internal
switches instead of changing the existing load code:

- pipe_receiver_require_user_identity (default false): an asynchronous
  LOAD of the pipe receiver without an authenticated sender is executed
  as the default user instead of being rejected.
- load_require_user_identity (default false): an asynchronous LOAD
  without a user is queued with the default user instead of letting the
  strict LoadUtil check reject it.

The switches are not added to iotdb-system.properties.template and
therefore stay internal. ActiveLoadTsFileLoader, LoadTsFileDataTypeConverter,
LoadUtil and TreeSchemaAutoCreatorAndVerifier keep the behavior of this
branch.

Tests: LoadUtilTest, ActiveLoadTsFileLoaderTest, IoTDBDataNodeReceiverTest
and LoadTsFileAnalyzerTest.

* Keep the load user identity switches minimal

Reduce the diff of the configurable load identity requirement:

- Move the LOAD switch back into LoadUtil so LoadTsFileAnalyzer keeps the
  behavior of this branch.
- Read the two switches only when the properties are loaded; they are not
  hot reloadable.
- Shorten the comments and javadocs of the new switches.

ActiveLoadTsFileLoader, LoadTsFileDataTypeConverter, TreeSchemaAutoCreatorAndVerifier
and the asynchronous LOAD acknowledgement semantics are unchanged.

Tests: LoadUtilTest.

* Drop the redundant user resolution in the LOAD pipeline

The load path already carries a locally resolved user and LoadUtil always
queues a user when the strict switch is enabled, so the extra resolution
layers are redundant:

- TreeSchemaAutoCreatorAndVerifier uses the user of the load session again
  instead of resolving the id per measurement.
- ActiveLoadTsFileLoader uses the user attribute with the default user as
  fallback, instead of walking the listening directory and failing files.
- Remove the i18n constants that are no longer used.

Tests: LoadUtilTest, ActiveLoadTsFileLoaderTest, IoTDBDataNodeReceiverTest,
LoadTsFileAnalyzerTest.

* Keep the legacy user fallback of LoadUtil

LoadUtil only needs to use the user name of the session when it is present
and fall back to the default user otherwise, which is what the original
code does. Drop the strict switch and restore the original method.

Remove the now unused i18n constant.

Tests: LoadUtilTest, ActiveLoadTsFileLoaderTest, IoTDBDataNodeReceiverTest,
LoadTsFileAnalyzerTest.

* Require a logged in user when the load identity switches are enabled

- pipe_receiver_require_user_identity: the asynchronous LOAD of the pipe
  receiver requires the authenticated sender to be the logged in user of
  the connection.
- load_require_user_identity: a LOAD without a user requires a logged in
  session that carries a user name instead of falling back to the default
  user.

Both switches keep the legacy fallback when they are disabled.

Tests: LoadUtilTest, ActiveLoadTsFileLoaderTest, IoTDBDataNodeReceiverTest,
LoadTsFileAnalyzerTest.

* Align the legacy receiver runtime test and cover local LOAD options

- A V1 handshake carries no credentials, so the receiver runtime reports an
  unknown user instead of the default user. Update the legacy receiver
  expectation accordingly, keeping the removal of privileged defaults.
- Add a regression test for a locally dropped TsFile under an attribute
  directory without a user; such a local input must still be loaded with the
  default user.

* Require an authenticated sender for asynchronous LOAD by default

Enable pipe_receiver_require_user_identity by default: the receiver only
queues an asynchronous LOAD when the sender is authenticated and the
connection is logged in as that user. Set the switch to false to restore
the legacy default user fallback. The load_require_user_identity switch
keeps its default (false).

Tests: LoadUtilTest, ActiveLoadTsFileLoaderTest, IoTDBDataNodeReceiverTest,
LoadTsFileAnalyzerTest.

* Compare the number of points of the replicas instead of their TsFile sizes

The restart case waited until every data region replica on the restarted
DataNode had the same on-disk size as the other replicas of that region. A
replica that replayed the LOAD can keep the same points in fewer physical
TsFiles than the replicas that stayed up, so the sizes never converge and
the case fails with a size mismatch that says nothing about the data.

Hard-link the TsFiles of every replica of a region into
<temp dir>/<region>/<DataNode>/, read them back point by point, and wait
until all replicas of the region hold the same number of points. Duplicate
physical copies collapse in the point set, so the check no longer depends on
how the same data is spread over files. Every waiting attempt deletes its
snapshots, and tearDownClass removes what an interrupted attempt left behind.

Author: luoluoyuyu

integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileClusterIT.java

@@ -452,7 +452,9 @@ private boolean hasOldSenderReceiverRuntime(
             && receiverDataNodeId == getInt(resultSet, "ReceiverNodeId", "receiver_node_id")
             && "thrift".equals(getString(resultSet, "Protocol", "protocol"))
             && "Unknown".equals(getString(resultSet, "SenderClusterId", "sender_cluster_id"))
-            && SessionConfig.DEFAULT_USER.equals(getString(resultSet, "UserName", "user_name"))
+            // A V1 handshake carries no credentials, so the receiver reports no user instead of
+            // the privileged default user.
+            && "Unknown".equals(getString(resultSet, "UserName", "user_name"))
             && getString(resultSet, "SenderAddress", "sender_address") != null
             && getString(resultSet, "SenderPorts", "sender_ports") != null
             && getInt(resultSet, "ConnectionCount", "connection_count") >= 1

integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBPipeReceiverAsyncLoadAuthIT.java

@@ -720,4 +720,7 @@ private ManagerMessages() {}
 
   public static final String MESSAGE_CONFIGNODE_LEADER_IS_WAITING_FOR_NODE_STATUS_PERSISTENCE_8CA96809 =
       "ConfigNode leader is waiting for node status persistence.";
+
+  public static final String LOG_FAILED_TO_RESOLVE_THE_USER_ID_OF_THE_PIPE_SENDER_ARG_THE_AUDIT_RECORDS_OF_THIS_TRANSFER_WILL_NOT_CONTAIN_A_USER_ID_0F8C4256 =
+      "Failed to resolve the user id of the pipe sender {}, the audit records of this transfer will not contain a user id.";
 }

integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBShowReceiversIT.java

@@ -699,4 +699,7 @@ private ManagerMessages() {}
 
   public static final String MESSAGE_CONFIGNODE_LEADER_IS_WAITING_FOR_NODE_STATUS_PERSISTENCE_8CA96809 =
       "ConfigNode leader 正在等待节点状态持久化。";
+
+  public static final String LOG_FAILED_TO_RESOLVE_THE_USER_ID_OF_THE_PIPE_SENDER_ARG_THE_AUDIT_RECORDS_OF_THIS_TRANSFER_WILL_NOT_CONTAIN_A_USER_ID_0F8C4256 =
+      "无法解析管道发送端 {} 的用户 ID，本次传输的审计记录将不包含用户 ID。";
 }

iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java

@@ -135,6 +135,7 @@
 import org.apache.iotdb.confignode.rpc.thrift.TDropTriggerReq;
 import org.apache.iotdb.confignode.rpc.thrift.TSetSchemaTemplateReq;
 import org.apache.iotdb.confignode.rpc.thrift.TUnsetSchemaTemplateReq;
+import org.apache.iotdb.confignode.rpc.thrift.TUserResp;
 import org.apache.iotdb.confignode.service.ConfigNode;
 import org.apache.iotdb.consensus.exception.ConsensusException;
 import org.apache.iotdb.db.protocol.session.IClientSession;
@@ -1395,6 +1396,34 @@ protected TSStatus login() {
     return configManager.login(username, password, false).getStatus();
   }
 
+  @Override
+  protected String getLoggedInSessionUsername() {
+    final IClientSession clientSession = SESSION_MANAGER.getCurrSession();
+    return clientSession != null && clientSession.isLogin() ? clientSession.getUsername() : null;
+  }
+
+  @Override
+  protected IAuditEntity createAuditEntity(final String username, final String cliHostname) {
+    // The user id must be resolved from the local permission manager instead of the handshake,
+    // because the id supplied by the sender is not authoritative.
+    long userId = -1L;
+    if (username != null) {
+      try {
+        final TUserResp user = configManager.getPermissionManager().getUser(username).getUserInfo();
+        if (user != null) {
+          userId = user.getUserId();
+        }
+      } catch (final Exception e) {
+        LOGGER.warn(
+            ManagerMessages
+                .LOG_FAILED_TO_RESOLVE_THE_USER_ID_OF_THE_PIPE_SENDER_ARG_THE_AUDIT_RECORDS_OF_THIS_TRANSFER_WILL_NOT_CONTAIN_A_USER_ID_0F8C4256,
+            username,
+            e);
+      }
+    }
+    return new UserEntity(userId, username, cliHostname);
+  }
+
   @Override
   protected String getReceiverFileBaseDir() {
     return ConfigNodeDescriptor.getInstance().getConf().getPipeReceiverFileDir();

iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java

@@ -1762,4 +1762,9 @@ private StorageEngineMessages() {}
       "Failed to sweep the staged directories of the finished LOAD tasks.";
   public static final String LOG_FAILED_TO_CREATE_SNAPSHOT_FOR_DATABASE_ARG_REGION_ARG_1A7C8DEA =
       "Failed to create snapshot for database {}, region {}";
+
+  public static final String EXCEPTION_FAILED_TO_DETERMINE_THE_USER_THAT_LOADS_THE_TSFILE_E47F7207 =
+      "Failed to determine the user that loads the TsFile";
+  public static final String EXCEPTION_FAILED_TO_DETERMINE_THE_USER_FOR_THE_TSFILE_DATA_TYPE_CONVERSION_A221793E =
+      "Failed to determine the user for the TsFile data type conversion";
 }

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/receiver/protocol/IoTDBConfigNodeReceiver.java

@@ -1681,4 +1681,9 @@ private StorageEngineMessages() {}
       "扫描已完成的 LOAD 任务暂存目录失败。";
   public static final String LOG_FAILED_TO_CREATE_SNAPSHOT_FOR_DATABASE_ARG_REGION_ARG_1A7C8DEA =
       "为 database {}、region {} 创建快照失败";
+
+  public static final String EXCEPTION_FAILED_TO_DETERMINE_THE_USER_THAT_LOADS_THE_TSFILE_E47F7207 =
+      "无法确定加载该 TsFile 的用户";
+  public static final String EXCEPTION_FAILED_TO_DETERMINE_THE_USER_FOR_THE_TSFILE_DATA_TYPE_CONVERSION_A221793E =
+      "无法确定执行 TsFile 数据类型转换的用户";
 }

iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java

@@ -1240,6 +1240,18 @@ public class IoTDBConfig implements DefaultEncodingProvider {
 
   private boolean loadActiveListeningVerifyEnable = true;
 
+  /**
+   * Whether a LOAD requires a logged in session with a user name instead of falling back to the
+   * default user. Disabled by default to keep the legacy fallback.
+   */
+  private boolean loadRequireUserIdentity = false;
+
+  /**
+   * Whether the pipe receiver rejects an asynchronous LOAD without an authenticated sender. Enabled
+   * by default; set to false to fall back to the default user.
+   */
+  private boolean pipeReceiverRequireUserIdentity = true;
+
   private String loadDiskSelectStrategy = LoadDiskSelectorType.MIN_IO_FIRST.getValue();
 
   private String loadDiskSelectStrategyForIoTV2AndPipe =
@@ -4365,6 +4377,22 @@ public void setLoadActiveListeningVerifyEnable(boolean loadActiveListeningVerify
     this.loadActiveListeningVerifyEnable = loadActiveListeningVerifyEnable;
   }
 
+  public boolean isLoadRequireUserIdentity() {
+    return loadRequireUserIdentity;
+  }
+
+  public void setLoadRequireUserIdentity(boolean loadRequireUserIdentity) {
+    this.loadRequireUserIdentity = loadRequireUserIdentity;
+  }
+
+  public boolean isPipeReceiverRequireUserIdentity() {
+    return pipeReceiverRequireUserIdentity;
+  }
+
+  public void setPipeReceiverRequireUserIdentity(boolean pipeReceiverRequireUserIdentity) {
+    this.pipeReceiverRequireUserIdentity = pipeReceiverRequireUserIdentity;
+  }
+
   public String getLoadDiskSelectStrategy() {
     return loadDiskSelectStrategy;
   }

iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java

@@ -2709,6 +2709,17 @@ private void loadLoadTsFileProps(TrimProperties properties) {
                 "load_active_listening_verify_enable",
                 Boolean.toString(conf.isLoadActiveListeningVerifyEnable()))));
 
+    conf.setLoadRequireUserIdentity(
+        Boolean.parseBoolean(
+            properties.getProperty(
+                "load_require_user_identity", Boolean.toString(conf.isLoadRequireUserIdentity()))));
+
+    conf.setPipeReceiverRequireUserIdentity(
+        Boolean.parseBoolean(
+            properties.getProperty(
+                "pipe_receiver_require_user_identity",
+                Boolean.toString(conf.isPipeReceiverRequireUserIdentity()))));
+
     conf.setLoadDiskSelectStrategy(
         properties.getProperty(
             "load_disk_select_strategy",