You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sa...@apache.org on 2025/06/20 10:31:16 UTC
(ignite-3) branch main updated: IGNITE-23702 Fix incorrect HB in deferred replica stop on partition restart (#6031)
This is an automated email from the ASF dual-hosted git repository.
sanpwc pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new 0e57a194931 IGNITE-23702 Fix incorrect HB in deferred replica stop on partition restart (#6031)
0e57a194931 is described below
commit 0e57a19493172a7f3091684e52872d588faefa71
Author: Denis Chudov <mo...@gmail.com>
AuthorDate: Fri Jun 20 13:31:10 2025 +0300
IGNITE-23702 Fix incorrect HB in deferred replica stop on partition restart (#6031)
---
.../internal/placementdriver/LeaseUpdater.java | 9 --
.../ignite/internal/replicator/ReplicaManager.java | 1 -
.../ReplicaReservationFailedException.java | 19 +---
.../internal/replicator/ReplicaStateManager.java | 102 ++++++++++++---------
.../replicator/PlacementDriverReplicaSideTest.java | 3 +-
.../internal/table/distributed/TableManager.java | 2 +-
6 files changed, 65 insertions(+), 71 deletions(-)
diff --git a/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/LeaseUpdater.java b/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/LeaseUpdater.java
index bff5bcac2aa..216566d87b2 100644
--- a/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/LeaseUpdater.java
+++ b/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/LeaseUpdater.java
@@ -21,7 +21,6 @@ import static java.util.Objects.hash;
import static java.util.Objects.requireNonNullElse;
import static java.util.concurrent.CompletableFuture.completedFuture;
import static org.apache.ignite.internal.hlc.HybridTimestamp.NULL_HYBRID_TIMESTAMP;
-import static org.apache.ignite.internal.lang.IgniteStringFormatter.format;
import static org.apache.ignite.internal.metastorage.dsl.Conditions.notExists;
import static org.apache.ignite.internal.metastorage.dsl.Conditions.or;
import static org.apache.ignite.internal.metastorage.dsl.Conditions.value;
@@ -475,14 +474,6 @@ public class LeaseUpdater {
if (lease.isProlongable() && agreement.isAccepted()) {
Lease negotiatedLease = agreement.getLease();
- // Lease information is taken from lease tracker, where it appears on meta storage watch updates, so it can
- // contain stale leases, if watch processing was delayed for some reason. It is ok: negotiated lease is
- // guaranteed to be already written to meta storage before negotiation begins, and in this case its start time
- // would be greater than lease's.
- assert negotiatedLease.getStartTime().longValue() >= lease.getStartTime().longValue()
- : format("Can't publish the lease that was not negotiated [groupId={}, startTime={}, "
- + "agreementLeaseStartTime={}].", grpId, lease.getStartTime(), agreement.getLease().getStartTime());
-
publishLease(grpId, negotiatedLease, renewedLeases, leaseExpirationInterval);
continue;
diff --git a/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java b/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java
index e7c58af7cf1..1492ef5a37b 100644
--- a/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java
+++ b/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaManager.java
@@ -280,7 +280,6 @@ public class ReplicaManager extends AbstractEventProducer<LocalReplicaEvent, Loc
this.replicaStateManager = new ReplicaStateManager(
replicaStartStopExecutor,
- clockService,
placementDriver,
this,
failureProcessor
diff --git a/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaReservationFailedException.java b/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaReservationFailedException.java
index cc5708701c4..97c493646a4 100644
--- a/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaReservationFailedException.java
+++ b/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaReservationFailedException.java
@@ -17,29 +17,18 @@
package org.apache.ignite.internal.replicator;
-import static org.apache.ignite.internal.lang.IgniteStringFormatter.format;
-
-import org.apache.ignite.internal.hlc.HybridTimestamp;
-
/**
* Exception thrown when a replica reservation fails.
*/
public class ReplicaReservationFailedException extends RuntimeException {
- public static final long serialVersionUID = 33897490775763932L;
+ private static final long serialVersionUID = 33897490775763932L;
/**
* Constructor.
*
- * @param groupId Replication group ID.
- * @param leaseStartTime Lease start time.
- * @param currentReplicaState Current state of the replica.
+ * @param message Exception message.
*/
- public ReplicaReservationFailedException(ReplicationGroupId groupId, HybridTimestamp leaseStartTime, String currentReplicaState) {
- super(format(
- "Replica reservation failed [groupId={}, leaseStartTime={}, currentReplicaState={}].",
- groupId,
- leaseStartTime,
- currentReplicaState
- ));
+ ReplicaReservationFailedException(String message) {
+ super(message);
}
}
diff --git a/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaStateManager.java b/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaStateManager.java
index 602f8d6e830..da446ed54cb 100644
--- a/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaStateManager.java
+++ b/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaStateManager.java
@@ -33,7 +33,6 @@ import java.util.concurrent.Executor;
import java.util.function.Supplier;
import org.apache.ignite.internal.failure.FailureContext;
import org.apache.ignite.internal.failure.FailureProcessor;
-import org.apache.ignite.internal.hlc.ClockService;
import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.lang.NodeStoppingException;
import org.apache.ignite.internal.logger.IgniteLogger;
@@ -56,8 +55,6 @@ class ReplicaStateManager {
private final Executor replicaStartStopExecutor;
- private final ClockService clockService;
-
private final PlacementDriver placementDriver;
private final ReplicaManager replicaManager;
@@ -70,13 +67,11 @@ class ReplicaStateManager {
ReplicaStateManager(
Executor replicaStartStopExecutor,
- ClockService clockService,
PlacementDriver placementDriver,
ReplicaManager replicaManager,
FailureProcessor failureProcessor
) {
this.replicaStartStopExecutor = replicaStartStopExecutor;
- this.clockService = clockService;
this.placementDriver = placementDriver;
this.replicaManager = replicaManager;
this.failureProcessor = failureProcessor;
@@ -109,16 +104,12 @@ class ReplicaStateManager {
+ ", leaseStartTime=" + parameters.startTime() + ", reservedForPrimary=" + context.reservedForPrimary
+ ", contextLeaseStartTime=" + context.leaseStartTime + "].";
} else if (context.reservedForPrimary) {
- context.assertReservation(replicationGroupId, parameters.startTime());
+ context.assertReservation(replicationGroupId);
// Unreserve if another replica was elected as primary, only if its lease start time is greater,
// otherwise it means that event is too late relatively to lease negotiation start and should be ignored.
if (parameters.startTime().compareTo(context.leaseStartTime) > 0) {
context.releaseReservation();
-
- if (context.replicaState == ReplicaState.PRIMARY_ONLY) {
- executeDeferredReplicaStop(context);
- }
}
}
}
@@ -141,15 +132,13 @@ class ReplicaStateManager {
if (context != null) {
synchronized (context) {
- context.assertReservation(parameters.groupId(), parameters.startTime());
+ if (context.reservedForPrimary) {
+ context.assertReservation(parameters.groupId());
- // Unreserve if primary replica expired, only if its lease start time is equal to reservation time,
- // otherwise it means that event is too late relatively to lease negotiation start and should be ignored.
- if (parameters.startTime().equals(context.leaseStartTime)) {
- context.releaseReservation();
-
- if (context.replicaState == ReplicaState.RESTART_PLANNED) {
- executeDeferredReplicaStop(context);
+ // Unreserve if primary replica expired, only if its lease start time is equal to reservation time,
+ // otherwise it means that event is too late relatively to lease negotiation start and should be ignored.
+ if (parameters.startTime().equals(context.leaseStartTime)) {
+ context.releaseReservation();
}
}
}
@@ -276,7 +265,7 @@ class ReplicaStateManager {
if (context.reservedForPrimary) {
context.replicaState = ReplicaState.PRIMARY_ONLY;
// Intentionally do not return future here: it can freeze the handling of assignment changes.
- planDeferredReplicaStop(groupId, context, null, stopOperation);
+ planDeferredReplicaStop(groupId, context, stopOperation);
} else {
return stopReplica(groupId, context, stopOperation);
}
@@ -292,9 +281,7 @@ class ReplicaStateManager {
// If is primary, turning off the primary first.
context.replicaState = ReplicaState.RESTART_PLANNED;
return replicaManager.stopLeaseProlongation(groupId, null)
- .thenCompose(leaseExpirationTime ->
- planDeferredReplicaStop(groupId, context, leaseExpirationTime, stopOperation)
- );
+ .thenCompose(unused -> planDeferredReplicaStop(groupId, context, stopOperation));
} else {
return stopReplica(groupId, context, stopOperation);
}
@@ -313,11 +300,26 @@ class ReplicaStateManager {
}
}
+ /**
+ * Should be called within {@code synchronized (context)}.
+ *
+ * @param groupId Group id.
+ * @param context Replica state context.
+ * @param stopOperation Replica stop operation.
+ * @return Completable future that completes when the replica is stopped.
+ */
private CompletableFuture<Void> stopReplica(
ReplicationGroupId groupId,
ReplicaStateContext context,
Supplier<CompletableFuture<Void>> stopOperation
) {
+ // These is some probability that the replica would be reserved after the previous lease is expired and before this method
+ // is called, so reservation state needs to be checked again.
+ if (context.reservedForPrimary) {
+ return replicaManager.stopLeaseProlongation(groupId, null)
+ .thenCompose(unused -> planDeferredReplicaStop(groupId, context, stopOperation));
+ }
+
context.replicaState = ReplicaState.STOPPING;
context.previousOperationFuture = context.previousOperationFuture
.handleAsync((v, e) -> stopOperation.get(), replicaStartStopExecutor)
@@ -342,32 +344,25 @@ class ReplicaStateManager {
private CompletableFuture<Void> planDeferredReplicaStop(
ReplicationGroupId groupId,
ReplicaStateContext context,
- @Nullable HybridTimestamp leaseExpirationTime,
Supplier<CompletableFuture<Void>> deferredStopOperation
) {
synchronized (context) {
- // TODO IGNITE-23702: proper sync with waiting of expiration event, and proper deferred stop after cancellation of
- // reservation made by a lease that was not negotiated.
+ LOG.debug("Planning deferred replica stop [groupId={}, reservedForPrimary={}].", groupId, context.reservedForPrimary);
// No parallel actions affected this, continue.
if (context.reservedForPrimary) {
- context.deferredStopReadyFuture = leaseExpirationTime == null
- ? new CompletableFuture<>()
- : clockService.waitFor(leaseExpirationTime);
+ context.deferredStopReadyFuture = new CompletableFuture<>();
} else {
// context.unreserve() has been called in parallel, no need to wait.
context.deferredStopReadyFuture = nullCompletedFuture();
}
return context.deferredStopReadyFuture
- .thenCompose(unused -> stopReplica(groupId, context, deferredStopOperation));
- }
- }
-
- private static void executeDeferredReplicaStop(ReplicaStateContext context) {
- if (context.deferredStopReadyFuture != null) {
- context.deferredStopReadyFuture.complete(null);
- context.deferredStopReadyFuture = null;
+ .thenCompose(unused -> {
+ synchronized (context) {
+ return stopReplica(groupId, context, deferredStopOperation);
+ }
+ });
}
}
@@ -380,6 +375,9 @@ class ReplicaStateManager {
ReplicaStateContext context = getContext(groupId);
synchronized (context) {
+ LOG.debug("Trying to reserve replica [groupId={}, leaseStartTime={}, replicaState={}, reservedForPrimary={}].",
+ groupId, leaseStartTime, context.replicaState, context.reservedForPrimary);
+
ReplicaState state = context.replicaState;
if (state == ReplicaState.STOPPING || state == ReplicaState.STOPPED) {
@@ -390,12 +388,21 @@ class ReplicaStateManager {
if (context.reservedForPrimary) {
throw new AssertionError("Unexpected replica reservation with " + state + " state [groupId=" + groupId + "].");
}
- } else if (state != ReplicaState.RESTART_PLANNED) {
+ } else if (state == ReplicaState.RESTART_PLANNED) {
+ // Explicitly release reservation if the replica is going to restart, otherwise placement driver will always try to reserve
+ // it again before the replica stopping begins.
+ context.releaseReservation();
+ } else {
context.reserve(groupId, leaseStartTime);
}
if (!context.reservedForPrimary) {
- throw new ReplicaReservationFailedException(groupId, leaseStartTime, state.name());
+ throw new ReplicaReservationFailedException(format(
+ "Replica reservation failed [groupId={}, leaseStartTime={}, currentReplicaState={}].",
+ groupId,
+ leaseStartTime,
+ state
+ ));
}
}
}
@@ -511,21 +518,30 @@ class ReplicaStateManager {
// Newer lease may reserve this replica when it's already reserved by older one: this means than older one is no longer
// valid and most likely has not been negotiated and is discarded. By the same reason we shouldn't process the attempt
// of reservation by older lease, which is not likely and means reordering of message handling.
- throw new IllegalArgumentException(format("Replica reservation failed: newer lease has already reserved this replica ["
- + "groupId={}, requestedLeaseStartTime={}, newerLeaseStartTime={}].", groupId, leaseStartTime,
+ throw new ReplicaReservationFailedException(format("Replica reservation failed: newer lease has already reserved this "
+ + "replica [groupId={}, requestedLeaseStartTime={}, newerLeaseStartTime={}].", groupId, leaseStartTime,
this.leaseStartTime));
}
+
+ LOG.debug("Reserving replica [groupId={}, leaseStartTime={}].", groupId, leaseStartTime);
+
this.leaseStartTime = leaseStartTime;
- reservedForPrimary = true;
+ this.reservedForPrimary = true;
}
void releaseReservation() {
- // TODO IGNITE-23702: should also lead to replica stop if it is PRIMARY_ONLY.
reservedForPrimary = false;
leaseStartTime = null;
+
+ // Unblock the deferred replica stop, if needed.
+ if (deferredStopReadyFuture != null
+ && (replicaState == ReplicaState.PRIMARY_ONLY || replicaState == ReplicaState.RESTART_PLANNED)) {
+ deferredStopReadyFuture.complete(null);
+ deferredStopReadyFuture = null;
+ }
}
- void assertReservation(ReplicationGroupId groupId, HybridTimestamp leaseStartTime) {
+ void assertReservation(ReplicationGroupId groupId) {
assert reservedForPrimary : "Replica is elected as primary but not reserved [groupId="
+ groupId + ", leaseStartTime=" + leaseStartTime + "].";
assert leaseStartTime != null : "Replica is reserved but lease start time is null [groupId="
diff --git a/modules/replicator/src/test/java/org/apache/ignite/internal/replicator/PlacementDriverReplicaSideTest.java b/modules/replicator/src/test/java/org/apache/ignite/internal/replicator/PlacementDriverReplicaSideTest.java
index 728a33a4e7e..209e228c166 100644
--- a/modules/replicator/src/test/java/org/apache/ignite/internal/replicator/PlacementDriverReplicaSideTest.java
+++ b/modules/replicator/src/test/java/org/apache/ignite/internal/replicator/PlacementDriverReplicaSideTest.java
@@ -21,7 +21,6 @@ import static java.lang.System.currentTimeMillis;
import static java.util.UUID.randomUUID;
import static java.util.concurrent.CompletableFuture.completedFuture;
import static java.util.concurrent.CompletableFuture.failedFuture;
-import static org.apache.ignite.internal.hlc.HybridTimestamp.hybridTimestamp;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willSucceedFast;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willSucceedIn;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
@@ -385,7 +384,7 @@ public class PlacementDriverReplicaSideTest extends BaseIgniteAbstractTest {
long leaseStartTime = 10;
reservationClosure = () -> {
- throw new ReplicaReservationFailedException(GRP_ID, hybridTimestamp(leaseStartTime), "STOPPING");
+ throw new ReplicaReservationFailedException("Test: replica reservation failed");
};
leaderElection(LOCAL_NODE);
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
index 5877bb51861..01548b5f91b 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
@@ -581,7 +581,7 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
schemaVersions = new SchemaVersionsImpl(executorInclinedSchemaSyncService, catalogService, clockService);
- tablesVv = new IncrementalVersionedValue<>(registry);
+ tablesVv = new IncrementalVersionedValue<>(registry, 100, null);
localPartitionsVv = new IncrementalVersionedValue<>(dependingOn(tablesVv));