IGNITE-23741 Partition storages are destroyed on zone partition stop (#5173)

Author: JAkutenshi

modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/LocalPartitionReplicaEvent.java

@@ -29,7 +29,7 @@ public enum LocalPartitionReplicaEvent implements Event {
     AFTER_REPLICA_STARTED,
 
     /**
-     * Fired when partition replica has stopped.
+     * Fired when partition replica has been destroyed.
      */
-    AFTER_REPLICA_STOPPED
+    AFTER_REPLICA_DESTROYED
 }

modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/PartitionReplicaLifecycleManager.java

@@ -914,16 +914,12 @@ private CompletableFuture<Void> stopAndDestroyPartitionAndUpdateClients(
                 .noneMatch(assignment -> assignment.consistentId().equals(localNode().name()));
 
         if (shouldStopLocalServices) {
-            return clientUpdateFuture.thenCompose(v -> stopAndDestroyPartition(zonePartitionId, revision));
+            return clientUpdateFuture.thenCompose(v -> weakStopAndDestroyPartition(zonePartitionId, revision));
         } else {
             return clientUpdateFuture;
         }
     }
 
-    private CompletableFuture<Void> stopAndDestroyPartition(ZonePartitionId zonePartitionId, long revision) {
-        return weakStopPartition(zonePartitionId, revision);
-    }
-
     private CompletableFuture<Void> handleChangePendingAssignmentEvent(Entry pendingAssignmentsEntry, long revision) {
         if (pendingAssignmentsEntry.value() == null || pendingAssignmentsEntry.empty()) {
             return nullCompletedFuture();
@@ -1249,40 +1245,45 @@ public CompletableFuture<Void> stopAsync(ComponentContext componentContext) {
         return Assignments.fromBytes(entry.value());
     }
 
-    private CompletableFuture<Void> weakStopPartition(ZonePartitionId zonePartitionId, long revision) {
+    private CompletableFuture<Void> weakStopAndDestroyPartition(ZonePartitionId zonePartitionId, long revision) {
         return replicaMgr.weakStopReplica(
                 zonePartitionId,
                 WeakReplicaStopReason.EXCLUDED_FROM_ASSIGNMENTS,
-                () -> stopPartition(zonePartitionId, revision)
+                () -> stopPartition(zonePartitionId).thenCompose(replicaWasStopped -> {
+                    if (!replicaWasStopped) {
+                        return nullCompletedFuture();
+                    }
+
+                    return fireEvent(
+                            LocalPartitionReplicaEvent.AFTER_REPLICA_DESTROYED,
+                            new LocalPartitionReplicaEventParameters(zonePartitionId, revision)
+                    );
+                })
         );
     }
 
     /**
      * Stops all resources associated with a given partition, like replicas and partition trackers.
      *
      * @param zonePartitionId Partition ID.
-     * @return Future that will be completed after all resources have been closed.
+     * @return Future that will be completed after all resources have been closed and the future's result answers was replica was stopped
+     *      correctly or not.
      */
-    private CompletableFuture<Void> stopPartition(ZonePartitionId zonePartitionId, long revision) {
+    private CompletableFuture<Boolean> stopPartition(ZonePartitionId zonePartitionId) {
         return executeUnderZoneWriteLock(zonePartitionId.zoneId(), () -> {
             try {
                 return replicaMgr.stopReplica(zonePartitionId)
-                        .thenCompose((replicaWasStopped) -> {
+                        .thenApply((replicaWasStopped) -> {
                             if (replicaWasStopped) {
                                 zonePartitionRaftListeners.remove(zonePartitionId);
                                 replicationGroupIds.remove(zonePartitionId);
-
-                                return fireEvent(
-                                        LocalPartitionReplicaEvent.AFTER_REPLICA_STOPPED,
-                                        new LocalPartitionReplicaEventParameters(zonePartitionId, revision)
-                                );
-                            } else {
-                                return nullCompletedFuture();
                             }
+
+                            return replicaWasStopped;
                         });
             } catch (NodeStoppingException e) {
                 // No-op.
-                return nullCompletedFuture();
+                return falseCompletedFuture();
             }
         });
     }
@@ -1293,11 +1294,8 @@ private CompletableFuture<Void> stopPartition(ZonePartitionId zonePartitionId, l
      * @param partitionIds Partitions to stop.
      */
     private void cleanUpPartitionsResources(Set<ZonePartitionId> partitionIds) {
-        // TODO: Due to IGNITE-23741 we shouldn't destroy partitions on node stop thus the revision will be removed.
-        long revision = catalogMgr.latestCatalogVersion();
-
         CompletableFuture<?>[] stopPartitionsFuture = partitionIds.stream()
-                .map(partId -> stopPartition(partId, revision))
+                .map(this::stopPartition)
                 .toArray(CompletableFuture[]::new);
 
         try {
@@ -1354,7 +1352,7 @@ public void loadTableListenerToZoneReplica(
         zonePartitionRaftListeners.get(zonePartitionId).addTablePartitionRaftListener(tablePartitionId, tablePartitionRaftListener);
     }
 
-    private CompletableFuture<Void> executeUnderZoneWriteLock(int zoneId, Supplier<CompletableFuture<Void>> action) {
+    private <T> CompletableFuture<T> executeUnderZoneWriteLock(int zoneId, Supplier<CompletableFuture<T>> action) {
         StampedLock lock = zonePartitionsLocks.computeIfAbsent(zoneId, id -> new StampedLock());
 
         long stamp = lock.writeLock();

modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java

@@ -606,8 +606,8 @@ public TableManager(
         );
 
         partitionReplicaLifecycleManager.listen(
-                LocalPartitionReplicaEvent.AFTER_REPLICA_STOPPED,
-                this::onZoneReplicaStopped
+                LocalPartitionReplicaEvent.AFTER_REPLICA_DESTROYED,
+                this::onZoneReplicaDestroyed
         );
 
         rebalanceRetryDelayConfiguration = new SystemDistributedConfigurationPropertyHolder<>(
@@ -718,7 +718,7 @@ private CompletableFuture<Boolean> onZoneReplicaCreated(LocalPartitionReplicaEve
         );
     }
 
-    private CompletableFuture<Boolean> onZoneReplicaStopped(LocalPartitionReplicaEventParameters parameters) {
+    private CompletableFuture<Boolean> onZoneReplicaDestroyed(LocalPartitionReplicaEventParameters parameters) {
         if (!enabledColocationFeature) {
             return falseCompletedFuture();
         }