Skip to content

Commit 2501efb

Browse files
CRZbulabulaCRZbulabula
committed
Fix ConfigNode leader warm-up recovery
1 parent a57680d commit 2501efb

12 files changed

Lines changed: 230 additions & 87 deletions

File tree

‎iotdb-core/ainode/iotdb/ainode/core/constant.py‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -80,6 +80,7 @@
8080
class TSStatusCode(Enum):
8181
SUCCESS_STATUS = 200
8282
REDIRECTION_RECOMMEND = 400
83+
CONFIG_NODE_LEADER_WARMING_UP = 1014
8384
MODEL_EXISTED_ERROR = 1503
8485
MODEL_NOT_EXIST_ERROR = 1504
8586
CREATE_MODEL_ERROR = 1505

‎iotdb-core/ainode/iotdb/ainode/core/rpc/client.py‎

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -66,6 +66,7 @@ def __init__(self, config_leader: TEndPoint):
6666
"Fail to connect to any config node. Please check status of ConfigNodes"
6767
)
6868
self._RETRY_NUM = 10
69+
self._STARTUP_RETRY_NUM = 60
6970
self._RETRY_INTERVAL_IN_S = 1
7071

7172
self._try_to_connect()
@@ -163,6 +164,12 @@ def _update_config_node_leader(self, status: TSStatus) -> bool:
163164
else:
164165
self._config_leader = None
165166
return True
167+
if status.code == TSStatusCode.CONFIG_NODE_LEADER_WARMING_UP.get_status_code():
168+
logger.info(
169+
"ConfigNode leader is warming up before serving AINode, will wait and retry. Reason: {}",
170+
status.message,
171+
)
172+
return True
166173
return False
167174

168175
def node_register(
@@ -177,7 +184,7 @@ def node_register(
177184
versionInfo=version_info,
178185
)
179186

180-
for _ in range(0, self._RETRY_NUM):
187+
for _ in range(0, self._STARTUP_RETRY_NUM):
181188
try:
182189
resp = self._client.registerAINode(req)
183190
if not self._update_config_node_leader(resp.status):
@@ -208,7 +215,7 @@ def node_restart(
208215
versionInfo=version_info,
209216
)
210217

211-
for _ in range(0, self._RETRY_NUM):
218+
for _ in range(0, self._STARTUP_RETRY_NUM):
212219
try:
213220
resp = self._client.restartAINode(req)
214221
if not self._update_config_node_leader(resp.status):

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

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -120,9 +120,11 @@ public void onComplete(TDataNodeHeartbeatResp heartbeatResp) {
120120
&& SCHEMA_REGION_SHOULD_CACHE_CONSENSUS_SAMPLE)
121121
|| (TConsensusGroupType.DataRegion.equals(regionGroupId.getType())
122122
&& DATA_REGION_SHOULD_CACHE_CONSENSUS_SAMPLE);
123-
long logicalTimestamp =
123+
boolean hasConsensusLogicalTimestamp =
124124
heartbeatResp.isSetConsensusLogicalTimeMap()
125-
&& heartbeatResp.getConsensusLogicalTimeMap().containsKey(regionGroupId)
125+
&& heartbeatResp.getConsensusLogicalTimeMap().containsKey(regionGroupId);
126+
long logicalTimestamp =
127+
hasConsensusLogicalTimestamp
126128
? heartbeatResp.getConsensusLogicalTimeMap().get(regionGroupId)
127129
: heartbeatResp.getHeartbeatTimestamp();
128130
loadManager
@@ -132,7 +134,7 @@ public void onComplete(TDataNodeHeartbeatResp heartbeatResp) {
132134
nodeId,
133135
Boolean.TRUE.equals(isLeader),
134136
logicalTimestamp,
135-
shouldCacheConsensusSample);
137+
shouldCacheConsensusSample && hasConsensusLogicalTimestamp);
136138
});
137139
loadManager
138140
.getLoadCache()

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

Lines changed: 128 additions & 72 deletions
Original file line numberDiff line numberDiff line change
@@ -65,6 +65,7 @@
6565
import java.util.concurrent.ScheduledExecutorService;
6666
import java.util.concurrent.TimeUnit;
6767
import java.util.concurrent.atomic.AtomicBoolean;
68+
import java.util.concurrent.atomic.AtomicLong;
6869
import java.util.regex.Matcher;
6970
import java.util.regex.Pattern;
7071

@@ -78,6 +79,8 @@ public class ConfigRegionStateMachine implements IStateMachine, IStateMachine.Ev
7879
private static final ConfigNodeConfig CONF = ConfigNodeDescriptor.getInstance().getConf();
7980
private final ConfigPlanExecutor executor;
8081
private final AtomicBoolean leaderServicesReady;
82+
private final AtomicLong leaderServicesEpoch;
83+
private final Object leaderServicesLock;
8184
private ConfigManager configManager;
8285

8386
/** Variables for {@link ConfigNode} Simple Consensus. */
@@ -101,6 +104,8 @@ public class ConfigRegionStateMachine implements IStateMachine, IStateMachine.Ev
101104
public ConfigRegionStateMachine(ConfigManager configManager, ConfigPlanExecutor executor) {
102105
this.executor = executor;
103106
this.leaderServicesReady = new AtomicBoolean(false);
107+
this.leaderServicesEpoch = new AtomicLong(0);
108+
this.leaderServicesLock = new Object();
104109
this.configManager = configManager;
105110
this.currentNodeTEndPoint =
106111
new TEndPoint()
@@ -234,11 +239,11 @@ public void loadSnapshot(final File latestSnapshotRootDir) {
234239

235240
@Override
236241
public void notifyLeaderChanged(ConsensusGroupId groupId, int newLeaderId) {
237-
leaderServicesReady.set(false);
238242
// We get currentNodeId here because the currentNodeId
239243
// couldn't initialize earlier than the ConfigRegionStateMachine
240244
int currentNodeId = ConfigNodeDescriptor.getInstance().getConf().getConfigNodeId();
241245
if (currentNodeId != newLeaderId) {
246+
invalidateLeaderServices();
242247
LOGGER.info(
243248
ConfigNodeMessages.CURRENT_NODE_NODEID_IP_PORT_IS_NO_LONGER_THE_LEADER
244249
+ "the new leader is [nodeId:{}]",
@@ -250,7 +255,7 @@ public void notifyLeaderChanged(ConsensusGroupId groupId, int newLeaderId) {
250255

251256
@Override
252257
public void notifyNotLeader() {
253-
leaderServicesReady.set(false);
258+
invalidateLeaderServices();
254259
// We get currentNodeId here because the currentNodeId
255260
// couldn't initialize earlier than the ConfigRegionStateMachine
256261
int currentNodeId = ConfigNodeDescriptor.getInstance().getConf().getConfigNodeId();
@@ -259,25 +264,30 @@ public void notifyNotLeader() {
259264
+ "start cleaning up related services",
260265
currentNodeId,
261266
currentNodeTEndPoint);
262-
// Stop leader scheduling services
263-
configManager.getPipeManager().getPipeRuntimeCoordinator().stopPipeMetaSync();
264-
configManager.getPipeManager().getPipeRuntimeCoordinator().stopPipeHeartbeat();
265-
configManager.getSubscriptionManager().getSubscriptionCoordinator().stopSubscriptionMetaSync();
266-
configManager.getLoadManager().stopTopologyService();
267-
configManager.getLoadManager().stopLoadServices();
268-
configManager.getProcedureManager().stopExecutor();
269-
configManager.getRetryFailedTasksThread().stopRetryFailedTasksService();
270-
configManager.getPartitionManager().stopRegionCleaner();
271-
configManager.getCQManager().stopCQScheduler();
272-
configManager.getClusterSchemaManager().clearSchemaQuotaCache();
273-
// Remove Metric after leader change
274-
configManager.removeMetrics();
275-
276-
// Shutdown leader related service for config pipe
277-
PipeConfigNodeAgent.runtime().notifyLeaderUnavailable();
278-
279-
// Clean receiver file dir
280-
PipeConfigNodeAgent.receiver().cleanPipeReceiverDir();
267+
synchronized (leaderServicesLock) {
268+
// Stop leader scheduling services
269+
configManager.getPipeManager().getPipeRuntimeCoordinator().stopPipeMetaSync();
270+
configManager.getPipeManager().getPipeRuntimeCoordinator().stopPipeHeartbeat();
271+
configManager
272+
.getSubscriptionManager()
273+
.getSubscriptionCoordinator()
274+
.stopSubscriptionMetaSync();
275+
configManager.getLoadManager().stopTopologyService();
276+
configManager.getLoadManager().stopLoadServices();
277+
configManager.getProcedureManager().stopExecutor();
278+
configManager.getRetryFailedTasksThread().stopRetryFailedTasksService();
279+
configManager.getPartitionManager().stopRegionCleaner();
280+
configManager.getCQManager().stopCQScheduler();
281+
configManager.getClusterSchemaManager().clearSchemaQuotaCache();
282+
// Remove Metric after leader change
283+
configManager.removeMetrics();
284+
285+
// Shutdown leader related service for config pipe
286+
PipeConfigNodeAgent.runtime().notifyLeaderUnavailable();
287+
288+
// Clean receiver file dir
289+
PipeConfigNodeAgent.receiver().cleanPipeReceiverDir();
290+
}
281291

282292
LOGGER.info(
283293
ConfigNodeMessages.CURRENT_NODE_NODEID_IP_PORT_IS_NO_LONGER_THE_LEADER
@@ -288,7 +298,7 @@ public void notifyNotLeader() {
288298

289299
@Override
290300
public void notifyLeaderReady() {
291-
leaderServicesReady.set(false);
301+
final long epoch = nextLeaderServicesEpoch();
292302
LOGGER.info(
293303
ConfigNodeMessages.CURRENT_NODE_NODEID_IP_PORT_BECOMES_CONFIG_REGION_LEADER,
294304
ConfigNodeDescriptor.getInstance().getConf().getConfigNodeId(),
@@ -301,75 +311,121 @@ public void notifyLeaderReady() {
301311
configManager.getLoadManager().startTopologyService();
302312
}
303313

304-
threadPool.submit(this::startLeaderServicesAfterLoadReady);
314+
threadPool.submit(() -> startLeaderServicesAfterLoadReady(epoch));
305315
}
306316

307-
private void startLeaderServicesAfterLoadReady() {
317+
private void startLeaderServicesAfterLoadReady(final long epoch) {
308318
if (!configManager.getLoadManager().isLoadReady()) {
309319
LOGGER.info(
310320
"Current ConfigNode(nodeId: {}, ip: {}) starts leader services while load warm-up is"
311321
+ " still in progress.",
312322
ConfigNodeDescriptor.getInstance().getConf().getConfigNodeId(),
313323
currentNodeTEndPoint);
314324
}
315-
if (!configManager.getConsensusManager().isLeaderReady()) {
325+
synchronized (leaderServicesLock) {
326+
if (!isCurrentLeaderServicesEpoch(epoch)) {
327+
LOGGER.info(
328+
ConfigNodeMessages.CURRENT_NODE_NODEID_IP_PORT_IS_NO_LONGER_THE_LEADER
329+
+ "skip starting leader services because the leader epoch is stale",
330+
ConfigNodeDescriptor.getInstance().getConf().getConfigNodeId(),
331+
currentNodeTEndPoint);
332+
return;
333+
}
334+
335+
// Start leader scheduling services
336+
submitIfLeaderServicesEpochCurrent(
337+
epoch, () -> configManager.getProcedureManager().getStore().getProcedureInfo().upgrade());
338+
configManager.getRetryFailedTasksThread().startRetryFailedTasksService();
339+
configManager.getPartitionManager().startRegionCleaner();
340+
// Add Metric after leader ready
341+
configManager.addMetrics();
342+
343+
// Activate leader related service for config pipe
344+
PipeConfigNodeAgent.runtime().notifyLeaderReady();
345+
346+
// we do cq recovery async for performance:
347+
// cq recovery may be time-consuming, we use another thread to do it in
348+
// make notifyLeaderChanged not blocked by it
349+
submitIfLeaderServicesEpochCurrent(
350+
epoch, () -> configManager.getCQManager().startCQScheduler());
351+
352+
submitIfLeaderServicesEpochCurrent(
353+
epoch,
354+
() -> configManager.getPipeManager().getPipeRuntimeCoordinator().startPipeMetaSync());
355+
submitIfLeaderServicesEpochCurrent(
356+
epoch,
357+
() -> configManager.getPipeManager().getPipeRuntimeCoordinator().startPipeHeartbeat());
358+
submitIfLeaderServicesEpochCurrent(
359+
epoch,
360+
() ->
361+
configManager
362+
.getPipeManager()
363+
.getPipeRuntimeCoordinator()
364+
.onConfigRegionGroupLeaderChanged());
365+
366+
submitIfLeaderServicesEpochCurrent(
367+
epoch,
368+
() ->
369+
configManager
370+
.getSubscriptionManager()
371+
.getSubscriptionCoordinator()
372+
.startSubscriptionMetaSync());
373+
374+
// To adapt old version, we check cluster ID after state machine has been fully recovered.
375+
// Do check async because sync will be slow and block every other things.
376+
submitIfLeaderServicesEpochCurrent(
377+
epoch, () -> configManager.getClusterManager().checkClusterId());
378+
379+
if (!isCurrentLeaderServicesEpoch(epoch)) {
380+
return;
381+
}
382+
configManager
383+
.getProcedureManager()
384+
.startExecutor(
385+
() -> {
386+
if (isCurrentLeaderServicesEpoch(epoch)) {
387+
leaderServicesReady.set(true);
388+
}
389+
});
390+
if (!leaderServicesReady.get()) {
391+
return;
392+
}
393+
}
394+
395+
if (isCurrentLeaderServicesEpoch(epoch)) {
316396
LOGGER.info(
317-
ConfigNodeMessages.CURRENT_NODE_NODEID_IP_PORT_IS_NO_LONGER_THE_LEADER
318-
+ "skip starting leader services because consensus leader is no longer ready",
397+
ConfigNodeMessages.CURRENT_NODE_NODEID_IP_PORT_AS_CONFIG_REGION_LEADER_IS,
319398
ConfigNodeDescriptor.getInstance().getConf().getConfigNodeId(),
320399
currentNodeTEndPoint);
321-
return;
322400
}
401+
}
323402

324-
// Start leader scheduling services
325-
configManager.getProcedureManager().startExecutor();
326-
threadPool.submit(
327-
() -> configManager.getProcedureManager().getStore().getProcedureInfo().upgrade());
328-
configManager.getRetryFailedTasksThread().startRetryFailedTasksService();
329-
configManager.getPartitionManager().startRegionCleaner();
330-
// Add Metric after leader ready
331-
configManager.addMetrics();
332-
333-
// Activate leader related service for config pipe
334-
PipeConfigNodeAgent.runtime().notifyLeaderReady();
335-
336-
// we do cq recovery async for performance:
337-
// cq recovery may be time-consuming, we use another thread to do it in
338-
// make notifyLeaderChanged not blocked by it
339-
threadPool.submit(() -> configManager.getCQManager().startCQScheduler());
340-
341-
threadPool.submit(
342-
() -> configManager.getPipeManager().getPipeRuntimeCoordinator().startPipeMetaSync());
343-
threadPool.submit(
344-
() -> configManager.getPipeManager().getPipeRuntimeCoordinator().startPipeHeartbeat());
345-
threadPool.submit(
346-
() ->
347-
configManager
348-
.getPipeManager()
349-
.getPipeRuntimeCoordinator()
350-
.onConfigRegionGroupLeaderChanged());
351-
352-
threadPool.submit(
353-
() ->
354-
configManager
355-
.getSubscriptionManager()
356-
.getSubscriptionCoordinator()
357-
.startSubscriptionMetaSync());
403+
public boolean areLeaderServicesReady() {
404+
return leaderServicesReady.get();
405+
}
358406

359-
// To adapt old version, we check cluster ID after state machine has been fully recovered.
360-
// Do check async because sync will be slow and block every other things.
361-
threadPool.submit(() -> configManager.getClusterManager().checkClusterId());
407+
private long nextLeaderServicesEpoch() {
408+
leaderServicesReady.set(false);
409+
return leaderServicesEpoch.incrementAndGet();
410+
}
362411

363-
leaderServicesReady.set(true);
412+
private void invalidateLeaderServices() {
413+
leaderServicesReady.set(false);
414+
leaderServicesEpoch.incrementAndGet();
415+
}
364416

365-
LOGGER.info(
366-
ConfigNodeMessages.CURRENT_NODE_NODEID_IP_PORT_AS_CONFIG_REGION_LEADER_IS,
367-
ConfigNodeDescriptor.getInstance().getConf().getConfigNodeId(),
368-
currentNodeTEndPoint);
417+
private boolean isCurrentLeaderServicesEpoch(final long epoch) {
418+
return leaderServicesEpoch.get() == epoch
419+
&& configManager.getConsensusManager().isLeaderReady();
369420
}
370421

371-
public boolean areLeaderServicesReady() {
372-
return leaderServicesReady.get();
422+
private void submitIfLeaderServicesEpochCurrent(final long epoch, final Runnable task) {
423+
threadPool.submit(
424+
() -> {
425+
if (isCurrentLeaderServicesEpoch(epoch)) {
426+
task.run();
427+
}
428+
});
373429
}
374430

375431
@Override

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

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1247,6 +1247,10 @@ protected TSStatus confirmLeader() {
12471247
"ConsensusManager of target-ConfigNode is not initialized, "
12481248
+ "please make sure the target-ConfigNode has been started successfully.");
12491249
}
1250+
// Procedure recovery replays metadata writes before external load warm-up is complete.
1251+
if (procedureManager.isProcedureExecutionThread()) {
1252+
return getConsensusManager().confirmLeaderForInternalProcedure();
1253+
}
12501254
return getConsensusManager().confirmLeader();
12511255
}
12521256

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

Lines changed: 12 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -226,14 +226,21 @@ public ProcedureManager(ConfigManager configManager, ProcedureInfo procedureInfo
226226
}
227227

228228
public void startExecutor() {
229+
startExecutor(null);
230+
}
231+
232+
public void startExecutor(final Runnable beforeStartingWorkers) {
229233
if (!executor.isRunning()) {
230234
executor.init(CONFIG_NODE_CONFIG.getProcedureCoreWorkerThreadsCount());
231-
executor.startWorkers();
232235
executor.startCompletedCleaner(
233236
CONFIG_NODE_CONFIG.getProcedureCompletedCleanInterval(),
234237
CONFIG_NODE_CONFIG.getProcedureCompletedEvictTTL());
235238
executor.addInternalProcedure(partitionTableCleaner);
236239
store.start();
240+
if (beforeStartingWorkers != null) {
241+
beforeStartingWorkers.run();
242+
}
243+
executor.startWorkers();
237244
LOGGER.info(ManagerMessages.PROCEDUREMANAGER_IS_STARTED_SUCCESSFULLY);
238245
}
239246
}
@@ -250,6 +257,10 @@ public void stopExecutor() {
250257
}
251258
}
252259

260+
public boolean isProcedureExecutionThread() {
261+
return ProcedureExecutor.isProcedureExecutionThread();
262+
}
263+
253264
@TestOnly
254265
public TSStatus createManyDatabases() {
255266
this.executor.submitProcedure(new CreateManyDatabasesProcedure());

0 commit comments

Comments
 (0)