diff --git a/iotdb-client/isession/src/main/java/org/apache/iotdb/isession/IInternalSessionBuilder.java b/iotdb-client/isession/src/main/java/org/apache/iotdb/isession/IInternalSessionBuilder.java new file mode 100644 index 0000000000000..ae67fa4093da0 --- /dev/null +++ b/iotdb-client/isession/src/main/java/org/apache/iotdb/isession/IInternalSessionBuilder.java @@ -0,0 +1,32 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.isession; + +import org.apache.iotdb.rpc.IoTDBConnectionException; + +/** Capability implemented by builders that can construct sessions on internal endpoints. */ +public interface IInternalSessionBuilder { + + /** + * Builds a session using the caller-provided internal node supplier. Whether the returned session + * is already open follows the builder's ordinary build contract. + */ + T buildInternal(INodeSupplier nodeSupplier) throws IoTDBConnectionException; +} diff --git a/iotdb-client/session/src/main/i18n/en/org/apache/iotdb/session/i18n/SessionMessages.java b/iotdb-client/session/src/main/i18n/en/org/apache/iotdb/session/i18n/SessionMessages.java index ded0deac31484..382567f378d61 100644 --- a/iotdb-client/session/src/main/i18n/en/org/apache/iotdb/session/i18n/SessionMessages.java +++ b/iotdb-client/session/src/main/i18n/en/org/apache/iotdb/session/i18n/SessionMessages.java @@ -20,6 +20,11 @@ package org.apache.iotdb.session.i18n; public final class SessionMessages { + public static final String EXCEPTION_NO_DATANODE_INTERNAL_ENDPOINT_IS_AVAILABLE_249D1081 = + "No DataNode internal endpoint is available"; + + public static final String EXCEPTION_INTERNAL_NODE_SUPPLIER_MUST_NOT_BE_NULL_C5CAF392 = + "Internal node supplier must not be null"; // Session public static final String NODE_URLS_EMPTY = "nodeUrls shouldn't be empty."; diff --git a/iotdb-client/session/src/main/i18n/zh/org/apache/iotdb/session/i18n/SessionMessages.java b/iotdb-client/session/src/main/i18n/zh/org/apache/iotdb/session/i18n/SessionMessages.java index 838a7e2148329..9975c43a74cd7 100644 --- a/iotdb-client/session/src/main/i18n/zh/org/apache/iotdb/session/i18n/SessionMessages.java +++ b/iotdb-client/session/src/main/i18n/zh/org/apache/iotdb/session/i18n/SessionMessages.java @@ -20,6 +20,11 @@ package org.apache.iotdb.session.i18n; public final class SessionMessages { + public static final String EXCEPTION_NO_DATANODE_INTERNAL_ENDPOINT_IS_AVAILABLE_249D1081 = + "没有可用的 DataNode 内部端点"; + + public static final String EXCEPTION_INTERNAL_NODE_SUPPLIER_MUST_NOT_BE_NULL_C5CAF392 = + "内部节点 supplier 不能为空"; // Session public static final String NODE_URLS_EMPTY = "nodeUrls 不能为空。"; diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/InternalSession.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/InternalSession.java new file mode 100644 index 0000000000000..fdcaed13f3964 --- /dev/null +++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/InternalSession.java @@ -0,0 +1,77 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.session; + +import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.isession.INodeSupplier; +import org.apache.iotdb.rpc.IoTDBConnectionException; +import org.apache.iotdb.session.i18n.SessionMessages; + +import java.time.ZoneId; +import java.util.List; +import java.util.Objects; + +/** Session using caller-supplied internal endpoints and the caller's username. */ +public class InternalSession extends Session { + + private final INodeSupplier nodeSupplier; + + public InternalSession(AbstractSessionBuilder builder, INodeSupplier nodeSupplier) { + super(builder); + this.nodeSupplier = + Objects.requireNonNull( + nodeSupplier, + SessionMessages.EXCEPTION_INTERNAL_NODE_SUPPLIER_MUST_NOT_BE_NULL_C5CAF392); + password = null; + useEncryptedPassword = false; + enableAutoFetch = false; + enableRedirection = false; + enableQueryRedirection = false; + } + + @Override + public synchronized void open() throws IoTDBConnectionException { + open(enableThriftRpcCompaction, connectionTimeoutInMs); + } + + @Override + protected INodeSupplier createNodeSupplier(boolean enableRPCCompaction, int timeoutInMs) { + return nodeSupplier; + } + + @Override + public SessionConnection constructSessionConnection( + Session session, TEndPoint endpoint, ZoneId zoneId) throws IoTDBConnectionException { + final List candidates = availableNodes.get(); + if (candidates.isEmpty()) { + throw new IoTDBConnectionException( + SessionMessages.EXCEPTION_NO_DATANODE_INTERNAL_ENDPOINT_IS_AVAILABLE_249D1081); + } + defaultEndPoint = + candidates.contains(endpoint) + ? endpoint.deepCopy() + : nodeSupplier + .getQueryEndPoint() + .filter(candidates::contains) + .orElse(candidates.get(0)) + .deepCopy(); + return super.constructSessionConnection(session, defaultEndPoint, zoneId); + } +} diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java index de3ee417afc0d..f415a63246490 100644 --- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java +++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java @@ -21,6 +21,7 @@ import org.apache.iotdb.common.rpc.thrift.TAggregationType; import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.isession.IInternalSessionBuilder; import org.apache.iotdb.isession.INodeSupplier; import org.apache.iotdb.isession.ISession; import org.apache.iotdb.isession.SessionConfig; @@ -549,33 +550,7 @@ public synchronized void open(boolean enableRPCCompaction, int connectionTimeout this.availableNodes = null; } - if (enableAutoFetch) { - initThreadPool(); - this.availableNodes = - NodesSupplier.createNodeSupplier( - getNodeUrls(), - SessionUtils.parseSeedNodeUrls(initialNodeUrls), - executorService, - username, - password, - zoneId, - thriftDefaultBufferSize, - thriftMaxFrameSize, - connectionTimeoutInMs, - useSSL, - trustStore, - trustStorePwd, - keyStore, - keyStorePwd, - sslProtocol, - enableRPCCompaction, - version.toString()); - } else { - // A static supplier has no separate seed tier. Keep every configured address after reopen. - Set staticNodes = new LinkedHashSet<>(getNodeUrls()); - staticNodes.addAll(SessionUtils.parseSeedNodeUrls(initialNodeUrls)); - this.availableNodes = new DummyNodesSupplier(new ArrayList<>(staticNodes)); - } + this.availableNodes = createNodeSupplier(enableRPCCompaction, connectionTimeoutInMs); this.connectionTimeoutInMs = connectionTimeoutInMs; setDefaultSessionConnection(constructSessionConnection(this, defaultEndPoint, zoneId)); @@ -589,6 +564,37 @@ public synchronized void open(boolean enableRPCCompaction, int connectionTimeout } } + /** Creates the endpoint source used by initial connections and reconnects. */ + protected INodeSupplier createNodeSupplier( + boolean enableRPCCompaction, int connectionTimeoutInMs) { + if (enableAutoFetch) { + initThreadPool(); + return NodesSupplier.createNodeSupplier( + getNodeUrls(), + SessionUtils.parseSeedNodeUrls(initialNodeUrls), + executorService, + username, + password, + zoneId, + thriftDefaultBufferSize, + thriftMaxFrameSize, + connectionTimeoutInMs, + useSSL, + trustStore, + trustStorePwd, + keyStore, + keyStorePwd, + sslProtocol, + enableRPCCompaction, + version.toString()); + } else { + // A static supplier has no separate seed tier. Keep every configured address after reopen. + Set staticNodes = new LinkedHashSet<>(getNodeUrls()); + staticNodes.addAll(SessionUtils.parseSeedNodeUrls(initialNodeUrls)); + return new DummyNodesSupplier(new ArrayList<>(staticNodes)); + } + } + private void initThreadPool() { this.executorService = Executors.newSingleThreadScheduledExecutor( @@ -4445,7 +4451,8 @@ public void setEnableThriftRpcCompaction(boolean enableThriftRpcCompaction) { this.enableThriftRpcCompaction = enableThriftRpcCompaction; } - public static class Builder extends AbstractSessionBuilder { + public static class Builder extends AbstractSessionBuilder + implements IInternalSessionBuilder { public Builder host(String host) { this.host = host; @@ -4595,5 +4602,11 @@ public Session build() { } return new Session(this); } + + /** Builds an unopened internal session using caller-supplied internal endpoints. */ + @Override + public InternalSession buildInternal(INodeSupplier nodeSupplier) { + return new InternalSession(this, nodeSupplier); + } } } diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/TableSessionBuilder.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/TableSessionBuilder.java index a39af4c847a95..856ed7a91bc6b 100644 --- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/TableSessionBuilder.java +++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/TableSessionBuilder.java @@ -19,6 +19,8 @@ package org.apache.iotdb.session; +import org.apache.iotdb.isession.IInternalSessionBuilder; +import org.apache.iotdb.isession.INodeSupplier; import org.apache.iotdb.isession.ITableSession; import org.apache.iotdb.isession.SessionConfig; import org.apache.iotdb.rpc.IoTDBConnectionException; @@ -41,7 +43,8 @@ * *

All configurations have reasonable default values, which can be overridden as needed. */ -public class TableSessionBuilder extends AbstractSessionBuilder { +public class TableSessionBuilder extends AbstractSessionBuilder + implements IInternalSessionBuilder { private boolean enableCompression = false; private int connectionTimeoutInMs = SessionConfig.DEFAULT_CONNECTION_TIMEOUT_MS; @@ -437,7 +440,17 @@ public ITableSession build() throws IoTDBConnectionException { Collections.singletonList(SessionConfig.DEFAULT_HOST + ":" + SessionConfig.DEFAULT_PORT); } this.sqlDialect = TABLE; - Session newSession = new Session(this); + return buildSession(new Session(this)); + } + + /** Builds and opens a table session using caller-supplied internal endpoints. */ + @Override + public ITableSession buildInternal(INodeSupplier nodeSupplier) throws IoTDBConnectionException { + this.sqlDialect = TABLE; + return buildSession(new InternalSession(this, nodeSupplier)); + } + + private ITableSession buildSession(Session newSession) throws IoTDBConnectionException { newSession.enableIoTDBRpcCompression = isIoTDBRpcCompressionEnabled; newSession.tabletCompressionMinRowSize = tabletCompressionMinRowSize; diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/InternalTableSessionPool.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/InternalTableSessionPool.java new file mode 100644 index 0000000000000..b4d8ef74a6cd1 --- /dev/null +++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/InternalTableSessionPool.java @@ -0,0 +1,38 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.session.pool; + +import org.apache.iotdb.isession.INodeSupplier; +import org.apache.iotdb.session.i18n.SessionMessages; + +import java.util.Objects; + +/** Table session pool using caller-supplied internal endpoints and the caller's username. */ +public class InternalTableSessionPool extends TableSessionPool { + + InternalTableSessionPool(AbstractSessionPoolBuilder builder, INodeSupplier nodeSupplier) { + super( + new SessionPool( + builder, + Objects.requireNonNull( + nodeSupplier, + SessionMessages.EXCEPTION_INTERNAL_NODE_SUPPLIER_MUST_NOT_BE_NULL_C5CAF392))); + } +} diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java index 16b79c6415c4b..088afed4e2352 100644 --- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java +++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/SessionPool.java @@ -173,6 +173,8 @@ public class SessionPool implements ISessionPool { private INodeSupplier availableNodes; + private INodeSupplier internalNodeSupplier; + // set to true, means that we will start a background thread to fetch all available (Status is // not Removing) datanodes in cluster, and these available nodes will be used in retrying stage private boolean enableAutoFetch = true; @@ -526,15 +528,20 @@ public SessionPool( } public SessionPool(AbstractSessionPoolBuilder builder) { + this(builder, null); + } + + SessionPool(AbstractSessionPoolBuilder builder, INodeSupplier internalNodeSupplier) { + this.internalNodeSupplier = internalNodeSupplier; this.maxSize = builder.maxSize; this.user = builder.username; - this.password = builder.pw; + this.password = internalNodeSupplier == null ? builder.pw : null; this.fetchSize = builder.fetchSize; this.waitToGetSessionTimeoutInMs = builder.waitToGetSessionTimeoutInMs; this.enableThriftCompression = builder.isThriftRpcCompactionEnabled; this.enableIoTDBRpcCompression = builder.isIoTDBRpcCompressionEnabled; this.zoneId = builder.zoneId; - this.enableRedirection = builder.enableRedirection; + this.enableRedirection = internalNodeSupplier == null && builder.enableRedirection; if (this.enableRedirection) { deviceIdToEndpoint = new ConcurrentHashMap<>(); tableModelDeviceIdToEndpoint = new ConcurrentHashMap<>(); @@ -544,7 +551,7 @@ public SessionPool(AbstractSessionPoolBuilder builder) { this.version = builder.version; this.thriftDefaultBufferSize = builder.thriftDefaultBufferSize; this.thriftMaxFrameSize = builder.thriftMaxFrameSize; - this.enableAutoFetch = builder.enableAutoFetch; + this.enableAutoFetch = internalNodeSupplier == null && builder.enableAutoFetch; this.useSSL = builder.useSSL; this.trustStore = builder.trustStore; this.trustStorePwd = builder.trustStorePwd; @@ -561,7 +568,13 @@ public SessionPool(AbstractSessionPoolBuilder builder) { if (enableAutoFetch) { initThreadPool(); } - if (builder.nodeUrls != null) { + if (internalNodeSupplier != null) { + this.host = null; + this.port = -1; + this.nodeUrls = null; + this.formattedNodeUrls = internalNodeSupplier.getClass().getSimpleName(); + this.availableNodes = internalNodeSupplier; + } else if (builder.nodeUrls != null) { if (builder.nodeUrls.isEmpty()) { throw new IllegalArgumentException(SessionMessages.NODE_URLS_EMPTY); } @@ -590,67 +603,40 @@ public SessionPool(AbstractSessionPoolBuilder builder) { } private Session constructNewSession() { - Session session; + Session.Builder builder = + new Session.Builder() + .username(user) + .password(password) + .fetchSize(fetchSize) + .zoneId(zoneId) + .thriftDefaultBufferSize(thriftDefaultBufferSize) + .thriftMaxFrameSize(thriftMaxFrameSize) + .enableRedirection(enableRedirection) + .enableRecordsAutoConvertTablet(enableRecordsAutoConvertTablet) + .version(version) + .useSSL(useSSL) + .trustStore(trustStore) + .trustStorePwd(trustStorePwd) + .keyStore(keyStore) + .keyStorePwd(keyStorePwd) + .sslProtocol(sslProtocol) + .maxRetryCount(maxRetryCount) + .retryIntervalInMs(retryIntervalInMs) + .sqlDialect(sqlDialect) + .database(database) + .enableAutoCreateSchema(enableAutoCreateSchema) + .timeOut(queryTimeoutInMs) + .enableIoTDBRpcCompression(enableIoTDBRpcCompression) + .enableThriftRpcCompression(enableThriftCompression); + if (internalNodeSupplier != null) { + return builder.buildInternal(internalNodeSupplier); + } if (nodeUrls == null) { - // Construct custom Session - session = - new Session.Builder() - .host(host) - .port(port) - .username(user) - .password(password) - .fetchSize(fetchSize) - .zoneId(zoneId) - .thriftDefaultBufferSize(thriftDefaultBufferSize) - .thriftMaxFrameSize(thriftMaxFrameSize) - .enableRedirection(enableRedirection) - .enableRecordsAutoConvertTablet(enableRecordsAutoConvertTablet) - .version(version) - .useSSL(useSSL) - .trustStore(trustStore) - .trustStorePwd(trustStorePwd) - .keyStore(keyStore) - .keyStorePwd(keyStorePwd) - .sslProtocol(sslProtocol) - .maxRetryCount(maxRetryCount) - .retryIntervalInMs(retryIntervalInMs) - .sqlDialect(sqlDialect) - .database(database) - .enableAutoCreateSchema(enableAutoCreateSchema) - .timeOut(queryTimeoutInMs) - .enableIoTDBRpcCompression(enableIoTDBRpcCompression) - .enableThriftRpcCompression(enableThriftCompression) - .build(); + builder.host(host).port(port); } else { - // Construct redirect-able Session - session = - new Session.Builder() - .nodeUrls(nodeUrls) - .username(user) - .password(password) - .fetchSize(fetchSize) - .zoneId(zoneId) - .thriftDefaultBufferSize(thriftDefaultBufferSize) - .thriftMaxFrameSize(thriftMaxFrameSize) - .enableRedirection(enableRedirection) - .enableRecordsAutoConvertTablet(enableRecordsAutoConvertTablet) - .version(version) - .useSSL(useSSL) - .trustStore(trustStore) - .trustStorePwd(trustStorePwd) - .keyStore(keyStore) - .keyStorePwd(keyStorePwd) - .sslProtocol(sslProtocol) - .maxRetryCount(maxRetryCount) - .retryIntervalInMs(retryIntervalInMs) - .sqlDialect(sqlDialect) - .database(database) - .enableAutoCreateSchema(enableAutoCreateSchema) - .timeOut(queryTimeoutInMs) - .enableIoTDBRpcCompression(enableIoTDBRpcCompression) - .enableThriftRpcCompression(enableThriftCompression) - .build(); + builder.nodeUrls(nodeUrls); } + Session session = builder.build(); session.setEnableQueryRedirection(enableQueryRedirection); return session; } @@ -849,7 +835,9 @@ public synchronized void close() { this.executorService = null; } if (availableNodes != null) { - this.availableNodes.close(); + if (internalNodeSupplier == null) { + this.availableNodes.close(); + } this.availableNodes = null; } LOGGER.info(SessionMessages.CLOSING_SESSION_POOL); @@ -3544,16 +3532,16 @@ public boolean isEnableThriftCompression() { @Override public void setEnableRedirection(boolean enableRedirection) { - this.enableRedirection = enableRedirection; + this.enableRedirection = internalNodeSupplier == null && enableRedirection; if (this.enableRedirection) { deviceIdToEndpoint = new ConcurrentHashMap<>(); tableModelDeviceIdToEndpoint = new ConcurrentHashMap<>(); } for (ISession session : queue) { - session.setEnableRedirection(enableRedirection); + session.setEnableRedirection(this.enableRedirection); } for (ISession session : occupied.keySet()) { - session.setEnableRedirection(enableRedirection); + session.setEnableRedirection(this.enableRedirection); } } @@ -3564,12 +3552,12 @@ public boolean isEnableRedirection() { @Override public void setEnableQueryRedirection(boolean enableQueryRedirection) { - this.enableQueryRedirection = enableQueryRedirection; + this.enableQueryRedirection = internalNodeSupplier == null && enableQueryRedirection; for (ISession session : queue) { - session.setEnableQueryRedirection(enableQueryRedirection); + session.setEnableQueryRedirection(this.enableQueryRedirection); } for (ISession session : occupied.keySet()) { - session.setEnableQueryRedirection(enableQueryRedirection); + session.setEnableQueryRedirection(this.enableQueryRedirection); } } diff --git a/iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/TableSessionPoolBuilder.java b/iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/TableSessionPoolBuilder.java index f19c5fc8878a6..294f40b8190c7 100644 --- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/TableSessionPoolBuilder.java +++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/pool/TableSessionPoolBuilder.java @@ -19,6 +19,8 @@ package org.apache.iotdb.session.pool; +import org.apache.iotdb.isession.IInternalSessionBuilder; +import org.apache.iotdb.isession.INodeSupplier; import org.apache.iotdb.isession.SessionConfig; import org.apache.iotdb.isession.pool.ITableSessionPool; @@ -34,7 +36,8 @@ * *

All configurations have reasonable default values, which can be overridden as needed. */ -public class TableSessionPoolBuilder extends AbstractSessionPoolBuilder { +public class TableSessionPoolBuilder extends AbstractSessionPoolBuilder + implements IInternalSessionBuilder { /** * Sets the list of node URLs for the IoTDB cluster. @@ -346,4 +349,11 @@ public ITableSessionPool build() { SessionPool sessionPool = new SessionPool(this); return new TableSessionPool(sessionPool); } + + /** Builds a pool using the caller's internal endpoints and username; connections open lazily. */ + @Override + public ITableSessionPool buildInternal(INodeSupplier nodeSupplier) { + this.sqlDialect = "table"; + return new InternalTableSessionPool(this, nodeSupplier); + } } diff --git a/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/InternalSubscriptionTableSession.java b/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/InternalSubscriptionTableSession.java new file mode 100644 index 0000000000000..13cf8c659ed94 --- /dev/null +++ b/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/InternalSubscriptionTableSession.java @@ -0,0 +1,83 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.session.subscription; + +import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.isession.INodeSupplier; +import org.apache.iotdb.rpc.IoTDBConnectionException; +import org.apache.iotdb.session.AbstractSessionBuilder; +import org.apache.iotdb.session.Session; +import org.apache.iotdb.session.SessionConnection; +import org.apache.iotdb.session.i18n.SessionMessages; + +import java.time.ZoneId; +import java.util.List; +import java.util.Objects; + +/** Table subscription-management session using the caller's identity and internal endpoints. */ +public class InternalSubscriptionTableSession extends SubscriptionTableSession { + + public InternalSubscriptionTableSession( + AbstractSessionBuilder builder, INodeSupplier nodeSupplier) { + super(new InternalSessionWrapper(builder, nodeSupplier)); + } + + private static class InternalSessionWrapper extends SubscriptionSessionWrapper { + + private final INodeSupplier nodeSupplier; + + private InternalSessionWrapper(AbstractSessionBuilder builder, INodeSupplier nodeSupplier) { + super(builder); + this.nodeSupplier = + Objects.requireNonNull( + nodeSupplier, + SessionMessages.EXCEPTION_INTERNAL_NODE_SUPPLIER_MUST_NOT_BE_NULL_C5CAF392); + password = null; + useEncryptedPassword = false; + enableAutoFetch = false; + enableRedirection = false; + enableQueryRedirection = false; + } + + @Override + protected INodeSupplier createNodeSupplier(boolean enableRPCCompaction, int timeoutInMs) { + return nodeSupplier; + } + + @Override + public SessionConnection constructSessionConnection( + Session session, TEndPoint endpoint, ZoneId zoneId) throws IoTDBConnectionException { + final List candidates = nodeSupplier.get(); + if (candidates.isEmpty()) { + throw new IoTDBConnectionException( + SessionMessages.EXCEPTION_NO_DATANODE_INTERNAL_ENDPOINT_IS_AVAILABLE_249D1081); + } + defaultEndPoint = + candidates.contains(endpoint) + ? endpoint.deepCopy() + : nodeSupplier + .getQueryEndPoint() + .filter(candidates::contains) + .orElse(candidates.get(0)) + .deepCopy(); + return super.constructSessionConnection(session, defaultEndPoint, zoneId); + } + } +} diff --git a/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/SubscriptionSessionWrapper.java b/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/SubscriptionSessionWrapper.java index d7e715fa62762..018f4ef3a185d 100644 --- a/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/SubscriptionSessionWrapper.java +++ b/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/SubscriptionSessionWrapper.java @@ -30,7 +30,7 @@ import java.time.ZoneId; import java.util.Objects; -public final class SubscriptionSessionWrapper extends Session { +public class SubscriptionSessionWrapper extends Session { public SubscriptionSessionWrapper(final AbstractSessionBuilder builder) { super(builder); diff --git a/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/SubscriptionTableSession.java b/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/SubscriptionTableSession.java index 123deea2fa45d..8949c46ff83ad 100644 --- a/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/SubscriptionTableSession.java +++ b/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/SubscriptionTableSession.java @@ -36,6 +36,10 @@ public SubscriptionTableSession(final AbstractSessionBuilder builder) { super(new SubscriptionSessionWrapper(builder)); } + protected SubscriptionTableSession(final SubscriptionSessionWrapper session) { + super(session); + } + /////////////////////////////// open & close /////////////////////////////// @Override diff --git a/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/SubscriptionTableSessionBuilder.java b/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/SubscriptionTableSessionBuilder.java index e0c8e95968dc9..0f34d27ee4980 100644 --- a/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/SubscriptionTableSessionBuilder.java +++ b/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/SubscriptionTableSessionBuilder.java @@ -19,10 +19,13 @@ package org.apache.iotdb.session.subscription; +import org.apache.iotdb.isession.IInternalSessionBuilder; +import org.apache.iotdb.isession.INodeSupplier; import org.apache.iotdb.rpc.IoTDBConnectionException; import org.apache.iotdb.session.AbstractSessionBuilder; -public class SubscriptionTableSessionBuilder extends AbstractSessionBuilder { +public class SubscriptionTableSessionBuilder extends AbstractSessionBuilder + implements IInternalSessionBuilder { public SubscriptionTableSessionBuilder() { // use table model @@ -73,4 +76,23 @@ public ISubscriptionTableSession build() throws IoTDBConnectionException { session.open(); return session; } + + /** Builds and opens a table subscription-management session on internal endpoints. */ + @Override + public ISubscriptionTableSession buildInternal(INodeSupplier nodeSupplier) + throws IoTDBConnectionException { + final ISubscriptionTableSession session = + new InternalSubscriptionTableSession(this, nodeSupplier); + try { + session.open(); + } catch (IoTDBConnectionException e) { + try { + session.close(); + } catch (Exception closeException) { + e.addSuppressed(closeException); + } + throw e; + } + return session; + } } diff --git a/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java b/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java index 7f60b82a0e225..4a284c44255cc 100644 --- a/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java +++ b/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java @@ -20,6 +20,7 @@ package org.apache.iotdb.session.subscription.consumer.base; import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.isession.INodeSupplier; import org.apache.iotdb.isession.SessionConfig; import org.apache.iotdb.rpc.subscription.config.ConsumerConstant; import org.apache.iotdb.rpc.subscription.config.TopicConfig; @@ -572,6 +573,11 @@ private void seekAfterInternalTopicProgress( /////////////////////////////// subscription provider /////////////////////////////// + /** Returns the complete internal endpoint source, or null to use server discovery. */ + protected INodeSupplier getInternalNodeSupplier() { + return null; + } + protected abstract AbstractSubscriptionProvider constructSubscriptionProvider( final TEndPoint endPoint, final String username, diff --git a/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionProviders.java b/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionProviders.java index afb5285b5f577..e332b6c7931fa 100644 --- a/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionProviders.java +++ b/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionProviders.java @@ -34,6 +34,7 @@ import java.util.ArrayList; import java.util.LinkedHashMap; +import java.util.LinkedHashSet; import java.util.List; import java.util.Map; import java.util.Objects; @@ -92,6 +93,11 @@ void openProviders(final AbstractSubscriptionConsumer consumer) throws Subscript // close stale providers closeProviders(false); + if (consumer.getInternalNodeSupplier() != null) { + syncInternalProviders(consumer); + return; + } + final Map connectionFailures = new LinkedHashMap<>(); final List connectionFailureCauses = new ArrayList<>(); for (final TEndPoint endPoint : initialEndpoints) { @@ -457,6 +463,16 @@ private void syncInternal(final AbstractSubscriptionConsumer consumer) { if (consumer.isClosed() || consumer.isFenced()) { return; } + if (consumer.getInternalNodeSupplier() != null) { + try { + syncInternalProviders(consumer); + } catch (final SubscriptionConsumerFencedException e) { + consumer.fence(e); + } catch (final Exception e) { + LOGGER.warn(SubscriptionMessages.FETCH_ENDPOINTS_FAILED, consumer, e, e); + } + return; + } if (hasNoAvailableProviders()) { try { openProviders(consumer); @@ -567,6 +583,73 @@ private void syncInternal(final AbstractSubscriptionConsumer consumer) { } } + /** Reconciles providers against a complete endpoint list without server-side discovery. */ + private void syncInternalProviders(final AbstractSubscriptionConsumer consumer) { + final Set endpoints = + consumer.getInternalNodeSupplier().get().stream() + .map(TEndPoint::deepCopy) + .collect(Collectors.toCollection(LinkedHashSet::new)); + + for (final AbstractSubscriptionProvider provider : getAllProviders()) { + if (!provider.isAvailable() || !endpoints.contains(provider.getEndPoint())) { + try { + closeAndRemoveProvider(provider.getDataNodeId()); + } catch (final Exception e) { + LOGGER.warn(SubscriptionMessages.PROVIDER_CLOSE_FAILED, provider, e, e); + } + } + } + + final Set connectedEndpoints = + getAllAvailableProviders().stream() + .map(AbstractSubscriptionProvider::getEndPoint) + .collect(Collectors.toSet()); + Exception lastFailure = null; + for (final TEndPoint endpoint : endpoints) { + if (isClosing.get() || consumer.isFenced()) { + return; + } + if (connectedEndpoints.contains(endpoint)) { + continue; + } + try { + final AbstractSubscriptionProvider provider = + consumer.constructProviderAndHandshake(endpoint); + if (containsProvider(provider.getDataNodeId())) { + try { + closeAndRemoveProvider(provider.getDataNodeId()); + } catch (final Exception e) { + LOGGER.warn(SubscriptionMessages.PROVIDER_CLOSE_FAILED, provider, e, e); + } + } + addProvider(provider.getDataNodeId(), provider); + } catch (final SubscriptionConsumerFencedException e) { + consumer.fence(e); + throw e; + } catch (final Exception e) { + lastFailure = e; + LOGGER.warn( + SubscriptionMessages.LOG_ARG_FAILED_CREATE_CONNECTION_ARG_BECAUSE_ARG_E536E22A, + consumer, + endpoint, + e, + e); + } + } + + if (hasNoAvailableProviders()) { + throw new SubscriptionConnectionException( + String.format( + SubscriptionMessages + .EXCEPTION_CLUSTER_HAS_NO_AVAILABLE_SUBSCRIPTION_PROVIDERS_CONNECT_INITIAL_ENDPOINTS_ARG_5DB83198, + endpoints), + lastFailure); + } + if (!subscriptionProviders.containsKey(nextDataNodeId)) { + nextDataNodeId = subscriptionProviders.firstKey(); + } + } + @Override public String toString() { final StringBuilder sb = new StringBuilder(); diff --git a/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/table/InternalSubscriptionTablePullConsumer.java b/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/table/InternalSubscriptionTablePullConsumer.java new file mode 100644 index 0000000000000..32987f0b5c836 --- /dev/null +++ b/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/table/InternalSubscriptionTablePullConsumer.java @@ -0,0 +1,76 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.session.subscription.consumer.table; + +import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.isession.INodeSupplier; +import org.apache.iotdb.session.i18n.SessionMessages; +import org.apache.iotdb.session.subscription.consumer.base.AbstractSubscriptionProvider; + +import java.util.Objects; + +/** Table pull consumer using the caller's complete internal endpoint source and username. */ +public class InternalSubscriptionTablePullConsumer extends SubscriptionTablePullConsumer { + + private final INodeSupplier nodeSupplier; + + public InternalSubscriptionTablePullConsumer( + final SubscriptionTablePullConsumerBuilder builder, final INodeSupplier nodeSupplier) { + super(builder); + this.nodeSupplier = + Objects.requireNonNull( + nodeSupplier, + SessionMessages.EXCEPTION_INTERNAL_NODE_SUPPLIER_MUST_NOT_BE_NULL_C5CAF392); + } + + @Override + protected INodeSupplier getInternalNodeSupplier() { + return nodeSupplier; + } + + @Override + protected AbstractSubscriptionProvider constructSubscriptionProvider( + final TEndPoint endPoint, + final String username, + final String password, + final String encryptedPassword, + final String consumerId, + final String consumerGroupId, + final String ownerId, + final Long ownerEpoch, + final int thriftMaxFrameSize, + final long heartbeatIntervalMs, + final int connectionTimeoutInMs) { + // The internal RPC service authenticates the username without a password. Each provider's + // existing session wrapper stays pinned to this endpoint, including during reconnects. + return super.constructSubscriptionProvider( + endPoint, + username, + "", + null, + consumerId, + consumerGroupId, + ownerId, + ownerEpoch, + thriftMaxFrameSize, + heartbeatIntervalMs, + connectionTimeoutInMs); + } +} diff --git a/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/table/SubscriptionTablePullConsumerBuilder.java b/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/table/SubscriptionTablePullConsumerBuilder.java index efc99debb4977..7c5cbcdb8261d 100644 --- a/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/table/SubscriptionTablePullConsumerBuilder.java +++ b/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/table/SubscriptionTablePullConsumerBuilder.java @@ -19,12 +19,15 @@ package org.apache.iotdb.session.subscription.consumer.table; +import org.apache.iotdb.isession.IInternalSessionBuilder; +import org.apache.iotdb.isession.INodeSupplier; import org.apache.iotdb.session.subscription.consumer.ISubscriptionTablePullConsumer; import org.apache.iotdb.session.subscription.consumer.base.AbstractSubscriptionPullConsumerBuilder; import java.util.List; -public class SubscriptionTablePullConsumerBuilder extends AbstractSubscriptionPullConsumerBuilder { +public class SubscriptionTablePullConsumerBuilder extends AbstractSubscriptionPullConsumerBuilder + implements IInternalSessionBuilder { @Override public SubscriptionTablePullConsumerBuilder host(final String host) { @@ -152,4 +155,10 @@ public SubscriptionTablePullConsumerBuilder autoCommitIntervalMs( public ISubscriptionTablePullConsumer build() { return new SubscriptionTablePullConsumer(this); } + + /** Builds an unopened consumer using the caller's internal endpoint source and username. */ + @Override + public ISubscriptionTablePullConsumer buildInternal(final INodeSupplier nodeSupplier) { + return new InternalSubscriptionTablePullConsumer(this, nodeSupplier); + } } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnInternalServiceAsyncRequestManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnInternalServiceAsyncRequestManager.java index 3010f8b524c04..3cb2468c27893 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnInternalServiceAsyncRequestManager.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnInternalServiceAsyncRequestManager.java @@ -449,7 +449,8 @@ protected void initActionMapBuilder() { actionMapBuilder.put( CnToDnAsyncRequestType.INSERT_RECORD, (req, client, handler) -> - client.insertRecord((TSInsertRecordReq) req, (DataNodeTSStatusRPCHandler) handler)); + client.insertRecordInternal( + (TSInsertRecordReq) req, (DataNodeTSStatusRPCHandler) handler)); actionMapBuilder.put( CnToDnAsyncRequestType.UPDATE_TABLE, (req, client, handler) -> diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/sync/SyncDataNodeClientPool.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/sync/SyncDataNodeClientPool.java index f35b6d906a525..60c58ddea0d2b 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/sync/SyncDataNodeClientPool.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/sync/SyncDataNodeClientPool.java @@ -137,7 +137,8 @@ private void buildActionMap() { CnToDnSyncRequestType.RESET_PEER_LIST, (req, client) -> client.resetPeerList((TResetPeerListReq) req)); actionMapBuilder.put( - CnToDnSyncRequestType.SHOW_CONFIGURATION, (req, client) -> client.showConfiguration()); + CnToDnSyncRequestType.SHOW_CONFIGURATION, + (req, client) -> client.showConfigurationInternal()); actionMapBuilder.put( CnToDnSyncRequestType.SHOW_APPLIED_CONFIGURATIONS, (req, client) -> client.showAppliedConfigurations()); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/session/SessionManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/session/SessionManager.java index 85d9acae276ec..7017bda16794f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/session/SessionManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/session/SessionManager.java @@ -157,75 +157,118 @@ public BasicOpenSessionResp login( final TSStatus loginStatus = AuthorityChecker.checkUser(username, password, useEncryptedPassword); if (loginStatus.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) { - // check the version compatibility - if (!tsProtocolVersion.equals(CURRENT_RPC_VERSION)) { - openSessionResp - .sessionId(-1) - .setCode(TSStatusCode.INCOMPATIBLE_VERSION.getStatusCode()) - .setMessage( - DataNodeMiscMessages.VERSION_INCOMPATIBLE_PLEASE_UPGRADE_TO - + IoTDBConstant.VERSION); - } else { - session.setSqlDialect(sqlDialect); - supplySession(session, userId, username, ZoneId.of(zoneId), clientVersion); - String logInMessage = DataNodeMiscMessages.EXCEPTION_LOGIN_SUCCESSFULLY_1C2EF670; - openSessionResp - .sessionId(session.getId()) - .setCode(TSStatusCode.SUCCESS_STATUS.getStatusCode()) - .setMessage(logInMessage); - AUDIT_LOGGER.log( - new AuditLogFields( - userId, - username, - session.getClientAddress(), - AuditEventType.LOGIN, - AuditLogOperation.CONTROL, - true), - () -> - String.format( - DataNodeMiscMessages - .EXCEPTION_ARG_LOGIN_STATUS_ARG_USER_ARG_ID_ARG_OPENS_SESSION_ARG_70586F85, - IoTDBConstant.GLOBAL_DB_NAME, - openSessionResp.getMessage(), - username, - userId, - session)); - LOGGER.info( - DataNodeMiscMessages.LOGIN_STATUS, - IoTDBConstant.GLOBAL_DB_NAME, - openSessionResp.getMessage(), - username, - session); - updateIdleTime(); - if (enableLoginLock) { - loginLockManager.clearFailure(userId, session.getClientAddress()); - } + BasicOpenSessionResp response = + completeLogin( + session, userId, username, zoneId, tsProtocolVersion, clientVersion, sqlDialect); + if (response.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode() && enableLoginLock) { + loginLockManager.clearFailure(userId, session.getClientAddress()); } + return response; } else { openSessionResp.sessionId(-1).setMessage(loginStatus.message).setCode(loginStatus.code); if (enableLoginLock) { loginLockManager.recordFailure(userId, session.getClientAddress()); } - AUDIT_LOGGER.log( - new AuditLogFields( - userId, - username, - session.getClientAddress(), - AuditEventType.LOGIN, - AuditLogOperation.CONTROL, - false), - () -> - String.format( - DataNodeMiscMessages - .EXCEPTION_USER_ARG_ID_ARG_LOGIN_FAILED_WITH_CODE_ARG_ARG_93FAE4FD, - username, - userId, - loginStatus.getCode(), - loginStatus.getMessage())); + logFailedLogin(session, userId, username, loginStatus); } return openSessionResp; } + /** Opens a session for a trusted internal caller without checking a password. */ + public BasicOpenSessionResp loginInternal( + IClientSession session, + String username, + String zoneId, + TSProtocolVersion tsProtocolVersion, + IoTDBConstant.ClientVersion clientVersion, + SqlDialect sqlDialect) { + final long userId = AuthorityChecker.getUserId(username).orElse(-1L); + if (userId == -1) { + BasicOpenSessionResp response = new BasicOpenSessionResp(); + response + .sessionId(-1) + .setCode(TSStatusCode.USER_NOT_EXIST.getStatusCode()) + .setMessage( + String.format( + DataNodeMiscMessages.MISC_EXCEPTION_USER_S_DOES_NOT_EXIST_0CE725D8, username)); + logFailedLogin(session, userId, username, response); + return response; + } + return completeLogin( + session, userId, username, zoneId, tsProtocolVersion, clientVersion, sqlDialect); + } + + private BasicOpenSessionResp completeLogin( + IClientSession session, + long userId, + String username, + String zoneId, + TSProtocolVersion tsProtocolVersion, + IoTDBConstant.ClientVersion clientVersion, + SqlDialect sqlDialect) { + BasicOpenSessionResp response = new BasicOpenSessionResp(); + if (!tsProtocolVersion.equals(CURRENT_RPC_VERSION)) { + response + .sessionId(-1) + .setCode(TSStatusCode.INCOMPATIBLE_VERSION.getStatusCode()) + .setMessage( + DataNodeMiscMessages.VERSION_INCOMPATIBLE_PLEASE_UPGRADE_TO + IoTDBConstant.VERSION); + return response; + } + + session.setSqlDialect(sqlDialect); + supplySession(session, userId, username, ZoneId.of(zoneId), clientVersion); + response + .sessionId(session.getId()) + .setCode(TSStatusCode.SUCCESS_STATUS.getStatusCode()) + .setMessage(DataNodeMiscMessages.EXCEPTION_LOGIN_SUCCESSFULLY_1C2EF670); + AUDIT_LOGGER.log( + new AuditLogFields( + userId, + username, + session.getClientAddress(), + AuditEventType.LOGIN, + AuditLogOperation.CONTROL, + true), + () -> + String.format( + DataNodeMiscMessages + .EXCEPTION_ARG_LOGIN_STATUS_ARG_USER_ARG_ID_ARG_OPENS_SESSION_ARG_70586F85, + IoTDBConstant.GLOBAL_DB_NAME, + response.getMessage(), + username, + userId, + session)); + LOGGER.info( + DataNodeMiscMessages.LOGIN_STATUS, + IoTDBConstant.GLOBAL_DB_NAME, + response.getMessage(), + username, + session); + updateIdleTime(); + return response; + } + + private void logFailedLogin( + IClientSession session, long userId, String username, TSStatus status) { + AUDIT_LOGGER.log( + new AuditLogFields( + userId, + username, + session.getClientAddress(), + AuditEventType.LOGIN, + AuditLogOperation.CONTROL, + false), + () -> + String.format( + DataNodeMiscMessages + .EXCEPTION_USER_ARG_ID_ARG_LOGIN_FAILED_WITH_CODE_ARG_ARG_93FAE4FD, + username, + userId, + status.getCode(), + status.getMessage())); + } + public boolean closeSession(IClientSession session, LongConsumer releaseByQueryId) { return closeSession(session, releaseByQueryId, true); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/handler/InternalServiceThriftHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/handler/InternalServiceThriftHandler.java index 58130eb91cc58..831e12ea17bbd 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/handler/InternalServiceThriftHandler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/handler/InternalServiceThriftHandler.java @@ -21,19 +21,27 @@ import org.apache.iotdb.commons.service.NoopServerContext; import org.apache.iotdb.commons.service.metric.MetricService; +import org.apache.iotdb.db.protocol.session.ClientSession; +import org.apache.iotdb.db.protocol.session.SessionManager; +import org.apache.iotdb.db.protocol.thrift.impl.IClientRPCServiceWithHandler; +import org.apache.iotdb.rpc.TElasticFramedTransport; import org.apache.thrift.protocol.TProtocol; import org.apache.thrift.server.ServerContext; import org.apache.thrift.server.TServerEventHandler; +import org.apache.thrift.transport.TSocket; import org.apache.thrift.transport.TTransport; +import java.net.Socket; import java.util.concurrent.atomic.AtomicLong; public class InternalServiceThriftHandler implements TServerEventHandler { private AtomicLong thriftConnectionNumber = new AtomicLong(0); + private final IClientRPCServiceWithHandler eventHandler; - public InternalServiceThriftHandler() { + public InternalServiceThriftHandler(IClientRPCServiceWithHandler eventHandler) { + this.eventHandler = eventHandler; MetricService.getInstance() .addMetricSet(new InternalServiceThriftHandlerMetrics(thriftConnectionNumber)); } @@ -46,13 +54,21 @@ public void preServe() { @Override public ServerContext createContext(TProtocol tProtocol, TProtocol tProtocol1) { thriftConnectionNumber.incrementAndGet(); + Socket socket = + ((TSocket) ((TElasticFramedTransport) tProtocol1.getTransport()).getSocket()).getSocket(); + SessionManager.getInstance().registerSession(new ClientSession(socket)); return NoopServerContext.INSTANCE; } @Override public void deleteContext( ServerContext serverContext, TProtocol tProtocol, TProtocol tProtocol1) { - thriftConnectionNumber.decrementAndGet(); + try { + eventHandler.handleClientExit(); + } finally { + SessionManager.getInstance().removeCurrSession(); + thriftConnectionNumber.decrementAndGet(); + } } @Override diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java index fcb8fc23999ba..3e8927bdf6057 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/ClientRPCServiceImpl.java @@ -1626,16 +1626,7 @@ public TSOpenSessionResp openSession(TSOpenSessionReq req) throws TException { } Optional database = parseDatabase(req); IClientSession clientSession = SESSION_MANAGER.getCurrSession(); - BasicOpenSessionResp openSessionResp = - SESSION_MANAGER.login( - SESSION_MANAGER.getCurrSession(), - req.username, - req.password, - req.zoneId, - req.client_protocol, - clientVersion, - sqlDialect, - parseUseEncryptedPassword(req)); + BasicOpenSessionResp openSessionResp = login(clientSession, req, clientVersion, sqlDialect); TSStatus tsStatus = RpcUtils.getStatus(openSessionResp.getCode(), openSessionResp.getMessage()); if (tsStatus.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) { @@ -1651,6 +1642,22 @@ public TSOpenSessionResp openSession(TSOpenSessionReq req) throws TException { return resp.setSessionId(openSessionResp.getSessionId()).setConfiguration(configuration); } + protected BasicOpenSessionResp login( + IClientSession session, + TSOpenSessionReq req, + IoTDBConstant.ClientVersion clientVersion, + SqlDialect sqlDialect) { + return SESSION_MANAGER.login( + session, + req.username, + req.password, + req.zoneId, + req.client_protocol, + clientVersion, + sqlDialect, + parseUseEncryptedPassword(req)); + } + private IoTDBConstant.ClientVersion parseClientVersion(TSOpenSessionReq req) { Map configuration = req.configuration; if (configuration != null && configuration.containsKey("version")) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java index 4b1e05f4e857d..2523be59d7096 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java @@ -77,6 +77,7 @@ import org.apache.iotdb.commons.pipe.agent.plugin.meta.PipePluginMeta; import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMeta; import org.apache.iotdb.commons.queryengine.common.SessionInfo; +import org.apache.iotdb.commons.queryengine.common.SqlDialect; import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNode; import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeType; @@ -119,6 +120,7 @@ import org.apache.iotdb.db.i18n.DataNodeSchemaMessages; import org.apache.iotdb.db.partition.DataPartitionTableGenerator; import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent; +import org.apache.iotdb.db.protocol.basic.BasicOpenSessionResp; import org.apache.iotdb.db.protocol.client.ConfigNodeInfo; import org.apache.iotdb.db.protocol.client.cn.DnToCnInternalServiceAsyncRequestManager; import org.apache.iotdb.db.protocol.client.cn.DnToCnRequestType; @@ -350,6 +352,7 @@ import org.apache.iotdb.rpc.TSStatusCode; import org.apache.iotdb.rpc.subscription.exception.SubscriptionException; import org.apache.iotdb.service.rpc.thrift.TSInsertRecordReq; +import org.apache.iotdb.service.rpc.thrift.TSOpenSessionReq; import org.apache.iotdb.trigger.api.enums.FailureStrategy; import org.apache.iotdb.trigger.api.enums.TriggerEvent; @@ -416,7 +419,8 @@ import static org.apache.iotdb.db.utils.ErrorHandlingUtils.onIoTDBException; import static org.apache.iotdb.db.utils.ErrorHandlingUtils.onQueryException; -public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface { +public class DataNodeInternalRPCServiceImpl extends ClientRPCServiceImpl + implements IDataNodeRPCService.Iface { private static final Logger LOGGER = LoggerFactory.getLogger(DataNodeInternalRPCServiceImpl.class); @@ -473,6 +477,8 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface private static final String SYSTEM = "system"; + private long consensusWaitTimeoutSeconds = 30; + public DataNodeInternalRPCServiceImpl(DataNodeContext dataNodeContext) { this(dataNodeContext, SystemMetrics.getInstance()); } @@ -485,8 +491,6 @@ public DataNodeInternalRPCServiceImpl(DataNodeContext dataNodeContext) { this.dataNodeContext = dataNodeContext; } - private long consensusWaitTimeoutSeconds = 30; - private TSStatus waitForConsensusStarted() { if (dataNodeContext.isAllConsensusStarted()) { return null; @@ -508,6 +512,16 @@ private TSStatus waitForConsensusStarted() { } } + @Override + protected BasicOpenSessionResp login( + IClientSession session, + TSOpenSessionReq req, + ClientVersion clientVersion, + SqlDialect sqlDialect) { + return SESSION_MANAGER.loginInternal( + session, req.username, req.zoneId, req.client_protocol, clientVersion, sqlDialect); + } + @Override public TSendFragmentInstanceResp sendFragmentInstance(final TSendFragmentInstanceReq req) { LOGGER.debug(DataNodeMiscMessages.RECEIVE_FRAGMENT_INSTANCE, req.getConsensusGroupId()); @@ -1839,9 +1853,10 @@ private TSStatus executeInternalSchemaTask( @Override public TSStatus executeCQ(TExecuteCQ req) { + IClientSession previousSession = SESSION_MANAGER.getCurrSession(); IClientSession session = new InternalClientSession(req.cqId); - SESSION_MANAGER.registerSession(session); + SESSION_MANAGER.setCurrSession(session); SESSION_MANAGER.supplySession( session, @@ -1923,8 +1938,11 @@ public TSStatus executeCQ(TExecuteCQ req) { } catch (Exception e) { return onQueryException(e, "\"" + executedSQL + "\". " + OperationType.EXECUTE_STATEMENT); } finally { - SESSION_MANAGER.closeSession(session, COORDINATOR::cleanupQueryExecution); - SESSION_MANAGER.removeCurrSession(); + try { + SESSION_MANAGER.closeSession(session, COORDINATOR::cleanupQueryExecution); + } finally { + SESSION_MANAGER.restoreSession(previousSession, session); + } } } @@ -2379,11 +2397,6 @@ private List testAllDataNodeExternalServiceConnection( handler, TEST_CONNECTION_RETRY_NUM, TEST_CONNECTION_TIMEOUT_MS, true)); } - @Override - public TSStatus testConnectionEmptyRPC() throws TException { - return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()); - } - @Override public TSStatus updateClusterTopology(TUpdateClusterTopologyReq req) throws TException { clusterTopology.updateTopology(req.getDataNodes(), req.getTopology()); @@ -2810,7 +2823,7 @@ public TSStatus loadConfiguration() throws TException { } @Override - public TShowConfigurationResp showConfiguration() { + public TShowConfigurationResp showConfigurationInternal() { TShowConfigurationResp resp = new TShowConfigurationResp(RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS), ""); try { @@ -3562,7 +3575,7 @@ public TSStatus stopAndClearDataNode() { } @Override - public TSStatus insertRecord(TSInsertRecordReq req) throws TException { + public TSStatus insertRecordInternal(TSInsertRecordReq req) { try { InsertRowStatement statement = StatementGenerator.createStatement(req); SessionInfo sessionInfo = @@ -3606,10 +3619,6 @@ public TSStatus writeAuditLog(TAuditLogReq req) { return RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS); } - public void handleClientExit() { - // Do nothing - } - // ==================================================== // Data Partition Table Integrity Check Implementation // ==================================================== diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeInternalRPCService.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeInternalRPCService.java index f9b9c6e010110..ef0afbf295651 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeInternalRPCService.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeInternalRPCService.java @@ -78,7 +78,7 @@ public void initThriftServiceThread() throws IllegalAccessException { config.getRpcMaxConcurrentClientNum(), config.getThriftServerAwaitTimeForStopService(), new TrustedChannelAuditServerEventHandler( - new InternalServiceThriftHandler(), + new InternalServiceThriftHandler(getImpl()), new TEndPoint(getBindIP(), getBindPort()), DNAuditLogger.getInstance()::recordTrustedChannelFailureAuditLogIfNecessary), config.isRpcThriftCompressionEnable(), @@ -95,7 +95,7 @@ public void initThriftServiceThread() throws IllegalAccessException { getBindPort(), config.getRpcMaxConcurrentClientNum(), config.getThriftServerAwaitTimeForStopService(), - new InternalServiceThriftHandler(), + new InternalServiceThriftHandler(getImpl()), config.isRpcThriftCompressionEnable(), DeepCopyRpcTransportFactory.INSTANCE); } catch (RPCServiceException e) { diff --git a/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift b/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift index 57afa70e72846..7415b96bd652e 100644 --- a/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift +++ b/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift @@ -931,7 +931,7 @@ struct TFetchDeviceEntrySegmentResp { * END: Used for EXPLAIN ANALYZE **/ -service IDataNodeRPCService { +service IDataNodeRPCService extends client.IClientRPCService { // -----------------------------------For Data Node----------------------------------------------- @@ -1186,7 +1186,7 @@ service IDataNodeRPCService { common.TSStatus clearCache(set cacheClearOptions) - common.TShowConfigurationResp showConfiguration() + common.TShowConfigurationResp showConfigurationInternal() common.TShowAppliedConfigurationsResp showAppliedConfigurations() @@ -1434,17 +1434,14 @@ service IDataNodeRPCService { common.TTestConnectionResp submitInternalTestConnectionTask(common.TNodeLocations nodeLocations) - /** Empty rpc, only for connection test */ - common.TSStatus testConnectionEmptyRPC() - /** * Push cluster topology to this DataNode. * Each DataNode receives only its own reachable set. */ common.TSStatus updateClusterTopology(1:TUpdateClusterTopologyReq req) - /** to write audit log or other events as time series **/ - common.TSStatus insertRecord(1:client.TSInsertRecordReq req); + /** Write internal audit records independently of client session identity. */ + common.TSStatus insertRecordInternal(1:client.TSInsertRecordReq req); /** * Write an audit log entry to the DataNode's AuditEventLogger