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/06/06 14:07:22 UTC
(ignite-3) branch main updated: IGNITE-25597 Add new events to PartitionReplicaLifecycleManager (#5991)
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 5bacbe86377 IGNITE-25597 Add new events to PartitionReplicaLifecycleManager (#5991)
5bacbe86377 is described below
commit 5bacbe863770e87c9b1a13b33394e567bf438637
Author: Alexander Polovtcev <al...@gmail.com>
AuthorDate: Fri Jun 6 17:07:16 2025 +0300
IGNITE-25597 Add new events to PartitionReplicaLifecycleManager (#5991)
---
.../replicator/ItReplicaLifecycleTest.java | 2 +
.../replicator/LocalPartitionReplicaEvent.java | 14 ++-
.../PartitionReplicaLifecycleManager.java | 71 ++++++-------
.../PartitionReplicaLifecycleManagerTest.java | 112 ++++++++++++++++-----
4 files changed, 139 insertions(+), 60 deletions(-)
diff --git a/modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/ItReplicaLifecycleTest.java b/modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/ItReplicaLifecycleTest.java
index d2c5a309762..09e02102f67 100644
--- a/modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/ItReplicaLifecycleTest.java
+++ b/modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/ItReplicaLifecycleTest.java
@@ -26,6 +26,7 @@ import static org.apache.ignite.internal.distributionzones.rebalance.ZoneRebalan
import static org.apache.ignite.internal.lang.IgniteSystemProperties.COLOCATION_FEATURE_FLAG;
import static org.apache.ignite.internal.lang.IgniteSystemProperties.enabledColocation;
import static org.apache.ignite.internal.partition.replicator.LocalPartitionReplicaEvent.AFTER_REPLICA_STOPPED;
+import static org.apache.ignite.internal.partition.replicator.LocalPartitionReplicaEvent.BEFORE_REPLICA_STOPPED;
import static org.apache.ignite.internal.partitiondistribution.PartitionDistributionUtils.calculateAssignmentForPartition;
import static org.apache.ignite.internal.sql.SqlCommon.DEFAULT_SCHEMA_NAME;
import static org.apache.ignite.internal.table.TableTestUtils.dropTable;
@@ -752,6 +753,7 @@ public class ItReplicaLifecycleTest extends ItAbstractColocationTest {
assertThat(
node.partitionReplicaLifecycleManager.stopPartitionInternal(
zonePartitionId,
+ BEFORE_REPLICA_STOPPED,
AFTER_REPLICA_STOPPED,
-1L,
replicaWasStopped -> {}
diff --git a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/LocalPartitionReplicaEvent.java b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/LocalPartitionReplicaEvent.java
index f4e39bce659..43ec2ef0212 100644
--- a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/LocalPartitionReplicaEvent.java
+++ b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/LocalPartitionReplicaEvent.java
@@ -29,12 +29,22 @@ public enum LocalPartitionReplicaEvent implements Event {
BEFORE_REPLICA_STARTED,
/**
- * Fired when partition replica has just been stopped and the related partition shouldn't be destroyed.
+ * Fired before a partition replica is stopped.
+ */
+ BEFORE_REPLICA_STOPPED,
+
+ /**
+ * Fired after a partition replica has been stopped (but not destroyed).
*/
AFTER_REPLICA_STOPPED,
/**
- * Fired when partition replica has been destroyed.
+ * Fired before a partition replica is stopped and destroyed ({@link #BEFORE_REPLICA_STOPPED} will not be produced in this case).
+ */
+ BEFORE_REPLICA_DESTROYED,
+
+ /**
+ * Fired after a partition replica has been stopped and destroyed ({@link #AFTER_REPLICA_STOPPED} will not be produced in this case).
*/
AFTER_REPLICA_DESTROYED
}
diff --git a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/PartitionReplicaLifecycleManager.java b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/PartitionReplicaLifecycleManager.java
index e7f1d22561a..a735bb055f6 100644
--- a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/PartitionReplicaLifecycleManager.java
+++ b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/PartitionReplicaLifecycleManager.java
@@ -48,7 +48,10 @@ import static org.apache.ignite.internal.hlc.HybridTimestamp.nullableHybridTimes
import static org.apache.ignite.internal.lang.IgniteSystemProperties.enabledColocation;
import static org.apache.ignite.internal.metastorage.dsl.Conditions.notExists;
import static org.apache.ignite.internal.metastorage.dsl.Operations.put;
+import static org.apache.ignite.internal.partition.replicator.LocalPartitionReplicaEvent.AFTER_REPLICA_DESTROYED;
import static org.apache.ignite.internal.partition.replicator.LocalPartitionReplicaEvent.AFTER_REPLICA_STOPPED;
+import static org.apache.ignite.internal.partition.replicator.LocalPartitionReplicaEvent.BEFORE_REPLICA_DESTROYED;
+import static org.apache.ignite.internal.partition.replicator.LocalPartitionReplicaEvent.BEFORE_REPLICA_STOPPED;
import static org.apache.ignite.internal.partitiondistribution.Assignments.assignmentListToString;
import static org.apache.ignite.internal.partitiondistribution.PartitionDistributionUtils.calculateAssignmentForPartition;
import static org.apache.ignite.internal.partitiondistribution.PartitionDistributionUtils.calculateAssignments;
@@ -648,8 +651,8 @@ public class PartitionReplicaLifecycleManager extends
}
/**
- * Start a replica for the corresponding {@code zoneId} and {@code partId} on the local node if {@code localAssignment} is not
- * null, meaning that the local node is part of the assignment.
+ * Start a replica for the corresponding {@code zoneId} and {@code partId} on the local node if {@code localAssignment} is not null,
+ * meaning that the local node is part of the assignment.
*
* @param zonePartitionId Zone Partition ID.
* @param localAssignment Assignment of the local member, or null if local member is not part of the assignment.
@@ -1197,7 +1200,7 @@ public class PartitionReplicaLifecycleManager extends
return replicaMgr.weakStopReplica(
zonePartitionId,
WeakReplicaStopReason.RESTART,
- () -> stopPartitionInternal(zonePartitionId, AFTER_REPLICA_STOPPED, revision, replica -> {})
+ () -> stopPartitionInternal(zonePartitionId, BEFORE_REPLICA_STOPPED, AFTER_REPLICA_STOPPED, revision, replica -> {})
);
}
@@ -1576,47 +1579,41 @@ public class PartitionReplicaLifecycleManager extends
}
/**
- * Stops zone replica, executes a given action and fires a given local event after.
+ * Stops zone replica, executes a given action and fires given local events.
*
- * @param zonePartitionId Zone's replication group identifier.
- * @param afterReplicaStopAction A consumer that will be executed if it is not null after zone replica stop process will be finished and
- * stopping result will be given to the consumer.
- * @param afterReplicaStoppedEvent A local event type that if not null should be fired in the end of the method.
- * @param afterReplicaStoppedEventRevision A revision parameter of the local event if the last is presented. Must not be null if the
- * event isn't null too.
- *
- * @return Future that will be completed after zone replica was stopped and all given non-null actions are done, the future's result
- * answers was replica was stopped correctly or not.
+ * @return Future that will be completed after zone replica was stopped and event handling was finished.
*/
@VisibleForTesting
public CompletableFuture<Void> stopPartitionInternal(
ZonePartitionId zonePartitionId,
+ LocalPartitionReplicaEvent beforeReplicaStoppedEvent,
LocalPartitionReplicaEvent afterReplicaStoppedEvent,
- long afterReplicaStoppedEventRevision,
+ long eventRevision,
Consumer<Boolean> afterReplicaStopAction
) {
+ // Not using the busy lock here, because this method is called on component stop.
return executeUnderZoneWriteLock(zonePartitionId.zoneId(), () -> {
- try {
- return replicaMgr.stopReplica(zonePartitionId)
- .thenCompose(replicaWasStopped -> {
- afterReplicaStopAction.accept(replicaWasStopped);
+ var eventParameters = new LocalPartitionReplicaEventParameters(zonePartitionId, eventRevision, false);
- if (!replicaWasStopped) {
- return nullCompletedFuture();
- }
+ return fireEvent(beforeReplicaStoppedEvent, eventParameters)
+ .thenCompose(v -> {
+ try {
+ return replicaMgr.stopReplica(zonePartitionId)
+ .thenCompose(replicaWasStopped -> {
+ afterReplicaStopAction.accept(replicaWasStopped);
- replicationGroupIds.remove(zonePartitionId);
+ if (!replicaWasStopped) {
+ return nullCompletedFuture();
+ }
- assert afterReplicaStoppedEvent != null;
+ replicationGroupIds.remove(zonePartitionId);
- return fireEvent(
- afterReplicaStoppedEvent,
- new LocalPartitionReplicaEventParameters(zonePartitionId, afterReplicaStoppedEventRevision, false)
- );
- });
- } catch (NodeStoppingException e) {
- return nullCompletedFuture();
- }
+ return fireEvent(afterReplicaStoppedEvent, eventParameters);
+ });
+ } catch (NodeStoppingException e) {
+ return nullCompletedFuture();
+ }
+ });
});
}
@@ -1627,7 +1624,13 @@ public class PartitionReplicaLifecycleManager extends
*/
private void cleanUpPartitionsResources(Set<ZonePartitionId> partitionIds) {
CompletableFuture<?>[] stopPartitionsFuture = partitionIds.stream()
- .map(zonePartitionId -> stopPartitionInternal(zonePartitionId, AFTER_REPLICA_STOPPED, -1L, replicaWasStopped -> {}))
+ .map(zonePartitionId -> stopPartitionInternal(
+ zonePartitionId,
+ BEFORE_REPLICA_STOPPED,
+ AFTER_REPLICA_STOPPED,
+ -1L,
+ replicaWasStopped -> {}
+ ))
.toArray(CompletableFuture[]::new);
try {
@@ -1703,7 +1706,6 @@ public class PartitionReplicaLifecycleManager extends
*
* @param zonePartitionId Zone partition id.
* @param tableId Table's identifier.
- *
* @return A future that gets completed after all resources have been unloaded.
*/
public CompletableFuture<Void> unloadTableResourcesFromZoneReplica(ZonePartitionId zonePartitionId, int tableId) {
@@ -1781,7 +1783,8 @@ public class PartitionReplicaLifecycleManager extends
private CompletableFuture<Void> stopAndDestroyPartition(ZonePartitionId zonePartitionId, long revision) {
return stopPartitionInternal(
zonePartitionId,
- LocalPartitionReplicaEvent.AFTER_REPLICA_DESTROYED,
+ BEFORE_REPLICA_DESTROYED,
+ AFTER_REPLICA_DESTROYED,
revision,
replicaWasStopped -> {
if (replicaWasStopped) {
diff --git a/modules/partition-replicator/src/test/java/org/apache/ignite/internal/partition/replicator/PartitionReplicaLifecycleManagerTest.java b/modules/partition-replicator/src/test/java/org/apache/ignite/internal/partition/replicator/PartitionReplicaLifecycleManagerTest.java
index 524b16a9909..14f79fcc5a1 100644
--- a/modules/partition-replicator/src/test/java/org/apache/ignite/internal/partition/replicator/PartitionReplicaLifecycleManagerTest.java
+++ b/modules/partition-replicator/src/test/java/org/apache/ignite/internal/partition/replicator/PartitionReplicaLifecycleManagerTest.java
@@ -23,21 +23,31 @@ import static org.apache.ignite.internal.catalog.CatalogTestUtils.TEST_DELAY_DUR
import static org.apache.ignite.internal.distributionzones.rebalance.ZoneRebalanceUtil.stablePartAssignmentsKey;
import static org.apache.ignite.internal.lang.IgniteSystemProperties.COLOCATION_FEATURE_FLAG;
import static org.apache.ignite.internal.lang.IgniteSystemProperties.THREAD_ASSERTIONS_ENABLED;
+import static org.apache.ignite.internal.partition.replicator.LocalPartitionReplicaEvent.AFTER_REPLICA_DESTROYED;
+import static org.apache.ignite.internal.partition.replicator.LocalPartitionReplicaEvent.AFTER_REPLICA_STOPPED;
+import static org.apache.ignite.internal.partition.replicator.LocalPartitionReplicaEvent.BEFORE_REPLICA_DESTROYED;
+import static org.apache.ignite.internal.partition.replicator.LocalPartitionReplicaEvent.BEFORE_REPLICA_STOPPED;
import static org.apache.ignite.internal.testframework.IgniteTestUtils.testNodeName;
+import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willBe;
import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
import static org.apache.ignite.internal.util.CompletableFutures.emptySetCompletedFuture;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+import static org.apache.ignite.internal.util.IgniteUtils.startAsync;
+import static org.apache.ignite.internal.util.IgniteUtils.stopAsync;
import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Answers.RETURNS_DEEP_STUBS;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.Mockito.inOrder;
+import static org.mockito.Mockito.lenient;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.timeout;
import static org.mockito.Mockito.when;
import java.util.List;
import java.util.Set;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
import org.apache.ignite.internal.catalog.CatalogManager;
@@ -48,8 +58,11 @@ import org.apache.ignite.internal.configuration.SystemDistributedConfiguration;
import org.apache.ignite.internal.configuration.testframework.ConfigurationExtension;
import org.apache.ignite.internal.configuration.testframework.InjectConfiguration;
import org.apache.ignite.internal.distributionzones.DistributionZoneManager;
+import org.apache.ignite.internal.event.EventListener;
import org.apache.ignite.internal.failure.FailureManager;
import org.apache.ignite.internal.hlc.ClockService;
+import org.apache.ignite.internal.hlc.HybridClock;
+import org.apache.ignite.internal.hlc.HybridClockImpl;
import org.apache.ignite.internal.lang.NodeStoppingException;
import org.apache.ignite.internal.lowwatermark.LowWatermark;
import org.apache.ignite.internal.manager.ComponentContext;
@@ -80,7 +93,6 @@ import org.apache.ignite.internal.testframework.InjectExecutorService;
import org.apache.ignite.internal.testframework.WithSystemProperty;
import org.apache.ignite.internal.tx.TxManager;
import org.apache.ignite.internal.tx.storage.state.TxStatePartitionStorage;
-import org.apache.ignite.internal.util.IgniteUtils;
import org.apache.ignite.internal.util.PendingComparableValuesTracker;
import org.apache.ignite.network.NetworkAddress;
import org.junit.jupiter.api.AfterEach;
@@ -106,24 +118,14 @@ class PartitionReplicaLifecycleManagerTest extends BaseIgniteAbstractTest {
private PartitionReplicaLifecycleManager partitionReplicaLifecycleManager;
+ private final HybridClock clock = new HybridClockImpl();
+
@Mock
private Loza raftManager;
@Mock
private ZoneResourcesManager zoneResourcesManager;
- @Mock
- private TopologyAwareRaftGroupService topologyAwareRaftGroupService;
-
- @InjectExecutorService
- private ExecutorService executorService;
-
- @InjectExecutorService
- private ScheduledExecutorService scheduledExecutorService;
-
- @InjectConfiguration
- private SystemDistributedConfiguration systemDistributedConfiguration;
-
@BeforeEach
void setUp(
TestInfo testInfo,
@@ -132,6 +134,7 @@ class PartitionReplicaLifecycleManagerTest extends BaseIgniteAbstractTest {
@Mock LowWatermark lowWatermark,
@Mock ClockService clockService,
@Mock PlacementDriver placementDriver,
+ @Mock TopologyAwareRaftGroupService topologyAwareRaftGroupService,
@Mock SchemaSyncService schemaSyncService,
@Mock TxManager txManager,
@Mock SchemaManager schemaManager,
@@ -142,14 +145,18 @@ class PartitionReplicaLifecycleManagerTest extends BaseIgniteAbstractTest {
@Mock PartitionSnapshotStorage partitionSnapshotStorage,
@Mock TxStatePartitionStorage txStatePartitionStorage,
@Mock ZonePartitionRaftListener raftGroupListener,
- @Mock DataStorageManager dataStorageManager
+ @Mock DataStorageManager dataStorageManager,
+ @InjectExecutorService ExecutorService executorService,
+ @InjectExecutorService ScheduledExecutorService scheduledExecutorService,
+ @InjectConfiguration SystemDistributedConfiguration systemDistributedConfiguration
+
) throws NodeStoppingException {
String nodeName = testNodeName(testInfo, 0);
when(clusterService.topologyService().localMember())
.thenReturn(new ClusterNodeImpl(randomUUID(), nodeName, new NetworkAddress("localhost", 0)));
- when(placementDriver.getPrimaryReplica(any(), any())).thenReturn(nullCompletedFuture());
+ lenient().when(placementDriver.getPrimaryReplica(any(), any())).thenReturn(nullCompletedFuture());
when(cmgManager.metaStorageNodes()).thenReturn(emptySetCompletedFuture());
@@ -166,6 +173,8 @@ class PartitionReplicaLifecycleManagerTest extends BaseIgniteAbstractTest {
when(raftManager.startRaftGroupNode(any(), any(), any(), any(), any(RaftGroupOptions.class), any()))
.thenReturn(topologyAwareRaftGroupService);
+ lenient().when(schemaSyncService.waitForMetadataCompleteness(any())).thenReturn(nullCompletedFuture());
+
metaStorageManager = StandaloneMetaStorageManager.create();
catalogManager = new CatalogManagerImpl(
@@ -217,15 +226,13 @@ class PartitionReplicaLifecycleManagerTest extends BaseIgniteAbstractTest {
var componentContext = new ComponentContext();
- assertThat(metaStorageManager.startAsync(componentContext), willCompleteSuccessfully());
- assertThat(metaStorageManager.recoveryFinishedFuture(), willCompleteSuccessfully());
-
- assertThat(catalogManager.startAsync(componentContext), willCompleteSuccessfully());
- assertThat(replicaManager.startAsync(componentContext), willCompleteSuccessfully());
- assertThat(partitionReplicaLifecycleManager.startAsync(componentContext), willCompleteSuccessfully());
+ CompletableFuture<Void> startFuture = metaStorageManager.startAsync(componentContext)
+ .thenCompose(v -> metaStorageManager.recoveryFinishedFuture())
+ .thenCompose(v -> startAsync(componentContext, catalogManager, replicaManager, partitionReplicaLifecycleManager))
+ .thenCompose(v -> metaStorageManager.deployWatches())
+ .thenCompose(v -> catalogManager.catalogInitializationFuture());
- assertThat(metaStorageManager.deployWatches(), willCompleteSuccessfully());
- assertThat(catalogManager.catalogInitializationFuture(), willCompleteSuccessfully());
+ assertThat(startFuture, willCompleteSuccessfully());
}
@AfterEach
@@ -234,7 +241,7 @@ class PartitionReplicaLifecycleManagerTest extends BaseIgniteAbstractTest {
components.forEach(IgniteComponent::beforeNodeStop);
- IgniteUtils.stopAsync(new ComponentContext(), components);
+ assertThat(stopAsync(new ComponentContext(), components), willCompleteSuccessfully());
}
/**
@@ -260,4 +267,61 @@ class PartitionReplicaLifecycleManagerTest extends BaseIgniteAbstractTest {
inOrder.verify(zoneResourcesManager, timeout(1_000)).destroyZonePartitionResources(zonePartitionId);
inOrder.verify(replicaManager, timeout(1_000)).destroyReplicationProtocolStorages(zonePartitionId, false);
}
+
+ @Test
+ void producesEventsOnPartitionRestart() {
+ var beforeReplicaStoppedFuture = new CompletableFuture<LocalPartitionReplicaEventParameters>();
+ var afterReplicaStoppedFuture = new CompletableFuture<LocalPartitionReplicaEventParameters>();
+
+ partitionReplicaLifecycleManager.listen(BEFORE_REPLICA_STOPPED, EventListener.fromConsumer(beforeReplicaStoppedFuture::complete));
+ partitionReplicaLifecycleManager.listen(AFTER_REPLICA_STOPPED, EventListener.fromConsumer(params -> {
+ assertTrue(beforeReplicaStoppedFuture.isDone(), "AFTER_REPLICA_STOPPED event received before BEFORE_REPLICA_STOPPED");
+
+ afterReplicaStoppedFuture.complete(params);
+ }));
+
+ int zoneId = catalogManager.catalog(catalogManager.latestCatalogVersion()).defaultZone().id();
+
+ var zonePartitionId = new ZonePartitionId(zoneId, 0);
+
+ assertThat(
+ partitionReplicaLifecycleManager.restartPartition(zonePartitionId, Long.MAX_VALUE, clock.nowLong()),
+ willCompleteSuccessfully()
+ );
+
+ assertThat(beforeReplicaStoppedFuture.thenApply(LocalPartitionReplicaEventParameters::zonePartitionId), willBe(zonePartitionId));
+ assertThat(afterReplicaStoppedFuture.thenApply(LocalPartitionReplicaEventParameters::zonePartitionId), willBe(zonePartitionId));
+ }
+
+ @Test
+ void producesEventsOnPartitionDestroy() {
+ var beforeReplicaDestroyedFuture = new CompletableFuture<LocalPartitionReplicaEventParameters>();
+ var afterReplicaDestroyedFuture = new CompletableFuture<LocalPartitionReplicaEventParameters>();
+
+ partitionReplicaLifecycleManager.listen(
+ BEFORE_REPLICA_DESTROYED,
+ EventListener.fromConsumer(beforeReplicaDestroyedFuture::complete)
+ );
+
+ partitionReplicaLifecycleManager.listen(AFTER_REPLICA_DESTROYED, EventListener.fromConsumer(params -> {
+ assertTrue(beforeReplicaDestroyedFuture.isDone(), "AFTER_REPLICA_DESTROYED event received before BEFORE_REPLICA_DESTROYED");
+
+ afterReplicaDestroyedFuture.complete(params);
+ }));
+
+ int zoneId = catalogManager.catalog(catalogManager.latestCatalogVersion()).defaultZone().id();
+
+ var zonePartitionId = new ZonePartitionId(zoneId, 0);
+
+ // Put empty stable assignments to force the replica manager to stop the running replicas.
+ Assignments assignments = Assignments.of(Set.of(), 1);
+
+ assertThat(
+ metaStorageManager.put(stablePartAssignmentsKey(zonePartitionId), assignments.toBytes()),
+ willCompleteSuccessfully()
+ );
+
+ assertThat(beforeReplicaDestroyedFuture.thenApply(LocalPartitionReplicaEventParameters::zonePartitionId), willBe(zonePartitionId));
+ assertThat(afterReplicaDestroyedFuture.thenApply(LocalPartitionReplicaEventParameters::zonePartitionId), willBe(zonePartitionId));
+ }
}