JackieTien97 commented on code in PR #18561:
URL: https://github.com/apache/iotdb/pull/18561#discussion_r3910294289
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java:
##########
@@ -203,299 +199,53 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import javax.net.ssl.SSLHandshakeException;
-
-import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
-import java.util.function.Predicate;
-
-public class ConfigNodeClient implements IConfigNodeRPCService.Iface,
ThriftClient, AutoCloseable {
+public class ConfigNodeClient extends AbstractConfigNodeClient {
private static final Logger logger =
LoggerFactory.getLogger(ConfigNodeClient.class);
- private static final int RETRY_NUM = 15;
-
public static final String MSG_RECONNECTION_FAIL =
- "Fail to connect to any config node. Please check status of ConfigNodes
or logs of connected DataNode";
+ String.format(ConfigMessages.MSG_RECONNECTION_FAIL, "DataNode");
- private static final String MSG_RECONNECTION_DATANODE_FAIL =
- "Failed to connect to ConfigNode %s from DataNode %s when executing %s,
Exception:";
- private static final long RETRY_INTERVAL_MS = 1000L;
- private static final long WAIT_CN_LEADER_ELECTION_INTERVAL_MS = 2000L;
private static final long REGISTER_LEADER_WARMING_UP_RETRY_TIMEOUT_MS =
60_000L;
private static final String UNSUPPORTED_INVOCATION =
DataNodeMiscMessages.UNSUPPORTED_INVOCATION_BY_DATANODE;
- private final ThriftClientProperty property;
-
- private IConfigNodeRPCService.Iface client;
-
- private TTransport transport;
-
- private TEndPoint configLeader;
-
- private List<TEndPoint> configNodes;
-
- private TEndPoint configNode;
-
- private int cursor = 0;
-
- private boolean isFirstInitiated;
-
private final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
- private final CommonConfig commonConfig =
CommonDescriptor.getInstance().getConfig();
-
- ClientManager<ConfigRegionId, ConfigNodeClient> clientManager;
-
- ConfigRegionId configRegionId = ConfigNodeInfo.CONFIG_REGION_ID;
-
public ConfigNodeClient(
List<TEndPoint> configNodes,
ThriftClientProperty property,
- ClientManager<ConfigRegionId, ConfigNodeClient> clientManager)
+ 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 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));
- }
-
- private void connectAndSync() throws TException {
- try {
- tryToConnect(property.getConnectionTimeoutMs());
- } catch (TException e) {
- // Can not connect to each config node
- syncLatestConfigNodeList();
- tryToConnect(property.getConnectionTimeoutMs());
- }
- }
-
- private 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(DataNodeMiscMessages.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(DataNodeMiscMessages.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(DataNodeMiscMessages.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(MSG_RECONNECTION_FAIL);
- }
-
- public TTransport getTransport() {
- return transport;
- }
-
- public void syncLatestConfigNodeList() {
- configNodes = ConfigNodeInfo.getInstance().getLatestConfigNodes();
- cursor = 0;
- }
-
- @Override
- public void close() {
- clientManager.returnClient(configRegionId, this);
+ super(configNodes, property, clientManager);
}
@Override
- public void invalidate() {
- Optional.ofNullable(transport).ifPresent(TTransport::close);
+ protected String getNodeTypeName() {
+ return "DataNode";
Review Comment:
better define a string constant
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeInfo.java:
##########
@@ -19,137 +19,18 @@
package org.apache.iotdb.db.protocol.client;
-import org.apache.iotdb.common.rpc.thrift.TEndPoint;
-import org.apache.iotdb.commons.consensus.ConfigRegionId;
-import org.apache.iotdb.commons.exception.BadNodeUrlException;
-import org.apache.iotdb.commons.exception.StartupException;
-import org.apache.iotdb.commons.file.SystemPropertiesHandler;
-import org.apache.iotdb.commons.utils.NodeUrlUtils;
+import org.apache.iotdb.commons.client.AbstractConfigNodeInfo;
import org.apache.iotdb.db.conf.DataNodeSystemPropertiesHandler;
-import org.apache.iotdb.db.i18n.DataNodeMiscMessages;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Properties;
-import java.util.Set;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
-
-public class ConfigNodeInfo {
- private static final Logger logger =
LoggerFactory.getLogger(ConfigNodeInfo.class);
-
- private static final String CONFIG_NODE_LIST = "config_node_list";
-
- private final ReentrantReadWriteLock configNodeInfoReadWriteLock;
-
- /** latest config nodes. */
- private final Set<TEndPoint> onlineConfigNodes;
-
- public static final ConfigRegionId CONFIG_REGION_ID = new ConfigRegionId(0);
-
- SystemPropertiesHandler systemPropertiesHandler =
DataNodeSystemPropertiesHandler.getInstance();
+public class ConfigNodeInfo extends AbstractConfigNodeInfo {
private ConfigNodeInfo() {
- this.configNodeInfoReadWriteLock = new ReentrantReadWriteLock();
- this.onlineConfigNodes = new HashSet<>();
- }
-
- public static void reinitializeStatics() {
- ConfigNodeInfoHolder.INSTANCE = new ConfigNodeInfo();
- }
-
- /** Update ConfigNodeList both in memory and system.properties file */
- public boolean updateConfigNodeList(List<TEndPoint> latestConfigNodes) {
- long startTime = System.currentTimeMillis();
- // Check whether the config nodes are latest or not
- configNodeInfoReadWriteLock.readLock().lock();
- try {
- if (onlineConfigNodes.size() == latestConfigNodes.size()
- && onlineConfigNodes.containsAll(latestConfigNodes)) {
- return true;
- }
- } finally {
- configNodeInfoReadWriteLock.readLock().unlock();
- }
-
- // Update config nodes
- configNodeInfoReadWriteLock.writeLock().lock();
- try {
- onlineConfigNodes.clear();
- onlineConfigNodes.addAll(latestConfigNodes);
- storeConfigNodeList();
- long endTime = System.currentTimeMillis();
- logger.info(
- DataNodeMiscMessages.UPDATE_CONFIG_NODE_SUCCESSFULLY,
- onlineConfigNodes,
- (endTime - startTime));
- } catch (IOException e) {
- logger.error(DataNodeMiscMessages.UPDATE_CONFIG_NODE_FAILED, e);
- return false;
- } finally {
- configNodeInfoReadWriteLock.writeLock().unlock();
- }
- return true;
- }
-
- /**
- * Call this method to store config node list.
- *
- * @throws IOException if properties deserialization or configNode list
serialization failed.
- */
- public void storeConfigNodeList() throws IOException {
- if (!systemPropertiesHandler.fileExist()) {
- logger.info(DataNodeMiscMessages.SYSTEM_PROPERTIES_NOT_EXIST);
- return;
- }
- systemPropertiesHandler.put(
- CONFIG_NODE_LIST, NodeUrlUtils.convertTEndPointUrls(new
ArrayList<>(onlineConfigNodes)));
+ super(DataNodeSystemPropertiesHandler.getInstance());
}
- public void loadConfigNodeList() throws StartupException {
- long startTime = System.currentTimeMillis();
- // properties contain CONFIG_NODE_LIST only when start as Data node
- configNodeInfoReadWriteLock.writeLock().lock();
- try {
- Properties properties = systemPropertiesHandler.read();
-
- if (properties.containsKey(CONFIG_NODE_LIST)) {
- onlineConfigNodes.clear();
- onlineConfigNodes.addAll(
-
NodeUrlUtils.parseTEndPointUrls(properties.getProperty(CONFIG_NODE_LIST)));
- }
- if (onlineConfigNodes.isEmpty()) {
- throw new StartupException(
- DataNodeMiscMessages
-
.MISC_EXCEPTION_REMOVING_IS_ONLY_ALLOWED_IN_AN_ENVIRONMENT_WHERE_THE_DATANODE_5A3E1FEA);
- }
- long endTime = System.currentTimeMillis();
- logger.info(
- DataNodeMiscMessages.LOAD_CONFIG_NODE_SUCCESSFULLY,
- onlineConfigNodes,
- (endTime - startTime));
- } catch (IOException e) {
- throw new RuntimeException(e);
- } catch (BadNodeUrlException e) {
- logger.error(DataNodeMiscMessages.CANNOT_PARSE_CONFIG_NODE_LIST);
- } finally {
- configNodeInfoReadWriteLock.writeLock().unlock();
- }
- }
-
- public List<TEndPoint> getLatestConfigNodes() {
- List<TEndPoint> result;
- configNodeInfoReadWriteLock.readLock().lock();
- try {
- result = new ArrayList<>(onlineConfigNodes);
- } finally {
- configNodeInfoReadWriteLock.readLock().unlock();
- }
- return result;
+ @Override
+ protected String getNodeTypeName() {
+ return "datanode";
Review Comment:
reuse the string constant, and why here lower case and previous is upper
case.
--
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]
