This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new a4d1a508100 Pipe: Applied the login function to config receiver
(#15258)
a4d1a508100 is described below
commit a4d1a50810010d83f04d8241856164d887ed8b90
Author: Caideyipi <[email protected]>
AuthorDate: Wed Apr 2 17:29:20 2025 +0800
Pipe: Applied the login function to config receiver (#15258)
---
.../receiver/protocol/IoTDBConfigNodeReceiver.java | 12 ++--
.../protocol/thrift/IoTDBDataNodeReceiver.java | 82 ++++++----------------
.../commons/pipe/receiver/IoTDBFileReceiver.java | 31 +++++++-
3 files changed, 57 insertions(+), 68 deletions(-)
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/receiver/protocol/IoTDBConfigNodeReceiver.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/receiver/protocol/IoTDBConfigNodeReceiver.java
index 8a86d679b14..79f6326ffac 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/receiver/protocol/IoTDBConfigNodeReceiver.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/receiver/protocol/IoTDBConfigNodeReceiver.java
@@ -827,13 +827,17 @@ public class IoTDBConfigNodeReceiver extends
IoTDBFileReceiver {
@Override
protected String getClusterId() {
- return
ConfigNode.getInstance().getConfigManager().getClusterManager().getClusterId();
+ return configManager.getClusterManager().getClusterId();
}
@Override
- protected TSStatus tryLogin() {
- // Do nothing. Login check will be done in the data node receiver.
- return StatusUtils.getStatus(TSStatusCode.SUCCESS_STATUS);
+ protected boolean shouldLogin() {
+ return lastSuccessfulLoginTime == Long.MIN_VALUE || super.shouldLogin();
+ }
+
+ @Override
+ protected TSStatus login() {
+ return configManager.login(username, password).getStatus();
}
@Override
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java
index ea2574f47ae..513c7b0790f 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/receiver/protocol/thrift/IoTDBDataNodeReceiver.java
@@ -39,7 +39,6 @@ import
org.apache.iotdb.commons.pipe.receiver.PipeReceiverStatusHandler;
import org.apache.iotdb.commons.schema.column.ColumnHeaderConstant;
import org.apache.iotdb.commons.utils.FileUtils;
import org.apache.iotdb.commons.utils.RetryUtils;
-import org.apache.iotdb.commons.utils.StatusUtils;
import org.apache.iotdb.confignode.rpc.thrift.TDatabaseSchema;
import org.apache.iotdb.db.auth.AuthorityChecker;
import org.apache.iotdb.db.conf.IoTDBConfig;
@@ -178,9 +177,6 @@ public class IoTDBDataNodeReceiver extends
IoTDBFileReceiver {
private final SqlParser tableSqlParser = new SqlParser();
private static final SessionManager SESSION_MANAGER =
SessionManager.getInstance();
- private static final long LOGIN_PERIODIC_VERIFICATION_INTERVAL_MS =
-
PipeConfig.getInstance().getPipeReceiverLoginPeriodicVerificationIntervalMs();
- private long lastSuccessfulLoginTime = Long.MIN_VALUE;
private static final double ACTUAL_TO_ESTIMATED_MEMORY_RATIO =
PipeConfig.getInstance().getPipeReceiverActualToEstimatedMemoryRatio();
@@ -500,27 +496,10 @@ public class IoTDBDataNodeReceiver extends
IoTDBFileReceiver {
}
@Override
- protected TSStatus tryLogin() {
- final IClientSession clientSession = SESSION_MANAGER.getCurrSession();
- if (clientSession == null
- || !clientSession.isLogin()
- || (LOGIN_PERIODIC_VERIFICATION_INTERVAL_MS >= 0
- && lastSuccessfulLoginTime
- < System.currentTimeMillis() -
LOGIN_PERIODIC_VERIFICATION_INTERVAL_MS)) {
- final TSStatus status =
- SESSION_MANAGER.login(
- SESSION_MANAGER.getCurrSession(),
- username,
- password,
- ZoneId.systemDefault().toString(),
- SessionManager.CURRENT_RPC_VERSION,
- IoTDBConstant.ClientVersion.V_1_0);
- if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- lastSuccessfulLoginTime = System.currentTimeMillis();
- }
- return status;
- }
- return StatusUtils.getStatus(TSStatusCode.SUCCESS_STATUS);
+ protected boolean shouldLogin() {
+ // The idle time is updated per request
+ final IClientSession clientSession =
SESSION_MANAGER.getCurrSessionAndUpdateIdleTime();
+ return clientSession == null || !clientSession.isLogin() ||
super.shouldLogin();
}
@Override
@@ -869,23 +848,18 @@ public class IoTDBDataNodeReceiver extends
IoTDBFileReceiver {
databaseName = null;
}
+ // Permission check
+ final TSStatus loginStatus = loginIfNecessary();
+ if (loginStatus.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ return loginStatus;
+ }
+
+ final IClientSession clientSession = SESSION_MANAGER.getCurrSession();
+
// For table model, the authority check is done in inner execution. No
need to check here
if (!isTableModelStatement) {
- // Permission check
- TSStatus permissionCheckStatus;
- IClientSession clientSession =
SESSION_MANAGER.getCurrSessionAndUpdateIdleTime();
- if (clientSession == null
- || !clientSession.isLogin()
- || (LOGIN_PERIODIC_VERIFICATION_INTERVAL_MS >= 0
- && lastSuccessfulLoginTime
- < System.currentTimeMillis() -
LOGIN_PERIODIC_VERIFICATION_INTERVAL_MS)) {
- permissionCheckStatus = login();
- if (permissionCheckStatus.getCode() !=
TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- return permissionCheckStatus;
- }
- clientSession = SESSION_MANAGER.getCurrSession();
- }
- permissionCheckStatus = AuthorityChecker.checkAuthority(statement,
clientSession);
+ final TSStatus permissionCheckStatus =
+ AuthorityChecker.checkAuthority(statement, clientSession);
if (permissionCheckStatus.getCode() !=
TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
LOGGER.warn(
"Receiver id = {}: Failed to check authority for statement {},
username = {}, response = {}.",
@@ -920,7 +894,8 @@ public class IoTDBDataNodeReceiver extends
IoTDBFileReceiver {
: status;
}
- private TSStatus login() {
+ @Override
+ protected TSStatus login() {
final BasicOpenSessionResp openSessionResp =
SESSION_MANAGER.login(
SESSION_MANAGER.getCurrSession(),
@@ -929,16 +904,7 @@ public class IoTDBDataNodeReceiver extends
IoTDBFileReceiver {
ZoneId.systemDefault().toString(),
SessionManager.CURRENT_RPC_VERSION,
IoTDBConstant.ClientVersion.V_1_0);
- if (openSessionResp.getCode() !=
TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- LOGGER.warn(
- "Receiver id = {}: Failed to open session, username = {}, response =
{}.",
- receiverId.get(),
- username,
- openSessionResp);
- return RpcUtils.getStatus(openSessionResp.getCode(),
openSessionResp.getMessage());
- }
- lastSuccessfulLoginTime = System.currentTimeMillis();
- return RpcUtils.SUCCESS_STATUS;
+ return RpcUtils.getStatus(openSessionResp.getCode(),
openSessionResp.getMessage());
}
private TSStatus executeStatementForTableModel(
@@ -1039,17 +1005,9 @@ public class IoTDBDataNodeReceiver extends
IoTDBFileReceiver {
final org.apache.iotdb.db.queryengine.plan.relational.sql.ast.Statement
statement,
final String databaseName) {
try {
- // Permission check
- final IClientSession clientSession =
SESSION_MANAGER.getCurrSessionAndUpdateIdleTime();
- if (clientSession == null
- || !clientSession.isLogin()
- || (LOGIN_PERIODIC_VERIFICATION_INTERVAL_MS >= 0
- && lastSuccessfulLoginTime
- < System.currentTimeMillis() -
LOGIN_PERIODIC_VERIFICATION_INTERVAL_MS)) {
- final TSStatus result = login();
- if (result.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
- return result;
- }
+ final TSStatus status = loginIfNecessary();
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ return status;
}
final TSStatus result =
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java
index 069f4a6a15d..a733fcde0a6 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/receiver/IoTDBFileReceiver.java
@@ -74,6 +74,10 @@ public abstract class IoTDBFileReceiver implements
IoTDBReceiver {
protected String username = CONNECTOR_IOTDB_USER_DEFAULT_VALUE;
protected String password = CONNECTOR_IOTDB_PASSWORD_DEFAULT_VALUE;
+ private static final long LOGIN_PERIODIC_VERIFICATION_INTERVAL_MS =
+
PipeConfig.getInstance().getPipeReceiverLoginPeriodicVerificationIntervalMs();
+ protected long lastSuccessfulLoginTime = Long.MIN_VALUE;
+
private static final boolean IS_FSYNC_ENABLED =
PipeConfig.getInstance().getPipeFileReceiverFsyncEnabled();
private File writingFile;
@@ -258,7 +262,7 @@ public abstract class IoTDBFileReceiver implements
IoTDBReceiver {
if (passwordString != null) {
password = passwordString;
}
- final TSStatus status = tryLogin();
+ final TSStatus status = loginIfNecessary();
if (status.code != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
LOGGER.warn(
"Receiver id = {}: Handshake failed because login failed, response
status = {}.",
@@ -311,7 +315,30 @@ public abstract class IoTDBFileReceiver implements
IoTDBReceiver {
protected abstract String getClusterId();
- protected abstract TSStatus tryLogin();
+ protected boolean shouldLogin() {
+ return LOGIN_PERIODIC_VERIFICATION_INTERVAL_MS >= 0
+ && lastSuccessfulLoginTime
+ < System.currentTimeMillis() -
LOGIN_PERIODIC_VERIFICATION_INTERVAL_MS;
+ }
+
+ protected TSStatus loginIfNecessary() {
+ if (shouldLogin()) {
+ final TSStatus permissionCheckStatus = login();
+ if (permissionCheckStatus.getCode() !=
TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ LOGGER.warn(
+ "Receiver id = {}: Failed to login, username = {}, response = {}.",
+ receiverId.get(),
+ username,
+ permissionCheckStatus);
+ return permissionCheckStatus;
+ } else {
+ lastSuccessfulLoginTime = System.currentTimeMillis();
+ }
+ }
+ return StatusUtils.OK;
+ }
+
+ protected abstract TSStatus login();
protected final TPipeTransferResp handleTransferFilePiece(
final PipeTransferFilePieceReq req,
