You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by vp...@apache.org on 2025/06/04 19:58:51 UTC
(ignite-3) branch main updated: IGNITE-25434 Fixed critical system error on processPlacementDriverMessage (#5957)
This is an automated email from the ASF dual-hosted git repository.
vpyatkov 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 26642304f4d IGNITE-25434 Fixed critical system error on processPlacementDriverMessage (#5957)
26642304f4d is described below
commit 26642304f4d9c34975577e418d118523b01e0f60
Author: Denis Chudov <mo...@gmail.com>
AuthorDate: Wed Jun 4 22:58:46 2025 +0300
IGNITE-25434 Fixed critical system error on processPlacementDriverMessage (#5957)
---
.../PlacementDriverMessageProcessor.java | 13 +++----
.../ignite/internal/replicator/ReplicaManager.java | 2 +
.../ReplicaReservationFailedException.java | 45 ++++++++++++++++++++++
.../internal/replicator/ReplicaStateManager.java | 7 ++--
.../replicator/PlacementDriverReplicaSideTest.java | 15 +++++---
5 files changed, 66 insertions(+), 16 deletions(-)
diff --git a/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/PlacementDriverMessageProcessor.java b/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/PlacementDriverMessageProcessor.java
index df9d90ca172..75701268f8f 100644
--- a/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/PlacementDriverMessageProcessor.java
+++ b/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/PlacementDriverMessageProcessor.java
@@ -28,7 +28,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
-import java.util.function.BiFunction;
+import java.util.function.BiConsumer;
import org.apache.ignite.internal.failure.FailureContext;
import org.apache.ignite.internal.failure.FailureProcessor;
import org.apache.ignite.internal.hlc.ClockService;
@@ -72,7 +72,7 @@ public class PlacementDriverMessageProcessor {
private final ClockService clockService;
- private final BiFunction<ReplicationGroupId, HybridTimestamp, Boolean> replicaReservationClosure;
+ private final BiConsumer<ReplicationGroupId, HybridTimestamp> replicaReservationClosure;
// TODO: IGNITE-20063 Maybe get rid of it
private final Executor executor;
@@ -113,7 +113,7 @@ public class PlacementDriverMessageProcessor {
ClusterNode localNode,
PlacementDriver placementDriver,
ClockService clockService,
- BiFunction<ReplicationGroupId, HybridTimestamp, Boolean> replicaReservationClosure,
+ BiConsumer<ReplicationGroupId, HybridTimestamp> replicaReservationClosure,
Executor executor,
PendingComparableValuesTracker<Long, Void> storageIndexTracker,
TopologyAwareRaftGroupService raftClient,
@@ -150,7 +150,8 @@ public class PlacementDriverMessageProcessor {
TrackerClosedException.class,
TimeoutException.class,
ReplicationException.class,
- ReplicationTimeoutException.class
+ ReplicationTimeoutException.class,
+ ReplicaReservationFailedException.class
)) {
String errorMessage = String.format("Failed to process the lease granted message [msg=%s].", msg);
failureProcessor.process(new FailureContext(e, errorMessage));
@@ -278,9 +279,7 @@ public class PlacementDriverMessageProcessor {
private CompletableFuture<Void> waitForActualState(HybridTimestamp startTime, long expirationTime) {
LOG.info("Waiting for actual storage state, group=" + groupId);
- if (!replicaReservationClosure.apply(groupId, startTime)) {
- throw new IllegalStateException("Replica reservation failed [groupId=" + groupId + ", leaseStartTime=" + startTime + "].");
- }
+ replicaReservationClosure.accept(groupId, startTime);
long timeout = expirationTime - currentTimeMillis();
if (timeout <= 0) {
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 22cd09a300c..e524002b547 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
@@ -840,6 +840,8 @@ public class ReplicaManager extends AbstractEventProducer<LocalReplicaEvent, Loc
var eventParams = new LocalReplicaEventParameters(replicaGrpId);
+ LOG.info("Replica is stopping [replicationGroupId={}].", replicaGrpId);
+
fireEvent(BEFORE_REPLICA_STOPPED, eventParams).whenComplete((v, e) -> {
if (e != null) {
failureProcessor.process(new FailureContext(e, "Error when notifying about BEFORE_REPLICA_STOPPED event."));
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
new file mode 100644
index 00000000000..cc5708701c4
--- /dev/null
+++ b/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/ReplicaReservationFailedException.java
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+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;
+
+ /**
+ * Constructor.
+ *
+ * @param groupId Replication group ID.
+ * @param leaseStartTime Lease start time.
+ * @param currentReplicaState Current state of the replica.
+ */
+ public ReplicaReservationFailedException(ReplicationGroupId groupId, HybridTimestamp leaseStartTime, String currentReplicaState) {
+ super(format(
+ "Replica reservation failed [groupId={}, leaseStartTime={}, currentReplicaState={}].",
+ groupId,
+ leaseStartTime,
+ currentReplicaState
+ ));
+ }
+}
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 7b91f8d9fb6..602f8d6e830 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
@@ -375,9 +375,8 @@ class ReplicaStateManager {
* Reserve replica as primary.
*
* @param groupId Group id.
- * @return Whether the replica was successfully reserved.
*/
- boolean reserveReplica(ReplicationGroupId groupId, HybridTimestamp leaseStartTime) {
+ void reserveReplica(ReplicationGroupId groupId, HybridTimestamp leaseStartTime) {
ReplicaStateContext context = getContext(groupId);
synchronized (context) {
@@ -395,7 +394,9 @@ class ReplicaStateManager {
context.reserve(groupId, leaseStartTime);
}
- return context.reservedForPrimary;
+ if (!context.reservedForPrimary) {
+ throw new ReplicaReservationFailedException(groupId, leaseStartTime, state.name());
+ }
}
}
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 33fe0efbb7a..728a33a4e7e 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,6 +21,7 @@ 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;
@@ -55,7 +56,6 @@ import org.apache.ignite.internal.raft.client.TopologyAwareRaftGroupService;
import org.apache.ignite.internal.replicator.listener.ReplicaListener;
import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
import org.apache.ignite.internal.testframework.failure.FailureManagerExtension;
-import org.apache.ignite.internal.testframework.failure.MuteFailureManagerLogging;
import org.apache.ignite.internal.thread.NamedThreadFactory;
import org.apache.ignite.internal.util.IgniteUtils;
import org.apache.ignite.internal.util.PendingComparableValuesTracker;
@@ -90,7 +90,7 @@ public class PlacementDriverReplicaSideTest extends BaseIgniteAbstractTest {
private int countOfTimeoutExceptionsOnReadIndexToThrow = 0;
- private boolean reservationSuccess = true;
+ private Runnable reservationClosure;
private final ExecutorService executor = Executors.newSingleThreadExecutor(
NamedThreadFactory.create("common", "replica", log)
@@ -134,7 +134,7 @@ public class PlacementDriverReplicaSideTest extends BaseIgniteAbstractTest {
LOCAL_NODE,
placementDriver,
new TestClockService(new HybridClockImpl()),
- (unused0, unused1) -> reservationSuccess,
+ (unused0, unused1) -> reservationClosure.run(),
executor,
storageIndexTracker,
raftClient,
@@ -158,6 +158,7 @@ public class PlacementDriverReplicaSideTest extends BaseIgniteAbstractTest {
indexOnLeader.set(1L);
currentLeader = null;
countOfTimeoutExceptionsOnReadIndexToThrow = 0;
+ reservationClosure = () -> { };
replica = startReplica();
}
@@ -380,11 +381,13 @@ public class PlacementDriverReplicaSideTest extends BaseIgniteAbstractTest {
}
@Test
- @MuteFailureManagerLogging // Failure is expected.
public void testReservationFailed() {
- reservationSuccess = false;
-
long leaseStartTime = 10;
+
+ reservationClosure = () -> {
+ throw new ReplicaReservationFailedException(GRP_ID, hybridTimestamp(leaseStartTime), "STOPPING");
+ };
+
leaderElection(LOCAL_NODE);
CompletableFuture<LeaseGrantedMessageResponse> respFut = sendLeaseGranted(hts(leaseStartTime), hts(leaseStartTime + 10), false);