Skip to content

Commit a3b1b0b

Browse files
CaideyipiCaideyipi
committed
fix
1 parent 01f12e7 commit a3b1b0b

10 files changed

Lines changed: 159 additions & 165 deletions

File tree

‎iotdb-core/consensus/src/main/i18n/en/org/apache/iotdb/consensus/i18n/IoTConsensusV2Messages.java‎

Lines changed: 28 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -30,77 +30,77 @@ private IoTConsensusV2Messages() {}
3030
// ===================== IoTConsensusV2 lifecycle =====================
3131

3232
public static final String RECOVER_TASK_CANCELLED =
33-
"IoTV2 Recover Task is cancelled";
33+
"IoTV2 recover task was cancelled";
3434
public static final String RECOVER_FUTURE_EXCEPTION =
3535
"Exception while waiting for recover future completion";
3636
public static final String RECOVER_TASK_INTERRUPTED =
37-
"IoTV2 Recover Task is interrupted";
37+
"IoTV2 recover task was interrupted";
3838
public static final String FAILED_RECOVER_CONSENSUS =
39-
"Failed to recover consensus from {} for {}, ignore it and continue recover other group, async backend checker thread will automatically deregister related pipe side effects for this failed consensus group.";
39+
"Failed to recover consensus group {} from {}; ignoring it and continuing to recover other groups. The async backend checker thread will automatically deregister related pipe side effects for this failed consensus group.";
4040
public static final String FAILED_RECOVER_CONSENSUS_READ_DIR =
41-
"Failed to recover consensus from {} because read dir failed";
41+
"Failed to recover consensus from {} because reading the directory failed";
4242
public static final String FAILED_RECOVER_CONSENSUS_SHORT =
4343
"Failed to recover consensus from {}";
4444

4545
// ===================== IoTConsensusV2 peer operations =====================
4646

4747
public static final String START_DELETE_LOCAL_PEER =
48-
"[{}] start to delete local peer for group {}";
48+
"[{}] start deleting local peer for group {}";
4949
public static final String FINISH_DELETE_LOCAL_PEER =
50-
"[{}] finish deleting local peer for group {}";
50+
"[{}] finished deleting local peer for group {}";
5151
public static final String INACTIVATE_NEW_PEER =
52-
"[{}] inactivate new peer: {}";
52+
"[{}] deactivate new peer: {}";
5353
public static final String NOTIFY_CREATE_CONSENSUS_PIPES =
5454
"[{}] notify current peers to create consensus pipes...";
5555
public static final String WAIT_PEERS_FINISH_TRANSFER =
56-
"[{}] wait until all the other peers finish transferring...";
56+
"[{}] wait until all other peers finish transferring...";
5757
public static final String ACTIVATE_NEW_PEER =
5858
"[{}] activate new peer...";
5959
public static final String ADD_REMOTE_PEER_FAILED_CLEANUP =
60-
"[{}] add remote peer failed, automatic cleanup side effects...";
60+
"[{}] failed to add remote peer, automatically cleaning up side effects...";
6161
public static final String FAILED_CLEANUP_SIDE_EFFECTS =
62-
"[{}] failed to cleanup side effects after failed to add remote peer";
62+
"[{}] failed to clean up side effects after failing to add remote peer";
6363
public static final String NOTIFY_DROP_CONSENSUS_PIPES =
6464
"[{}] notify other peers to drop consensus pipes...";
6565
public static final String INACTIVATE_PEER =
66-
"[{}] inactivate peer {}";
66+
"[{}] deactivate peer {}";
6767
public static final String WAIT_TARGET_PEER_COMPLETE_TRANSFER =
68-
"[{}] wait target peer{} complete transfer...";
68+
"[{}] wait for target peer {} to complete transfer...";
6969
public static final String WAIT_PEER_RELEASE_RESOURCE =
70-
"[{}] wait {} to release all resource...";
70+
"[{}] wait for {} to release all resources...";
7171
public static final String NOT_SUPPORT_LEADER_TRANSFER =
7272
"%s does not support leader transfer";
7373

7474
// ===================== IoTConsensusV2ServerImpl =====================
7575

7676
public static final String ERROR_SET_PEER_ACTIVE =
77-
"error when set peer %s to active %s. result status: %s";
77+
"error when setting peer %s active status to %s. result status: %s";
7878
public static final String ERROR_SET_PEER_ACTIVE_SHORT =
79-
"error when set peer %s to active %s";
79+
"error when setting peer %s active status to %s";
8080
public static final String TARGET_PEER_MAY_BE_DOWN =
81-
"target peer may be down, error when set peer {} to active {}";
81+
"target peer may be down; error when setting peer {} active status to {}";
8282
public static final String CANNOT_NOTIFY_PEER_CREATE_PIPE =
83-
"{} cannot notify peer {} to create consensus pipe, may because that peer is unknown currently, please manually check!";
83+
"{} cannot notify peer {} to create consensus pipe, possibly because that peer is currently unknown. Please check it manually.";
8484
public static final String CANNOT_CREATE_CONSENSUS_PIPE =
85-
"{} cannot create consensus pipe to {}, may because target peer is unknown currently, please manually check!";
85+
"{} cannot create consensus pipe to {}, possibly because the target peer is currently unknown. Please check it manually.";
8686
public static final String ERROR_NOTIFY_PEER_CREATE_PIPE =
87-
"error when notify peer %s to create consensus pipe";
87+
"error when notifying peer %s to create consensus pipe";
8888
public static final String CANNOT_NOTIFY_PEER_DROP_PIPE =
89-
"{} cannot notify peer {} to drop consensus pipe, may because that peer is unknown currently, please manually check!";
89+
"{} cannot notify peer {} to drop consensus pipe, possibly because that peer is currently unknown. Please check it manually.";
9090
public static final String CANNOT_DROP_CONSENSUS_PIPE =
91-
"{} cannot drop consensus pipe to {}, may because target peer is unknown currently, please manually check!";
91+
"{} cannot drop consensus pipe to {}, possibly because the target peer is currently unknown. Please check it manually.";
9292
public static final String ERROR_NOTIFY_PEER_DROP_PIPE =
93-
"error when notify peer %s to drop consensus pipe";
93+
"error when notifying peer %s to drop consensus pipe";
9494
public static final String INTERRUPTED_WAITING_TRANSFER =
95-
"{} is interrupted when waiting for transfer completed";
95+
"{} was interrupted while waiting for transfer completion";
9696
public static final String INTERRUPTED_WAITING_TRANSFER_FMT =
97-
"%s is interrupted when waiting for transfer completed";
97+
"%s was interrupted while waiting for transfer completion";
9898
public static final String CANNOT_CHECK_PIPE_TRANSMISSION =
99-
"{} cannot check consensus pipes transmission completed to peer {}";
99+
"{} cannot check whether consensus pipe transmission to peer {} has completed";
100100
public static final String ERROR_CHECK_PIPE_TRANSMISSION =
101-
"error when check consensus pipes transmission completed to peer %s";
101+
"error when checking whether consensus pipe transmission to peer %s has completed";
102102
public static final String CANNOT_CHECK_PIPE_TRANSMISSION_SHORT =
103-
"{} cannot check consensus pipes transmission completed";
103+
"{} cannot check whether consensus pipe transmission has completed";
104104

105105
// ===================== IoTConsensusV2RPCServiceProcessor =====================
106106

@@ -119,5 +119,5 @@ private IoTConsensusV2Messages() {}
119119
public static final String FAILED_DROP_CONSENSUS_PIPE =
120120
"Failed to drop consensus pipe to target peer with req {}";
121121
public static final String FAILED_CHECK_CONSENSUS_PIPE =
122-
"Failed to check consensus pipe completed with req {}, set is completed to {}";
122+
"Failed to check whether consensus pipe has completed with req {}; set isCompleted to {}";
123123
}

‎iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/IoTConsensusV2.java‎

Lines changed: 13 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -146,7 +146,7 @@ private Future<Void> initAndRecover() throws IOException {
146146
}
147147
return CompletableFuture.completedFuture(null);
148148
} else {
149-
// asynchronously recover, retry logic is implemented at IoTConsensusV2Impl
149+
// Recover asynchronously; retry logic is handled by the async backend checker.
150150
return CompletableFuture.runAsync(
151151
() -> {
152152
try (DirectoryStream<Path> stream = Files.newDirectoryStream(storageDir.toPath())) {
@@ -166,8 +166,8 @@ private Future<Void> initAndRecover() throws IOException {
166166
} catch (Exception e) {
167167
LOGGER.error(
168168
IoTConsensusV2Messages.FAILED_RECOVER_CONSENSUS,
169-
storageDir,
170169
consensusGroupId,
170+
storageDir,
171171
e);
172172
}
173173
}
@@ -199,12 +199,12 @@ private void checkPeerListAndStartIfEligible(
199199

200200
if (correctPeerListBeforeStart != null) {
201201
if (correctPeerListBeforeStart.containsKey(consensusGroupId)) {
202-
// make peers which are in list correct
202+
// Correct peers listed in correctPeerListBeforeStart.
203203
resetPeerListWithoutThrow.accept(
204204
consensusGroupId, correctPeerListBeforeStart.get(consensusGroupId));
205205
consensus.start();
206206
} else {
207-
// clear peers which are not in the list
207+
// Clear peers that are not listed in correctPeerListBeforeStart.
208208
resetPeerListWithoutThrow.accept(consensusGroupId, Collections.emptyList());
209209
}
210210

@@ -344,21 +344,21 @@ public void addRemotePeer(ConsensusGroupId groupId, Peer peer) throws ConsensusE
344344
throw new PeerAlreadyInConsensusGroupException(groupId, peer);
345345
}
346346
try {
347-
// step 1: inactive new Peer to prepare for following steps
347+
// Step 1: deactivate the new peer to prepare for the following steps.
348348
LOGGER.info(IoTConsensusV2Messages.INACTIVATE_NEW_PEER, CLASS_NAME, peer);
349349
impl.setRemotePeerActive(peer, false, false);
350350

351-
// step 2: notify all the other Peers to create consensus pipes to newPeer
352-
// NOTE: For this step, all the other peers will try to transfer its user write data to target
351+
// Step 2: notify all other peers to create consensus pipes to the new peer.
352+
// Each peer will transfer its user-written data to the target peer.
353353
LOGGER.info(IoTConsensusV2Messages.NOTIFY_CREATE_CONSENSUS_PIPES, CLASS_NAME);
354354
impl.notifyPeersToCreateConsensusPipes(peer);
355355
KillPoint.setKillPoint(DataNodeKillPoints.COORDINATOR_ADD_PEER_TRANSITION);
356356

357-
// step 3: wait until all other Peers finish transferring
357+
// Step 3: wait until all other peers finish transferring.
358358
LOGGER.info(IoTConsensusV2Messages.WAIT_PEERS_FINISH_TRANSFER, CLASS_NAME);
359359
impl.waitPeersToTargetPeerTransmissionCompleted(peer);
360360

361-
// step 4: active new Peer to let new Peer receive client requests
361+
// Step 4: activate the new peer so it can receive client requests.
362362
LOGGER.info(IoTConsensusV2Messages.ACTIVATE_NEW_PEER, CLASS_NAME);
363363
impl.setRemotePeerActive(peer, true, false);
364364
KillPoint.setKillPoint(DataNodeKillPoints.COORDINATOR_ADD_PEER_DONE);
@@ -387,23 +387,23 @@ public void removeRemotePeer(ConsensusGroupId groupId, Peer peer) throws Consens
387387
KillPoint.setKillPoint(IoTConsensusRemovePeerCoordinatorKillPoints.INIT);
388388

389389
try {
390-
// let other peers to drop consensus pipes to target
390+
// Let other peers drop consensus pipes to the target.
391391
LOGGER.info(IoTConsensusV2Messages.NOTIFY_DROP_CONSENSUS_PIPES, CLASS_NAME);
392392
impl.notifyPeersToDropConsensusPipe(peer);
393393
KillPoint.setKillPoint(
394394
IoTConsensusRemovePeerCoordinatorKillPoints
395395
.AFTER_NOTIFY_PEERS_TO_REMOVE_REPLICATE_CHANNEL);
396396

397-
// let target peer reject new write
397+
// Let the target peer reject new writes.
398398
LOGGER.info(IoTConsensusV2Messages.INACTIVATE_PEER, CLASS_NAME, peer);
399399
impl.setRemotePeerActive(peer, false, true);
400400
KillPoint.setKillPoint(IoTConsensusRemovePeerCoordinatorKillPoints.AFTER_INACTIVE_PEER);
401401

402-
// wait its consensus pipes to complete
402+
// Wait for its consensus pipes to complete.
403403
LOGGER.info(IoTConsensusV2Messages.WAIT_TARGET_PEER_COMPLETE_TRANSFER, CLASS_NAME, peer);
404404
impl.waitTargetPeerToPeersTransmissionCompleted(peer);
405405

406-
// wait target peer to release all resource
406+
// Wait for the target peer to release all resources.
407407
LOGGER.info(IoTConsensusV2Messages.WAIT_PEER_RELEASE_RESOURCE, CLASS_NAME, peer);
408408
impl.waitReleaseAllRegionRelatedResource(peer);
409409
} catch (ConsensusGroupModifyPeerException e) {

‎iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/IoTConsensusV2ServerImpl.java‎

Lines changed: 10 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -139,7 +139,7 @@ public TSStatus write(IConsensusRequest request) {
139139
try {
140140
long consensusWriteStartTime = System.nanoTime();
141141
long getStateMachineLockTime = System.nanoTime();
142-
// statistic the time of acquiring stateMachine lock
142+
// Record the time spent acquiring the stateMachine lock.
143143
iotConsensusV2ServerMetrics.recordGetStateMachineLockTime(
144144
getStateMachineLockTime - consensusWriteStartTime);
145145
long writeToStateMachineStartTime = System.nanoTime();
@@ -151,7 +151,7 @@ public TSStatus write(IConsensusRequest request) {
151151
long writeToStateMachineEndTime = System.nanoTime();
152152
PERFORMANCE_OVERVIEW_METRICS.recordEngineCost(
153153
writeToStateMachineEndTime - writeToStateMachineStartTime);
154-
// statistic the time of writing request into stateMachine
154+
// Record the time spent writing the request into the stateMachine.
155155
iotConsensusV2ServerMetrics.recordUserWriteStateMachineTime(
156156
writeToStateMachineEndTime - writeToStateMachineStartTime);
157157
return result;
@@ -165,7 +165,7 @@ public TSStatus writeOnFollowerReplica(IConsensusRequest request) {
165165
try {
166166
long consensusWriteStartTime = System.nanoTime();
167167
long getStateMachineLockTime = System.nanoTime();
168-
// statistic the time of acquiring stateMachine lock
168+
// Record the time spent acquiring the stateMachine lock.
169169
iotConsensusV2ServerMetrics.recordGetStateMachineLockTime(
170170
getStateMachineLockTime - consensusWriteStartTime);
171171

@@ -175,7 +175,7 @@ public TSStatus writeOnFollowerReplica(IConsensusRequest request) {
175175

176176
PERFORMANCE_OVERVIEW_METRICS.recordEngineCost(
177177
writeToStateMachineEndTime - writeToStateMachineStartTime);
178-
// statistic the time of writing request into stateMachine
178+
// Record the time spent writing the request into the stateMachine.
179179
iotConsensusV2ServerMetrics.recordReplicaWriteStateMachineTime(
180180
writeToStateMachineEndTime - writeToStateMachineStartTime);
181181
return result;
@@ -210,11 +210,10 @@ public void setRemotePeerActive(Peer peer, boolean isActive, boolean isForDeleti
210210
}
211211
} catch (ClientManagerException e) {
212212
if (isForDeletionPurpose) {
213-
// for remove peer, if target peer is already down, we can skip this step.
213+
// For removePeer, skip this step if the target peer is already down.
214214
LOGGER.warn(IoTConsensusV2Messages.TARGET_PEER_MAY_BE_DOWN, peer, isActive, e);
215215
} else {
216-
// for add peer, if target peer is down, we need to throw exception to identify the failure
217-
// of this addPeerProcedure.
216+
// For addPeer, fail the procedure if the target peer is down.
218217
throw new ConsensusGroupModifyPeerException(e);
219218
}
220219
}
@@ -245,7 +244,7 @@ public void notifyPeersToCreateConsensusPipes(Peer targetPeer)
245244
}
246245

247246
try {
248-
// This node which acts as coordinator will transfer complete historical snapshot to new
247+
// This node acts as the coordinator and transfers the complete historical snapshot to the new
249248
// target.
250249
createConsensusPipeToTargetPeer(targetPeer);
251250
} catch (Exception e) {
@@ -299,7 +298,7 @@ public synchronized void dropConsensusPipeToTargetPeer(Peer targetPeer)
299298
peerManager.removePeer(targetPeer);
300299
}
301300

302-
/** Wait for the user written data up to firstCheck to be replicated */
301+
/** Wait for user-written data up to the first check to be replicated. */
303302
public void waitPeersToTargetPeerTransmissionCompleted(Peer targetPeer)
304303
throws ConsensusGroupModifyPeerException {
305304
boolean isTransmissionCompleted = false;
@@ -337,7 +336,7 @@ public void waitPeersToTargetPeerTransmissionCompleted(Peer targetPeer)
337336
}
338337
}
339338

340-
/** Wait for the user written data up to firstCheck to be replicated */
339+
/** Wait for user-written data up to the first check to be replicated. */
341340
public void waitTargetPeerToPeersTransmissionCompleted(Peer targetPeer)
342341
throws ConsensusGroupModifyPeerException {
343342
boolean isTransmissionCompleted = false;
@@ -435,7 +434,7 @@ public void waitReleaseAllRegionRelatedResource(Peer targetPeer)
435434
Thread.sleep(checkIntervalInMs);
436435
}
437436
} catch (ClientManagerException | TException e) {
438-
// in case of target peer is down or can not serve, we simply skip it.
437+
// Skip this step if the target peer is down or cannot serve requests.
439438
LOGGER.warn(
440439
String.format(
441440
ConsensusMessages.ERROR_WAITING_RELEASE_RESOURCE, targetPeer, e.getMessage()),

0 commit comments

Comments
 (0)