You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ta...@apache.org on 2022/05/05 09:35:24 UTC
[iotdb] branch addEventApi created (now 0c22d3feb5)
This is an automated email from the ASF dual-hosted git repository.
tanxinyu pushed a change to branch addEventApi
in repository https://gitbox.apache.org/repos/asf/iotdb.git
at 0c22d3feb5 init
This branch includes the following new commits:
new 0c22d3feb5 init
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[iotdb] 01/01: init
Posted by ta...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tanxinyu pushed a commit to branch addEventApi
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 0c22d3feb5c033700044ee5b8fb5a12283eaf3c0
Author: LebronAl <TX...@gmail.com>
AuthorDate: Thu May 5 17:35:07 2022 +0800
init
---
.../org/apache/iotdb/consensus/IStateMachine.java | 41 ++++++++++++++++++-
.../ratis/ApplicationStateMachineProxy.java | 28 ++++++++++++-
.../iotdb/consensus/ratis/RatisConsensus.java | 47 +++++++++++-----------
.../org/apache/iotdb/consensus/ratis/Utils.java | 47 +++++++++++++++-------
.../apache/iotdb/consensus/ratis/SnapshotTest.java | 2 +-
.../apache/iotdb/consensus/ratis/TestUtils.java | 2 +-
.../apache/iotdb/consensus/ratis/UtilsTest.java | 4 +-
7 files changed, 126 insertions(+), 45 deletions(-)
diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/IStateMachine.java b/consensus/src/main/java/org/apache/iotdb/consensus/IStateMachine.java
index e12622f2ad..322c6a55b4 100644
--- a/consensus/src/main/java/org/apache/iotdb/consensus/IStateMachine.java
+++ b/consensus/src/main/java/org/apache/iotdb/consensus/IStateMachine.java
@@ -19,14 +19,17 @@
package org.apache.iotdb.consensus;
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.consensus.ConsensusGroupId;
import org.apache.iotdb.consensus.common.DataSet;
+import org.apache.iotdb.consensus.common.Peer;
import org.apache.iotdb.consensus.common.request.IConsensusRequest;
import javax.annotation.concurrent.ThreadSafe;
import java.io.File;
+import java.util.List;
import java.util.function.Function;
@ThreadSafe
@@ -54,7 +57,7 @@ public interface IStateMachine {
/**
* Take a snapshot of current statemachine. All files are required to be stored under snapshotDir,
- * which is a sub-directory of the StorageDir in Consensus
+ * which is a subdirectory of the StorageDir in Consensus
*
* @param snapshotDir required storage dir
* @return true if snapshot is successfully taken
@@ -67,4 +70,40 @@ public interface IStateMachine {
* @param latestSnapshotRootDir dir where the latest snapshot sits
*/
void loadSnapshot(File latestSnapshotRootDir);
+
+ /** An optional API for event notifications. */
+ interface EventApi {
+ /**
+ * Notify the {@link IStateMachine} that a new leader has been elected. Note that the new leader
+ * can possibly be this server.
+ *
+ * @param groupId The id of this consensus group.
+ * @param newLeader The id of the new leader.
+ */
+ default void notifyLeaderChanged(ConsensusGroupId groupId, TEndPoint newLeader) {}
+
+ /**
+ * Notify the {@link IStateMachine} a configuration change. This method will be invoked when a
+ * newConfiguration is processed.
+ *
+ * @param term term of the current log entry
+ * @param index index which is being updated
+ * @param newRaftConfiguration new configuration
+ */
+ default void notifyConfigurationChanged(
+ long term, long index, List<Peer> newRaftConfiguration) {}
+ }
+
+ /**
+ * Get the {@link IStateMachine.EventApi} object.
+ *
+ * <p>If this {@link IStateMachine} chooses to support the optional {@link
+ * IStateMachine.EventApi}, it may either implement {@link IStateMachine.EventApi} directly or
+ * override this method to return an {@link IStateMachine.EventApi} object.
+ *
+ * @return The {@link IStateMachine.EventApi} object.
+ */
+ default IStateMachine.EventApi event() {
+ return (IStateMachine.EventApi) this;
+ }
}
diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/ratis/ApplicationStateMachineProxy.java b/consensus/src/main/java/org/apache/iotdb/consensus/ratis/ApplicationStateMachineProxy.java
index 123bd4f13d..3a53925b7e 100644
--- a/consensus/src/main/java/org/apache/iotdb/consensus/ratis/ApplicationStateMachineProxy.java
+++ b/consensus/src/main/java/org/apache/iotdb/consensus/ratis/ApplicationStateMachineProxy.java
@@ -25,8 +25,11 @@ import org.apache.iotdb.consensus.common.request.ByteBufferConsensusRequest;
import org.apache.iotdb.consensus.common.request.IConsensusRequest;
import org.apache.ratis.proto.RaftProtos;
+import org.apache.ratis.proto.RaftProtos.RaftConfigurationProto;
import org.apache.ratis.protocol.Message;
import org.apache.ratis.protocol.RaftGroupId;
+import org.apache.ratis.protocol.RaftGroupMemberId;
+import org.apache.ratis.protocol.RaftPeerId;
import org.apache.ratis.server.RaftServer;
import org.apache.ratis.server.protocol.TermIndex;
import org.apache.ratis.server.raftlog.RaftLog;
@@ -49,11 +52,13 @@ public class ApplicationStateMachineProxy extends BaseStateMachine {
// Raft Storage sub dir for statemachine data, default (_sm)
private File statemachineDir;
private final SnapshotStorage snapshotStorage;
+ private final RaftGroupId groupId;
- public ApplicationStateMachineProxy(IStateMachine stateMachine) {
+ public ApplicationStateMachineProxy(IStateMachine stateMachine, RaftGroupId id) {
applicationStateMachine = stateMachine;
snapshotStorage = new SnapshotStorage(applicationStateMachine);
applicationStateMachine.start();
+ groupId = id;
}
@Override
@@ -69,7 +74,7 @@ public class ApplicationStateMachineProxy extends BaseStateMachine {
}
@Override
- public void reinitialize() throws IOException {
+ public void reinitialize() {
setLastAppliedTermIndex(null);
loadSnapshot(snapshotStorage.findLatestSnapshotDir());
if (getLifeCycleState() == LifeCycle.State.PAUSED) {
@@ -159,4 +164,23 @@ public class ApplicationStateMachineProxy extends BaseStateMachine {
public StateMachineStorage getStateMachineStorage() {
return snapshotStorage;
}
+
+ public void notifyLeaderChanged(RaftGroupMemberId groupMemberId, RaftPeerId newLeaderId) {
+ applicationStateMachine
+ .event()
+ .notifyLeaderChanged(
+ Utils.fromRaftGroupIdToConsensusGroupId(groupMemberId.getGroupId()),
+ Utils.formRaftPeerIdToTEndPoint(newLeaderId));
+ }
+
+ public void notifyConfigurationChanged(
+ long term, long index, RaftConfigurationProto newRaftConfiguration) {
+ applicationStateMachine
+ .event()
+ .notifyConfigurationChanged(
+ term,
+ index,
+ Utils.fromRaftProtoListAndRaftGroupIdToPeers(
+ newRaftConfiguration.getPeersList(), groupId));
+ }
}
diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java b/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java
index 96b3b766e3..aff5378afa 100644
--- a/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java
+++ b/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java
@@ -109,7 +109,7 @@ class RatisConsensus implements IConsensus {
public RatisConsensus(TEndPoint endpoint, File ratisStorageDir, IStateMachine.Registry registry)
throws IOException {
String address = Utils.IPAddress(endpoint);
- myself = Utils.toRaftPeer(endpoint, DEFAULT_PRIORITY);
+ myself = Utils.fromTEndPointAndPriorityToRaftPeer(endpoint, DEFAULT_PRIORITY);
RaftServerConfigKeys.setStorageDir(properties, Collections.singletonList(ratisStorageDir));
RaftServerConfigKeys.Snapshot.setAutoTriggerEnabled(properties, true);
@@ -126,7 +126,8 @@ class RatisConsensus implements IConsensus {
.setStateMachineRegistry(
raftGroupId ->
new ApplicationStateMachineProxy(
- registry.apply(Utils.toConsensusGroupId(raftGroupId))))
+ registry.apply(Utils.fromRaftGroupIdToConsensusGroupId(raftGroupId)),
+ raftGroupId))
.build();
}
@@ -150,7 +151,7 @@ class RatisConsensus implements IConsensus {
ConsensusGroupId groupId, IConsensusRequest IConsensusRequest) {
// pre-condition: group exists and myself server serves this group
- RaftGroupId raftGroupId = Utils.toRatisGroupId(groupId);
+ RaftGroupId raftGroupId = Utils.fromConsensusGroupIdToRaftGroupId(groupId);
RaftGroup raftGroup = getGroupInfo(raftGroupId);
if (raftGroup == null || !raftGroup.getPeers().contains(myself)) {
return failedWrite(new ConsensusGroupNotExistException(groupId));
@@ -198,7 +199,7 @@ class RatisConsensus implements IConsensus {
}
if (suggestedLeader != null) {
- TEndPoint leaderEndPoint = Utils.getEndpoint(suggestedLeader);
+ TEndPoint leaderEndPoint = Utils.formRaftPeerIdToTEndPoint(suggestedLeader.getId());
writeResult.setRedirectNode(new TEndPoint(leaderEndPoint.getIp(), leaderEndPoint.getPort()));
}
@@ -209,7 +210,7 @@ class RatisConsensus implements IConsensus {
@Override
public ConsensusReadResponse read(ConsensusGroupId groupId, IConsensusRequest IConsensusRequest) {
- RaftGroup group = getGroupInfo(Utils.toRatisGroupId(groupId));
+ RaftGroup group = getGroupInfo(Utils.fromConsensusGroupIdToRaftGroupId(groupId));
if (group == null || !group.getPeers().contains(myself)) {
return failedRead(new ConsensusGroupNotExistException(groupId));
}
@@ -278,7 +279,7 @@ class RatisConsensus implements IConsensus {
*/
@Override
public ConsensusGenericResponse removeConsensusGroup(ConsensusGroupId groupId) {
- RaftGroupId raftGroupId = Utils.toRatisGroupId(groupId);
+ RaftGroupId raftGroupId = Utils.fromConsensusGroupIdToRaftGroupId(groupId);
RaftGroup raftGroup = getGroupInfo(raftGroupId);
// pre-conditions: group exists and myself in this group
@@ -318,9 +319,9 @@ class RatisConsensus implements IConsensus {
*/
@Override
public ConsensusGenericResponse addPeer(ConsensusGroupId groupId, Peer peer) {
- RaftGroupId raftGroupId = Utils.toRatisGroupId(groupId);
+ RaftGroupId raftGroupId = Utils.fromConsensusGroupIdToRaftGroupId(groupId);
RaftGroup group = getGroupInfo(raftGroupId);
- RaftPeer peerToAdd = Utils.toRaftPeer(peer, DEFAULT_PRIORITY);
+ RaftPeer peerToAdd = Utils.fromTEndPointAndPriorityToRaftPeer(peer, DEFAULT_PRIORITY);
// pre-conditions: group exists and myself in this group
if (group == null || !group.getPeers().contains(myself)) {
@@ -353,9 +354,9 @@ class RatisConsensus implements IConsensus {
*/
@Override
public ConsensusGenericResponse removePeer(ConsensusGroupId groupId, Peer peer) {
- RaftGroupId raftGroupId = Utils.toRatisGroupId(groupId);
+ RaftGroupId raftGroupId = Utils.fromConsensusGroupIdToRaftGroupId(groupId);
RaftGroup group = getGroupInfo(raftGroupId);
- RaftPeer peerToRemove = Utils.toRaftPeer(peer, DEFAULT_PRIORITY);
+ RaftPeer peerToRemove = Utils.fromTEndPointAndPriorityToRaftPeer(peer, DEFAULT_PRIORITY);
// pre-conditions: group exists and myself in this group
if (group == null || !group.getPeers().contains(myself)) {
@@ -413,14 +414,14 @@ class RatisConsensus implements IConsensus {
// first fetch the newest information
- RaftGroupId raftGroupId = Utils.toRatisGroupId(groupId);
+ RaftGroupId raftGroupId = Utils.fromConsensusGroupIdToRaftGroupId(groupId);
RaftGroup raftGroup = getGroupInfo(raftGroupId);
if (raftGroup == null) {
return failed(new ConsensusGroupNotExistException(groupId));
}
- RaftPeer newRaftLeader = Utils.toRaftPeer(newLeader, LEADER_PRIORITY);
+ RaftPeer newRaftLeader = Utils.fromTEndPointAndPriorityToRaftPeer(newLeader, LEADER_PRIORITY);
ArrayList<RaftPeer> newConfiguration = new ArrayList<>();
for (RaftPeer raftPeer : raftGroup.getPeers()) {
@@ -428,7 +429,9 @@ class RatisConsensus implements IConsensus {
newConfiguration.add(newRaftLeader);
} else {
// degrade every other peer to default priority
- newConfiguration.add(Utils.toRaftPeer(Utils.getEndpoint(raftPeer), DEFAULT_PRIORITY));
+ newConfiguration.add(
+ Utils.fromTEndPointAndPriorityToRaftPeer(
+ Utils.formRaftPeerIdToTEndPoint(raftPeer.getId()), DEFAULT_PRIORITY));
}
}
@@ -459,7 +462,7 @@ class RatisConsensus implements IConsensus {
@Override
public boolean isLeader(ConsensusGroupId groupId) {
- RaftGroupId raftGroupId = Utils.toRatisGroupId(groupId);
+ RaftGroupId raftGroupId = Utils.fromConsensusGroupIdToRaftGroupId(groupId);
boolean isLeader;
try {
@@ -475,10 +478,10 @@ class RatisConsensus implements IConsensus {
@Override
public Peer getLeader(ConsensusGroupId groupId) {
if (isLeader(groupId)) {
- return new Peer(groupId, Utils.parseFromRatisId(myself.getId().toString()));
+ return new Peer(groupId, Utils.formRaftPeerIdToTEndPoint(myself.getId()));
}
- RaftGroupId raftGroupId = Utils.toRatisGroupId(groupId);
+ RaftGroupId raftGroupId = Utils.fromConsensusGroupIdToRaftGroupId(groupId);
RaftClient client;
try {
client = server.getDivision(raftGroupId).getRaftClient();
@@ -486,7 +489,7 @@ class RatisConsensus implements IConsensus {
logger.warn("cannot find raft client for group " + groupId);
return null;
}
- TEndPoint leaderEndpoint = Utils.parseFromRatisId(client.getLeaderId().toString());
+ TEndPoint leaderEndpoint = Utils.formRaftPeerIdToTEndPoint(client.getLeaderId());
return new Peer(groupId, leaderEndpoint);
}
@@ -513,7 +516,7 @@ class RatisConsensus implements IConsensus {
.setServerId(server.getId())
.setClientId(localFakeId)
.setCallId(localFakeCallId.incrementAndGet())
- .setGroupId(Utils.toRatisGroupId(groupId))
+ .setGroupId(Utils.fromConsensusGroupIdToRaftGroupId(groupId))
.setType(type)
.setMessage(message)
.build();
@@ -536,11 +539,9 @@ class RatisConsensus implements IConsensus {
}
private RaftGroup buildRaftGroup(ConsensusGroupId groupId, List<Peer> peers) {
- List<RaftPeer> raftPeers =
- peers.stream()
- .map(peer -> Utils.toRaftPeer(peer, DEFAULT_PRIORITY))
- .collect(Collectors.toList());
- return RaftGroup.valueOf(Utils.toRatisGroupId(groupId), raftPeers);
+ return RaftGroup.valueOf(
+ Utils.fromConsensusGroupIdToRaftGroupId(groupId),
+ Utils.fromPeersAndPriorityToRaftPeers(peers, DEFAULT_PRIORITY));
}
private RatisClient getRaftClient(RaftGroup group) throws IOException {
diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/ratis/Utils.java b/consensus/src/main/java/org/apache/iotdb/consensus/ratis/Utils.java
index 0c3e5d984f..e1773c7f80 100644
--- a/consensus/src/main/java/org/apache/iotdb/consensus/ratis/Utils.java
+++ b/consensus/src/main/java/org/apache/iotdb/consensus/ratis/Utils.java
@@ -23,8 +23,10 @@ import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.consensus.ConsensusGroupId;
import org.apache.iotdb.consensus.common.Peer;
+import org.apache.ratis.proto.RaftProtos.RaftPeerProto;
import org.apache.ratis.protocol.RaftGroupId;
import org.apache.ratis.protocol.RaftPeer;
+import org.apache.ratis.protocol.RaftPeerId;
import org.apache.ratis.server.protocol.TermIndex;
import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
import org.apache.thrift.TException;
@@ -33,6 +35,8 @@ import org.apache.thrift.transport.TByteBuffer;
import java.io.File;
import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.stream.Collectors;
public class Utils {
private static final int tempBufferSize = 1024;
@@ -42,7 +46,7 @@ public class Utils {
return String.format("%s:%d", endpoint.getIp(), endpoint.getPort());
}
- /** Encode the ConsensusGroupId into 6 bytes 2 Bytes for Group Type 4 Bytes for Group ID */
+ /** Encode the ConsensusGroupId into 6 bytes: 2 Bytes for Group Type and 4 Bytes for Group ID */
public static long groupEncode(ConsensusGroupId consensusGroupId) {
// use abbreviations to prevent overflow
long groupType = consensusGroupId.getType().getValue();
@@ -51,36 +55,49 @@ public class Utils {
return groupCode;
}
- public static String RatisPeerId(TEndPoint endpoint) {
- return String.format("%s-%d", endpoint.getIp(), endpoint.getPort());
+ public static RaftPeerId fromTEndPointToRaftPeerId(TEndPoint endpoint) {
+ return RaftPeerId.valueOf(String.format("%s-%d", endpoint.getIp(), endpoint.getPort()));
}
- public static TEndPoint parseFromRatisId(String ratisId) {
- String[] items = ratisId.split("-");
+ public static TEndPoint formRaftPeerIdToTEndPoint(RaftPeerId id) {
+ String[] items = id.toString().split("-");
+ return new TEndPoint(items[0], Integer.parseInt(items[1]));
+ }
+
+ public static TEndPoint formRaftPeerProtoToTEndPoint(RaftPeerProto proto) {
+ String[] items = proto.getAddress().split(":");
return new TEndPoint(items[0], Integer.parseInt(items[1]));
}
// priority is used as ordinal of leader election
- public static RaftPeer toRaftPeer(TEndPoint endpoint, int priority) {
+ public static RaftPeer fromTEndPointAndPriorityToRaftPeer(TEndPoint endpoint, int priority) {
return RaftPeer.newBuilder()
- .setId(RatisPeerId(endpoint))
+ .setId(fromTEndPointToRaftPeerId(endpoint))
.setAddress(IPAddress(endpoint))
.setPriority(priority)
.build();
}
- public static RaftPeer toRaftPeer(Peer peer, int priority) {
- return toRaftPeer(peer.getEndpoint(), priority);
+ public static RaftPeer fromTEndPointAndPriorityToRaftPeer(Peer peer, int priority) {
+ return fromTEndPointAndPriorityToRaftPeer(peer.getEndpoint(), priority);
+ }
+
+ public static List<RaftPeer> fromPeersAndPriorityToRaftPeers(List<Peer> peers, int priority) {
+ return peers.stream()
+ .map(peer -> Utils.fromTEndPointAndPriorityToRaftPeer(peer, priority))
+ .collect(Collectors.toList());
}
- public static TEndPoint getEndpoint(RaftPeer raftPeer) {
- String address = raftPeer.getAddress(); // ip:port
- String[] split = address.split(":");
- return new TEndPoint(split[0], Integer.parseInt(split[1]));
+ public static List<Peer> fromRaftProtoListAndRaftGroupIdToPeers(
+ List<RaftPeerProto> raftProtoList, RaftGroupId id) {
+ ConsensusGroupId consensusGroupId = Utils.fromRaftGroupIdToConsensusGroupId(id);
+ return raftProtoList.stream()
+ .map(peer -> new Peer(consensusGroupId, Utils.formRaftPeerProtoToTEndPoint(peer)))
+ .collect(Collectors.toList());
}
/** Given ConsensusGroupId, generate a deterministic RaftGroupId current scheme: */
- public static RaftGroupId toRatisGroupId(ConsensusGroupId consensusGroupId) {
+ public static RaftGroupId fromConsensusGroupIdToRaftGroupId(ConsensusGroupId consensusGroupId) {
long groupCode = groupEncode(consensusGroupId);
byte[] bGroupCode = ByteBuffer.allocate(Long.BYTES).putLong(groupCode).array();
byte[] bPaddedGroupName = new byte[16];
@@ -93,7 +110,7 @@ public class Utils {
}
/** Given raftGroupId, decrypt ConsensusGroupId out of it */
- public static ConsensusGroupId toConsensusGroupId(RaftGroupId raftGroupId) {
+ public static ConsensusGroupId fromRaftGroupIdToConsensusGroupId(RaftGroupId raftGroupId) {
byte[] padded = raftGroupId.toByteString().toByteArray();
long type = (padded[10] << 8) + padded[11];
ByteBuffer byteBuffer = ByteBuffer.allocate(Integer.BYTES);
diff --git a/consensus/src/test/java/org/apache/iotdb/consensus/ratis/SnapshotTest.java b/consensus/src/test/java/org/apache/iotdb/consensus/ratis/SnapshotTest.java
index 3af8c5df5e..e42f981706 100644
--- a/consensus/src/test/java/org/apache/iotdb/consensus/ratis/SnapshotTest.java
+++ b/consensus/src/test/java/org/apache/iotdb/consensus/ratis/SnapshotTest.java
@@ -87,7 +87,7 @@ public class SnapshotTest {
@Test
public void testSnapshot() throws Exception {
ApplicationStateMachineProxy proxy =
- new ApplicationStateMachineProxy(new TestUtils.IntegerCounter());
+ new ApplicationStateMachineProxy(new TestUtils.IntegerCounter(), null);
proxy.initialize(null, null, new EmptyStorageWithOnlySMDir());
diff --git a/consensus/src/test/java/org/apache/iotdb/consensus/ratis/TestUtils.java b/consensus/src/test/java/org/apache/iotdb/consensus/ratis/TestUtils.java
index eca57c4a19..d6b378af97 100644
--- a/consensus/src/test/java/org/apache/iotdb/consensus/ratis/TestUtils.java
+++ b/consensus/src/test/java/org/apache/iotdb/consensus/ratis/TestUtils.java
@@ -61,7 +61,7 @@ public class TestUtils {
}
}
- static class IntegerCounter implements IStateMachine {
+ static class IntegerCounter implements IStateMachine, IStateMachine.EventApi {
private AtomicInteger integer;
private final Logger logger = LoggerFactory.getLogger(IntegerCounter.class);
diff --git a/consensus/src/test/java/org/apache/iotdb/consensus/ratis/UtilsTest.java b/consensus/src/test/java/org/apache/iotdb/consensus/ratis/UtilsTest.java
index 9d11b7967e..794c23289a 100644
--- a/consensus/src/test/java/org/apache/iotdb/consensus/ratis/UtilsTest.java
+++ b/consensus/src/test/java/org/apache/iotdb/consensus/ratis/UtilsTest.java
@@ -29,8 +29,8 @@ public class UtilsTest {
@Test
public void testEncryption() {
ConsensusGroupId raw = new PartitionRegionId(100);
- RaftGroupId id = Utils.toRatisGroupId(raw);
- ConsensusGroupId cgid = Utils.toConsensusGroupId(id);
+ RaftGroupId id = Utils.fromConsensusGroupIdToRaftGroupId(raw);
+ ConsensusGroupId cgid = Utils.fromRaftGroupIdToConsensusGroupId(id);
Assert.assertEquals(raw.getId(), cgid.getId());
Assert.assertEquals(raw.getType(), cgid.getType());
}