Skip to content

Commit a57680d

Browse files
CRZbulabulaCRZbulabula
committed
Refine ConfigNode leader warm-up readiness
1 parent 4e6d45f commit a57680d

9 files changed

Lines changed: 136 additions & 198 deletions

File tree

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/heartbeat/ConfigNodeHeartbeatHandler.java‎

Lines changed: 2 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,6 @@
1919

2020
package org.apache.iotdb.confignode.client.async.handlers.heartbeat;
2121

22-
import org.apache.iotdb.commons.client.ThriftClient;
2322
import org.apache.iotdb.commons.cluster.NodeStatus;
2423
import org.apache.iotdb.commons.cluster.NodeType;
2524
import org.apache.iotdb.confignode.manager.load.LoadManager;
@@ -47,10 +46,8 @@ public void onComplete(TConfigNodeHeartbeatResp resp) {
4746

4847
@Override
4948
public void onError(Exception e) {
50-
if (ThriftClient.isConnectionBroken(e)) {
51-
loadManager.forceUpdateNodeCache(
52-
NodeType.ConfigNode, nodeId, new NodeHeartbeatSample(NodeStatus.Unknown));
53-
}
49+
loadManager.forceUpdateNodeCache(
50+
NodeType.ConfigNode, nodeId, new NodeHeartbeatSample(NodeStatus.Unknown));
5451
loadManager.getLoadCache().resetHeartbeatProcessing(nodeId);
5552
}
5653
}

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/heartbeat/DataNodeHeartbeatHandler.java‎

Lines changed: 3 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,6 @@
2121

2222
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
2323
import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
24-
import org.apache.iotdb.commons.client.ThriftClient;
2524
import org.apache.iotdb.commons.cluster.NodeStatus;
2625
import org.apache.iotdb.commons.cluster.NodeType;
2726
import org.apache.iotdb.commons.cluster.RegionStatus;
@@ -177,11 +176,9 @@ public void onComplete(TDataNodeHeartbeatResp heartbeatResp) {
177176

178177
@Override
179178
public void onError(Exception e) {
180-
if (ThriftClient.isConnectionBroken(e)) {
181-
loadManager.forceUpdateNodeCache(
182-
NodeType.DataNode, nodeId, new NodeHeartbeatSample(NodeStatus.Unknown));
183-
loadManager.getLoadCache().cacheDataNodeHeartbeatFailureSample(nodeId, System.nanoTime());
184-
}
179+
loadManager.forceUpdateNodeCache(
180+
NodeType.DataNode, nodeId, new NodeHeartbeatSample(NodeStatus.Unknown));
181+
loadManager.getLoadCache().cacheDataNodeHeartbeatFailureSample(nodeId, System.nanoTime());
185182
loadManager.getLoadCache().resetHeartbeatProcessing(nodeId);
186183
}
187184
}

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/statemachine/ConfigRegionStateMachine.java‎

Lines changed: 19 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -64,6 +64,7 @@
6464
import java.util.concurrent.ExecutorService;
6565
import java.util.concurrent.ScheduledExecutorService;
6666
import java.util.concurrent.TimeUnit;
67+
import java.util.concurrent.atomic.AtomicBoolean;
6768
import java.util.regex.Matcher;
6869
import java.util.regex.Pattern;
6970

@@ -76,6 +77,7 @@ public class ConfigRegionStateMachine implements IStateMachine, IStateMachine.Ev
7677
IoTDBThreadPoolFactory.newCachedThreadPool(ThreadName.CONFIG_NODE_RECOVER.getName());
7778
private static final ConfigNodeConfig CONF = ConfigNodeDescriptor.getInstance().getConf();
7879
private final ConfigPlanExecutor executor;
80+
private final AtomicBoolean leaderServicesReady;
7981
private ConfigManager configManager;
8082

8183
/** Variables for {@link ConfigNode} Simple Consensus. */
@@ -98,6 +100,7 @@ public class ConfigRegionStateMachine implements IStateMachine, IStateMachine.Ev
98100

99101
public ConfigRegionStateMachine(ConfigManager configManager, ConfigPlanExecutor executor) {
100102
this.executor = executor;
103+
this.leaderServicesReady = new AtomicBoolean(false);
101104
this.configManager = configManager;
102105
this.currentNodeTEndPoint =
103106
new TEndPoint()
@@ -231,6 +234,7 @@ public void loadSnapshot(final File latestSnapshotRootDir) {
231234

232235
@Override
233236
public void notifyLeaderChanged(ConsensusGroupId groupId, int newLeaderId) {
237+
leaderServicesReady.set(false);
234238
// We get currentNodeId here because the currentNodeId
235239
// couldn't initialize earlier than the ConfigRegionStateMachine
236240
int currentNodeId = ConfigNodeDescriptor.getInstance().getConf().getConfigNodeId();
@@ -246,6 +250,7 @@ public void notifyLeaderChanged(ConsensusGroupId groupId, int newLeaderId) {
246250

247251
@Override
248252
public void notifyNotLeader() {
253+
leaderServicesReady.set(false);
249254
// We get currentNodeId here because the currentNodeId
250255
// couldn't initialize earlier than the ConfigRegionStateMachine
251256
int currentNodeId = ConfigNodeDescriptor.getInstance().getConf().getConfigNodeId();
@@ -283,29 +288,29 @@ public void notifyNotLeader() {
283288

284289
@Override
285290
public void notifyLeaderReady() {
291+
leaderServicesReady.set(false);
286292
LOGGER.info(
287293
ConfigNodeMessages.CURRENT_NODE_NODEID_IP_PORT_BECOMES_CONFIG_REGION_LEADER,
288294
ConfigNodeDescriptor.getInstance().getConf().getConfigNodeId(),
289295
currentNodeTEndPoint);
290296

291-
// Always start load services first and wait for its first full warm-up before serving.
292-
long loadReadyEpoch = configManager.getLoadManager().startLoadServices();
297+
// Always start load services first. ConsensusManager gates external serving until warm-up.
298+
configManager.getLoadManager().startLoadServices();
293299

294300
if (CONF.isEnableTopologyProbing()) {
295301
configManager.getLoadManager().startTopologyService();
296302
}
297303

298-
threadPool.submit(() -> startLeaderServicesAfterLoadReady(loadReadyEpoch));
304+
threadPool.submit(this::startLeaderServicesAfterLoadReady);
299305
}
300306

301-
private void startLeaderServicesAfterLoadReady(long loadReadyEpoch) {
302-
if (!configManager.getLoadManager().waitForLoadReady(loadReadyEpoch)) {
307+
private void startLeaderServicesAfterLoadReady() {
308+
if (!configManager.getLoadManager().isLoadReady()) {
303309
LOGGER.info(
304-
ConfigNodeMessages.CURRENT_NODE_NODEID_IP_PORT_IS_NO_LONGER_THE_LEADER
305-
+ "skip starting leader services because load warm-up is interrupted",
310+
"Current ConfigNode(nodeId: {}, ip: {}) starts leader services while load warm-up is"
311+
+ " still in progress.",
306312
ConfigNodeDescriptor.getInstance().getConf().getConfigNodeId(),
307313
currentNodeTEndPoint);
308-
return;
309314
}
310315
if (!configManager.getConsensusManager().isLeaderReady()) {
311316
LOGGER.info(
@@ -355,12 +360,18 @@ private void startLeaderServicesAfterLoadReady(long loadReadyEpoch) {
355360
// Do check async because sync will be slow and block every other things.
356361
threadPool.submit(() -> configManager.getClusterManager().checkClusterId());
357362

363+
leaderServicesReady.set(true);
364+
358365
LOGGER.info(
359366
ConfigNodeMessages.CURRENT_NODE_NODEID_IP_PORT_AS_CONFIG_REGION_LEADER_IS,
360367
ConfigNodeDescriptor.getInstance().getConf().getConfigNodeId(),
361368
currentNodeTEndPoint);
362369
}
363370

371+
public boolean areLeaderServicesReady() {
372+
return leaderServicesReady.get();
373+
}
374+
364375
@Override
365376
public void start() {
366377
if (ConsensusFactory.SIMPLE_CONSENSUS.equals(CONF.getConfigNodeConsensusProtocolClass())) {

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java‎

Lines changed: 1 addition & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1247,13 +1247,7 @@ protected TSStatus confirmLeader() {
12471247
"ConsensusManager of target-ConfigNode is not initialized, "
12481248
+ "please make sure the target-ConfigNode has been started successfully.");
12491249
}
1250-
TSStatus status = getConsensusManager().confirmLeader();
1251-
if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()
1252-
&& !getLoadManager().isLoadReady()) {
1253-
return new TSStatus(TSStatusCode.CONFIG_NODE_LEADER_WARMING_UP.getStatusCode())
1254-
.setMessage(getLoadManager().getLoadReadyReason());
1255-
}
1256-
return status;
1250+
return getConsensusManager().confirmLeader();
12571251
}
12581252

12591253
@Override

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/consensus/ConsensusManager.java‎

Lines changed: 41 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -77,12 +77,14 @@ public class ConsensusManager {
7777
new ConfigRegionId(CONF.getConfigRegionId());
7878

7979
private final IManager configManager;
80+
private final ConfigRegionStateMachine stateMachine;
8081
private IConsensus consensusImpl;
8182

8283
private boolean isInitialized;
8384

8485
public ConsensusManager(IManager configManager, ConfigRegionStateMachine stateMachine) {
8586
this.configManager = configManager;
87+
this.stateMachine = stateMachine;
8688
setConsensusLayer(stateMachine);
8789
}
8890

@@ -424,39 +426,50 @@ public boolean isLeaderExist() {
424426
* NEED_REDIRECTION otherwise
425427
*/
426428
public TSStatus confirmLeader() {
427-
TSStatus result = new TSStatus();
428-
if (isLeaderReady()) {
429-
result.setCode(TSStatusCode.SUCCESS_STATUS.getStatusCode());
430-
} else {
431-
result.setCode(TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode());
432-
if (isLeader()) {
433-
long startTime = System.currentTimeMillis();
434-
while (System.currentTimeMillis() - startTime < MAX_WAIT_READY_TIME_MS) {
435-
if (isLeaderReady()) {
436-
result.setCode(TSStatusCode.SUCCESS_STATUS.getStatusCode());
437-
return result;
438-
}
439-
try {
440-
Thread.sleep(RETRY_WAIT_TIME_MS);
441-
} catch (InterruptedException e) {
442-
Thread.currentThread().interrupt();
443-
LOGGER.warn(
444-
ManagerMessages.UNEXPECTED_INTERRUPTION_DURING_WAITING_FOR_CONFIGNODE_LEADER_READY);
445-
break;
446-
}
447-
}
448-
result.setMessage(
449-
"The current ConfigNode is leader but not ready yet, please try again later.");
450-
} else {
451-
result.setMessage(
452-
"The current ConfigNode is not leader, please redirect to a new ConfigNode.");
453-
}
429+
if (!isLeader()) {
430+
TSStatus result = new TSStatus(TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode());
431+
result.setMessage(
432+
"The current ConfigNode is not leader, please redirect to a new ConfigNode.");
454433
TConfigNodeLocation leaderLocation = getLeaderLocation();
455434
if (leaderLocation != null) {
456435
result.setRedirectNode(leaderLocation.getInternalEndPoint());
457436
}
437+
return result;
458438
}
459-
return result;
439+
440+
long startTime = System.currentTimeMillis();
441+
while (System.currentTimeMillis() - startTime < MAX_WAIT_READY_TIME_MS) {
442+
if (isLeaderReady()) {
443+
break;
444+
}
445+
try {
446+
Thread.sleep(RETRY_WAIT_TIME_MS);
447+
} catch (InterruptedException e) {
448+
Thread.currentThread().interrupt();
449+
LOGGER.warn(
450+
ManagerMessages.UNEXPECTED_INTERRUPTION_DURING_WAITING_FOR_CONFIGNODE_LEADER_READY);
451+
break;
452+
}
453+
}
454+
455+
if (!isLeaderReady()) {
456+
return getLeaderWarmingUpStatus(
457+
"The current ConfigNode is leader but consensus is not ready yet.");
458+
}
459+
if (!stateMachine.areLeaderServicesReady()) {
460+
return getLeaderWarmingUpStatus(
461+
"The current ConfigNode is leader but leader services are not ready yet.");
462+
}
463+
if (!configManager.getLoadManager().isLoadReady()) {
464+
return getLeaderWarmingUpStatus(configManager.getLoadManager().getLoadReadyReason());
465+
}
466+
467+
return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
468+
}
469+
470+
private TSStatus getLeaderWarmingUpStatus(String message) {
471+
return new TSStatus(TSStatusCode.CONFIG_NODE_LEADER_WARMING_UP.getStatusCode())
472+
.setMessage(message);
460473
}
461474

462475
public ConsensusGroupId getConsensusGroupId() {

0 commit comments

Comments
 (0)