You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by xi...@apache.org on 2022/10/25 13:19:19 UTC
[iotdb] 01/01: add logic to wail log sync to be completed when some peer is going to be removed
This is an automated email from the ASF dual-hosted git repository.
xingtanzjr pushed a commit to branch xingtanzjr/migration_add_wait_sync
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 133f22ef276146cd9f4edd4e22cc48a51d930268
Author: Jinrui.Zhang <xi...@gmail.com>
AuthorDate: Tue Oct 25 21:19:03 2022 +0800
add logic to wail log sync to be completed when some peer is going to be removed
---
.../multileader/MultiLeaderConsensus.java | 12 ++++++++
.../multileader/MultiLeaderServerImpl.java | 36 ++++++++++++++++++++++
.../service/MultiLeaderRPCServiceProcessor.java | 15 +++++++++
.../src/main/thrift/mutlileader.thrift | 11 +++++++
4 files changed, 74 insertions(+)
diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/MultiLeaderConsensus.java b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/MultiLeaderConsensus.java
index fc8039c979..10fd222713 100644
--- a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/MultiLeaderConsensus.java
+++ b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/MultiLeaderConsensus.java
@@ -298,6 +298,7 @@ public class MultiLeaderConsensus implements IConsensus {
.build();
}
try {
+ // let other peers remove the sync channel with target peer
impl.notifyPeersToRemoveSyncLogChannel(peer);
} catch (ConsensusGroupAddPeerException e) {
return ConsensusGenericResponse.newBuilder()
@@ -305,6 +306,17 @@ public class MultiLeaderConsensus implements IConsensus {
.setException(new ConsensusException(e.getMessage()))
.build();
}
+
+ try {
+ // let target peer reject new write
+ impl.inactivePeer(peer);
+ // wait its SyncLog to complete
+ impl.waitTargetPeerUntilSyncLogCompleted(peer);
+ } catch (ConsensusGroupAddPeerException e) {
+ // we only log warning here because sometimes the target peer may already be down
+ logger.warn("cannot wait {} to complete SyncLog. error message: {}", peer, e.getMessage());
+ }
+
return ConsensusGenericResponse.newBuilder().setSuccess(true).build();
}
diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/MultiLeaderServerImpl.java b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/MultiLeaderServerImpl.java
index aaeb63d788..5e2cb5fb7e 100644
--- a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/MultiLeaderServerImpl.java
+++ b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/MultiLeaderServerImpl.java
@@ -45,6 +45,8 @@ import org.apache.iotdb.consensus.multileader.thrift.TSendSnapshotFragmentReq;
import org.apache.iotdb.consensus.multileader.thrift.TSendSnapshotFragmentRes;
import org.apache.iotdb.consensus.multileader.thrift.TTriggerSnapshotLoadReq;
import org.apache.iotdb.consensus.multileader.thrift.TTriggerSnapshotLoadRes;
+import org.apache.iotdb.consensus.multileader.thrift.TWaitSyncLogCompleteReq;
+import org.apache.iotdb.consensus.multileader.thrift.TWaitSyncLogCompleteRes;
import org.apache.iotdb.consensus.multileader.wal.ConsensusReqReader;
import org.apache.iotdb.consensus.multileader.wal.GetConsensusReqReaderPlan;
import org.apache.iotdb.rpc.RpcUtils;
@@ -412,6 +414,40 @@ public class MultiLeaderServerImpl {
}
}
+ public void waitTargetPeerUntilSyncLogCompleted(Peer targetPeer)
+ throws ConsensusGroupAddPeerException {
+ long checkIntervalInMs = 10_000L;
+ try (SyncMultiLeaderServiceClient client =
+ syncClientManager.borrowClient(targetPeer.getEndpoint())) {
+ while (true) {
+ TWaitSyncLogCompleteRes res =
+ client.waitSyncLogComplete(
+ new TWaitSyncLogCompleteReq(targetPeer.getGroupId().convertToTConsensusGroupId()));
+ if (res.complete) {
+ return;
+ }
+ logger.info(
+ "{} SyncLog is still in progress. TargetIndex: {}, CurrentSyncIndex: {}",
+ targetPeer,
+ res.searchIndex,
+ res.safeIndex);
+ Thread.sleep(checkIntervalInMs);
+ }
+ } catch (IOException | TException e) {
+ throw new ConsensusGroupAddPeerException(
+ String.format(
+ "error when waiting %s to complete SyncLog. %s", targetPeer, e.getMessage()),
+ e);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new ConsensusGroupAddPeerException(
+ String.format(
+ "thread interrupted when waiting %s to complete SyncLog. %s",
+ targetPeer, e.getMessage()),
+ e);
+ }
+ }
+
private boolean isSuccess(TSStatus status) {
return status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode();
}
diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/service/MultiLeaderRPCServiceProcessor.java b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/service/MultiLeaderRPCServiceProcessor.java
index 3f045e5e66..20b6aad421 100644
--- a/consensus/src/main/java/org/apache/iotdb/consensus/multileader/service/MultiLeaderRPCServiceProcessor.java
+++ b/consensus/src/main/java/org/apache/iotdb/consensus/multileader/service/MultiLeaderRPCServiceProcessor.java
@@ -45,6 +45,8 @@ import org.apache.iotdb.consensus.multileader.thrift.TSyncLogReq;
import org.apache.iotdb.consensus.multileader.thrift.TSyncLogRes;
import org.apache.iotdb.consensus.multileader.thrift.TTriggerSnapshotLoadReq;
import org.apache.iotdb.consensus.multileader.thrift.TTriggerSnapshotLoadRes;
+import org.apache.iotdb.consensus.multileader.thrift.TWaitSyncLogCompleteReq;
+import org.apache.iotdb.consensus.multileader.thrift.TWaitSyncLogCompleteRes;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.thrift.TException;
@@ -229,6 +231,19 @@ public class MultiLeaderRPCServiceProcessor implements MultiLeaderConsensusIServ
resultHandler.onComplete(new TRemoveSyncLogChannelRes(responseStatus));
}
+ @Override
+ public void waitSyncLogComplete(
+ TWaitSyncLogCompleteReq req, AsyncMethodCallback<TWaitSyncLogCompleteRes> resultHandler)
+ throws TException {
+ ConsensusGroupId groupId =
+ ConsensusGroupId.Factory.createFromTConsensusGroupId(req.getConsensusGroupId());
+ MultiLeaderServerImpl impl = consensus.getImpl(groupId);
+ long searchIndex = impl.getIndex();
+ long safeIndex = impl.getCurrentSafelyDeletedSearchIndex();
+ resultHandler.onComplete(
+ new TWaitSyncLogCompleteRes(searchIndex == safeIndex, searchIndex, safeIndex));
+ }
+
@Override
public void sendSnapshotFragment(
TSendSnapshotFragmentReq req, AsyncMethodCallback<TSendSnapshotFragmentRes> resultHandler)
diff --git a/thrift-multi-leader-consensus/src/main/thrift/mutlileader.thrift b/thrift-multi-leader-consensus/src/main/thrift/mutlileader.thrift
index 345fc70900..81252ce353 100644
--- a/thrift-multi-leader-consensus/src/main/thrift/mutlileader.thrift
+++ b/thrift-multi-leader-consensus/src/main/thrift/mutlileader.thrift
@@ -81,6 +81,16 @@ struct TSendSnapshotFragmentReq {
5: required binary fileChunk
}
+struct TWaitSyncLogCompleteReq {
+ 1: required common.TConsensusGroupId consensusGroupId
+}
+
+struct TWaitSyncLogCompleteRes {
+ 1: required bool complete
+ 2: required i64 searchIndex
+ 3: required i64 safeIndex
+}
+
struct TSendSnapshotFragmentRes {
1: required common.TSStatus status
}
@@ -100,6 +110,7 @@ service MultiLeaderConsensusIService {
TActivatePeerRes activatePeer(TActivatePeerReq req)
TBuildSyncLogChannelRes buildSyncLogChannel(TBuildSyncLogChannelReq req)
TRemoveSyncLogChannelRes removeSyncLogChannel(TRemoveSyncLogChannelReq req)
+ TWaitSyncLogCompleteRes waitSyncLogComplete(TWaitSyncLogCompleteReq req)
TSendSnapshotFragmentRes sendSnapshotFragment(TSendSnapshotFragmentReq req)
TTriggerSnapshotLoadRes triggerSnapshotLoad(TTriggerSnapshotLoadReq req)
}
\ No newline at end of file