ibessonov commented on code in PR #2239:
URL: https://github.com/apache/ignite-3/pull/2239#discussion_r1244913159
##########
modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/impl/MetaStorageManagerImpl.java:
##########
@@ -137,66 +144,123 @@ public MetaStorageManagerImpl(
LogicalTopologyService logicalTopologyService,
RaftManager raftMgr,
KeyValueStorage storage,
- HybridClock clock
+ HybridClock clock,
+ TopologyAwareRaftGroupServiceFactory
topologyAwareRaftGroupServiceFactory
) {
this.vaultMgr = vaultMgr;
this.clusterService = clusterService;
this.raftMgr = raftMgr;
this.cmgMgr = cmgMgr;
this.logicalTopologyService = logicalTopologyService;
this.storage = storage;
- this.clusterTime = new ClusterTimeImpl(busyLock, clock);
+ this.clusterTime = new ClusterTimeImpl(clusterService.nodeName(),
busyLock, clock);
+ this.topologyAwareRaftGroupServiceFactory =
topologyAwareRaftGroupServiceFactory;
}
- private CompletableFuture<MetaStorageServiceImpl>
initializeMetaStorage(Set<String> metaStorageNodes) {
- String thisNodeName = clusterService.nodeName();
+ /**
+ * Constructor for tests, that allows to pass Meta Storage configuration.
+ */
+ @TestOnly
+ public MetaStorageManagerImpl(
+ VaultManager vaultMgr,
+ ClusterService clusterService,
+ ClusterManagementGroupManager cmgMgr,
+ LogicalTopologyService logicalTopologyService,
+ RaftManager raftMgr,
+ KeyValueStorage storage,
+ HybridClock clock,
+ TopologyAwareRaftGroupServiceFactory
topologyAwareRaftGroupServiceFactory,
+ MetaStorageConfiguration configuration
+ ) {
+ this(vaultMgr, clusterService, cmgMgr, logicalTopologyService,
raftMgr, storage, clock, topologyAwareRaftGroupServiceFactory);
+
+ configure(configuration);
+ }
- CompletableFuture<RaftGroupService> raftServiceFuture;
+ private CompletableFuture<MetaStorageServiceImpl> initializeMetaStorage(
+ Set<String> metaStorageNodes, MetaStorageConfiguration
metaStorageConfig
+ ) {
+ assert metaStorageConfig != null : "Meta Storage configuration has not
been set";
try {
- var ownFsmCallerExecutorDisruptorConfig = new
RaftNodeDisruptorConfiguration("metastorage", 1);
+ String thisNodeName = clusterService.nodeName();
- // We need to configure the replication protocol differently
whether this node is a synchronous or asynchronous replica.
- if (metaStorageNodes.contains(thisNodeName)) {
- PeersAndLearners configuration =
PeersAndLearners.fromConsistentIds(metaStorageNodes);
+ var disruptorConfig = new
RaftNodeDisruptorConfiguration("metastorage", 1);
- Peer localPeer = configuration.peer(thisNodeName);
+ CompletableFuture<? extends RaftGroupService> raftServiceFuture =
metaStorageNodes.contains(thisNodeName)
+ ? startFollowerNode(metaStorageNodes, disruptorConfig,
metaStorageConfig)
+ : startLearnerNode(metaStorageNodes, disruptorConfig);
- assert localPeer != null;
+ return raftServiceFuture.thenApply(raftService -> new
MetaStorageServiceImpl(thisNodeName, raftService, busyLock, clusterTime));
+ } catch (NodeStoppingException e) {
+ return CompletableFuture.failedFuture(e);
+ }
+ }
- raftServiceFuture =
raftMgr.startRaftGroupNodeAndWaitNodeReadyFuture(
- new RaftNodeId(MetastorageGroupId.INSTANCE, localPeer),
- configuration,
- new MetaStorageListener(storage, clusterTime),
- new MetaStorageRaftGroupEventsListener(
- busyLock,
- clusterService,
- logicalTopologyService,
- metaStorageSvcFut,
- clusterTime
- ),
- ownFsmCallerExecutorDisruptorConfig
- );
- } else {
- PeersAndLearners configuration =
PeersAndLearners.fromConsistentIds(metaStorageNodes, Set.of(thisNodeName));
+ private CompletableFuture<? extends RaftGroupService> startFollowerNode(
+ Set<String> metaStorageNodes, RaftNodeDisruptorConfiguration
disruptorConfig, MetaStorageConfiguration metaStorageConfig
+ ) throws NodeStoppingException {
+ String thisNodeName = clusterService.nodeName();
- Peer localPeer = configuration.learner(thisNodeName);
+ PeersAndLearners configuration =
PeersAndLearners.fromConsistentIds(metaStorageNodes);
- assert localPeer != null;
+ Peer localPeer = configuration.peer(thisNodeName);
- raftServiceFuture =
raftMgr.startRaftGroupNodeAndWaitNodeReadyFuture(
+ assert localPeer != null;
+
+ CompletableFuture<TopologyAwareRaftGroupService> raftServiceFuture =
+ raftMgr.startRaftGroupNodeAndWaitNodeReadyFuture(
new RaftNodeId(MetastorageGroupId.INSTANCE, localPeer),
configuration,
new MetaStorageListener(storage, clusterTime),
RaftGroupEventsListener.noopLsnr,
- ownFsmCallerExecutorDisruptorConfig
+ disruptorConfig,
+ topologyAwareRaftGroupServiceFactory
);
- }
- } catch (NodeStoppingException e) {
- return CompletableFuture.failedFuture(e);
- }
- return raftServiceFuture.thenApply(raftService -> new
MetaStorageServiceImpl(thisNodeName, raftService, busyLock, clusterTime));
+ raftServiceFuture
+ .thenAccept(service -> service.subscribeLeader(new
MetaStorageLeaderElectionListener(
+ busyLock,
+ clusterService,
+ logicalTopologyService,
+ metaStorageSvcFut,
+ clusterTime,
+ metaStorageConfig
+ )));
+
+ return raftServiceFuture;
+ }
+
+ private CompletableFuture<? extends RaftGroupService> startLearnerNode(
+ Set<String> metaStorageNodes, RaftNodeDisruptorConfiguration
disruptorConfig
+ ) throws NodeStoppingException {
+ String thisNodeName = clusterService.nodeName();
+
+ PeersAndLearners configuration =
PeersAndLearners.fromConsistentIds(metaStorageNodes, Set.of(thisNodeName));
Review Comment:
The fact that we pass a single node as a learner in configuration
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
