Skip to content

Commit 8f9f175

Browse files
CRZbulabulaCRZbulabula
committed
Clean up leader warm-up heartbeat flow
1 parent bac0c60 commit 8f9f175

5 files changed

Lines changed: 140 additions & 137 deletions

File tree

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

Lines changed: 72 additions & 47 deletions
Original file line numberDiff line numberDiff line change
@@ -39,7 +39,6 @@
3939

4040
import java.util.Collections;
4141
import java.util.Map;
42-
import java.util.Set;
4342
import java.util.function.Consumer;
4443

4544
public class DataNodeHeartbeatHandler implements AsyncMethodCallback<TDataNodeHeartbeatResp> {
@@ -53,7 +52,6 @@ public class DataNodeHeartbeatHandler implements AsyncMethodCallback<TDataNodeHe
5352
private final int nodeId;
5453

5554
private final LoadManager loadManager;
56-
private final Set<TConsensusGroupId> regionGroupIdsOnDataNode;
5755

5856
private final Map<Integer, Long> deviceNum;
5957
private final Map<Integer, Long> timeSeriesNum;
@@ -67,7 +65,6 @@ public class DataNodeHeartbeatHandler implements AsyncMethodCallback<TDataNodeHe
6765
public DataNodeHeartbeatHandler(
6866
int nodeId,
6967
LoadManager loadManager,
70-
Set<TConsensusGroupId> regionGroupIdsOnDataNode,
7168
Map<Integer, Long> deviceNum,
7269
Map<Integer, Long> timeSeriesNum,
7370
Map<Integer, Long> regionDisk,
@@ -77,7 +74,6 @@ public DataNodeHeartbeatHandler(
7774

7875
this.nodeId = nodeId;
7976
this.loadManager = loadManager;
80-
this.regionGroupIdsOnDataNode = regionGroupIdsOnDataNode;
8177
this.deviceNum = deviceNum;
8278
this.timeSeriesNum = timeSeriesNum;
8379
this.regionDisk = regionDisk;
@@ -88,11 +84,21 @@ public DataNodeHeartbeatHandler(
8884

8985
@Override
9086
public void onComplete(TDataNodeHeartbeatResp heartbeatResp) {
91-
// Update NodeCache
87+
cacheNodeHeartbeatSample(heartbeatResp);
88+
cacheRegionGroupHeartbeatSamples(heartbeatResp);
89+
cacheUsageSamples(heartbeatResp);
90+
cachePipeHeartbeat(heartbeatResp);
91+
cacheConfirmedConfigNodeEndPoints(heartbeatResp);
92+
cacheRegionSizeSamples(heartbeatResp);
93+
}
94+
95+
private void cacheNodeHeartbeatSample(TDataNodeHeartbeatResp heartbeatResp) {
9296
loadManager
9397
.getLoadCache()
9498
.cacheDataNodeHeartbeatSample(nodeId, new NodeHeartbeatSample(heartbeatResp));
99+
}
95100

101+
private void cacheRegionGroupHeartbeatSamples(TDataNodeHeartbeatResp heartbeatResp) {
96102
RegionStatus regionStatus = RegionStatus.valueOf(heartbeatResp.getStatus());
97103

98104
Map<TConsensusGroupId, Boolean> judgedLeaders =
@@ -101,53 +107,63 @@ public void onComplete(TDataNodeHeartbeatResp heartbeatResp) {
101107
: Collections.emptyMap();
102108
judgedLeaders.forEach(
103109
(regionGroupId, isLeader) -> {
104-
105-
// Do not allow regions to inherit the Removing state from datanode
106-
RegionStatus nextRegionStatus = regionStatus;
107-
if (nextRegionStatus == RegionStatus.Removing) {
108-
nextRegionStatus =
109-
loadManager.getLoadCache().getRegionCacheLastSampleStatus(regionGroupId, nodeId);
110-
}
111-
112-
// Update RegionGroupCache
113-
loadManager
114-
.getLoadCache()
115-
.cacheRegionHeartbeatSample(
116-
regionGroupId,
117-
nodeId,
118-
new RegionHeartbeatSample(
119-
heartbeatResp.getHeartbeatTimestamp(),
120-
// Region will inherit DataNode's status
121-
nextRegionStatus),
122-
false);
123-
124-
boolean shouldCacheConsensusSample =
125-
(TConsensusGroupType.SchemaRegion.equals(regionGroupId.getType())
126-
&& SCHEMA_REGION_SHOULD_CACHE_CONSENSUS_SAMPLE)
127-
|| (TConsensusGroupType.DataRegion.equals(regionGroupId.getType())
128-
&& DATA_REGION_SHOULD_CACHE_CONSENSUS_SAMPLE);
129-
boolean hasConsensusLogicalTimestamp =
130-
heartbeatResp.isSetConsensusLogicalTimeMap()
131-
&& heartbeatResp.getConsensusLogicalTimeMap().containsKey(regionGroupId);
132-
if (shouldCacheConsensusSample
133-
&& hasConsensusLogicalTimestamp
134-
&& Boolean.TRUE.equals(isLeader)) {
135-
loadManager
136-
.getLoadCache()
137-
.cacheConsensusSample(
138-
regionGroupId,
139-
new ConsensusGroupHeartbeatSample(
140-
heartbeatResp.getConsensusLogicalTimeMap().get(regionGroupId), nodeId));
141-
}
110+
cacheRegionHeartbeatSample(heartbeatResp, regionStatus, regionGroupId);
111+
cacheConsensusSampleIfNeeded(heartbeatResp, regionGroupId, isLeader);
142112
});
113+
}
114+
115+
private void cacheRegionHeartbeatSample(
116+
TDataNodeHeartbeatResp heartbeatResp,
117+
RegionStatus dataNodeRegionStatus,
118+
TConsensusGroupId regionGroupId) {
143119
loadManager
144120
.getLoadCache()
145-
.cacheUnreportedDataNodeRegionHeartbeatSamples(
121+
.cacheRegionHeartbeatSample(
122+
regionGroupId,
146123
nodeId,
147-
regionGroupIdsOnDataNode,
148-
judgedLeaders.keySet(),
149-
heartbeatResp.getHeartbeatTimestamp());
124+
new RegionHeartbeatSample(
125+
heartbeatResp.getHeartbeatTimestamp(),
126+
getRegionHeartbeatStatus(regionGroupId, dataNodeRegionStatus)),
127+
false);
128+
}
150129

130+
private RegionStatus getRegionHeartbeatStatus(
131+
TConsensusGroupId regionGroupId, RegionStatus dataNodeRegionStatus) {
132+
return dataNodeRegionStatus == RegionStatus.Removing
133+
? loadManager.getLoadCache().getRegionCacheLastSampleStatus(regionGroupId, nodeId)
134+
: dataNodeRegionStatus;
135+
}
136+
137+
private void cacheConsensusSampleIfNeeded(
138+
TDataNodeHeartbeatResp heartbeatResp, TConsensusGroupId regionGroupId, Boolean isLeader) {
139+
if (!Boolean.TRUE.equals(isLeader)
140+
|| !shouldCacheConsensusSample(regionGroupId)
141+
|| !hasConsensusLogicalTimestamp(heartbeatResp, regionGroupId)) {
142+
return;
143+
}
144+
145+
loadManager
146+
.getLoadCache()
147+
.cacheConsensusSample(
148+
regionGroupId,
149+
new ConsensusGroupHeartbeatSample(
150+
heartbeatResp.getConsensusLogicalTimeMap().get(regionGroupId), nodeId));
151+
}
152+
153+
private boolean shouldCacheConsensusSample(TConsensusGroupId regionGroupId) {
154+
return (TConsensusGroupType.SchemaRegion.equals(regionGroupId.getType())
155+
&& SCHEMA_REGION_SHOULD_CACHE_CONSENSUS_SAMPLE)
156+
|| (TConsensusGroupType.DataRegion.equals(regionGroupId.getType())
157+
&& DATA_REGION_SHOULD_CACHE_CONSENSUS_SAMPLE);
158+
}
159+
160+
private boolean hasConsensusLogicalTimestamp(
161+
TDataNodeHeartbeatResp heartbeatResp, TConsensusGroupId regionGroupId) {
162+
return heartbeatResp.isSetConsensusLogicalTimeMap()
163+
&& heartbeatResp.getConsensusLogicalTimeMap().containsKey(regionGroupId);
164+
}
165+
166+
private void cacheUsageSamples(TDataNodeHeartbeatResp heartbeatResp) {
151167
if (heartbeatResp.getRegionDeviceUsageMap() != null) {
152168
deviceNum.putAll(heartbeatResp.getRegionDeviceUsageMap());
153169
deviceUsageRespProcess.accept(heartbeatResp.getRegionDeviceUsageMap());
@@ -159,6 +175,9 @@ public void onComplete(TDataNodeHeartbeatResp heartbeatResp) {
159175
if (heartbeatResp.getRegionDisk() != null) {
160176
regionDisk.putAll(heartbeatResp.getRegionDisk());
161177
}
178+
}
179+
180+
private void cachePipeHeartbeat(TDataNodeHeartbeatResp heartbeatResp) {
162181
if (heartbeatResp.getPipeMetaList() != null) {
163182
pipeRuntimeCoordinator.parseHeartbeat(
164183
nodeId,
@@ -167,12 +186,18 @@ public void onComplete(TDataNodeHeartbeatResp heartbeatResp) {
167186
heartbeatResp.getPipeRemainingEventCountList(),
168187
heartbeatResp.getPipeRemainingTimeList());
169188
}
189+
}
190+
191+
private void cacheConfirmedConfigNodeEndPoints(TDataNodeHeartbeatResp heartbeatResp) {
170192
if (heartbeatResp.isSetConfirmedConfigNodeEndPoints()) {
171193
loadManager
172194
.getLoadCache()
173195
.updateConfirmedConfigNodeEndPoints(
174196
nodeId, heartbeatResp.getConfirmedConfigNodeEndPoints());
175197
}
198+
}
199+
200+
private void cacheRegionSizeSamples(TDataNodeHeartbeatResp heartbeatResp) {
176201
if (heartbeatResp.isSetRegionDisk()) {
177202
loadManager.getLoadCache().updateRegionSizeMap(nodeId, heartbeatResp.getRegionDisk());
178203
}

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

Lines changed: 49 additions & 46 deletions
Original file line numberDiff line numberDiff line change
@@ -60,7 +60,6 @@
6060
import java.nio.file.StandardCopyOption;
6161
import java.util.Arrays;
6262
import java.util.Comparator;
63-
import java.util.Optional;
6463
import java.util.concurrent.ExecutorService;
6564
import java.util.concurrent.ScheduledExecutorService;
6665
import java.util.concurrent.TimeUnit;
@@ -101,8 +100,8 @@ public class ConfigRegionStateMachine implements IStateMachine, IStateMachine.Ev
101100
private static final long LOG_FILE_MAX_SIZE =
102101
CONF.getConfigNodeSimpleConsensusLogSegmentSizeMax();
103102
private final TEndPoint currentNodeTEndPoint;
104-
private static Pattern LOG_INPROGRESS_PATTERN = Pattern.compile("\\d+");
105-
private static Pattern LOG_PATTERN = Pattern.compile("(?<=_)(\\d+)$");
103+
private static final Pattern LOG_INPROGRESS_PATTERN = Pattern.compile("\\d+");
104+
private static final Pattern LOG_PATTERN = Pattern.compile("(?<=_)(\\d+)$");
106105

107106
public ConfigRegionStateMachine(ConfigManager configManager, ConfigPlanExecutor executor) {
108107
this.executor = executor;
@@ -126,9 +125,9 @@ public void setConfigManager(ConfigManager configManager) {
126125

127126
@Override
128127
public TSStatus write(IConsensusRequest request) {
129-
return Optional.ofNullable(request)
130-
.map(o -> write((ConfigPhysicalPlan) request))
131-
.orElseGet(() -> new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode()));
128+
return request == null
129+
? new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode())
130+
: write((ConfigPhysicalPlan) request);
132131
}
133132

134133
/** Transmit {@link ConfigPhysicalPlan} to {@link ConfigPlanExecutor} */
@@ -314,18 +313,34 @@ public void notifyLeaderReady() {
314313
configManager.getLoadManager().startTopologyService();
315314
}
316315

317-
threadPool.submit(() -> startLeaderServicesAndWaitLoadReady(epoch));
316+
threadPool.submit(() -> startLeaderServicesAndWaitForLoadReady(epoch));
318317
}
319318

320-
private void startLeaderServicesAndWaitLoadReady(final long epoch) {
319+
private void startLeaderServicesAndWaitForLoadReady(final long epoch) {
320+
if (!startLeaderServices(epoch)) {
321+
return;
322+
}
323+
324+
boolean loadReady = waitForLoadReady(epoch);
325+
if (!isCurrentLeaderServicesEpoch(epoch)) {
326+
return;
327+
}
328+
logLoadWarmUpIfNeeded(loadReady);
329+
LOGGER.info(
330+
ConfigNodeMessages.CURRENT_NODE_NODEID_IP_PORT_AS_CONFIG_REGION_LEADER_IS,
331+
ConfigNodeDescriptor.getInstance().getConf().getConfigNodeId(),
332+
currentNodeTEndPoint);
333+
}
334+
335+
private boolean startLeaderServices(final long epoch) {
321336
synchronized (leaderServicesLock) {
322337
if (!isCurrentLeaderServicesEpoch(epoch)) {
323338
LOGGER.info(
324339
ConfigNodeMessages.CURRENT_NODE_NODEID_IP_PORT_IS_NO_LONGER_THE_LEADER
325340
+ "skip starting leader services because the leader epoch is stale",
326341
ConfigNodeDescriptor.getInstance().getConf().getConfigNodeId(),
327342
currentNodeTEndPoint);
328-
return;
343+
return false;
329344
}
330345

331346
// Start leader scheduling services
@@ -373,25 +388,23 @@ private void startLeaderServicesAndWaitLoadReady(final long epoch) {
373388
epoch, () -> configManager.getClusterManager().checkClusterId());
374389

375390
if (!isCurrentLeaderServicesEpoch(epoch)) {
376-
return;
391+
return false;
377392
}
378393
configManager
379394
.getProcedureManager()
380-
.startExecutor(
381-
() -> {
382-
if (isCurrentLeaderServicesEpoch(epoch)) {
383-
leaderServicesReady.set(true);
384-
}
385-
});
386-
if (!leaderServicesReady.get()) {
387-
return;
388-
}
395+
.startExecutor(() -> markLeaderServicesReadyIfEpochCurrent(epoch));
396+
markLeaderServicesReadyIfEpochCurrent(epoch);
397+
return leaderServicesReady.get();
389398
}
399+
}
390400

391-
boolean loadReady = waitForLoadReady(epoch);
392-
if (!isCurrentLeaderServicesEpoch(epoch)) {
393-
return;
401+
private void markLeaderServicesReadyIfEpochCurrent(final long epoch) {
402+
if (isCurrentLeaderServicesEpoch(epoch)) {
403+
leaderServicesReady.set(true);
394404
}
405+
}
406+
407+
private void logLoadWarmUpIfNeeded(final boolean loadReady) {
395408
if (!loadReady) {
396409
LOGGER.info(
397410
"Current ConfigNode(nodeId: {}, ip: {}) finished starting leader services while load"
@@ -400,13 +413,6 @@ private void startLeaderServicesAndWaitLoadReady(final long epoch) {
400413
currentNodeTEndPoint,
401414
configManager.getLoadManager().getLoadReadyReason());
402415
}
403-
404-
if (isCurrentLeaderServicesEpoch(epoch)) {
405-
LOGGER.info(
406-
ConfigNodeMessages.CURRENT_NODE_NODEID_IP_PORT_AS_CONFIG_REGION_LEADER_IS,
407-
ConfigNodeDescriptor.getInstance().getConf().getConfigNodeId(),
408-
currentNodeTEndPoint);
409-
}
410416
}
411417

412418
private boolean waitForLoadReady(final long epoch) {
@@ -416,17 +422,24 @@ private boolean waitForLoadReady(final long epoch) {
416422
if (configManager.getLoadManager().isLoadReady()) {
417423
return true;
418424
}
419-
try {
420-
TimeUnit.MILLISECONDS.sleep(WAIT_LOAD_READY_INTERVAL_MS);
421-
} catch (InterruptedException e) {
422-
Thread.currentThread().interrupt();
423-
LOGGER.warn("Unexpected interruption while waiting for ConfigNode leader load warm-up.", e);
425+
if (!sleepForLoadReady()) {
424426
return false;
425427
}
426428
}
427429
return isCurrentLeaderServicesEpoch(epoch) && configManager.getLoadManager().isLoadReady();
428430
}
429431

432+
private boolean sleepForLoadReady() {
433+
try {
434+
TimeUnit.MILLISECONDS.sleep(WAIT_LOAD_READY_INTERVAL_MS);
435+
return true;
436+
} catch (InterruptedException e) {
437+
Thread.currentThread().interrupt();
438+
LOGGER.warn("Unexpected interruption while waiting for ConfigNode leader load warm-up.", e);
439+
return false;
440+
}
441+
}
442+
430443
public boolean areLeaderServicesReady() {
431444
return leaderServicesReady.get();
432445
}
@@ -528,7 +541,7 @@ private void initStandAloneConfigNode() {
528541
dir.mkdirs();
529542
String[] list = new File(CURRENT_FILE_DIR).list();
530543
if (list != null && list.length != 0) {
531-
Arrays.sort(list, new FileComparator());
544+
Arrays.sort(list, Comparator.comparingLong(ConfigRegionStateMachine::parseEndIndex));
532545
for (String logFileName : list) {
533546
File logFile =
534547
SystemFileFactory.INSTANCE.getFile(CURRENT_FILE_DIR + File.separator + logFileName);
@@ -612,17 +625,7 @@ private void createLogFile(int startIndex) {
612625
}
613626
}
614627

615-
static class FileComparator implements Comparator<String> {
616-
617-
@Override
618-
public int compare(String filename1, String filename2) {
619-
long id1 = parseEndIndex(filename1);
620-
long id2 = parseEndIndex(filename2);
621-
return Long.compare(id1, id2);
622-
}
623-
}
624-
625-
static long parseEndIndex(String filename) {
628+
private static long parseEndIndex(String filename) {
626629
if (filename.startsWith("log_inprogress_")) {
627630
Matcher matcher = LOG_INPROGRESS_PATTERN.matcher(filename);
628631
if (matcher.find()) {

0 commit comments

Comments
 (0)