alievmirza commented on code in PR #3633:
URL: https://github.com/apache/ignite-3/pull/3633#discussion_r1576080458
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -493,45 +612,92 @@ public CompletableFuture<Replica> startReplica(
}
}
+ private CompletableFuture<Replica> startReplicaInternal(
+ ReplicationGroupId replicaGrpId,
+ ReplicaListener listener,
+ RaftGroupService raftClient,
+ PendingComparableValuesTracker<Long, Void> storageIndexTracker
+ ) throws NodeStoppingException {
+ LOG.info("Replica is about to start [replicationGroupId={}].",
replicaGrpId);
+
+ ClusterNode localNode = clusterNetSvc.topologyService().localMember();
+
+ Replica newReplica = new Replica(
+ replicaGrpId,
+ listener,
+ storageIndexTracker,
+ localNode,
+ executor,
+ placementDriver,
+ clockService);
+
+ CompletableFuture<Replica> replicaFuture =
replicas.compute(replicaGrpId, (k, existingReplicaFuture) -> {
+ if (existingReplicaFuture == null ||
existingReplicaFuture.isDone()) {
+ assert existingReplicaFuture == null ||
isCompletedSuccessfully(existingReplicaFuture);
+ LOG.info("Replica is started [replicationGroupId={}].",
replicaGrpId);
+
+ return CompletableFuture.completedFuture(newReplica);
+ } else {
+ existingReplicaFuture.complete(newReplica);
+ LOG.info("Replica is started, existing replica waiter was
completed [replicationGroupId={}].", replicaGrpId);
+
+ return existingReplicaFuture;
+ }
+ });
+
+ var eventParams = new LocalReplicaEventParameters(replicaGrpId);
+
+ return fireEvent(AFTER_REPLICA_STARTED, eventParams)
+ .exceptionally(e -> {
+ LOG.error("Error when notifying about
AFTER_REPLICA_STARTED event.", e);
+
+ return null;
+ })
+ .thenCompose(v -> replicaFuture);
+ }
+
/**
* Internal method for starting a replica.
*
* @param replicaGrpId Replication group id.
- * @param listener Replica listener.
- * @param raftClient Topology aware Raft client.
+ * @param newConfiguration TODO
+ * @param createListener TODO
* @param storageIndexTracker Storage index tracker.
*/
private CompletableFuture<Replica> startReplicaInternal(
ReplicationGroupId replicaGrpId,
- ReplicaListener listener,
- TopologyAwareRaftGroupService raftClient,
+ PeersAndLearners newConfiguration,
+ Function<RaftGroupService, ReplicaListener> createListener,
PendingComparableValuesTracker<Long, Void> storageIndexTracker
- ) {
+ ) throws NodeStoppingException {
LOG.info("Replica is about to start [replicationGroupId={}].",
replicaGrpId);
ClusterNode localNode = clusterNetSvc.topologyService().localMember();
- Replica newReplica = new Replica(
- replicaGrpId,
- listener,
- storageIndexTracker,
- raftClient,
- localNode,
- executor,
- placementDriver,
- clockService
- );
+ CompletableFuture<Replica> newReplicaFut = raftManager
+ // TODO IGNITE-19614 This procedure takes 10 seconds if
there's no majority online.
+ .startRaftGroupService(replicaGrpId, newConfiguration,
raftGroupServiceFactory, raftCommandsMarshaller)
+ .thenApply(createListener)
Review Comment:
let's name it so it could reflect the function entity of this param
--
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]
