You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ap...@apache.org on 2025/07/08 10:23:32 UTC
(ignite-3) branch main updated: IGNITE-25842 Propagate exceptions in LeaseTracker.awaitPrimaryReplica (#6198)
This is an automated email from the ASF dual-hosted git repository.
apolovtsev 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 ec76bd222bd IGNITE-25842 Propagate exceptions in LeaseTracker.awaitPrimaryReplica (#6198)
ec76bd222bd is described below
commit ec76bd222bdfd9590372e227430d8b0ee8afd33e
Author: Alexander Polovtcev <al...@gmail.com>
AuthorDate: Tue Jul 8 13:23:27 2025 +0300
IGNITE-25842 Propagate exceptions in LeaseTracker.awaitPrimaryReplica (#6198)
---
.../placementdriver/leases/LeaseTracker.java | 107 ++++++++++++---------
.../internal/placementdriver/LeaseTrackerTest.java | 68 ++++++++++++-
2 files changed, 126 insertions(+), 49 deletions(-)
diff --git a/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/leases/LeaseTracker.java b/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/leases/LeaseTracker.java
index 8bc148daaf7..bb90592d647 100644
--- a/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/leases/LeaseTracker.java
+++ b/modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/leases/LeaseTracker.java
@@ -20,6 +20,7 @@ package org.apache.ignite.internal.placementdriver.leases;
import static java.util.Collections.emptyMap;
import static java.util.concurrent.CompletableFuture.allOf;
import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.concurrent.CompletableFuture.failedFuture;
import static org.apache.ignite.internal.hlc.HybridTimestamp.MIN_VALUE;
import static org.apache.ignite.internal.placementdriver.PlacementDriverManager.PLACEMENTDRIVER_LEASES_KEY;
import static org.apache.ignite.internal.placementdriver.event.PrimaryReplicaEvent.PRIMARY_REPLICA_ELECTED;
@@ -27,16 +28,17 @@ import static org.apache.ignite.internal.placementdriver.event.PrimaryReplicaEve
import static org.apache.ignite.internal.placementdriver.leases.Lease.emptyLease;
import static org.apache.ignite.internal.util.ArrayUtils.BYTE_EMPTY_ARRAY;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+import static org.apache.ignite.internal.util.ExceptionUtils.hasCause;
import static org.apache.ignite.internal.util.IgniteUtils.inBusyLock;
import static org.apache.ignite.internal.util.IgniteUtils.inBusyLockAsync;
import static org.apache.ignite.internal.util.IgniteUtils.newHashMap;
-import static org.apache.ignite.lang.ErrorGroups.Common.NODE_STOPPING_ERR;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
@@ -45,7 +47,6 @@ import java.util.function.Supplier;
import org.apache.ignite.internal.event.AbstractEventProducer;
import org.apache.ignite.internal.hlc.ClockService;
import org.apache.ignite.internal.hlc.HybridTimestamp;
-import org.apache.ignite.internal.lang.IgniteInternalException;
import org.apache.ignite.internal.lang.NodeStoppingException;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
@@ -62,13 +63,13 @@ import org.apache.ignite.internal.placementdriver.event.PrimaryReplicaEvent;
import org.apache.ignite.internal.placementdriver.event.PrimaryReplicaEventParameters;
import org.apache.ignite.internal.replicator.ReplicationGroupId;
import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.internal.util.PendingComparableValuesTracker;
import org.apache.ignite.internal.util.PendingIndependentComparableValuesTracker;
-import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.internal.util.TrackerClosedException;
import org.jetbrains.annotations.Nullable;
/**
- * Class tracks cluster leases in memory.
- * At first, the class state recoveries from Vault, then updates on watch's listener.
+ * Class that tracks cluster leases in memory.
*/
public class LeaseTracker extends AbstractEventProducer<PrimaryReplicaEvent, PrimaryReplicaEventParameters> implements
LeasePlacementDriver {
@@ -135,7 +136,7 @@ public class LeaseTracker extends AbstractEventProducer<PrimaryReplicaEvent, Pri
busyLock.block();
- primaryReplicaWaiters.forEach((groupId, pendingTracker) -> pendingTracker.close());
+ primaryReplicaWaiters.values().forEach(PendingComparableValuesTracker::close);
primaryReplicaWaiters.clear();
msManager.unregisterWatch(updateListener);
@@ -255,28 +256,6 @@ public class LeaseTracker extends AbstractEventProducer<PrimaryReplicaEvent, Pri
}
}
- private void awaitPrimaryReplica(
- ReplicationGroupId groupId,
- HybridTimestamp timestamp,
- CompletableFuture<ReplicaMeta> resultFuture
- ) {
- inBusyLockAsync(busyLock, () -> getOrCreatePrimaryReplicaWaiter(groupId).waitFor(timestamp)
- .thenAccept(replicaMeta -> {
- ClusterNode leaseholderNode = clusterNodeResolver.getById(replicaMeta.getLeaseholderId());
-
- if (leaseholderNode == null && !resultFuture.isDone()) {
- awaitPrimaryReplica(
- groupId,
- replicaMeta.getExpirationTime().tick(),
- resultFuture
- );
- } else {
- resultFuture.complete(replicaMeta);
- }
- })
- );
- }
-
@Override
public CompletableFuture<ReplicaMeta> awaitPrimaryReplica(
ReplicationGroupId groupId,
@@ -284,34 +263,70 @@ public class LeaseTracker extends AbstractEventProducer<PrimaryReplicaEvent, Pri
long timeout,
TimeUnit unit
) {
- if (!busyLock.enterBusy()) {
- throw new IgniteInternalException(NODE_STOPPING_ERR, new NodeStoppingException());
- }
- try {
+ return inBusyLockAsync(busyLock, () -> {
ReplicaMeta currentMeta = getCurrentPrimaryReplica(groupId, timestamp);
- if (currentMeta != null && clusterNodeResolver.getById(currentMeta.getLeaseholderId()) != null) {
+ if (isValidReplicaMeta(currentMeta)) {
return completedFuture(currentMeta);
}
- } finally {
- busyLock.leaveBusy();
- }
-
- CompletableFuture<ReplicaMeta> future = new CompletableFuture<>();
- awaitPrimaryReplica(groupId, timestamp, future);
+ return awaitPrimaryReplicaImpl(groupId, timestamp, timeout, unit);
+ });
+ }
- return future
- .orTimeout(timeout, unit)
+ private CompletableFuture<ReplicaMeta> awaitPrimaryReplicaImpl(
+ ReplicationGroupId groupId,
+ HybridTimestamp timestamp,
+ long timeout,
+ TimeUnit unit
+ ) {
+ return awaitPrimaryReplicaImpl(groupId, timestamp, System.nanoTime(), unit.toNanos(timeout))
.exceptionally(e -> {
- if (e instanceof TimeoutException) {
+ if (hasCause(e, TimeoutException.class)) {
throw new PrimaryReplicaAwaitTimeoutException(groupId, timestamp, leases.leaseByGroupId().get(groupId), e);
+ } else if (hasCause(e, TrackerClosedException.class)) {
+ // TrackerClosedException is thrown when trackers are closed on node stop.
+ throw new CompletionException(new NodeStoppingException(e));
+ } else {
+ throw new PrimaryReplicaAwaitException(groupId, timestamp, e);
}
-
- throw new PrimaryReplicaAwaitException(groupId, timestamp, e);
});
}
+ private CompletableFuture<ReplicaMeta> awaitPrimaryReplicaImpl(
+ ReplicationGroupId groupId,
+ HybridTimestamp timestamp,
+ long startNanoTime,
+ long timeoutNanos
+ ) {
+ return inBusyLockAsync(busyLock, () -> {
+ long elapsedNanos = System.nanoTime() - startNanoTime;
+
+ long remainingTimeoutNanos = timeoutNanos - elapsedNanos;
+
+ if (remainingTimeoutNanos <= 0) {
+ return failedFuture(new TimeoutException());
+ }
+
+ return getOrCreatePrimaryReplicaWaiter(groupId)
+ .waitFor(timestamp)
+ .orTimeout(remainingTimeoutNanos, TimeUnit.NANOSECONDS)
+ .thenCompose(replicaMeta -> {
+ if (isValidReplicaMeta(replicaMeta)) {
+ return completedFuture(replicaMeta);
+ }
+
+ return awaitPrimaryReplicaImpl(groupId, replicaMeta.getExpirationTime().tick(), startNanoTime, timeoutNanos);
+ });
+ });
+ }
+
+ private boolean isValidReplicaMeta(@Nullable ReplicaMeta replicaMeta) {
+ UUID leaseholderId = replicaMeta == null ? null : replicaMeta.getLeaseholderId();
+
+ return leaseholderId != null && clusterNodeResolver.getById(leaseholderId) != null;
+ }
+
@Override
public CompletableFuture<ReplicaMeta> getPrimaryReplica(ReplicationGroupId replicationGroupId, HybridTimestamp timestamp) {
return inBusyLockAsync(busyLock, () -> {
@@ -348,8 +363,8 @@ public class LeaseTracker extends AbstractEventProducer<PrimaryReplicaEvent, Pri
}
/**
- * Helper method that checks whether tracker for given groupId is present in {@code primaryReplicaWaiters} map, whether it's empty
- * and removes it if it's true.
+ * Helper method that checks whether tracker for given groupId is present in {@code primaryReplicaWaiters} map, whether it's empty and
+ * removes it if it's true.
*
* @param groupId Replication group id.
*/
diff --git a/modules/placement-driver/src/test/java/org/apache/ignite/internal/placementdriver/LeaseTrackerTest.java b/modules/placement-driver/src/test/java/org/apache/ignite/internal/placementdriver/LeaseTrackerTest.java
index ae9ab93f6a3..dacd273e2d0 100644
--- a/modules/placement-driver/src/test/java/org/apache/ignite/internal/placementdriver/LeaseTrackerTest.java
+++ b/modules/placement-driver/src/test/java/org/apache/ignite/internal/placementdriver/LeaseTrackerTest.java
@@ -22,6 +22,8 @@ import static java.util.UUID.randomUUID;
import static org.apache.ignite.internal.placementdriver.PlacementDriverManager.PLACEMENTDRIVER_LEASES_KEY;
import static org.apache.ignite.internal.placementdriver.event.PrimaryReplicaEvent.PRIMARY_REPLICA_ELECTED;
import static org.apache.ignite.internal.placementdriver.event.PrimaryReplicaEvent.PRIMARY_REPLICA_EXPIRED;
+import static org.apache.ignite.internal.testframework.matchers.CompletableFutureExceptionMatcher.willThrow;
+import static org.apache.ignite.internal.testframework.matchers.CompletableFutureExceptionMatcher.willThrowWithCauseOrSuppressed;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
import static org.apache.ignite.internal.util.CompletableFutures.falseCompletedFuture;
import static org.awaitility.Awaitility.await;
@@ -29,15 +31,18 @@ import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.contains;
import static org.hamcrest.Matchers.hasSize;
import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.mockito.Mockito.mock;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.when;
import java.util.ArrayList;
import java.util.List;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.ignite.internal.hlc.HybridClockImpl;
import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.hlc.TestClockService;
+import org.apache.ignite.internal.lang.NodeStoppingException;
import org.apache.ignite.internal.manager.ComponentContext;
import org.apache.ignite.internal.metastorage.MetaStorageManager;
import org.apache.ignite.internal.metastorage.impl.StandaloneMetaStorageManager;
@@ -52,15 +57,22 @@ import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
/**
* Tests for lease tracker.
*/
+@ExtendWith(MockitoExtension.class)
public class LeaseTrackerTest extends BaseIgniteAbstractTest {
private MetaStorageManager msManager;
private LeaseTracker leaseTracker;
+ @Mock
+ private ClusterNodeResolver clusterNodeResolver;
+
@BeforeEach
void setUp() {
msManager = StandaloneMetaStorageManager.create();
@@ -69,7 +81,7 @@ public class LeaseTrackerTest extends BaseIgniteAbstractTest {
leaseTracker = new LeaseTracker(
msManager,
- mock(ClusterNodeResolver.class),
+ clusterNodeResolver,
new TestClockService(clock)
);
@@ -90,7 +102,7 @@ public class LeaseTrackerTest extends BaseIgniteAbstractTest {
}
@Test
- public void testLeaseCleanup() {
+ void testLeaseCleanup() {
AtomicReference<PrimaryReplicaEventParameters> parametersRef = new AtomicReference<>();
leaseTracker.listen(PRIMARY_REPLICA_EXPIRED, p -> {
@@ -189,4 +201,54 @@ public class LeaseTrackerTest extends BaseIgniteAbstractTest {
assertThat(events, contains(PRIMARY_REPLICA_ELECTED, PRIMARY_REPLICA_EXPIRED, PRIMARY_REPLICA_ELECTED));
}
+
+ @Test
+ void awaitPrimaryReplicaPropagatesExceptions() {
+ when(clusterNodeResolver.getById(any())).thenThrow(new RuntimeException("test"));
+
+ var groupId = new TablePartitionId(0, 0);
+
+ CompletableFuture<?> future = leaseTracker.awaitPrimaryReplica(
+ groupId,
+ HybridTimestamp.MAX_VALUE,
+ 30,
+ TimeUnit.SECONDS
+ );
+
+ var lease = new Lease("test", randomUUID(), new HybridTimestamp(1, 0), HybridTimestamp.MAX_VALUE, groupId)
+ .acceptLease(HybridTimestamp.MAX_VALUE);
+
+ assertThat(
+ msManager.put(PLACEMENTDRIVER_LEASES_KEY, new LeaseBatch(List.of(lease)).bytes()),
+ willCompleteSuccessfully()
+ );
+
+ assertThat(future, willThrowWithCauseOrSuppressed(RuntimeException.class, "test"));
+ }
+
+ @Test
+ void awaitPrimaryReplicaPropagatesExceptionsOnStop() {
+ CompletableFuture<?> future = leaseTracker.awaitPrimaryReplica(
+ new TablePartitionId(0, 0),
+ HybridTimestamp.MAX_VALUE,
+ 30,
+ TimeUnit.SECONDS
+ );
+
+ leaseTracker.stopTrack();
+
+ assertThat(future, willThrow(NodeStoppingException.class));
+ }
+
+ @Test
+ void awaitPrimaryReplicaThrowsOnTimeout() {
+ CompletableFuture<?> future = leaseTracker.awaitPrimaryReplica(
+ new TablePartitionId(0, 0),
+ HybridTimestamp.MAX_VALUE,
+ 1,
+ TimeUnit.MILLISECONDS
+ );
+
+ assertThat(future, willThrow(PrimaryReplicaAwaitTimeoutException.class));
+ }
}