You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ha...@apache.org on 2024/03/05 02:29:52 UTC
(iotdb) 21/22: Do not wait to retry when configLeader exists (#12075)
This is an automated email from the ASF dual-hosted git repository.
haonan pushed a commit to branch rc/1.3.1
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit d1c4159592bc2d2338fe70dff840fb7c6ad3a4d2
Author: Peng Junzhi <78...@users.noreply.github.com>
AuthorDate: Thu Feb 29 14:57:20 2024 +0800
Do not wait to retry when configLeader exists (#12075)
---
.../manager/consensus/ConsensusManager.java | 18 ++++++++++-
.../subtask/processor/PipeProcessorSubtask.java | 1 -
.../iotdb/db/protocol/client/ConfigNodeClient.java | 37 ++++++++--------------
3 files changed, 31 insertions(+), 25 deletions(-)
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/consensus/ConsensusManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/consensus/ConsensusManager.java
index 69647bf7488..2fb412a7d7e 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/consensus/ConsensusManager.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/consensus/ConsensusManager.java
@@ -65,6 +65,9 @@ public class ConsensusManager {
private static final ConfigNodeConfig CONF = ConfigNodeDescriptor.getInstance().getConf();
private static final CommonConfig COMMON_CONF = CommonDescriptor.getInstance().getConfig();
private static final int SEED_CONFIG_NODE_ID = 0;
+ private static final long MAX_WAIT_READY_TIME_MS =
+ CommonDescriptor.getInstance().getConfig().getConnectionTimeoutInMS() / 2;
+ private static final long RETRY_WAIT_TIME_MS = 100;
/** There is only one ConfigNodeGroup */
public static final ConsensusGroupId DEFAULT_CONSENSUS_GROUP_ID =
new ConfigRegionId(CONF.getConfigRegionId());
@@ -343,7 +346,7 @@ public class ConsensusManager {
return leaderPeer;
}
try {
- TimeUnit.MILLISECONDS.sleep(100);
+ TimeUnit.MILLISECONDS.sleep(RETRY_WAIT_TIME_MS);
} catch (InterruptedException e) {
LOGGER.warn("ConsensusManager getLeaderPeer been interrupted, ", e);
Thread.currentThread().interrupt();
@@ -382,6 +385,19 @@ public class ConsensusManager {
} else {
result.setCode(TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode());
if (isLeader()) {
+ long startTime = System.currentTimeMillis();
+ while (System.currentTimeMillis() - startTime < MAX_WAIT_READY_TIME_MS) {
+ if (isLeaderReady()) {
+ result.setCode(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ return result;
+ }
+ try {
+ Thread.sleep(RETRY_WAIT_TIME_MS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ LOGGER.warn("Unexpected interruption during waiting for configNode leader ready.");
+ }
+ }
result.setMessage(
"The current ConfigNode is leader but not ready yet, please try again later.");
} else {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/processor/PipeProcessorSubtask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/processor/PipeProcessorSubtask.java
index c8c3c2205b1..63fe58907b1 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/processor/PipeProcessorSubtask.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/task/subtask/processor/PipeProcessorSubtask.java
@@ -19,7 +19,6 @@
package org.apache.iotdb.db.pipe.task.subtask.processor;
-import org.apache.iotdb.commons.exception.pipe.PipeRuntimeOutOfMemoryCriticalException;
import org.apache.iotdb.commons.pipe.execution.scheduler.PipeSubtaskScheduler;
import org.apache.iotdb.commons.pipe.task.EventSupplier;
import org.apache.iotdb.db.pipe.event.EnrichedEvent;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java
index 12c6f1d3f90..6d7bd57fdaa 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java
@@ -183,18 +183,10 @@ public class ConfigNodeClient implements IConfigNodeRPCService.Iface, ThriftClie
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);
- init();
- }
-
- public void init() throws TException {
- try {
- tryToConnect();
- } catch (TException e) {
- // Can not connect to each config node
- syncLatestConfigNodeList();
- tryToConnect();
- }
+ connectAndSync();
}
public void connect(TEndPoint endpoint) throws TException {
@@ -214,16 +206,7 @@ public class ConfigNodeClient implements IConfigNodeRPCService.Iface, ThriftClie
client = new IConfigNodeRPCService.Client(property.getProtocolFactory().getProtocol(transport));
}
- private void waitAndReconnect() throws TException {
- try {
- // Wait to start the next try
- Thread.sleep(RETRY_INTERVAL_MS);
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new TException(
- "Unexpected interruption when waiting to retry to connect to ConfigNode");
- }
-
+ private void connectAndSync() throws TException {
try {
tryToConnect();
} catch (TException e) {
@@ -242,6 +225,14 @@ public class ConfigNodeClient implements IConfigNodeRPCService.Iface, ThriftClie
logger.warn("The current node may have been down {},try next node", configLeader);
configLeader = null;
}
+ } else {
+ try {
+ // Wait to start the next try
+ Thread.sleep(RETRY_INTERVAL_MS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ logger.warn("Unexpected interruption when waiting to try to connect to ConfigNode");
+ }
}
if (transport != null) {
@@ -337,7 +328,7 @@ public class ConfigNodeClient implements IConfigNodeRPCService.Iface, ThriftClie
logger.warn(message, e);
configLeader = null;
}
- waitAndReconnect();
+ connectAndSync();
}
throw new TException(MSG_RECONNECTION_FAIL);
}
@@ -385,7 +376,7 @@ public class ConfigNodeClient implements IConfigNodeRPCService.Iface, ThriftClie
logger.warn(message, e);
configLeader = null;
}
- waitAndReconnect();
+ connectAndSync();
}
throw new TException(MSG_RECONNECTION_FAIL);
}