You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@ignite.apache.org by "denis-chudov (via GitHub)" <gi...@apache.org> on 2024/06/05 15:13:51 UTC
[PR] IGNITE-22202 [ignite-3]
denis-chudov opened a new pull request, #3881:
URL: https://github.com/apache/ignite-3/pull/3881
https://issues.apache.org/jira/browse/IGNITE-22202
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22202 Primary replica reservation for nodes that are excluded from assignments [ignite-3]
Posted by "denis-chudov (via GitHub)" <gi...@apache.org>.
denis-chudov commented on code in PR #3881:
URL: https://github.com/apache/ignite-3/pull/3881#discussion_r1638046408
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1005,6 +1021,49 @@ public boolean isReplicaStarted(ReplicationGroupId replicaGrpId) {
return replicaFuture != null && isCompletedSuccessfully(replicaFuture);
}
+ /**
+ * Can possibly start replica if it's not running or is stopping.
+ *
+ * @param groupId Group id.
+ * @param startOperation Replica start operation.
+ * @param forcedAssignments Assignments to reset forcibly, if needed. Assignments reset is only available when replica is started.
+ */
+ public CompletableFuture<Boolean> weakReplicaStart(
Review Comment:
it's not related to rework of `ReplicaManager#startReplica` to `CompletableFuture<Replica>` - it's different future.
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22202 Primary replica reservation for nodes that are excluded from assignments [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc commented on code in PR #3881:
URL: https://github.com/apache/ignite-3/pull/3881#discussion_r1638050845
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1028,4 +1087,357 @@ public Set<ReplicationGroupId> startedGroups() {
.map(Entry::getKey)
.collect(toSet());
}
+
+ @TestOnly
+ public boolean isReplicaPrimaryOnly(ReplicationGroupId groupId) {
+ return replicaLifecycle.isReplicaPrimaryOnly(groupId);
+ }
+
+ private static class ReplicaLifecycle {
+ private final IgniteLogger log = Loggers.forClass(ReplicaLifecycle.class);
+
+ final Map<ReplicationGroupId, ReplicaLifecycleContext> replicaContexts = new ConcurrentHashMap<>();
+
+ final Executor replicaStartStopPool;
+
+ final ClockService clockService;
+
+ final PlacementDriver placementDriver;
+
+ final ReplicaManager replicaManager;
+
+ String localNodeId;
+
+ ReplicaLifecycle(Executor replicaStartStopPool, ClockService clockService, PlacementDriver placementDriver,
+ ReplicaManager replicaManager) {
+ this.replicaStartStopPool = replicaStartStopPool;
+ this.clockService = clockService;
+ this.placementDriver = placementDriver;
+ this.replicaManager = replicaManager;
+ }
+
+ void start(String localNodeId) {
+ this.localNodeId = localNodeId;
+ placementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_ELECTED, this::onPrimaryElected);
+ placementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_EXPIRED, this::onPrimaryExpired);
+ }
+
+ private CompletableFuture<Boolean> onPrimaryElected(PrimaryReplicaEventParameters parameters) {
+ ReplicaLifecycleContext context = getContext(parameters.groupId());
+
+ synchronized (context) {
+ if (localNodeId.equals(parameters.leaseholderId())) {
+ assert context.replicaState != ReplicaState.STOPPED : "Unexpected primary replica state STOPPED [groupId="
+ + parameters.groupId() + "].";
+
+ context.reservedForPrimary = true;
+ } else {
+ context.reservedForPrimary = false;
+
+ if (context.replicaState == ReplicaState.PRIMARY_ONLY) {
+ stopReplica(parameters.groupId(), context, context.deferredStopOperation);
+ }
+ }
+ }
+
+ return falseCompletedFuture();
+ }
+
+ private CompletableFuture<Boolean> onPrimaryExpired(PrimaryReplicaEventParameters parameters) {
+ if (localNodeId.equals(parameters.leaseholderId())) {
+ ReplicaLifecycleContext context = replicaContexts.get(parameters.groupId());
+
+ if (context != null) {
+ synchronized (context) {
+ context.reservedForPrimary = false;
+ }
+ }
+ }
+
+ return falseCompletedFuture();
+ }
+
+ ReplicaLifecycleContext getContext(ReplicationGroupId groupId) {
+ return replicaContexts.computeIfAbsent(groupId,
+ // Treat the absence in the map as STOPPED.
+ k -> new ReplicaLifecycleContext(ReplicaState.STOPPED, nullCompletedFuture()));
+ }
+
+ /**
+ * Can possibly start replica if it's not running or is stopping.
+ *
+ * @param groupId Group id.
+ * @param startOperation Replica start operation.
+ * @param forcedAssignments Assignments to reset forcibly, if needed. Assignments reset is only available when replica is started.
+ */
+ CompletableFuture<Boolean> weakReplicaStart(
+ ReplicationGroupId groupId,
+ Supplier<CompletableFuture<Boolean>> startOperation,
+ @Nullable Assignments forcedAssignments
+ ) {
+ ReplicaLifecycleContext context = getContext(groupId);
+
+ synchronized (context) {
+ ReplicaState state = context.replicaState;
+
+ log.debug("Weak replica start [grp={}, state={}, future={}].", groupId, state, context.previousOperationFuture);
+
+ if (state == ReplicaState.STOPPED || state == ReplicaState.STOPPING) {
+ return startReplica(groupId, context, startOperation);
+ } else if (state == ReplicaState.ASSIGNED) {
+ if (forcedAssignments != null) {
+ assert forcedAssignments.force() :
+ format("Unexpected assignments to force [assignments={}, groupId={}].", forcedAssignments, groupId);
+
+ replicaManager.resetPeers(groupId, fromAssignments(forcedAssignments.nodes()));
+ }
+
+ // Telling the caller that the replica is started.
+ return trueCompletedFuture();
+ } else if (state == ReplicaState.PRIMARY_ONLY) {
+ context.replicaState = ReplicaState.ASSIGNED;
+
+ return trueCompletedFuture();
+ } // else no-op.
+
+ throw new AssertionError("Replica start cannot begin while the replica is being started [groupId=" + groupId + "].");
+ }
+ }
+
+ private CompletableFuture<Boolean> startReplica(
+ ReplicationGroupId groupId,
+ ReplicaLifecycleContext context,
+ Supplier<CompletableFuture<Boolean>> startOperation
+ ) {
+ context.replicaState = ReplicaState.STARTING;
+ context.previousOperationFuture = context.previousOperationFuture
Review Comment:
I also believe that instead of exclusive states ASSIGNED and PRIMARY_ONLY that should be STARTED, STARTED_RESERVED, STARTED_PRIMARY or we may hide RESERVED/PRIMARY inside started.
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22202 Primary replica reservation for nodes that are excluded from assignments [ignite-3]
Posted by "denis-chudov (via GitHub)" <gi...@apache.org>.
denis-chudov commented on code in PR #3881:
URL: https://github.com/apache/ignite-3/pull/3881#discussion_r1638107004
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1005,6 +1021,49 @@ public boolean isReplicaStarted(ReplicationGroupId replicaGrpId) {
return replicaFuture != null && isCompletedSuccessfully(replicaFuture);
}
+ /**
+ * Can possibly start replica if it's not running or is stopping.
+ *
+ * @param groupId Group id.
+ * @param startOperation Replica start operation.
+ * @param forcedAssignments Assignments to reset forcibly, if needed. Assignments reset is only available when replica is started.
Review Comment:
fixed
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1005,6 +1021,49 @@ public boolean isReplicaStarted(ReplicationGroupId replicaGrpId) {
return replicaFuture != null && isCompletedSuccessfully(replicaFuture);
}
+ /**
+ * Can possibly start replica if it's not running or is stopping.
+ *
+ * @param groupId Group id.
+ * @param startOperation Replica start operation.
+ * @param forcedAssignments Assignments to reset forcibly, if needed. Assignments reset is only available when replica is started.
+ */
+ public CompletableFuture<Boolean> weakReplicaStart(
+ ReplicationGroupId groupId,
+ Supplier<CompletableFuture<Boolean>> startOperation,
+ @Nullable Assignments forcedAssignments
+ ) {
+ return replicaLifecycle.weakReplicaStart(groupId, startOperation, forcedAssignments);
+ }
+
+ /**
+ * Can possibly stop replica if it is running or starting, and is not a primary replica. Relies on the given reason. If
+ * the reason is {@link WeakReplicaStopReason#EXCLUDED_FROM_ASSIGNMENTS} then the replica can be not stopped if it is still
+ * a primary. If the reason is {@link WeakReplicaStopReason#PRIMARY_EXPIRED} then the replica is stopped only if its state
+ * is {@link ReplicaState#PRIMARY_ONLY}, because this assumes that it was excluded from assignments before.
+ *
+ * @param groupId Group id.
+ * @param reason Reason to stop replica.
+ * @param stopOperation Replica stop operation.
Review Comment:
fixed
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22202 Primary replica reservation for nodes that are excluded from assignments [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc merged PR #3881:
URL: https://github.com/apache/ignite-3/pull/3881
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22202 Primary replica reservation for nodes that are excluded from assignments [ignite-3]
Posted by "denis-chudov (via GitHub)" <gi...@apache.org>.
denis-chudov commented on code in PR #3881:
URL: https://github.com/apache/ignite-3/pull/3881#discussion_r1638073003
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1005,6 +1021,49 @@ public boolean isReplicaStarted(ReplicationGroupId replicaGrpId) {
return replicaFuture != null && isCompletedSuccessfully(replicaFuture);
}
+ /**
+ * Can possibly start replica if it's not running or is stopping.
+ *
+ * @param groupId Group id.
+ * @param startOperation Replica start operation.
+ * @param forcedAssignments Assignments to reset forcibly, if needed. Assignments reset is only available when replica is started.
+ */
+ public CompletableFuture<Boolean> weakReplicaStart(
Review Comment:
Yes, it was in closure before (I mean TableManager#startPartitionAndStartClient ) but I moved it and now it's outside the startOperation.
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22202 Primary replica reservation for nodes that are excluded from assignments [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc commented on code in PR #3881:
URL: https://github.com/apache/ignite-3/pull/3881#discussion_r1638020118
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1028,4 +1087,357 @@ public Set<ReplicationGroupId> startedGroups() {
.map(Entry::getKey)
.collect(toSet());
}
+
+ @TestOnly
+ public boolean isReplicaPrimaryOnly(ReplicationGroupId groupId) {
+ return replicaLifecycle.isReplicaPrimaryOnly(groupId);
+ }
+
+ private static class ReplicaLifecycle {
+ private final IgniteLogger log = Loggers.forClass(ReplicaLifecycle.class);
+
+ final Map<ReplicationGroupId, ReplicaLifecycleContext> replicaContexts = new ConcurrentHashMap<>();
+
+ final Executor replicaStartStopPool;
+
+ final ClockService clockService;
+
+ final PlacementDriver placementDriver;
+
+ final ReplicaManager replicaManager;
+
+ String localNodeId;
+
+ ReplicaLifecycle(Executor replicaStartStopPool, ClockService clockService, PlacementDriver placementDriver,
+ ReplicaManager replicaManager) {
+ this.replicaStartStopPool = replicaStartStopPool;
+ this.clockService = clockService;
+ this.placementDriver = placementDriver;
+ this.replicaManager = replicaManager;
+ }
+
+ void start(String localNodeId) {
+ this.localNodeId = localNodeId;
+ placementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_ELECTED, this::onPrimaryElected);
+ placementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_EXPIRED, this::onPrimaryExpired);
+ }
+
+ private CompletableFuture<Boolean> onPrimaryElected(PrimaryReplicaEventParameters parameters) {
+ ReplicaLifecycleContext context = getContext(parameters.groupId());
+
+ synchronized (context) {
+ if (localNodeId.equals(parameters.leaseholderId())) {
+ assert context.replicaState != ReplicaState.STOPPED : "Unexpected primary replica state STOPPED [groupId="
+ + parameters.groupId() + "].";
+
+ context.reservedForPrimary = true;
+ } else {
+ context.reservedForPrimary = false;
+
+ if (context.replicaState == ReplicaState.PRIMARY_ONLY) {
+ stopReplica(parameters.groupId(), context, context.deferredStopOperation);
+ }
+ }
+ }
+
+ return falseCompletedFuture();
+ }
+
+ private CompletableFuture<Boolean> onPrimaryExpired(PrimaryReplicaEventParameters parameters) {
+ if (localNodeId.equals(parameters.leaseholderId())) {
+ ReplicaLifecycleContext context = replicaContexts.get(parameters.groupId());
+
+ if (context != null) {
+ synchronized (context) {
+ context.reservedForPrimary = false;
+ }
+ }
+ }
+
+ return falseCompletedFuture();
+ }
+
+ ReplicaLifecycleContext getContext(ReplicationGroupId groupId) {
+ return replicaContexts.computeIfAbsent(groupId,
+ // Treat the absence in the map as STOPPED.
+ k -> new ReplicaLifecycleContext(ReplicaState.STOPPED, nullCompletedFuture()));
+ }
+
+ /**
+ * Can possibly start replica if it's not running or is stopping.
+ *
+ * @param groupId Group id.
+ * @param startOperation Replica start operation.
+ * @param forcedAssignments Assignments to reset forcibly, if needed. Assignments reset is only available when replica is started.
+ */
+ CompletableFuture<Boolean> weakReplicaStart(
+ ReplicationGroupId groupId,
+ Supplier<CompletableFuture<Boolean>> startOperation,
+ @Nullable Assignments forcedAssignments
+ ) {
+ ReplicaLifecycleContext context = getContext(groupId);
+
+ synchronized (context) {
+ ReplicaState state = context.replicaState;
+
+ log.debug("Weak replica start [grp={}, state={}, future={}].", groupId, state, context.previousOperationFuture);
+
+ if (state == ReplicaState.STOPPED || state == ReplicaState.STOPPING) {
+ return startReplica(groupId, context, startOperation);
+ } else if (state == ReplicaState.ASSIGNED) {
+ if (forcedAssignments != null) {
+ assert forcedAssignments.force() :
+ format("Unexpected assignments to force [assignments={}, groupId={}].", forcedAssignments, groupId);
+
+ replicaManager.resetPeers(groupId, fromAssignments(forcedAssignments.nodes()));
+ }
+
+ // Telling the caller that the replica is started.
+ return trueCompletedFuture();
Review Comment:
And what if we are not in force assignments and should the replica?
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22202 Primary replica reservation for nodes that are excluded from assignments [ignite-3]
Posted by "denis-chudov (via GitHub)" <gi...@apache.org>.
denis-chudov commented on code in PR #3881:
URL: https://github.com/apache/ignite-3/pull/3881#discussion_r1638086681
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1028,4 +1087,357 @@ public Set<ReplicationGroupId> startedGroups() {
.map(Entry::getKey)
.collect(toSet());
}
+
+ @TestOnly
+ public boolean isReplicaPrimaryOnly(ReplicationGroupId groupId) {
+ return replicaLifecycle.isReplicaPrimaryOnly(groupId);
+ }
+
+ private static class ReplicaLifecycle {
+ private final IgniteLogger log = Loggers.forClass(ReplicaLifecycle.class);
+
+ final Map<ReplicationGroupId, ReplicaLifecycleContext> replicaContexts = new ConcurrentHashMap<>();
+
+ final Executor replicaStartStopPool;
+
+ final ClockService clockService;
+
+ final PlacementDriver placementDriver;
+
+ final ReplicaManager replicaManager;
+
+ String localNodeId;
+
+ ReplicaLifecycle(Executor replicaStartStopPool, ClockService clockService, PlacementDriver placementDriver,
+ ReplicaManager replicaManager) {
+ this.replicaStartStopPool = replicaStartStopPool;
+ this.clockService = clockService;
+ this.placementDriver = placementDriver;
+ this.replicaManager = replicaManager;
+ }
+
+ void start(String localNodeId) {
+ this.localNodeId = localNodeId;
+ placementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_ELECTED, this::onPrimaryElected);
+ placementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_EXPIRED, this::onPrimaryExpired);
+ }
+
+ private CompletableFuture<Boolean> onPrimaryElected(PrimaryReplicaEventParameters parameters) {
+ ReplicaLifecycleContext context = getContext(parameters.groupId());
+
+ synchronized (context) {
+ if (localNodeId.equals(parameters.leaseholderId())) {
+ assert context.replicaState != ReplicaState.STOPPED : "Unexpected primary replica state STOPPED [groupId="
+ + parameters.groupId() + "].";
+
+ context.reservedForPrimary = true;
Review Comment:
It's expected that corresponding methods will do that
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22202 Primary replica reservation for nodes that are excluded from assignments [ignite-3]
Posted by "denis-chudov (via GitHub)" <gi...@apache.org>.
denis-chudov commented on code in PR #3881:
URL: https://github.com/apache/ignite-3/pull/3881#discussion_r1638157236
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1028,4 +1087,357 @@ public Set<ReplicationGroupId> startedGroups() {
.map(Entry::getKey)
.collect(toSet());
}
+
+ @TestOnly
+ public boolean isReplicaPrimaryOnly(ReplicationGroupId groupId) {
+ return replicaLifecycle.isReplicaPrimaryOnly(groupId);
+ }
+
+ private static class ReplicaLifecycle {
+ private final IgniteLogger log = Loggers.forClass(ReplicaLifecycle.class);
+
+ final Map<ReplicationGroupId, ReplicaLifecycleContext> replicaContexts = new ConcurrentHashMap<>();
+
+ final Executor replicaStartStopPool;
+
+ final ClockService clockService;
+
+ final PlacementDriver placementDriver;
+
+ final ReplicaManager replicaManager;
+
+ String localNodeId;
+
+ ReplicaLifecycle(Executor replicaStartStopPool, ClockService clockService, PlacementDriver placementDriver,
+ ReplicaManager replicaManager) {
+ this.replicaStartStopPool = replicaStartStopPool;
+ this.clockService = clockService;
+ this.placementDriver = placementDriver;
+ this.replicaManager = replicaManager;
+ }
+
+ void start(String localNodeId) {
+ this.localNodeId = localNodeId;
+ placementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_ELECTED, this::onPrimaryElected);
+ placementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_EXPIRED, this::onPrimaryExpired);
+ }
+
+ private CompletableFuture<Boolean> onPrimaryElected(PrimaryReplicaEventParameters parameters) {
+ ReplicaLifecycleContext context = getContext(parameters.groupId());
+
+ synchronized (context) {
+ if (localNodeId.equals(parameters.leaseholderId())) {
+ assert context.replicaState != ReplicaState.STOPPED : "Unexpected primary replica state STOPPED [groupId="
+ + parameters.groupId() + "].";
+
+ context.reservedForPrimary = true;
+ } else {
+ context.reservedForPrimary = false;
+
+ if (context.replicaState == ReplicaState.PRIMARY_ONLY) {
+ stopReplica(parameters.groupId(), context, context.deferredStopOperation);
Review Comment:
because there can be a scenario:
- lease negotiation begins
- replica is reserved
- replica is excluded from assignments and state is transfered to PRIMARY_ONLY
- negotiation fails and another node is elected as leaseholder, here we get a notification through PRIMARY_ELECTED event. Replica is neither in assignments nor primary nor going to become primary and needs to be stopped
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22202 Primary replica reservation for nodes that are excluded from assignments [ignite-3]
Posted by "denis-chudov (via GitHub)" <gi...@apache.org>.
denis-chudov commented on code in PR #3881:
URL: https://github.com/apache/ignite-3/pull/3881#discussion_r1638152440
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1028,4 +1087,357 @@ public Set<ReplicationGroupId> startedGroups() {
.map(Entry::getKey)
.collect(toSet());
}
+
+ @TestOnly
+ public boolean isReplicaPrimaryOnly(ReplicationGroupId groupId) {
+ return replicaLifecycle.isReplicaPrimaryOnly(groupId);
+ }
+
+ private static class ReplicaLifecycle {
Review Comment:
what about ReplicaStateManager?
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22202 Primary replica reservation for nodes that are excluded from assignments [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc commented on code in PR #3881:
URL: https://github.com/apache/ignite-3/pull/3881#discussion_r1638038699
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1028,4 +1087,357 @@ public Set<ReplicationGroupId> startedGroups() {
.map(Entry::getKey)
.collect(toSet());
}
+
+ @TestOnly
+ public boolean isReplicaPrimaryOnly(ReplicationGroupId groupId) {
+ return replicaLifecycle.isReplicaPrimaryOnly(groupId);
+ }
+
+ private static class ReplicaLifecycle {
+ private final IgniteLogger log = Loggers.forClass(ReplicaLifecycle.class);
+
+ final Map<ReplicationGroupId, ReplicaLifecycleContext> replicaContexts = new ConcurrentHashMap<>();
+
+ final Executor replicaStartStopPool;
+
+ final ClockService clockService;
+
+ final PlacementDriver placementDriver;
+
+ final ReplicaManager replicaManager;
+
+ String localNodeId;
+
+ ReplicaLifecycle(Executor replicaStartStopPool, ClockService clockService, PlacementDriver placementDriver,
+ ReplicaManager replicaManager) {
+ this.replicaStartStopPool = replicaStartStopPool;
+ this.clockService = clockService;
+ this.placementDriver = placementDriver;
+ this.replicaManager = replicaManager;
+ }
+
+ void start(String localNodeId) {
+ this.localNodeId = localNodeId;
+ placementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_ELECTED, this::onPrimaryElected);
+ placementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_EXPIRED, this::onPrimaryExpired);
+ }
+
+ private CompletableFuture<Boolean> onPrimaryElected(PrimaryReplicaEventParameters parameters) {
+ ReplicaLifecycleContext context = getContext(parameters.groupId());
+
+ synchronized (context) {
+ if (localNodeId.equals(parameters.leaseholderId())) {
+ assert context.replicaState != ReplicaState.STOPPED : "Unexpected primary replica state STOPPED [groupId="
+ + parameters.groupId() + "].";
+
+ context.reservedForPrimary = true;
+ } else {
+ context.reservedForPrimary = false;
+
+ if (context.replicaState == ReplicaState.PRIMARY_ONLY) {
+ stopReplica(parameters.groupId(), context, context.deferredStopOperation);
+ }
+ }
+ }
+
+ return falseCompletedFuture();
+ }
+
+ private CompletableFuture<Boolean> onPrimaryExpired(PrimaryReplicaEventParameters parameters) {
+ if (localNodeId.equals(parameters.leaseholderId())) {
+ ReplicaLifecycleContext context = replicaContexts.get(parameters.groupId());
+
+ if (context != null) {
+ synchronized (context) {
+ context.reservedForPrimary = false;
+ }
+ }
+ }
+
+ return falseCompletedFuture();
+ }
+
+ ReplicaLifecycleContext getContext(ReplicationGroupId groupId) {
+ return replicaContexts.computeIfAbsent(groupId,
+ // Treat the absence in the map as STOPPED.
+ k -> new ReplicaLifecycleContext(ReplicaState.STOPPED, nullCompletedFuture()));
+ }
+
+ /**
+ * Can possibly start replica if it's not running or is stopping.
+ *
+ * @param groupId Group id.
+ * @param startOperation Replica start operation.
+ * @param forcedAssignments Assignments to reset forcibly, if needed. Assignments reset is only available when replica is started.
+ */
+ CompletableFuture<Boolean> weakReplicaStart(
+ ReplicationGroupId groupId,
+ Supplier<CompletableFuture<Boolean>> startOperation,
+ @Nullable Assignments forcedAssignments
+ ) {
+ ReplicaLifecycleContext context = getContext(groupId);
+
+ synchronized (context) {
+ ReplicaState state = context.replicaState;
+
+ log.debug("Weak replica start [grp={}, state={}, future={}].", groupId, state, context.previousOperationFuture);
+
+ if (state == ReplicaState.STOPPED || state == ReplicaState.STOPPING) {
+ return startReplica(groupId, context, startOperation);
+ } else if (state == ReplicaState.ASSIGNED) {
+ if (forcedAssignments != null) {
+ assert forcedAssignments.force() :
+ format("Unexpected assignments to force [assignments={}, groupId={}].", forcedAssignments, groupId);
+
+ replicaManager.resetPeers(groupId, fromAssignments(forcedAssignments.nodes()));
+ }
+
+ // Telling the caller that the replica is started.
+ return trueCompletedFuture();
+ } else if (state == ReplicaState.PRIMARY_ONLY) {
+ context.replicaState = ReplicaState.ASSIGNED;
+
+ return trueCompletedFuture();
+ } // else no-op.
+
+ throw new AssertionError("Replica start cannot begin while the replica is being started [groupId=" + groupId + "].");
+ }
+ }
+
+ private CompletableFuture<Boolean> startReplica(
+ ReplicationGroupId groupId,
+ ReplicaLifecycleContext context,
+ Supplier<CompletableFuture<Boolean>> startOperation
+ ) {
+ context.replicaState = ReplicaState.STARTING;
Review Comment:
Why it's not implemented as a state machine? It might be simper to have ReplicaLifecycleStateMachine instead ReplicaLifecycle with single public method changeState(groupId, newState), internally it will conditionally change the state and call the closure if needed. Some states should await another some will be applied instantly, some are considered as impossible and will through an exception.
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22202 Primary replica reservation for nodes that are excluded from assignments [ignite-3]
Posted by "denis-chudov (via GitHub)" <gi...@apache.org>.
denis-chudov commented on code in PR #3881:
URL: https://github.com/apache/ignite-3/pull/3881#discussion_r1638078159
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1028,4 +1087,357 @@ public Set<ReplicationGroupId> startedGroups() {
.map(Entry::getKey)
.collect(toSet());
}
+
+ @TestOnly
+ public boolean isReplicaPrimaryOnly(ReplicationGroupId groupId) {
+ return replicaLifecycle.isReplicaPrimaryOnly(groupId);
+ }
+
+ private static class ReplicaLifecycle {
+ private final IgniteLogger log = Loggers.forClass(ReplicaLifecycle.class);
+
+ final Map<ReplicationGroupId, ReplicaLifecycleContext> replicaContexts = new ConcurrentHashMap<>();
+
+ final Executor replicaStartStopPool;
+
+ final ClockService clockService;
+
+ final PlacementDriver placementDriver;
+
+ final ReplicaManager replicaManager;
+
+ String localNodeId;
+
+ ReplicaLifecycle(Executor replicaStartStopPool, ClockService clockService, PlacementDriver placementDriver,
+ ReplicaManager replicaManager) {
+ this.replicaStartStopPool = replicaStartStopPool;
+ this.clockService = clockService;
+ this.placementDriver = placementDriver;
+ this.replicaManager = replicaManager;
+ }
+
+ void start(String localNodeId) {
+ this.localNodeId = localNodeId;
+ placementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_ELECTED, this::onPrimaryElected);
+ placementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_EXPIRED, this::onPrimaryExpired);
+ }
+
+ private CompletableFuture<Boolean> onPrimaryElected(PrimaryReplicaEventParameters parameters) {
+ ReplicaLifecycleContext context = getContext(parameters.groupId());
+
+ synchronized (context) {
+ if (localNodeId.equals(parameters.leaseholderId())) {
+ assert context.replicaState != ReplicaState.STOPPED : "Unexpected primary replica state STOPPED [groupId="
+ + parameters.groupId() + "].";
+
+ context.reservedForPrimary = true;
Review Comment:
It should. I replaced it with assertion (hope it will not fail :) )
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22202 Primary replica reservation for nodes that are excluded from assignments [ignite-3]
Posted by "denis-chudov (via GitHub)" <gi...@apache.org>.
denis-chudov commented on code in PR #3881:
URL: https://github.com/apache/ignite-3/pull/3881#discussion_r1638046408
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1005,6 +1021,49 @@ public boolean isReplicaStarted(ReplicationGroupId replicaGrpId) {
return replicaFuture != null && isCompletedSuccessfully(replicaFuture);
}
+ /**
+ * Can possibly start replica if it's not running or is stopping.
+ *
+ * @param groupId Group id.
+ * @param startOperation Replica start operation.
+ * @param forcedAssignments Assignments to reset forcibly, if needed. Assignments reset is only available when replica is started.
+ */
+ public CompletableFuture<Boolean> weakReplicaStart(
Review Comment:
it's not related to rework of `ReplicaManager#startReplica` to CompletableFuture<Replica> it's different future.
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22202 Primary replica reservation for nodes that are excluded from assignments [ignite-3]
Posted by "denis-chudov (via GitHub)" <gi...@apache.org>.
denis-chudov commented on code in PR #3881:
URL: https://github.com/apache/ignite-3/pull/3881#discussion_r1638074996
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1028,4 +1087,357 @@ public Set<ReplicationGroupId> startedGroups() {
.map(Entry::getKey)
.collect(toSet());
}
+
+ @TestOnly
+ public boolean isReplicaPrimaryOnly(ReplicationGroupId groupId) {
+ return replicaLifecycle.isReplicaPrimaryOnly(groupId);
+ }
+
+ private static class ReplicaLifecycle {
+ private final IgniteLogger log = Loggers.forClass(ReplicaLifecycle.class);
+
+ final Map<ReplicationGroupId, ReplicaLifecycleContext> replicaContexts = new ConcurrentHashMap<>();
+
+ final Executor replicaStartStopPool;
+
+ final ClockService clockService;
+
+ final PlacementDriver placementDriver;
+
+ final ReplicaManager replicaManager;
+
+ String localNodeId;
+
+ ReplicaLifecycle(Executor replicaStartStopPool, ClockService clockService, PlacementDriver placementDriver,
Review Comment:
fixed
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22202 Primary replica reservation for nodes that are excluded from assignments [ignite-3]
Posted by "denis-chudov (via GitHub)" <gi...@apache.org>.
denis-chudov commented on code in PR #3881:
URL: https://github.com/apache/ignite-3/pull/3881#discussion_r1638090405
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1028,4 +1087,357 @@ public Set<ReplicationGroupId> startedGroups() {
.map(Entry::getKey)
.collect(toSet());
}
+
+ @TestOnly
+ public boolean isReplicaPrimaryOnly(ReplicationGroupId groupId) {
+ return replicaLifecycle.isReplicaPrimaryOnly(groupId);
+ }
+
+ private static class ReplicaLifecycle {
+ private final IgniteLogger log = Loggers.forClass(ReplicaLifecycle.class);
+
+ final Map<ReplicationGroupId, ReplicaLifecycleContext> replicaContexts = new ConcurrentHashMap<>();
+
+ final Executor replicaStartStopPool;
+
+ final ClockService clockService;
+
+ final PlacementDriver placementDriver;
+
+ final ReplicaManager replicaManager;
+
+ String localNodeId;
+
+ ReplicaLifecycle(Executor replicaStartStopPool, ClockService clockService, PlacementDriver placementDriver,
+ ReplicaManager replicaManager) {
+ this.replicaStartStopPool = replicaStartStopPool;
+ this.clockService = clockService;
+ this.placementDriver = placementDriver;
+ this.replicaManager = replicaManager;
+ }
+
+ void start(String localNodeId) {
+ this.localNodeId = localNodeId;
+ placementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_ELECTED, this::onPrimaryElected);
+ placementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_EXPIRED, this::onPrimaryExpired);
+ }
+
+ private CompletableFuture<Boolean> onPrimaryElected(PrimaryReplicaEventParameters parameters) {
+ ReplicaLifecycleContext context = getContext(parameters.groupId());
+
+ synchronized (context) {
+ if (localNodeId.equals(parameters.leaseholderId())) {
+ assert context.replicaState != ReplicaState.STOPPED : "Unexpected primary replica state STOPPED [groupId="
+ + parameters.groupId() + "].";
+
+ context.reservedForPrimary = true;
+ } else {
+ context.reservedForPrimary = false;
+
+ if (context.replicaState == ReplicaState.PRIMARY_ONLY) {
+ stopReplica(parameters.groupId(), context, context.deferredStopOperation);
+ }
+ }
+ }
+
+ return falseCompletedFuture();
Review Comment:
It's required by `EventListener` interface, false means the listener shouldn't be removed
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22202 Primary replica reservation for nodes that are excluded from assignments [ignite-3]
Posted by "denis-chudov (via GitHub)" <gi...@apache.org>.
denis-chudov commented on code in PR #3881:
URL: https://github.com/apache/ignite-3/pull/3881#discussion_r1638045164
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1005,6 +1021,49 @@ public boolean isReplicaStarted(ReplicationGroupId replicaGrpId) {
return replicaFuture != null && isCompletedSuccessfully(replicaFuture);
}
+ /**
+ * Can possibly start replica if it's not running or is stopping.
+ *
+ * @param groupId Group id.
+ * @param startOperation Replica start operation.
Review Comment:
yes, added this to javadoc
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22202 Primary replica reservation for nodes that are excluded from assignments [ignite-3]
Posted by "denis-chudov (via GitHub)" <gi...@apache.org>.
denis-chudov commented on code in PR #3881:
URL: https://github.com/apache/ignite-3/pull/3881#discussion_r1638074563
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1028,4 +1087,357 @@ public Set<ReplicationGroupId> startedGroups() {
.map(Entry::getKey)
.collect(toSet());
}
+
+ @TestOnly
+ public boolean isReplicaPrimaryOnly(ReplicationGroupId groupId) {
+ return replicaLifecycle.isReplicaPrimaryOnly(groupId);
+ }
+
+ private static class ReplicaLifecycle {
+ private final IgniteLogger log = Loggers.forClass(ReplicaLifecycle.class);
+
+ final Map<ReplicationGroupId, ReplicaLifecycleContext> replicaContexts = new ConcurrentHashMap<>();
+
+ final Executor replicaStartStopPool;
+
+ final ClockService clockService;
+
+ final PlacementDriver placementDriver;
+
+ final ReplicaManager replicaManager;
+
+ String localNodeId;
Review Comment:
fixed
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22202 Primary replica reservation for nodes that are excluded from assignments [ignite-3]
Posted by "denis-chudov (via GitHub)" <gi...@apache.org>.
denis-chudov commented on code in PR #3881:
URL: https://github.com/apache/ignite-3/pull/3881#discussion_r1638071214
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1028,4 +1087,357 @@ public Set<ReplicationGroupId> startedGroups() {
.map(Entry::getKey)
.collect(toSet());
}
+
+ @TestOnly
+ public boolean isReplicaPrimaryOnly(ReplicationGroupId groupId) {
Review Comment:
seems it's different, it tells nothing about replica state in terms of lifecycle
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22202 Primary replica reservation for nodes that are excluded from assignments [ignite-3]
Posted by "denis-chudov (via GitHub)" <gi...@apache.org>.
denis-chudov commented on code in PR #3881:
URL: https://github.com/apache/ignite-3/pull/3881#discussion_r1638104385
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1028,4 +1087,357 @@ public Set<ReplicationGroupId> startedGroups() {
.map(Entry::getKey)
.collect(toSet());
}
+
+ @TestOnly
+ public boolean isReplicaPrimaryOnly(ReplicationGroupId groupId) {
+ return replicaLifecycle.isReplicaPrimaryOnly(groupId);
+ }
+
+ private static class ReplicaLifecycle {
+ private final IgniteLogger log = Loggers.forClass(ReplicaLifecycle.class);
+
+ final Map<ReplicationGroupId, ReplicaLifecycleContext> replicaContexts = new ConcurrentHashMap<>();
+
+ final Executor replicaStartStopPool;
+
+ final ClockService clockService;
+
+ final PlacementDriver placementDriver;
+
+ final ReplicaManager replicaManager;
+
+ String localNodeId;
+
+ ReplicaLifecycle(Executor replicaStartStopPool, ClockService clockService, PlacementDriver placementDriver,
+ ReplicaManager replicaManager) {
+ this.replicaStartStopPool = replicaStartStopPool;
+ this.clockService = clockService;
+ this.placementDriver = placementDriver;
+ this.replicaManager = replicaManager;
+ }
+
+ void start(String localNodeId) {
+ this.localNodeId = localNodeId;
+ placementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_ELECTED, this::onPrimaryElected);
+ placementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_EXPIRED, this::onPrimaryExpired);
+ }
+
+ private CompletableFuture<Boolean> onPrimaryElected(PrimaryReplicaEventParameters parameters) {
+ ReplicaLifecycleContext context = getContext(parameters.groupId());
+
+ synchronized (context) {
+ if (localNodeId.equals(parameters.leaseholderId())) {
+ assert context.replicaState != ReplicaState.STOPPED : "Unexpected primary replica state STOPPED [groupId="
+ + parameters.groupId() + "].";
+
+ context.reservedForPrimary = true;
+ } else {
+ context.reservedForPrimary = false;
+
+ if (context.replicaState == ReplicaState.PRIMARY_ONLY) {
+ stopReplica(parameters.groupId(), context, context.deferredStopOperation);
+ }
+ }
+ }
+
+ return falseCompletedFuture();
+ }
+
+ private CompletableFuture<Boolean> onPrimaryExpired(PrimaryReplicaEventParameters parameters) {
+ if (localNodeId.equals(parameters.leaseholderId())) {
+ ReplicaLifecycleContext context = replicaContexts.get(parameters.groupId());
+
+ if (context != null) {
+ synchronized (context) {
+ context.reservedForPrimary = false;
+ }
+ }
+ }
+
+ return falseCompletedFuture();
+ }
+
+ ReplicaLifecycleContext getContext(ReplicationGroupId groupId) {
+ return replicaContexts.computeIfAbsent(groupId,
+ // Treat the absence in the map as STOPPED.
+ k -> new ReplicaLifecycleContext(ReplicaState.STOPPED, nullCompletedFuture()));
+ }
+
+ /**
+ * Can possibly start replica if it's not running or is stopping.
+ *
+ * @param groupId Group id.
+ * @param startOperation Replica start operation.
+ * @param forcedAssignments Assignments to reset forcibly, if needed. Assignments reset is only available when replica is started.
+ */
+ CompletableFuture<Boolean> weakReplicaStart(
+ ReplicationGroupId groupId,
+ Supplier<CompletableFuture<Boolean>> startOperation,
+ @Nullable Assignments forcedAssignments
+ ) {
+ ReplicaLifecycleContext context = getContext(groupId);
+
+ synchronized (context) {
+ ReplicaState state = context.replicaState;
+
+ log.debug("Weak replica start [grp={}, state={}, future={}].", groupId, state, context.previousOperationFuture);
+
+ if (state == ReplicaState.STOPPED || state == ReplicaState.STOPPING) {
+ return startReplica(groupId, context, startOperation);
+ } else if (state == ReplicaState.ASSIGNED) {
+ if (forcedAssignments != null) {
+ assert forcedAssignments.force() :
+ format("Unexpected assignments to force [assignments={}, groupId={}].", forcedAssignments, groupId);
+
+ replicaManager.resetPeers(groupId, fromAssignments(forcedAssignments.nodes()));
+ }
+
+ // Telling the caller that the replica is started.
+ return trueCompletedFuture();
Review Comment:
I'm not sure I understand the question correctly, but both ASSIGNED and PRIMARY_ONLY states assume that the replica is already started, so we return true here (the same as startOperation was called and returned completable future with `true` result)
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22202 Primary replica reservation for nodes that are excluded from assignments [ignite-3]
Posted by "denis-chudov (via GitHub)" <gi...@apache.org>.
denis-chudov commented on code in PR #3881:
URL: https://github.com/apache/ignite-3/pull/3881#discussion_r1638043196
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/Replica.java:
##########
@@ -94,6 +94,8 @@ public class Replica {
private final ClockService clockService;
+ private final ReplicaManager replicaManager;
Review Comment:
replaced with closure
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22202 Primary replica reservation for nodes that are excluded from assignments [ignite-3]
Posted by "denis-chudov (via GitHub)" <gi...@apache.org>.
denis-chudov commented on code in PR #3881:
URL: https://github.com/apache/ignite-3/pull/3881#discussion_r1638032635
##########
modules/replicator/src/integrationTest/java/org/apache/ignite/internal/replicator/ItPlacementDriverReplicaSideTest.java:
##########
@@ -208,7 +209,8 @@ public void beforeTest(TestInfo testInfo) {
null,
topologyAwareRaftGroupServiceFactory,
raftManager,
- new VolatileLogStorageFactoryCreator(nodeName, workDir.resolve("volatile-log-spillout"))
+ new VolatileLogStorageFactoryCreator(nodeName, workDir.resolve("volatile-log-spillout")),
+ Executors.newSingleThreadExecutor()
Review Comment:
replaced with ForkJoinPool.commonPool()
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22202 Primary replica reservation for nodes that are excluded from assignments [ignite-3]
Posted by "denis-chudov (via GitHub)" <gi...@apache.org>.
denis-chudov commented on code in PR #3881:
URL: https://github.com/apache/ignite-3/pull/3881#discussion_r1638069288
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1005,6 +1021,49 @@ public boolean isReplicaStarted(ReplicationGroupId replicaGrpId) {
return replicaFuture != null && isCompletedSuccessfully(replicaFuture);
}
+ /**
+ * Can possibly start replica if it's not running or is stopping.
+ *
+ * @param groupId Group id.
+ * @param startOperation Replica start operation.
+ * @param forcedAssignments Assignments to reset forcibly, if needed. Assignments reset is only available when replica is started.
+ */
+ public CompletableFuture<Boolean> weakReplicaStart(
+ ReplicationGroupId groupId,
+ Supplier<CompletableFuture<Boolean>> startOperation,
+ @Nullable Assignments forcedAssignments
+ ) {
+ return replicaLifecycle.weakReplicaStart(groupId, startOperation, forcedAssignments);
+ }
+
+ /**
+ * Can possibly stop replica if it is running or starting, and is not a primary replica. Relies on the given reason. If
+ * the reason is {@link WeakReplicaStopReason#EXCLUDED_FROM_ASSIGNMENTS} then the replica can be not stopped if it is still
+ * a primary. If the reason is {@link WeakReplicaStopReason#PRIMARY_EXPIRED} then the replica is stopped only if its state
+ * is {@link ReplicaState#PRIMARY_ONLY}, because this assumes that it was excluded from assignments before.
+ *
+ * @param groupId Group id.
+ * @param reason Reason to stop replica.
+ * @param stopOperation Replica stop operation.
+ */
+ public CompletableFuture<Void> weakReplicaStop(
+ ReplicationGroupId groupId,
+ WeakReplicaStopReason reason,
+ Supplier<CompletableFuture<Void>> stopOperation
+ ) {
+ return replicaLifecycle.weakReplicaStop(groupId, reason, stopOperation);
+ }
+
+ /**
+ * Reserve replica as primary.
+ *
+ * @param groupId Group id.
+ * @return Whether the replica was successfully reserved.
+ */
+ boolean reserveReplica(ReplicationGroupId groupId) {
Review Comment:
removed the method
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22202 Primary replica reservation for nodes that are excluded from assignments [ignite-3]
Posted by "denis-chudov (via GitHub)" <gi...@apache.org>.
denis-chudov commented on code in PR #3881:
URL: https://github.com/apache/ignite-3/pull/3881#discussion_r1638067368
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1005,6 +1021,49 @@ public boolean isReplicaStarted(ReplicationGroupId replicaGrpId) {
return replicaFuture != null && isCompletedSuccessfully(replicaFuture);
}
+ /**
+ * Can possibly start replica if it's not running or is stopping.
+ *
+ * @param groupId Group id.
+ * @param startOperation Replica start operation.
+ * @param forcedAssignments Assignments to reset forcibly, if needed. Assignments reset is only available when replica is started.
+ */
+ public CompletableFuture<Boolean> weakReplicaStart(
+ ReplicationGroupId groupId,
+ Supplier<CompletableFuture<Boolean>> startOperation,
+ @Nullable Assignments forcedAssignments
+ ) {
+ return replicaLifecycle.weakReplicaStart(groupId, startOperation, forcedAssignments);
+ }
+
+ /**
+ * Can possibly stop replica if it is running or starting, and is not a primary replica. Relies on the given reason. If
+ * the reason is {@link WeakReplicaStopReason#EXCLUDED_FROM_ASSIGNMENTS} then the replica can be not stopped if it is still
Review Comment:
It's correct. "Can be not stopped" means that the replica probably will be not stopped; "can not be stopped" means it won't be stopped regardless of circumstances
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22202 Primary replica reservation for nodes that are excluded from assignments [ignite-3]
Posted by "denis-chudov (via GitHub)" <gi...@apache.org>.
denis-chudov commented on code in PR #3881:
URL: https://github.com/apache/ignite-3/pull/3881#discussion_r1638067769
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1005,6 +1021,49 @@ public boolean isReplicaStarted(ReplicationGroupId replicaGrpId) {
return replicaFuture != null && isCompletedSuccessfully(replicaFuture);
}
+ /**
+ * Can possibly start replica if it's not running or is stopping.
+ *
+ * @param groupId Group id.
+ * @param startOperation Replica start operation.
+ * @param forcedAssignments Assignments to reset forcibly, if needed. Assignments reset is only available when replica is started.
+ */
+ public CompletableFuture<Boolean> weakReplicaStart(
+ ReplicationGroupId groupId,
+ Supplier<CompletableFuture<Boolean>> startOperation,
+ @Nullable Assignments forcedAssignments
+ ) {
+ return replicaLifecycle.weakReplicaStart(groupId, startOperation, forcedAssignments);
+ }
+
+ /**
+ * Can possibly stop replica if it is running or starting, and is not a primary replica. Relies on the given reason. If
Review Comment:
extended the javadoc
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22202 Primary replica reservation for nodes that are excluded from assignments [ignite-3]
Posted by "denis-chudov (via GitHub)" <gi...@apache.org>.
denis-chudov commented on code in PR #3881:
URL: https://github.com/apache/ignite-3/pull/3881#discussion_r1638030372
##########
modules/raft-api/src/main/java/org/apache/ignite/internal/raft/PeersAndLearners.java:
##########
@@ -80,6 +82,24 @@ public static PeersAndLearners fromPeers(Collection<Peer> peers, Collection<Peer
return new PeersAndLearners(peers, learners);
}
+ /**
+ * Creates an instance using the given assignments.
+ */
+ public static PeersAndLearners fromAssignments(Collection<Assignment> assignments) {
Review Comment:
seems so, actually I just moved this method from TableManger
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22202 Primary replica reservation for nodes that are excluded from assignments [ignite-3]
Posted by "denis-chudov (via GitHub)" <gi...@apache.org>.
denis-chudov commented on code in PR #3881:
URL: https://github.com/apache/ignite-3/pull/3881#discussion_r1638029122
##########
modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildController.java:
##########
@@ -306,10 +306,7 @@ private void scheduleBuildIndex(
) {
MvPartitionStorage mvPartition = mvPartitionStorage(mvTableStorage, replicaId);
- // TODO: IGNITE-22202 Deal with this situation
- if (mvPartition == null) {
- return;
- }
+ assert mvPartition != null;
Review Comment:
added the message
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22202 Primary replica reservation for nodes that are excluded from assignments [ignite-3]
Posted by "denis-chudov (via GitHub)" <gi...@apache.org>.
denis-chudov commented on code in PR #3881:
URL: https://github.com/apache/ignite-3/pull/3881#discussion_r1638113196
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1028,4 +1087,357 @@ public Set<ReplicationGroupId> startedGroups() {
.map(Entry::getKey)
.collect(toSet());
}
+
+ @TestOnly
+ public boolean isReplicaPrimaryOnly(ReplicationGroupId groupId) {
+ return replicaLifecycle.isReplicaPrimaryOnly(groupId);
+ }
+
+ private static class ReplicaLifecycle {
+ private final IgniteLogger log = Loggers.forClass(ReplicaLifecycle.class);
+
+ final Map<ReplicationGroupId, ReplicaLifecycleContext> replicaContexts = new ConcurrentHashMap<>();
+
+ final Executor replicaStartStopPool;
+
+ final ClockService clockService;
+
+ final PlacementDriver placementDriver;
+
+ final ReplicaManager replicaManager;
+
+ String localNodeId;
+
+ ReplicaLifecycle(Executor replicaStartStopPool, ClockService clockService, PlacementDriver placementDriver,
+ ReplicaManager replicaManager) {
+ this.replicaStartStopPool = replicaStartStopPool;
+ this.clockService = clockService;
+ this.placementDriver = placementDriver;
+ this.replicaManager = replicaManager;
+ }
+
+ void start(String localNodeId) {
+ this.localNodeId = localNodeId;
+ placementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_ELECTED, this::onPrimaryElected);
+ placementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_EXPIRED, this::onPrimaryExpired);
+ }
+
+ private CompletableFuture<Boolean> onPrimaryElected(PrimaryReplicaEventParameters parameters) {
+ ReplicaLifecycleContext context = getContext(parameters.groupId());
+
+ synchronized (context) {
+ if (localNodeId.equals(parameters.leaseholderId())) {
+ assert context.replicaState != ReplicaState.STOPPED : "Unexpected primary replica state STOPPED [groupId="
+ + parameters.groupId() + "].";
+
+ context.reservedForPrimary = true;
+ } else {
+ context.reservedForPrimary = false;
+
+ if (context.replicaState == ReplicaState.PRIMARY_ONLY) {
+ stopReplica(parameters.groupId(), context, context.deferredStopOperation);
+ }
+ }
+ }
+
+ return falseCompletedFuture();
+ }
+
+ private CompletableFuture<Boolean> onPrimaryExpired(PrimaryReplicaEventParameters parameters) {
+ if (localNodeId.equals(parameters.leaseholderId())) {
+ ReplicaLifecycleContext context = replicaContexts.get(parameters.groupId());
+
+ if (context != null) {
+ synchronized (context) {
+ context.reservedForPrimary = false;
+ }
+ }
+ }
+
+ return falseCompletedFuture();
+ }
+
+ ReplicaLifecycleContext getContext(ReplicationGroupId groupId) {
+ return replicaContexts.computeIfAbsent(groupId,
+ // Treat the absence in the map as STOPPED.
+ k -> new ReplicaLifecycleContext(ReplicaState.STOPPED, nullCompletedFuture()));
+ }
+
+ /**
+ * Can possibly start replica if it's not running or is stopping.
+ *
+ * @param groupId Group id.
+ * @param startOperation Replica start operation.
+ * @param forcedAssignments Assignments to reset forcibly, if needed. Assignments reset is only available when replica is started.
+ */
+ CompletableFuture<Boolean> weakReplicaStart(
+ ReplicationGroupId groupId,
+ Supplier<CompletableFuture<Boolean>> startOperation,
+ @Nullable Assignments forcedAssignments
+ ) {
+ ReplicaLifecycleContext context = getContext(groupId);
+
+ synchronized (context) {
+ ReplicaState state = context.replicaState;
+
+ log.debug("Weak replica start [grp={}, state={}, future={}].", groupId, state, context.previousOperationFuture);
+
+ if (state == ReplicaState.STOPPED || state == ReplicaState.STOPPING) {
+ return startReplica(groupId, context, startOperation);
+ } else if (state == ReplicaState.ASSIGNED) {
+ if (forcedAssignments != null) {
+ assert forcedAssignments.force() :
+ format("Unexpected assignments to force [assignments={}, groupId={}].", forcedAssignments, groupId);
+
+ replicaManager.resetPeers(groupId, fromAssignments(forcedAssignments.nodes()));
+ }
+
+ // Telling the caller that the replica is started.
+ return trueCompletedFuture();
+ } else if (state == ReplicaState.PRIMARY_ONLY) {
+ context.replicaState = ReplicaState.ASSIGNED;
+
+ return trueCompletedFuture();
+ } // else no-op.
+
+ throw new AssertionError("Replica start cannot begin while the replica is being started [groupId=" + groupId + "].");
+ }
+ }
+
+ private CompletableFuture<Boolean> startReplica(
+ ReplicationGroupId groupId,
+ ReplicaLifecycleContext context,
+ Supplier<CompletableFuture<Boolean>> startOperation
+ ) {
+ context.replicaState = ReplicaState.STARTING;
Review Comment:
actually all the logic is in 2 methods now (weakStartReplica and weakStopReplica), maybe one single method would be too big; also it would have to accept the type of action (start or stop), because the transitions depend on it, so I am not sure the conditional logic will look more simple, just sheet of `if`-s in the number equal to cartesian product size
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22202 Primary replica reservation for nodes that are excluded from assignments [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc commented on code in PR #3881:
URL: https://github.com/apache/ignite-3/pull/3881#discussion_r1637611742
##########
modules/replicator/src/integrationTest/java/org/apache/ignite/internal/replicator/ItPlacementDriverReplicaSideTest.java:
##########
@@ -208,7 +209,8 @@ public void beforeTest(TestInfo testInfo) {
null,
topologyAwareRaftGroupServiceFactory,
raftManager,
- new VolatileLogStorageFactoryCreator(nodeName, workDir.resolve("volatile-log-spillout"))
+ new VolatileLogStorageFactoryCreator(nodeName, workDir.resolve("volatile-log-spillout")),
+ Executors.newSingleThreadExecutor()
Review Comment:
In tests I'd rather use NamedThreadFactory or ForkJoinPool.commonPool().
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/Replica.java:
##########
@@ -94,6 +94,8 @@ public class Replica {
private final ClockService clockService;
+ private final ReplicaManager replicaManager;
Review Comment:
Replica should not depend on ReplicaManager.
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1005,6 +1021,49 @@ public boolean isReplicaStarted(ReplicationGroupId replicaGrpId) {
return replicaFuture != null && isCompletedSuccessfully(replicaFuture);
}
+ /**
+ * Can possibly start replica if it's not running or is stopping.
+ *
+ * @param groupId Group id.
+ * @param startOperation Replica start operation.
+ * @param forcedAssignments Assignments to reset forcibly, if needed. Assignments reset is only available when replica is started.
+ */
+ public CompletableFuture<Boolean> weakReplicaStart(
+ ReplicationGroupId groupId,
+ Supplier<CompletableFuture<Boolean>> startOperation,
+ @Nullable Assignments forcedAssignments
+ ) {
+ return replicaLifecycle.weakReplicaStart(groupId, startOperation, forcedAssignments);
+ }
+
+ /**
+ * Can possibly stop replica if it is running or starting, and is not a primary replica. Relies on the given reason. If
+ * the reason is {@link WeakReplicaStopReason#EXCLUDED_FROM_ASSIGNMENTS} then the replica can be not stopped if it is still
+ * a primary. If the reason is {@link WeakReplicaStopReason#PRIMARY_EXPIRED} then the replica is stopped only if its state
+ * is {@link ReplicaState#PRIMARY_ONLY}, because this assumes that it was excluded from assignments before.
+ *
+ * @param groupId Group id.
+ * @param reason Reason to stop replica.
+ * @param stopOperation Replica stop operation.
+ */
+ public CompletableFuture<Void> weakReplicaStop(
+ ReplicationGroupId groupId,
+ WeakReplicaStopReason reason,
+ Supplier<CompletableFuture<Void>> stopOperation
+ ) {
+ return replicaLifecycle.weakReplicaStop(groupId, reason, stopOperation);
+ }
+
+ /**
+ * Reserve replica as primary.
+ *
+ * @param groupId Group id.
+ * @return Whether the replica was successfully reserved.
+ */
+ boolean reserveReplica(ReplicationGroupId groupId) {
Review Comment:
Here we can smell a bad design)) Ideally that should be hided inside ReplicaLifecycle. Any other better options?
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1028,4 +1087,357 @@ public Set<ReplicationGroupId> startedGroups() {
.map(Entry::getKey)
.collect(toSet());
}
+
+ @TestOnly
+ public boolean isReplicaPrimaryOnly(ReplicationGroupId groupId) {
+ return replicaLifecycle.isReplicaPrimaryOnly(groupId);
+ }
+
+ private static class ReplicaLifecycle {
+ private final IgniteLogger log = Loggers.forClass(ReplicaLifecycle.class);
+
+ final Map<ReplicationGroupId, ReplicaLifecycleContext> replicaContexts = new ConcurrentHashMap<>();
+
+ final Executor replicaStartStopPool;
+
+ final ClockService clockService;
+
+ final PlacementDriver placementDriver;
+
+ final ReplicaManager replicaManager;
+
+ String localNodeId;
+
+ ReplicaLifecycle(Executor replicaStartStopPool, ClockService clockService, PlacementDriver placementDriver,
Review Comment:
One at a row please.
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1005,6 +1021,49 @@ public boolean isReplicaStarted(ReplicationGroupId replicaGrpId) {
return replicaFuture != null && isCompletedSuccessfully(replicaFuture);
}
+ /**
+ * Can possibly start replica if it's not running or is stopping.
+ *
+ * @param groupId Group id.
+ * @param startOperation Replica start operation.
+ * @param forcedAssignments Assignments to reset forcibly, if needed. Assignments reset is only available when replica is started.
+ */
+ public CompletableFuture<Boolean> weakReplicaStart(
+ ReplicationGroupId groupId,
+ Supplier<CompletableFuture<Boolean>> startOperation,
+ @Nullable Assignments forcedAssignments
+ ) {
+ return replicaLifecycle.weakReplicaStart(groupId, startOperation, forcedAssignments);
+ }
+
+ /**
+ * Can possibly stop replica if it is running or starting, and is not a primary replica. Relies on the given reason. If
+ * the reason is {@link WeakReplicaStopReason#EXCLUDED_FROM_ASSIGNMENTS} then the replica can be not stopped if it is still
Review Comment:
can be not -> can not be
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1028,4 +1087,357 @@ public Set<ReplicationGroupId> startedGroups() {
.map(Entry::getKey)
.collect(toSet());
}
+
+ @TestOnly
+ public boolean isReplicaPrimaryOnly(ReplicationGroupId groupId) {
+ return replicaLifecycle.isReplicaPrimaryOnly(groupId);
+ }
+
+ private static class ReplicaLifecycle {
+ private final IgniteLogger log = Loggers.forClass(ReplicaLifecycle.class);
+
+ final Map<ReplicationGroupId, ReplicaLifecycleContext> replicaContexts = new ConcurrentHashMap<>();
+
+ final Executor replicaStartStopPool;
+
+ final ClockService clockService;
+
+ final PlacementDriver placementDriver;
+
+ final ReplicaManager replicaManager;
+
+ String localNodeId;
Review Comment:
volatile?
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1028,4 +1087,357 @@ public Set<ReplicationGroupId> startedGroups() {
.map(Entry::getKey)
.collect(toSet());
}
+
+ @TestOnly
+ public boolean isReplicaPrimaryOnly(ReplicationGroupId groupId) {
+ return replicaLifecycle.isReplicaPrimaryOnly(groupId);
+ }
+
+ private static class ReplicaLifecycle {
+ private final IgniteLogger log = Loggers.forClass(ReplicaLifecycle.class);
+
+ final Map<ReplicationGroupId, ReplicaLifecycleContext> replicaContexts = new ConcurrentHashMap<>();
+
+ final Executor replicaStartStopPool;
+
+ final ClockService clockService;
+
+ final PlacementDriver placementDriver;
+
+ final ReplicaManager replicaManager;
+
+ String localNodeId;
+
+ ReplicaLifecycle(Executor replicaStartStopPool, ClockService clockService, PlacementDriver placementDriver,
+ ReplicaManager replicaManager) {
+ this.replicaStartStopPool = replicaStartStopPool;
+ this.clockService = clockService;
+ this.placementDriver = placementDriver;
+ this.replicaManager = replicaManager;
+ }
+
+ void start(String localNodeId) {
+ this.localNodeId = localNodeId;
+ placementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_ELECTED, this::onPrimaryElected);
+ placementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_EXPIRED, this::onPrimaryExpired);
+ }
+
+ private CompletableFuture<Boolean> onPrimaryElected(PrimaryReplicaEventParameters parameters) {
+ ReplicaLifecycleContext context = getContext(parameters.groupId());
+
+ synchronized (context) {
+ if (localNodeId.equals(parameters.leaseholderId())) {
+ assert context.replicaState != ReplicaState.STOPPED : "Unexpected primary replica state STOPPED [groupId="
+ + parameters.groupId() + "].";
+
+ context.reservedForPrimary = true;
+ } else {
+ context.reservedForPrimary = false;
+
+ if (context.replicaState == ReplicaState.PRIMARY_ONLY) {
+ stopReplica(parameters.groupId(), context, context.deferredStopOperation);
Review Comment:
Why stopping on PRIMARY_REPLICA_EXIRED is not enough?
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1028,4 +1087,357 @@ public Set<ReplicationGroupId> startedGroups() {
.map(Entry::getKey)
.collect(toSet());
}
+
+ @TestOnly
+ public boolean isReplicaPrimaryOnly(ReplicationGroupId groupId) {
+ return replicaLifecycle.isReplicaPrimaryOnly(groupId);
+ }
+
+ private static class ReplicaLifecycle {
+ private final IgniteLogger log = Loggers.forClass(ReplicaLifecycle.class);
+
+ final Map<ReplicationGroupId, ReplicaLifecycleContext> replicaContexts = new ConcurrentHashMap<>();
+
+ final Executor replicaStartStopPool;
+
+ final ClockService clockService;
+
+ final PlacementDriver placementDriver;
+
+ final ReplicaManager replicaManager;
+
+ String localNodeId;
+
+ ReplicaLifecycle(Executor replicaStartStopPool, ClockService clockService, PlacementDriver placementDriver,
+ ReplicaManager replicaManager) {
+ this.replicaStartStopPool = replicaStartStopPool;
+ this.clockService = clockService;
+ this.placementDriver = placementDriver;
+ this.replicaManager = replicaManager;
+ }
+
+ void start(String localNodeId) {
+ this.localNodeId = localNodeId;
+ placementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_ELECTED, this::onPrimaryElected);
+ placementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_EXPIRED, this::onPrimaryExpired);
+ }
+
+ private CompletableFuture<Boolean> onPrimaryElected(PrimaryReplicaEventParameters parameters) {
+ ReplicaLifecycleContext context = getContext(parameters.groupId());
+
+ synchronized (context) {
+ if (localNodeId.equals(parameters.leaseholderId())) {
+ assert context.replicaState != ReplicaState.STOPPED : "Unexpected primary replica state STOPPED [groupId="
+ + parameters.groupId() + "].";
+
+ context.reservedForPrimary = true;
+ } else {
+ context.reservedForPrimary = false;
+
+ if (context.replicaState == ReplicaState.PRIMARY_ONLY) {
+ stopReplica(parameters.groupId(), context, context.deferredStopOperation);
+ }
+ }
+ }
+
+ return falseCompletedFuture();
Review Comment:
Why it's boolean? Where the boolean result is used?
##########
modules/index/src/main/java/org/apache/ignite/internal/index/IndexBuildController.java:
##########
@@ -306,10 +306,7 @@ private void scheduleBuildIndex(
) {
MvPartitionStorage mvPartition = mvPartitionStorage(mvTableStorage, replicaId);
- // TODO: IGNITE-22202 Deal with this situation
- if (mvPartition == null) {
- return;
- }
+ assert mvPartition != null;
Review Comment:
Further logic will depend on Kirill's answer. If it's fine to have and assertion here, let's have at least " : replicaId;" message.
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1005,6 +1021,49 @@ public boolean isReplicaStarted(ReplicationGroupId replicaGrpId) {
return replicaFuture != null && isCompletedSuccessfully(replicaFuture);
}
+ /**
+ * Can possibly start replica if it's not running or is stopping.
+ *
+ * @param groupId Group id.
+ * @param startOperation Replica start operation.
+ * @param forcedAssignments Assignments to reset forcibly, if needed. Assignments reset is only available when replica is started.
+ */
+ public CompletableFuture<Boolean> weakReplicaStart(
Review Comment:
Let's be consistent with naming and rename weakReplicaStart to weakStartReplica. Same for stop.
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1005,6 +1021,49 @@ public boolean isReplicaStarted(ReplicationGroupId replicaGrpId) {
return replicaFuture != null && isCompletedSuccessfully(replicaFuture);
}
+ /**
+ * Can possibly start replica if it's not running or is stopping.
Review Comment:
Could you please rephrase the javadoc? Behavior is deterministic here, let's explain it precisely.
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1005,6 +1021,49 @@ public boolean isReplicaStarted(ReplicationGroupId replicaGrpId) {
return replicaFuture != null && isCompletedSuccessfully(replicaFuture);
}
+ /**
+ * Can possibly start replica if it's not running or is stopping.
+ *
+ * @param groupId Group id.
+ * @param startOperation Replica start operation.
Review Comment:
Do you mean, that it's a closure that will actually start a replica and thus will be called if weekReplicaStart will internally decide that it required to start the replica?
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1005,6 +1021,49 @@ public boolean isReplicaStarted(ReplicationGroupId replicaGrpId) {
return replicaFuture != null && isCompletedSuccessfully(replicaFuture);
}
+ /**
+ * Can possibly start replica if it's not running or is stopping.
+ *
+ * @param groupId Group id.
+ * @param startOperation Replica start operation.
+ * @param forcedAssignments Assignments to reset forcibly, if needed. Assignments reset is only available when replica is started.
+ */
+ public CompletableFuture<Boolean> weakReplicaStart(
+ ReplicationGroupId groupId,
+ Supplier<CompletableFuture<Boolean>> startOperation,
+ @Nullable Assignments forcedAssignments
Review Comment:
I'd rather have an overloaded version with forcedAssignments, WDYT?
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1028,4 +1087,357 @@ public Set<ReplicationGroupId> startedGroups() {
.map(Entry::getKey)
.collect(toSet());
}
+
+ @TestOnly
+ public boolean isReplicaPrimaryOnly(ReplicationGroupId groupId) {
+ return replicaLifecycle.isReplicaPrimaryOnly(groupId);
+ }
+
+ private static class ReplicaLifecycle {
Review Comment:
The naming clashes with PartitionReplicaLifecycleManager. Any other ideas for the naming?
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1005,6 +1021,49 @@ public boolean isReplicaStarted(ReplicationGroupId replicaGrpId) {
return replicaFuture != null && isCompletedSuccessfully(replicaFuture);
}
+ /**
+ * Can possibly start replica if it's not running or is stopping.
+ *
+ * @param groupId Group id.
+ * @param startOperation Replica start operation.
+ * @param forcedAssignments Assignments to reset forcibly, if needed. Assignments reset is only available when replica is started.
+ */
+ public CompletableFuture<Boolean> weakReplicaStart(
+ ReplicationGroupId groupId,
+ Supplier<CompletableFuture<Boolean>> startOperation,
+ @Nullable Assignments forcedAssignments
+ ) {
+ return replicaLifecycle.weakReplicaStart(groupId, startOperation, forcedAssignments);
+ }
+
+ /**
+ * Can possibly stop replica if it is running or starting, and is not a primary replica. Relies on the given reason. If
+ * the reason is {@link WeakReplicaStopReason#EXCLUDED_FROM_ASSIGNMENTS} then the replica can be not stopped if it is still
+ * a primary. If the reason is {@link WeakReplicaStopReason#PRIMARY_EXPIRED} then the replica is stopped only if its state
+ * is {@link ReplicaState#PRIMARY_ONLY}, because this assumes that it was excluded from assignments before.
+ *
+ * @param groupId Group id.
+ * @param reason Reason to stop replica.
+ * @param stopOperation Replica stop operation.
Review Comment:
@return param is missing.
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1005,6 +1021,49 @@ public boolean isReplicaStarted(ReplicationGroupId replicaGrpId) {
return replicaFuture != null && isCompletedSuccessfully(replicaFuture);
}
+ /**
+ * Can possibly start replica if it's not running or is stopping.
+ *
+ * @param groupId Group id.
+ * @param startOperation Replica start operation.
+ * @param forcedAssignments Assignments to reset forcibly, if needed. Assignments reset is only available when replica is started.
Review Comment:
@return is missing.
##########
modules/raft-api/src/main/java/org/apache/ignite/internal/raft/PeersAndLearners.java:
##########
@@ -80,6 +82,24 @@ public static PeersAndLearners fromPeers(Collection<Peer> peers, Collection<Peer
return new PeersAndLearners(peers, learners);
}
+ /**
+ * Creates an instance using the given assignments.
+ */
+ public static PeersAndLearners fromAssignments(Collection<Assignment> assignments) {
Review Comment:
That will return PeersAndLearners with an empty inner peers and learners in case of an empty assignments, right?
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1005,6 +1021,49 @@ public boolean isReplicaStarted(ReplicationGroupId replicaGrpId) {
return replicaFuture != null && isCompletedSuccessfully(replicaFuture);
}
+ /**
+ * Can possibly start replica if it's not running or is stopping.
+ *
+ * @param groupId Group id.
+ * @param startOperation Replica start operation.
+ * @param forcedAssignments Assignments to reset forcibly, if needed. Assignments reset is only available when replica is started.
+ */
+ public CompletableFuture<Boolean> weakReplicaStart(
Review Comment:
As agreed let's discuss in person which CompletableFuture<Boolean> should be reworked to CompletableFuture<Replica>. @JAkutenshi FYI.
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1005,6 +1021,49 @@ public boolean isReplicaStarted(ReplicationGroupId replicaGrpId) {
return replicaFuture != null && isCompletedSuccessfully(replicaFuture);
}
+ /**
+ * Can possibly start replica if it's not running or is stopping.
+ *
+ * @param groupId Group id.
+ * @param startOperation Replica start operation.
+ * @param forcedAssignments Assignments to reset forcibly, if needed. Assignments reset is only available when replica is started.
+ */
+ public CompletableFuture<Boolean> weakReplicaStart(
Review Comment:
Didn't check that part with my own eyes. Do we update client peers and learners outside the closure? I mean that we should.
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1028,4 +1087,357 @@ public Set<ReplicationGroupId> startedGroups() {
.map(Entry::getKey)
.collect(toSet());
}
+
+ @TestOnly
+ public boolean isReplicaPrimaryOnly(ReplicationGroupId groupId) {
Review Comment:
Do we really need it? Why not to user. PD.getPrimary() check whether replica in assignments or not?
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1028,4 +1087,357 @@ public Set<ReplicationGroupId> startedGroups() {
.map(Entry::getKey)
.collect(toSet());
}
+
+ @TestOnly
+ public boolean isReplicaPrimaryOnly(ReplicationGroupId groupId) {
+ return replicaLifecycle.isReplicaPrimaryOnly(groupId);
+ }
+
+ private static class ReplicaLifecycle {
+ private final IgniteLogger log = Loggers.forClass(ReplicaLifecycle.class);
+
+ final Map<ReplicationGroupId, ReplicaLifecycleContext> replicaContexts = new ConcurrentHashMap<>();
+
+ final Executor replicaStartStopPool;
+
+ final ClockService clockService;
+
+ final PlacementDriver placementDriver;
+
+ final ReplicaManager replicaManager;
+
+ String localNodeId;
+
+ ReplicaLifecycle(Executor replicaStartStopPool, ClockService clockService, PlacementDriver placementDriver,
+ ReplicaManager replicaManager) {
+ this.replicaStartStopPool = replicaStartStopPool;
+ this.clockService = clockService;
+ this.placementDriver = placementDriver;
+ this.replicaManager = replicaManager;
+ }
+
+ void start(String localNodeId) {
+ this.localNodeId = localNodeId;
+ placementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_ELECTED, this::onPrimaryElected);
+ placementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_EXPIRED, this::onPrimaryExpired);
+ }
+
+ private CompletableFuture<Boolean> onPrimaryElected(PrimaryReplicaEventParameters parameters) {
+ ReplicaLifecycleContext context = getContext(parameters.groupId());
+
+ synchronized (context) {
+ if (localNodeId.equals(parameters.leaseholderId())) {
+ assert context.replicaState != ReplicaState.STOPPED : "Unexpected primary replica state STOPPED [groupId="
+ + parameters.groupId() + "].";
+
+ context.reservedForPrimary = true;
Review Comment:
I didn't get that part. Should it already be reserved because of reservation within waitForActualState?
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1028,4 +1087,357 @@ public Set<ReplicationGroupId> startedGroups() {
.map(Entry::getKey)
.collect(toSet());
}
+
+ @TestOnly
+ public boolean isReplicaPrimaryOnly(ReplicationGroupId groupId) {
+ return replicaLifecycle.isReplicaPrimaryOnly(groupId);
+ }
+
+ private static class ReplicaLifecycle {
+ private final IgniteLogger log = Loggers.forClass(ReplicaLifecycle.class);
+
+ final Map<ReplicationGroupId, ReplicaLifecycleContext> replicaContexts = new ConcurrentHashMap<>();
+
+ final Executor replicaStartStopPool;
+
+ final ClockService clockService;
+
+ final PlacementDriver placementDriver;
+
+ final ReplicaManager replicaManager;
+
+ String localNodeId;
+
+ ReplicaLifecycle(Executor replicaStartStopPool, ClockService clockService, PlacementDriver placementDriver,
+ ReplicaManager replicaManager) {
+ this.replicaStartStopPool = replicaStartStopPool;
+ this.clockService = clockService;
+ this.placementDriver = placementDriver;
+ this.replicaManager = replicaManager;
+ }
+
+ void start(String localNodeId) {
+ this.localNodeId = localNodeId;
+ placementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_ELECTED, this::onPrimaryElected);
+ placementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_EXPIRED, this::onPrimaryExpired);
+ }
+
+ private CompletableFuture<Boolean> onPrimaryElected(PrimaryReplicaEventParameters parameters) {
+ ReplicaLifecycleContext context = getContext(parameters.groupId());
+
+ synchronized (context) {
+ if (localNodeId.equals(parameters.leaseholderId())) {
+ assert context.replicaState != ReplicaState.STOPPED : "Unexpected primary replica state STOPPED [groupId="
+ + parameters.groupId() + "].";
+
+ context.reservedForPrimary = true;
+ } else {
+ context.reservedForPrimary = false;
Review Comment:
BTW, why reserved for primary is not the state?
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1028,4 +1087,357 @@ public Set<ReplicationGroupId> startedGroups() {
.map(Entry::getKey)
.collect(toSet());
}
+
+ @TestOnly
+ public boolean isReplicaPrimaryOnly(ReplicationGroupId groupId) {
+ return replicaLifecycle.isReplicaPrimaryOnly(groupId);
+ }
+
+ private static class ReplicaLifecycle {
+ private final IgniteLogger log = Loggers.forClass(ReplicaLifecycle.class);
+
+ final Map<ReplicationGroupId, ReplicaLifecycleContext> replicaContexts = new ConcurrentHashMap<>();
+
+ final Executor replicaStartStopPool;
+
+ final ClockService clockService;
+
+ final PlacementDriver placementDriver;
+
+ final ReplicaManager replicaManager;
+
+ String localNodeId;
+
+ ReplicaLifecycle(Executor replicaStartStopPool, ClockService clockService, PlacementDriver placementDriver,
+ ReplicaManager replicaManager) {
+ this.replicaStartStopPool = replicaStartStopPool;
+ this.clockService = clockService;
+ this.placementDriver = placementDriver;
+ this.replicaManager = replicaManager;
+ }
+
+ void start(String localNodeId) {
+ this.localNodeId = localNodeId;
+ placementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_ELECTED, this::onPrimaryElected);
+ placementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_EXPIRED, this::onPrimaryExpired);
+ }
+
+ private CompletableFuture<Boolean> onPrimaryElected(PrimaryReplicaEventParameters parameters) {
+ ReplicaLifecycleContext context = getContext(parameters.groupId());
+
+ synchronized (context) {
+ if (localNodeId.equals(parameters.leaseholderId())) {
+ assert context.replicaState != ReplicaState.STOPPED : "Unexpected primary replica state STOPPED [groupId="
+ + parameters.groupId() + "].";
+
+ context.reservedForPrimary = true;
Review Comment:
Hmm, is it possible to get onPrimaryElected if we are not in assignments? If it is, I'd say that we should update the state to either PRIMARY_ONLY or ASSIGNED.
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1005,6 +1021,49 @@ public boolean isReplicaStarted(ReplicationGroupId replicaGrpId) {
return replicaFuture != null && isCompletedSuccessfully(replicaFuture);
}
+ /**
+ * Can possibly start replica if it's not running or is stopping.
+ *
+ * @param groupId Group id.
+ * @param startOperation Replica start operation.
+ * @param forcedAssignments Assignments to reset forcibly, if needed. Assignments reset is only available when replica is started.
+ */
+ public CompletableFuture<Boolean> weakReplicaStart(
+ ReplicationGroupId groupId,
+ Supplier<CompletableFuture<Boolean>> startOperation,
+ @Nullable Assignments forcedAssignments
+ ) {
+ return replicaLifecycle.weakReplicaStart(groupId, startOperation, forcedAssignments);
+ }
+
+ /**
+ * Can possibly stop replica if it is running or starting, and is not a primary replica. Relies on the given reason. If
Review Comment:
Basically I'd expect similar javadoc in weakStart explaining the details.
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1028,4 +1087,357 @@ public Set<ReplicationGroupId> startedGroups() {
.map(Entry::getKey)
.collect(toSet());
}
+
+ @TestOnly
+ public boolean isReplicaPrimaryOnly(ReplicationGroupId groupId) {
+ return replicaLifecycle.isReplicaPrimaryOnly(groupId);
+ }
+
+ private static class ReplicaLifecycle {
+ private final IgniteLogger log = Loggers.forClass(ReplicaLifecycle.class);
+
+ final Map<ReplicationGroupId, ReplicaLifecycleContext> replicaContexts = new ConcurrentHashMap<>();
+
+ final Executor replicaStartStopPool;
+
+ final ClockService clockService;
+
+ final PlacementDriver placementDriver;
+
+ final ReplicaManager replicaManager;
+
+ String localNodeId;
+
+ ReplicaLifecycle(Executor replicaStartStopPool, ClockService clockService, PlacementDriver placementDriver,
+ ReplicaManager replicaManager) {
+ this.replicaStartStopPool = replicaStartStopPool;
+ this.clockService = clockService;
+ this.placementDriver = placementDriver;
+ this.replicaManager = replicaManager;
+ }
+
+ void start(String localNodeId) {
+ this.localNodeId = localNodeId;
+ placementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_ELECTED, this::onPrimaryElected);
+ placementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_EXPIRED, this::onPrimaryExpired);
+ }
+
+ private CompletableFuture<Boolean> onPrimaryElected(PrimaryReplicaEventParameters parameters) {
+ ReplicaLifecycleContext context = getContext(parameters.groupId());
+
+ synchronized (context) {
+ if (localNodeId.equals(parameters.leaseholderId())) {
+ assert context.replicaState != ReplicaState.STOPPED : "Unexpected primary replica state STOPPED [groupId="
+ + parameters.groupId() + "].";
+
+ context.reservedForPrimary = true;
+ } else {
+ context.reservedForPrimary = false;
+
+ if (context.replicaState == ReplicaState.PRIMARY_ONLY) {
+ stopReplica(parameters.groupId(), context, context.deferredStopOperation);
+ }
+ }
+ }
+
+ return falseCompletedFuture();
+ }
+
+ private CompletableFuture<Boolean> onPrimaryExpired(PrimaryReplicaEventParameters parameters) {
+ if (localNodeId.equals(parameters.leaseholderId())) {
Review Comment:
As mentioned above, I'd expect to see status update here.
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22202 Primary replica reservation for nodes that are excluded from assignments [ignite-3]
Posted by "denis-chudov (via GitHub)" <gi...@apache.org>.
denis-chudov commented on code in PR #3881:
URL: https://github.com/apache/ignite-3/pull/3881#discussion_r1638038257
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1005,6 +1021,49 @@ public boolean isReplicaStarted(ReplicationGroupId replicaGrpId) {
return replicaFuture != null && isCompletedSuccessfully(replicaFuture);
}
+ /**
+ * Can possibly start replica if it's not running or is stopping.
Review Comment:
done
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22202 Primary replica reservation for nodes that are excluded from assignments [ignite-3]
Posted by "denis-chudov (via GitHub)" <gi...@apache.org>.
denis-chudov commented on code in PR #3881:
URL: https://github.com/apache/ignite-3/pull/3881#discussion_r1638048804
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1005,6 +1021,49 @@ public boolean isReplicaStarted(ReplicationGroupId replicaGrpId) {
return replicaFuture != null && isCompletedSuccessfully(replicaFuture);
}
+ /**
+ * Can possibly start replica if it's not running or is stopping.
+ *
+ * @param groupId Group id.
+ * @param startOperation Replica start operation.
+ * @param forcedAssignments Assignments to reset forcibly, if needed. Assignments reset is only available when replica is started.
+ */
+ public CompletableFuture<Boolean> weakReplicaStart(
+ ReplicationGroupId groupId,
+ Supplier<CompletableFuture<Boolean>> startOperation,
+ @Nullable Assignments forcedAssignments
Review Comment:
it can be added in future when it's needed. Now this would be a method that is never used.
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22202 Primary replica reservation for nodes that are excluded from assignments [ignite-3]
Posted by "denis-chudov (via GitHub)" <gi...@apache.org>.
denis-chudov commented on code in PR #3881:
URL: https://github.com/apache/ignite-3/pull/3881#discussion_r1638050264
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1005,6 +1021,49 @@ public boolean isReplicaStarted(ReplicationGroupId replicaGrpId) {
return replicaFuture != null && isCompletedSuccessfully(replicaFuture);
}
+ /**
+ * Can possibly start replica if it's not running or is stopping.
+ *
+ * @param groupId Group id.
+ * @param startOperation Replica start operation.
+ * @param forcedAssignments Assignments to reset forcibly, if needed. Assignments reset is only available when replica is started.
+ */
+ public CompletableFuture<Boolean> weakReplicaStart(
Review Comment:
done
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-22202 Primary replica reservation for nodes that are excluded from assignments [ignite-3]
Posted by "denis-chudov (via GitHub)" <gi...@apache.org>.
denis-chudov commented on code in PR #3881:
URL: https://github.com/apache/ignite-3/pull/3881#discussion_r1638084224
##########
modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java:
##########
@@ -1028,4 +1087,357 @@ public Set<ReplicationGroupId> startedGroups() {
.map(Entry::getKey)
.collect(toSet());
}
+
+ @TestOnly
+ public boolean isReplicaPrimaryOnly(ReplicationGroupId groupId) {
+ return replicaLifecycle.isReplicaPrimaryOnly(groupId);
+ }
+
+ private static class ReplicaLifecycle {
+ private final IgniteLogger log = Loggers.forClass(ReplicaLifecycle.class);
+
+ final Map<ReplicationGroupId, ReplicaLifecycleContext> replicaContexts = new ConcurrentHashMap<>();
+
+ final Executor replicaStartStopPool;
+
+ final ClockService clockService;
+
+ final PlacementDriver placementDriver;
+
+ final ReplicaManager replicaManager;
+
+ String localNodeId;
+
+ ReplicaLifecycle(Executor replicaStartStopPool, ClockService clockService, PlacementDriver placementDriver,
+ ReplicaManager replicaManager) {
+ this.replicaStartStopPool = replicaStartStopPool;
+ this.clockService = clockService;
+ this.placementDriver = placementDriver;
+ this.replicaManager = replicaManager;
+ }
+
+ void start(String localNodeId) {
+ this.localNodeId = localNodeId;
+ placementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_ELECTED, this::onPrimaryElected);
+ placementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_EXPIRED, this::onPrimaryExpired);
+ }
+
+ private CompletableFuture<Boolean> onPrimaryElected(PrimaryReplicaEventParameters parameters) {
+ ReplicaLifecycleContext context = getContext(parameters.groupId());
+
+ synchronized (context) {
+ if (localNodeId.equals(parameters.leaseholderId())) {
+ assert context.replicaState != ReplicaState.STOPPED : "Unexpected primary replica state STOPPED [groupId="
+ + parameters.groupId() + "].";
+
+ context.reservedForPrimary = true;
+ } else {
+ context.reservedForPrimary = false;
Review Comment:
I've thought about it. It would require the split of ASSIGNED state to something like ASSIGNED and ASSIGNED_PRIMARY, this would complicate the transition between states and adding the state transition logic to #onPrimaryElected, #onPrimaryExpired and #reserveReplica method. Now the replica states reflect it's lifecycle, and `reservedForPrimary` exists in parallel with that, it seems to be more simple and intuitive.
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org