JAkutenshi commented on code in PR #3633:
URL: https://github.com/apache/ignite-3/pull/3633#discussion_r1594484506
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -497,41 +565,76 @@ public CompletableFuture<Replica> startReplica(
* 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(
+ // TODO: nonsense name
+ boolean shouldSkipReplicaStarting,
ReplicationGroupId replicaGrpId,
- ReplicaListener listener,
- TopologyAwareRaftGroupService raftClient,
+ PeersAndLearners newConfiguration,
+ Supplier<RaftGroupService> raftClientCache,
+ Function<RaftGroupService, ReplicaListener> createListener,
PendingComparableValuesTracker<Long, Void> storageIndexTracker
- ) {
+ ) throws NodeStoppingException {
LOG.info("Replica is about to start [replicationGroupId={}].",
replicaGrpId);
+ var raftClient = raftClientCache.get();
+ CompletableFuture<TopologyAwareRaftGroupService> newRaftClientFut;
+ if (raftClient == null) {
+ newRaftClientFut = createRaftClientAsync(replicaGrpId,
newConfiguration);
+ } else {
+ newRaftClientFut =
CompletableFuture.completedFuture((TopologyAwareRaftGroupService) raftClient);
+ }
+
+ // TODO: should be there for now because in TableManager:L978-990
there passing TableRaftService's updating
+ CompletableFuture<ReplicaListener> newReplicaListenerFut =
newRaftClientFut.thenApply(createListener);
+
+ if (shouldSkipReplicaStarting) {
+ return nullCompletedFuture();
+ }
+
+
+ return temporalInternalCreateReplica(replicaGrpId,
storageIndexTracker, newReplicaListenerFut);
+ }
+
+ /**
+ * TODO.
+ *
+ * @param replicaGrpId TODO
+ * @param storageIndexTracker TODO
+ * @param newReplicaListenerFut TODO
+ * @return TODO
+ */
+ public CompletableFuture<Replica> temporalInternalCreateReplica(
+ ReplicationGroupId replicaGrpId,
+ PendingComparableValuesTracker<Long, Void> storageIndexTracker,
+ CompletableFuture<ReplicaListener> newReplicaListenerFut
+ ) {
+
ClusterNode localNode = clusterNetSvc.topologyService().localMember();
- Replica newReplica = new Replica(
+ CompletableFuture<Replica> newReplicaFut =
newReplicaListenerFut.thenApply(listener -> new Replica(
replicaGrpId,
listener,
storageIndexTracker,
- raftClient,
localNode,
executor,
placementDriver,
- clockService
- );
+ 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 completedFuture(newReplica);
+ return newReplicaFut;
} else {
- existingReplicaFuture.complete(newReplica);
LOG.info("Replica is started, existing replica waiter was
completed [replicationGroupId={}].", replicaGrpId);
+ existingReplicaFuture.complete(newReplicaFut.join());
Review Comment:
✅ All joins are removed now
--
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: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
