JackieTien97 commented on code in PR #18561: URL: https://github.com/apache/iotdb/pull/18561#discussion_r3911247886
########## iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/AbstractConfigNodeClient.java: ########## @@ -0,0 +1,1331 @@ +/* + * 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.commons.client; + +import org.apache.iotdb.common.rpc.thrift.TConfigNodeLocation; +import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation; +import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.common.rpc.thrift.TExternalServiceListResp; +import org.apache.iotdb.common.rpc.thrift.TFlushReq; +import org.apache.iotdb.common.rpc.thrift.TNodeLocations; +import org.apache.iotdb.common.rpc.thrift.TPipeHeartbeatResp; +import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot; +import org.apache.iotdb.common.rpc.thrift.TSetConfigurationReq; +import org.apache.iotdb.common.rpc.thrift.TSetSpaceQuotaReq; +import org.apache.iotdb.common.rpc.thrift.TSetTTLReq; +import org.apache.iotdb.common.rpc.thrift.TSetThrottleQuotaReq; +import org.apache.iotdb.common.rpc.thrift.TShowAppliedConfigurationsResp; +import org.apache.iotdb.common.rpc.thrift.TShowConfigurationResp; +import org.apache.iotdb.common.rpc.thrift.TShowTTLReq; +import org.apache.iotdb.common.rpc.thrift.TTestConnectionResp; +import org.apache.iotdb.commons.client.property.ThriftClientProperty; +import org.apache.iotdb.commons.conf.CommonConfig; +import org.apache.iotdb.commons.conf.CommonDescriptor; +import org.apache.iotdb.commons.consensus.ConfigRegionId; +import org.apache.iotdb.commons.i18n.ConfigMessages; +import org.apache.iotdb.confignode.rpc.thrift.IConfigNodeRPCService; +import org.apache.iotdb.confignode.rpc.thrift.TAINodeConfigurationResp; +import org.apache.iotdb.confignode.rpc.thrift.TAINodeRegisterReq; +import org.apache.iotdb.confignode.rpc.thrift.TAINodeRegisterResp; +import org.apache.iotdb.confignode.rpc.thrift.TAINodeRemoveReq; +import org.apache.iotdb.confignode.rpc.thrift.TAINodeRestartReq; +import org.apache.iotdb.confignode.rpc.thrift.TAINodeRestartResp; +import org.apache.iotdb.confignode.rpc.thrift.TAddConsensusGroupReq; +import org.apache.iotdb.confignode.rpc.thrift.TAlterEncodingCompressorReq; +import org.apache.iotdb.confignode.rpc.thrift.TAlterLogicalViewReq; +import org.apache.iotdb.confignode.rpc.thrift.TAlterOrDropTableReq; +import org.apache.iotdb.confignode.rpc.thrift.TAlterPipeReq; +import org.apache.iotdb.confignode.rpc.thrift.TAlterSchemaTemplateReq; +import org.apache.iotdb.confignode.rpc.thrift.TAlterTimeSeriesReq; +import org.apache.iotdb.confignode.rpc.thrift.TAlterTopicReq; +import org.apache.iotdb.confignode.rpc.thrift.TAuthizedPatternTreeResp; +import org.apache.iotdb.confignode.rpc.thrift.TAuthorizerRelationalReq; +import org.apache.iotdb.confignode.rpc.thrift.TAuthorizerReq; +import org.apache.iotdb.confignode.rpc.thrift.TAuthorizerResp; +import org.apache.iotdb.confignode.rpc.thrift.TCheckUserPrivilegesReq; +import org.apache.iotdb.confignode.rpc.thrift.TCloseConsumerReq; +import org.apache.iotdb.confignode.rpc.thrift.TConfigNodeHeartbeatReq; +import org.apache.iotdb.confignode.rpc.thrift.TConfigNodeHeartbeatResp; +import org.apache.iotdb.confignode.rpc.thrift.TConfigNodeRegisterReq; +import org.apache.iotdb.confignode.rpc.thrift.TConfigNodeRegisterResp; +import org.apache.iotdb.confignode.rpc.thrift.TCountDatabaseResp; +import org.apache.iotdb.confignode.rpc.thrift.TCountTimeSlotListReq; +import org.apache.iotdb.confignode.rpc.thrift.TCountTimeSlotListResp; +import org.apache.iotdb.confignode.rpc.thrift.TCreateCQReq; +import org.apache.iotdb.confignode.rpc.thrift.TCreateConsumerReq; +import org.apache.iotdb.confignode.rpc.thrift.TCreateExternalServiceReq; +import org.apache.iotdb.confignode.rpc.thrift.TCreateFunctionReq; +import org.apache.iotdb.confignode.rpc.thrift.TCreatePipePluginReq; +import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq; +import org.apache.iotdb.confignode.rpc.thrift.TCreateSchemaTemplateReq; +import org.apache.iotdb.confignode.rpc.thrift.TCreateTableViewReq; +import org.apache.iotdb.confignode.rpc.thrift.TCreateTopicReq; +import org.apache.iotdb.confignode.rpc.thrift.TCreateTriggerReq; +import org.apache.iotdb.confignode.rpc.thrift.TDataNodeConfigurationResp; +import org.apache.iotdb.confignode.rpc.thrift.TDataNodeLeaseRecoveryResp; +import org.apache.iotdb.confignode.rpc.thrift.TDataNodeRegisterReq; +import org.apache.iotdb.confignode.rpc.thrift.TDataNodeRegisterResp; +import org.apache.iotdb.confignode.rpc.thrift.TDataNodeRemoveReq; +import org.apache.iotdb.confignode.rpc.thrift.TDataNodeRemoveResp; +import org.apache.iotdb.confignode.rpc.thrift.TDataNodeRestartReq; +import org.apache.iotdb.confignode.rpc.thrift.TDataNodeRestartResp; +import org.apache.iotdb.confignode.rpc.thrift.TDataPartitionReq; +import org.apache.iotdb.confignode.rpc.thrift.TDataPartitionTableResp; +import org.apache.iotdb.confignode.rpc.thrift.TDatabaseSchema; +import org.apache.iotdb.confignode.rpc.thrift.TDatabaseSchemaResp; +import org.apache.iotdb.confignode.rpc.thrift.TDeactivateSchemaTemplateReq; +import org.apache.iotdb.confignode.rpc.thrift.TDeleteDatabaseReq; +import org.apache.iotdb.confignode.rpc.thrift.TDeleteDatabasesReq; +import org.apache.iotdb.confignode.rpc.thrift.TDeleteLogicalViewReq; +import org.apache.iotdb.confignode.rpc.thrift.TDeleteTableDeviceReq; +import org.apache.iotdb.confignode.rpc.thrift.TDeleteTableDeviceResp; +import org.apache.iotdb.confignode.rpc.thrift.TDeleteTimeSeriesReq; +import org.apache.iotdb.confignode.rpc.thrift.TDescTable4InformationSchemaResp; +import org.apache.iotdb.confignode.rpc.thrift.TDescTableResp; +import org.apache.iotdb.confignode.rpc.thrift.TDropCQReq; +import org.apache.iotdb.confignode.rpc.thrift.TDropFunctionReq; +import org.apache.iotdb.confignode.rpc.thrift.TDropPipePluginReq; +import org.apache.iotdb.confignode.rpc.thrift.TDropPipeReq; +import org.apache.iotdb.confignode.rpc.thrift.TDropSubscriptionReq; +import org.apache.iotdb.confignode.rpc.thrift.TDropTopicReq; +import org.apache.iotdb.confignode.rpc.thrift.TDropTriggerReq; +import org.apache.iotdb.confignode.rpc.thrift.TExtendRegionReq; +import org.apache.iotdb.confignode.rpc.thrift.TFetchTableResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetAINodeLocationResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetAllPipeInfoResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetAllSubscriptionInfoResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetAllTemplatesResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetAllTopicInfoResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetClusterIdResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetCommitProgressReq; +import org.apache.iotdb.confignode.rpc.thrift.TGetCommitProgressResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetDataNodeLocationsResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetDatabaseReq; +import org.apache.iotdb.confignode.rpc.thrift.TGetJarInListReq; +import org.apache.iotdb.confignode.rpc.thrift.TGetJarInListResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetLocationForTriggerResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetPathsSetTemplatesReq; +import org.apache.iotdb.confignode.rpc.thrift.TGetPathsSetTemplatesResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetPipePluginTableResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetRegionGroupsByTimeReq; +import org.apache.iotdb.confignode.rpc.thrift.TGetRegionGroupsByTimeResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetRegionIdReq; +import org.apache.iotdb.confignode.rpc.thrift.TGetRegionIdResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetSeriesSlotListReq; +import org.apache.iotdb.confignode.rpc.thrift.TGetSeriesSlotListResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetTemplateResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetTimeSlotListReq; +import org.apache.iotdb.confignode.rpc.thrift.TGetTimeSlotListResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetTriggerTableResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetUDFTableResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetUdfTableReq; +import org.apache.iotdb.confignode.rpc.thrift.TLoginReq; +import org.apache.iotdb.confignode.rpc.thrift.TMigrateRegionReq; +import org.apache.iotdb.confignode.rpc.thrift.TPermissionInfoResp; +import org.apache.iotdb.confignode.rpc.thrift.TPipeConfigTransferReq; +import org.apache.iotdb.confignode.rpc.thrift.TPipeConfigTransferResp; +import org.apache.iotdb.confignode.rpc.thrift.TReconstructRegionReq; +import org.apache.iotdb.confignode.rpc.thrift.TRegionRouteMapResp; +import org.apache.iotdb.confignode.rpc.thrift.TRemoveRegionReq; +import org.apache.iotdb.confignode.rpc.thrift.TSchemaNodeManagementReq; +import org.apache.iotdb.confignode.rpc.thrift.TSchemaNodeManagementResp; +import org.apache.iotdb.confignode.rpc.thrift.TSchemaPartitionReq; +import org.apache.iotdb.confignode.rpc.thrift.TSchemaPartitionTableResp; +import org.apache.iotdb.confignode.rpc.thrift.TSetDataNodeStatusReq; +import org.apache.iotdb.confignode.rpc.thrift.TSetDataReplicationFactorReq; +import org.apache.iotdb.confignode.rpc.thrift.TSetSchemaReplicationFactorReq; +import org.apache.iotdb.confignode.rpc.thrift.TSetSchemaTemplateReq; +import org.apache.iotdb.confignode.rpc.thrift.TSetTimePartitionIntervalReq; +import org.apache.iotdb.confignode.rpc.thrift.TShowAINodesResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowCQResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowClusterResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowConfigNodes4InformationSchemaResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowConfigNodesResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowDataNodes4InformationSchemaResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowDataNodesResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowDatabaseResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowPipePluginReq; +import org.apache.iotdb.confignode.rpc.thrift.TShowPipeReq; +import org.apache.iotdb.confignode.rpc.thrift.TShowPipeResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowRegionReq; +import org.apache.iotdb.confignode.rpc.thrift.TShowRegionResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowRepairDataPartitionTableProgressResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowSubscriptionReq; +import org.apache.iotdb.confignode.rpc.thrift.TShowSubscriptionResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowTTLResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowTable4InformationSchemaResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowTableResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowThrottleReq; +import org.apache.iotdb.confignode.rpc.thrift.TShowTopicReq; +import org.apache.iotdb.confignode.rpc.thrift.TShowTopicResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowVariablesResp; +import org.apache.iotdb.confignode.rpc.thrift.TSpaceQuotaResp; +import org.apache.iotdb.confignode.rpc.thrift.TStartPipeReq; +import org.apache.iotdb.confignode.rpc.thrift.TStopPipeReq; +import org.apache.iotdb.confignode.rpc.thrift.TSubscribeReq; +import org.apache.iotdb.confignode.rpc.thrift.TSystemConfigurationResp; +import org.apache.iotdb.confignode.rpc.thrift.TTestOperation; +import org.apache.iotdb.confignode.rpc.thrift.TThrottleQuotaResp; +import org.apache.iotdb.confignode.rpc.thrift.TUnsetSchemaTemplateReq; +import org.apache.iotdb.confignode.rpc.thrift.TUnsubscribeReq; +import org.apache.iotdb.rpc.DeepCopyRpcTransportFactory; +import org.apache.iotdb.rpc.TSStatusCode; + +import org.apache.thrift.TException; +import org.apache.thrift.transport.TTransport; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.net.ssl.SSLHandshakeException; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.function.Predicate; + +public abstract class AbstractConfigNodeClient + implements IConfigNodeRPCService.Iface, ThriftClient, AutoCloseable { + + private static final Logger LOGGER = LoggerFactory.getLogger(AbstractConfigNodeClient.class); + + protected static final int RETRY_NUM = 15; + protected static final long RETRY_INTERVAL_MS = 1000L; + protected static final long WAIT_CN_LEADER_ELECTION_INTERVAL_MS = 2000L; + + protected final ThriftClientProperty property; + + protected IConfigNodeRPCService.Iface client; + + protected TTransport transport; + + protected TEndPoint configLeader; + + protected List<TEndPoint> configNodes; + + protected TEndPoint configNode; + + protected int cursor = 0; + + protected boolean isFirstInitiated; + + protected final CommonConfig commonConfig = CommonDescriptor.getInstance().getConfig(); + + protected final ClientManager<ConfigRegionId, ? super AbstractConfigNodeClient> clientManager; + + protected final ConfigRegionId configRegionId = AbstractConfigNodeInfo.CONFIG_REGION_ID; + + protected AbstractConfigNodeClient( + List<TEndPoint> configNodes, + ThriftClientProperty property, + ClientManager<ConfigRegionId, ? super AbstractConfigNodeClient> clientManager) + throws TException { + this.configNodes = configNodes; + this.property = property; + this.clientManager = clientManager; + // Set the first configNode as configLeader for a tentative connection + this.configLeader = this.configNodes.get(0); + this.isFirstInitiated = true; + + connectAndSync(); + } + + public TTransport getTransport() { + return transport; + } + + public void connect(TEndPoint endpoint, int timeoutMs) throws TException { + // Close existing transport before reassigning to prevent connection leaks. + if (transport != null) { + transport.close(); + } + transport = + commonConfig.isEnableInternalSSL() + ? DeepCopyRpcTransportFactory.INSTANCE.getTransport( + endpoint.getIp(), + endpoint.getPort(), + timeoutMs, + commonConfig.getTrustStorePath(), + commonConfig.getTrustStorePwd(), + commonConfig.getKeyStorePath(), + commonConfig.getKeyStorePwd()) + : DeepCopyRpcTransportFactory.INSTANCE.getTransport( + // As there is a try-catch already, we do not need to use TSocket.wrap + endpoint.getIp(), endpoint.getPort(), timeoutMs); + if (!transport.isOpen()) { + transport.open(); + } + configNode = endpoint; + + client = new IConfigNodeRPCService.Client(property.getProtocolFactory().getProtocol(transport)); + } + + protected void connectAndSync() throws TException { + try { + tryToConnect(property.getConnectionTimeoutMs()); + } catch (TException e) { + // Can not connect to each config node + syncLatestConfigNodeList(); + tryToConnect(property.getConnectionTimeoutMs()); + } + } + + protected void connectAndSync(int timeoutMs) throws TException { + try { + tryToConnect(timeoutMs); + } catch (TException e) { + // Can not connect to each config node + syncLatestConfigNodeList(); + tryToConnect(timeoutMs); + } + } + + private void tryToConnect(int timeoutMs) throws TException { + TException exception = null; + if (configLeader != null) { + try { + connect(configLeader, timeoutMs); + return; + } catch (TException e) { + LOGGER.warn(ConfigMessages.NODE_LEADER_MAY_DOWN_TRY_NEXT, configLeader); + configLeader = null; + exception = e; + } + } else { + try { + // Wait to start the next try + Thread.sleep(RETRY_INTERVAL_MS); + } catch (InterruptedException ignore) { + Thread.currentThread().interrupt(); + LOGGER.warn(ConfigMessages.UNEXPECTED_INTERRUPTION_CONNECT_CONFIG_NODE); + } + } + + for (int tryHostNum = 0; tryHostNum < configNodes.size(); tryHostNum++) { + cursor = (cursor + 1) % configNodes.size(); + TEndPoint tryEndpoint = configNodes.get(cursor); + + try { + connect(tryEndpoint, timeoutMs); + return; + } catch (TException e) { + LOGGER.warn(ConfigMessages.NODE_MAY_DOWN_TRY_NEXT, tryEndpoint); + exception = e; + } + } + if (exception != null + && exception.getCause() != null + && exception.getCause().getCause() != null + && exception.getCause().getCause() instanceof IOException) { + throw new TException(exception.getCause().getCause()); + } + + throw new TException(String.format(ConfigMessages.MSG_RECONNECTION_FAIL, getNodeTypeName())); + } + + public void syncLatestConfigNodeList() { + configNodes = getConfigNodeInfo().getLatestConfigNodes(); + cursor = 0; + } + + @Override + public void close() { + clientManager.returnClient(configRegionId, this); + } + + @Override + public void invalidate() { + Optional.ofNullable(transport).ifPresent(TTransport::close); + } + + @Override + public void invalidateAll() { + clientManager.clear(AbstractConfigNodeInfo.CONFIG_REGION_ID); + } + + @Override + public boolean printLogWhenEncounterException() { + return property.isPrintLogWhenEncounterException(); + } + + protected boolean updateConfigNodeLeader(TSStatus status) { + try { + if (status.getCode() == TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode()) { + if (status.isSetRedirectNode()) { + configLeader = + new TEndPoint(status.getRedirectNode().getIp(), status.getRedirectNode().getPort()); + } else { + configLeader = null; + } + if (!isFirstInitiated) { + LOGGER.info( + ConfigMessages.FAILED_CONNECT_CONFIG_NODE_NOT_LEADER, + configNode, + getNodeTypeName(), + getAddressAndPort()); + } + return true; + } + if (status.getCode() == TSStatusCode.CONFIG_NODE_LEADER_WARMING_UP.getStatusCode()) { + if (!isFirstInitiated) { + LOGGER.info( + ConfigMessages + .MESSAGE_CONFIGNODE_LEADER_ARG_IS_WARMING_UP_BEFORE_SERVING_NODE_ARG_WILL_WAIT_AND_RETRY_REASON_ARG_2ACA2BD0, + configNode, + getNodeTypeName(), + getAddressAndPort(), + status.getMessage()); + } + try { + Thread.sleep(WAIT_CN_LEADER_ELECTION_INTERVAL_MS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + LOGGER.warn(ConfigMessages.UNEXPECTED_INTERRUPTION_CONNECT_CONFIG_NODE_BREAK); + } + return true; + } + return false; + } finally { + isFirstInitiated = false; + } + } + + /** + * The frame of execute RPC, include logic of retry and exception handling. + * + * @param call which rpc should call + * @param check check the rpc's result + * @return rpc's result + * @param <R> the type of rpc result + * @throws TException if fails more than RETRY_NUM times, throw TException(MSG_RECONNECTION_FAIL) + */ + protected <R> R executeRemoteCallWithRetry(final Operation<R> call, final Predicate<R> check) Review Comment: [P1] Preserve the original dispatch semantics under CGLIB These helpers were `private` before the refactor, but `ConfigNodeClient` instances are CGLIB subclasses created by `SyncThriftClientWithErrorHandler`. Making `executeRemoteCallWithRetry` (and likewise `connectAndSync` and `updateConfigNodeLeader`) non-final `protected` methods means self-invocations are now intercepted, including during construction. I reproduced this with the CGLIB 3.3.0 version used by the project: `getStackTrace()[2]` becomes a `CGLIB$executeRemoteCallWithRetry$...` frame, terminal exceptions are resolved and wrapped twice, and an unchecked exception from `updateConfigNodeLeader` is converted to `TException` and retried instead of failing fast. Please keep these internal template methods non-overridable, and make the concrete hook overrides final, or filter the enhancer so that it intercepts only the intended public RPC entry points. ########## iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/AbstractConfigNodeClient.java: ########## @@ -0,0 +1,1331 @@ +/* + * 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.commons.client; + +import org.apache.iotdb.common.rpc.thrift.TConfigNodeLocation; +import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation; +import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.common.rpc.thrift.TExternalServiceListResp; +import org.apache.iotdb.common.rpc.thrift.TFlushReq; +import org.apache.iotdb.common.rpc.thrift.TNodeLocations; +import org.apache.iotdb.common.rpc.thrift.TPipeHeartbeatResp; +import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.common.rpc.thrift.TSeriesPartitionSlot; +import org.apache.iotdb.common.rpc.thrift.TSetConfigurationReq; +import org.apache.iotdb.common.rpc.thrift.TSetSpaceQuotaReq; +import org.apache.iotdb.common.rpc.thrift.TSetTTLReq; +import org.apache.iotdb.common.rpc.thrift.TSetThrottleQuotaReq; +import org.apache.iotdb.common.rpc.thrift.TShowAppliedConfigurationsResp; +import org.apache.iotdb.common.rpc.thrift.TShowConfigurationResp; +import org.apache.iotdb.common.rpc.thrift.TShowTTLReq; +import org.apache.iotdb.common.rpc.thrift.TTestConnectionResp; +import org.apache.iotdb.commons.client.property.ThriftClientProperty; +import org.apache.iotdb.commons.conf.CommonConfig; +import org.apache.iotdb.commons.conf.CommonDescriptor; +import org.apache.iotdb.commons.consensus.ConfigRegionId; +import org.apache.iotdb.commons.i18n.ConfigMessages; +import org.apache.iotdb.confignode.rpc.thrift.IConfigNodeRPCService; +import org.apache.iotdb.confignode.rpc.thrift.TAINodeConfigurationResp; +import org.apache.iotdb.confignode.rpc.thrift.TAINodeRegisterReq; +import org.apache.iotdb.confignode.rpc.thrift.TAINodeRegisterResp; +import org.apache.iotdb.confignode.rpc.thrift.TAINodeRemoveReq; +import org.apache.iotdb.confignode.rpc.thrift.TAINodeRestartReq; +import org.apache.iotdb.confignode.rpc.thrift.TAINodeRestartResp; +import org.apache.iotdb.confignode.rpc.thrift.TAddConsensusGroupReq; +import org.apache.iotdb.confignode.rpc.thrift.TAlterEncodingCompressorReq; +import org.apache.iotdb.confignode.rpc.thrift.TAlterLogicalViewReq; +import org.apache.iotdb.confignode.rpc.thrift.TAlterOrDropTableReq; +import org.apache.iotdb.confignode.rpc.thrift.TAlterPipeReq; +import org.apache.iotdb.confignode.rpc.thrift.TAlterSchemaTemplateReq; +import org.apache.iotdb.confignode.rpc.thrift.TAlterTimeSeriesReq; +import org.apache.iotdb.confignode.rpc.thrift.TAlterTopicReq; +import org.apache.iotdb.confignode.rpc.thrift.TAuthizedPatternTreeResp; +import org.apache.iotdb.confignode.rpc.thrift.TAuthorizerRelationalReq; +import org.apache.iotdb.confignode.rpc.thrift.TAuthorizerReq; +import org.apache.iotdb.confignode.rpc.thrift.TAuthorizerResp; +import org.apache.iotdb.confignode.rpc.thrift.TCheckUserPrivilegesReq; +import org.apache.iotdb.confignode.rpc.thrift.TCloseConsumerReq; +import org.apache.iotdb.confignode.rpc.thrift.TConfigNodeHeartbeatReq; +import org.apache.iotdb.confignode.rpc.thrift.TConfigNodeHeartbeatResp; +import org.apache.iotdb.confignode.rpc.thrift.TConfigNodeRegisterReq; +import org.apache.iotdb.confignode.rpc.thrift.TConfigNodeRegisterResp; +import org.apache.iotdb.confignode.rpc.thrift.TCountDatabaseResp; +import org.apache.iotdb.confignode.rpc.thrift.TCountTimeSlotListReq; +import org.apache.iotdb.confignode.rpc.thrift.TCountTimeSlotListResp; +import org.apache.iotdb.confignode.rpc.thrift.TCreateCQReq; +import org.apache.iotdb.confignode.rpc.thrift.TCreateConsumerReq; +import org.apache.iotdb.confignode.rpc.thrift.TCreateExternalServiceReq; +import org.apache.iotdb.confignode.rpc.thrift.TCreateFunctionReq; +import org.apache.iotdb.confignode.rpc.thrift.TCreatePipePluginReq; +import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq; +import org.apache.iotdb.confignode.rpc.thrift.TCreateSchemaTemplateReq; +import org.apache.iotdb.confignode.rpc.thrift.TCreateTableViewReq; +import org.apache.iotdb.confignode.rpc.thrift.TCreateTopicReq; +import org.apache.iotdb.confignode.rpc.thrift.TCreateTriggerReq; +import org.apache.iotdb.confignode.rpc.thrift.TDataNodeConfigurationResp; +import org.apache.iotdb.confignode.rpc.thrift.TDataNodeLeaseRecoveryResp; +import org.apache.iotdb.confignode.rpc.thrift.TDataNodeRegisterReq; +import org.apache.iotdb.confignode.rpc.thrift.TDataNodeRegisterResp; +import org.apache.iotdb.confignode.rpc.thrift.TDataNodeRemoveReq; +import org.apache.iotdb.confignode.rpc.thrift.TDataNodeRemoveResp; +import org.apache.iotdb.confignode.rpc.thrift.TDataNodeRestartReq; +import org.apache.iotdb.confignode.rpc.thrift.TDataNodeRestartResp; +import org.apache.iotdb.confignode.rpc.thrift.TDataPartitionReq; +import org.apache.iotdb.confignode.rpc.thrift.TDataPartitionTableResp; +import org.apache.iotdb.confignode.rpc.thrift.TDatabaseSchema; +import org.apache.iotdb.confignode.rpc.thrift.TDatabaseSchemaResp; +import org.apache.iotdb.confignode.rpc.thrift.TDeactivateSchemaTemplateReq; +import org.apache.iotdb.confignode.rpc.thrift.TDeleteDatabaseReq; +import org.apache.iotdb.confignode.rpc.thrift.TDeleteDatabasesReq; +import org.apache.iotdb.confignode.rpc.thrift.TDeleteLogicalViewReq; +import org.apache.iotdb.confignode.rpc.thrift.TDeleteTableDeviceReq; +import org.apache.iotdb.confignode.rpc.thrift.TDeleteTableDeviceResp; +import org.apache.iotdb.confignode.rpc.thrift.TDeleteTimeSeriesReq; +import org.apache.iotdb.confignode.rpc.thrift.TDescTable4InformationSchemaResp; +import org.apache.iotdb.confignode.rpc.thrift.TDescTableResp; +import org.apache.iotdb.confignode.rpc.thrift.TDropCQReq; +import org.apache.iotdb.confignode.rpc.thrift.TDropFunctionReq; +import org.apache.iotdb.confignode.rpc.thrift.TDropPipePluginReq; +import org.apache.iotdb.confignode.rpc.thrift.TDropPipeReq; +import org.apache.iotdb.confignode.rpc.thrift.TDropSubscriptionReq; +import org.apache.iotdb.confignode.rpc.thrift.TDropTopicReq; +import org.apache.iotdb.confignode.rpc.thrift.TDropTriggerReq; +import org.apache.iotdb.confignode.rpc.thrift.TExtendRegionReq; +import org.apache.iotdb.confignode.rpc.thrift.TFetchTableResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetAINodeLocationResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetAllPipeInfoResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetAllSubscriptionInfoResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetAllTemplatesResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetAllTopicInfoResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetClusterIdResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetCommitProgressReq; +import org.apache.iotdb.confignode.rpc.thrift.TGetCommitProgressResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetDataNodeLocationsResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetDatabaseReq; +import org.apache.iotdb.confignode.rpc.thrift.TGetJarInListReq; +import org.apache.iotdb.confignode.rpc.thrift.TGetJarInListResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetLocationForTriggerResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetPathsSetTemplatesReq; +import org.apache.iotdb.confignode.rpc.thrift.TGetPathsSetTemplatesResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetPipePluginTableResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetRegionGroupsByTimeReq; +import org.apache.iotdb.confignode.rpc.thrift.TGetRegionGroupsByTimeResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetRegionIdReq; +import org.apache.iotdb.confignode.rpc.thrift.TGetRegionIdResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetSeriesSlotListReq; +import org.apache.iotdb.confignode.rpc.thrift.TGetSeriesSlotListResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetTemplateResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetTimeSlotListReq; +import org.apache.iotdb.confignode.rpc.thrift.TGetTimeSlotListResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetTriggerTableResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetUDFTableResp; +import org.apache.iotdb.confignode.rpc.thrift.TGetUdfTableReq; +import org.apache.iotdb.confignode.rpc.thrift.TLoginReq; +import org.apache.iotdb.confignode.rpc.thrift.TMigrateRegionReq; +import org.apache.iotdb.confignode.rpc.thrift.TPermissionInfoResp; +import org.apache.iotdb.confignode.rpc.thrift.TPipeConfigTransferReq; +import org.apache.iotdb.confignode.rpc.thrift.TPipeConfigTransferResp; +import org.apache.iotdb.confignode.rpc.thrift.TReconstructRegionReq; +import org.apache.iotdb.confignode.rpc.thrift.TRegionRouteMapResp; +import org.apache.iotdb.confignode.rpc.thrift.TRemoveRegionReq; +import org.apache.iotdb.confignode.rpc.thrift.TSchemaNodeManagementReq; +import org.apache.iotdb.confignode.rpc.thrift.TSchemaNodeManagementResp; +import org.apache.iotdb.confignode.rpc.thrift.TSchemaPartitionReq; +import org.apache.iotdb.confignode.rpc.thrift.TSchemaPartitionTableResp; +import org.apache.iotdb.confignode.rpc.thrift.TSetDataNodeStatusReq; +import org.apache.iotdb.confignode.rpc.thrift.TSetDataReplicationFactorReq; +import org.apache.iotdb.confignode.rpc.thrift.TSetSchemaReplicationFactorReq; +import org.apache.iotdb.confignode.rpc.thrift.TSetSchemaTemplateReq; +import org.apache.iotdb.confignode.rpc.thrift.TSetTimePartitionIntervalReq; +import org.apache.iotdb.confignode.rpc.thrift.TShowAINodesResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowCQResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowClusterResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowConfigNodes4InformationSchemaResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowConfigNodesResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowDataNodes4InformationSchemaResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowDataNodesResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowDatabaseResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowPipePluginReq; +import org.apache.iotdb.confignode.rpc.thrift.TShowPipeReq; +import org.apache.iotdb.confignode.rpc.thrift.TShowPipeResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowRegionReq; +import org.apache.iotdb.confignode.rpc.thrift.TShowRegionResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowRepairDataPartitionTableProgressResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowSubscriptionReq; +import org.apache.iotdb.confignode.rpc.thrift.TShowSubscriptionResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowTTLResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowTable4InformationSchemaResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowTableResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowThrottleReq; +import org.apache.iotdb.confignode.rpc.thrift.TShowTopicReq; +import org.apache.iotdb.confignode.rpc.thrift.TShowTopicResp; +import org.apache.iotdb.confignode.rpc.thrift.TShowVariablesResp; +import org.apache.iotdb.confignode.rpc.thrift.TSpaceQuotaResp; +import org.apache.iotdb.confignode.rpc.thrift.TStartPipeReq; +import org.apache.iotdb.confignode.rpc.thrift.TStopPipeReq; +import org.apache.iotdb.confignode.rpc.thrift.TSubscribeReq; +import org.apache.iotdb.confignode.rpc.thrift.TSystemConfigurationResp; +import org.apache.iotdb.confignode.rpc.thrift.TTestOperation; +import org.apache.iotdb.confignode.rpc.thrift.TThrottleQuotaResp; +import org.apache.iotdb.confignode.rpc.thrift.TUnsetSchemaTemplateReq; +import org.apache.iotdb.confignode.rpc.thrift.TUnsubscribeReq; +import org.apache.iotdb.rpc.DeepCopyRpcTransportFactory; +import org.apache.iotdb.rpc.TSStatusCode; + +import org.apache.thrift.TException; +import org.apache.thrift.transport.TTransport; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.net.ssl.SSLHandshakeException; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.function.Predicate; + +public abstract class AbstractConfigNodeClient + implements IConfigNodeRPCService.Iface, ThriftClient, AutoCloseable { + + private static final Logger LOGGER = LoggerFactory.getLogger(AbstractConfigNodeClient.class); + + protected static final int RETRY_NUM = 15; + protected static final long RETRY_INTERVAL_MS = 1000L; + protected static final long WAIT_CN_LEADER_ELECTION_INTERVAL_MS = 2000L; + + protected final ThriftClientProperty property; + + protected IConfigNodeRPCService.Iface client; + + protected TTransport transport; + + protected TEndPoint configLeader; + + protected List<TEndPoint> configNodes; + + protected TEndPoint configNode; + + protected int cursor = 0; + + protected boolean isFirstInitiated; + + protected final CommonConfig commonConfig = CommonDescriptor.getInstance().getConfig(); + + protected final ClientManager<ConfigRegionId, ? super AbstractConfigNodeClient> clientManager; + + protected final ConfigRegionId configRegionId = AbstractConfigNodeInfo.CONFIG_REGION_ID; + + protected AbstractConfigNodeClient( + List<TEndPoint> configNodes, + ThriftClientProperty property, + ClientManager<ConfigRegionId, ? super AbstractConfigNodeClient> clientManager) Review Comment: [P2] Preserve the concrete client type in ClientManager `ClientManager<ConfigRegionId, ConfigNodeClient>`, which was accepted by the old public constructor and is the type used by `Factory`, is not assignable to `ClientManager<ConfigRegionId, ? super AbstractConfigNodeClient>`. The current factory works only because reflective construction erases this generic signature; direct callers and future concrete subclasses cannot pass their naturally typed manager. Please retain the concrete self type through the base class, for example `AbstractConfigNodeClient<C extends AbstractConfigNodeClient<C>>` with `ClientManager<ConfigRegionId, C>`. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
