IGNITE-29114 Fix chained distributed processes hang when the coordinator leaves between phases

modules/core/src/main/java/org/apache/ignite/internal/managers/encryption/GridEncryptionManager.java

@@ -1537,8 +1537,8 @@ private void finishPrepareMasterKeyChange(UUID id, Map<UUID, Message> res, Map<U
 
             completeMasterKeyChangeFuture(id, err);
         }
-        else if (U.isLocalNodeCoordinator(ctx.discovery()))
-            performMKChangeProc.start(id, masterKeyChangeRequest);
+        else
+            performMKChangeProc.startByCoordinator(id, masterKeyChangeRequest);
     }
 
     /**

modules/core/src/main/java/org/apache/ignite/internal/managers/encryption/GroupKeyChangeProcess.java

@@ -37,7 +37,6 @@
 import org.apache.ignite.internal.util.typedef.T2;
 import org.apache.ignite.internal.util.typedef.internal.CU;
 import org.apache.ignite.internal.util.typedef.internal.S;
-import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.lang.IgniteFuture;
 import org.apache.ignite.lang.IgniteFutureCancelledException;
 import org.apache.ignite.plugin.extensions.communication.Message;
@@ -275,8 +274,8 @@ private void finishPrepare(UUID id, Map<UUID, Message> res, Map<UUID, Throwable>
 
             completeFuture(id, err, fut);
         }
-        else if (U.isLocalNodeCoordinator(ctx.discovery()))
-            performGKChangeProc.start(id, req);
+        else
+            performGKChangeProc.startByCoordinator(id, req);
     }
 
     /**

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotCheckProcess.java

@@ -581,8 +581,7 @@ private void reducePreparationAndMetasCheck(
             if (clusterOpFut != null)
                 ctx.clusterMetas = metas;
 
-            if (U.isLocalNodeCoordinator(kctx.discovery()))
-                phase2PartsHashes.start(reqId, ctx.req);
+            phase2PartsHashes.startByCoordinator(reqId, ctx.req);
         }
         catch (Throwable th) {
             if (ctx != null) {

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotRestoreProcess.java

@@ -860,8 +860,7 @@ private void finishPrepare(UUID reqId, Map<UUID, SnapshotRestoreOperationRespons
 
         opCtx0.cfgs = globalCfgs;
 
-        if (U.isLocalNodeCoordinator(ctx.discovery()))
-            preloadProc.start(reqId, new SnapshotRestoreStartRequest(reqId));
+        preloadProc.startByCoordinator(reqId, new SnapshotRestoreStartRequest(reqId));
     }
 
     /**
@@ -1249,14 +1248,12 @@ private void finishPreload(UUID reqId, Map<UUID, Message> res, Map<UUID, Throwab
         opCtx0.errHnd.accept(failure);
 
         if (failure != null) {
-            if (U.isLocalNodeCoordinator(ctx.discovery()))
-                rollbackRestoreProc.start(reqId, new SnapshotRestoreStartRequest(reqId));
+            rollbackRestoreProc.startByCoordinator(reqId, new SnapshotRestoreStartRequest(reqId));
 
             return;
         }
 
-        if (U.isLocalNodeCoordinator(ctx.discovery()))
-            cacheStartProc.start(reqId, new SnapshotRestoreStartRequest(reqId));
+        cacheStartProc.startByCoordinator(reqId, new SnapshotRestoreStartRequest(reqId));
     }
 
     /**
@@ -1312,8 +1309,7 @@ private void finishCacheStart(UUID reqId, Map<UUID, Message> res, Map<UUID, Thro
 
         if (failure == null) {
             if (opCtx0.incIdx > 0) {
-                if (U.isLocalNodeCoordinator(ctx.discovery()))
-                    incSnpRestoreProc.start(reqId, new SnapshotRestoreStartRequest(reqId));
+                incSnpRestoreProc.startByCoordinator(reqId, new SnapshotRestoreStartRequest(reqId));
 
                 return;
             }
@@ -1325,8 +1321,7 @@ private void finishCacheStart(UUID reqId, Map<UUID, Message> res, Map<UUID, Thro
 
         opCtx0.err.compareAndSet(null, failure);
 
-        if (U.isLocalNodeCoordinator(ctx.discovery()))
-            cacheStopProc.start(reqId, new SnapshotRestoreStartRequest(reqId));
+        cacheStopProc.startByCoordinator(reqId, new SnapshotRestoreStartRequest(reqId));
     }
 
     /**
@@ -1337,7 +1332,7 @@ private IgniteInternalFuture<Message> cacheStop(UUID ignored, SnapshotRestoreSta
         if (!U.isLocalNodeCoordinator(ctx.discovery()))
             return new GridFinishedFuture<>();
 
-        assert opCtx.reqId == req.requestId();
+        assert opCtx.reqId.equals(req.requestId());
 
         SnapshotRestoreContext opCtx0 = opCtx;
 
@@ -1374,8 +1369,7 @@ private void finishCacheStop(UUID reqId, Map<UUID, Message> res, Map<UUID, Throw
             log.error("Failed to stop caches during a snapshot rollback routine [reqId=" + opCtx0.reqId + ", err=" + errs + ']');
         }
 
-        if (U.isLocalNodeCoordinator(ctx.discovery()))
-            rollbackRestoreProc.start(reqId, new SnapshotRestoreStartRequest(reqId));
+        rollbackRestoreProc.startByCoordinator(reqId, new SnapshotRestoreStartRequest(reqId));
     }
 
     /**
@@ -1576,8 +1570,7 @@ private void finishIncrementalSnapshotRestore(UUID reqId, Map<UUID, Message> res
 
         opCtx0.err.compareAndSet(null, failure);
 
-        if (U.isLocalNodeCoordinator(ctx.discovery()))
-            cacheStopProc.start(reqId, new SnapshotRestoreStartRequest(reqId));
+        cacheStopProc.startByCoordinator(reqId, new SnapshotRestoreStartRequest(reqId));
     }
 
     /**

modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/RollingUpgradeProcessor.java

@@ -535,8 +535,8 @@ private void finishPreparePhase(UUID reqId, Map<UUID, Message> responses, Map<UU
 
                 finishProcess(reqId, firstError(errors));
             }
-            else if (reqId.equals(curFinalizeProcId) && U.isLocalNodeCoordinator(ctx.discovery()))
-                completePhase.start(reqId, null);
+            else if (reqId.equals(curFinalizeProcId))
+                completePhase.startByCoordinator(reqId, null);
         }
 
         /** */

modules/core/src/main/java/org/apache/ignite/internal/util/distributed/DistributedProcess.java

@@ -41,6 +41,7 @@
 import org.apache.ignite.internal.util.typedef.F;
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.spi.discovery.IgniteDiscoveryThread;
 import org.jetbrains.annotations.Nullable;
 
 import static org.apache.ignite.events.EventType.EVT_NODE_FAILED;
@@ -62,6 +63,8 @@
  * </ol>
  * <p>
  * Several processes of one type can be started at the same time.
+ * <p>
+ * Processes can be chained: the finish closure starts the next process via {@link #startByCoordinator(UUID, Message)}.
  *
  * @param <I> Request type.
  * @param <R> Result type.
@@ -75,6 +78,9 @@ public class DistributedProcess<I extends Message, R extends Message> {
     /** Active processes. */
     private final ConcurrentHashMap<UUID, Process> processes = new ConcurrentHashMap<>(1);
 
+    /** Starts waiting for the coordinator to send the initial request, see {@link #startByCoordinator}. */
+    private final ConcurrentHashMap<UUID, PendingStart> pendingStarts = new ConcurrentHashMap<>(1);
+
     /** Synchronization mutex for coordinator initializing and the remaining collection operations. */
     private final Object mux = new Object();
 
@@ -126,6 +132,8 @@ public DistributedProcess(
             if (msg.type() != type.ordinal())
                 return;
 
+            pendingStarts.remove(msg.processId());
+
             Process p = processes.computeIfAbsent(msg.processId(), id -> new Process(msg.processId()));
 
             // May be completed in case of double delivering.
@@ -215,6 +223,8 @@ public DistributedProcess(
         ctx.event().addDiscoveryEventListener((evt, discoCache) -> {
             UUID leftNodeId = evt.eventNode().id();
 
+            reassignPendingStarts(leftNodeId);
+
             for (Process p : processes.values()) {
                 p.initFut.listen(() -> {
                     if (Objects.equals(leftNodeId, p.crdId)) {
@@ -265,6 +275,55 @@ public void start(UUID id, I req) {
         }
     }
 
+    /**
+     * Starts the process by the coordinator: it sends the initial request, and if it leaves before the request is
+     * delivered, the next coordinator sends it. Must be called on every server node while handling the same discovery
+     * event, e.g. from the finish closure of the previous process, so that the call precedes the initial request
+     * delivery.
+     *
+     * @param id Process id.
+     * @param req Initial request.
+     */
+    public void startByCoordinator(UUID id, I req) {
+        if (ctx.clientNode())
+            return;
+
+        assert Thread.currentThread() instanceof IgniteDiscoveryThread : "Must be called from the discovery thread";
+
+        ClusterNode crd = coordinator();
+
+        if (crd == null)
+            return;
+
+        pendingStarts.put(id, new PendingStart(crd.id(), req));
+
+        if (crd.isLocal())
+            start(id, req);
+    }
+
+    /** Reassigns the pending starts of the left coordinator to the new one and starts them if it is the local node. */
+    private void reassignPendingStarts(UUID leftNodeId) {
+        for (Map.Entry<UUID, PendingStart> e : pendingStarts.entrySet()) {
+            PendingStart pending = e.getValue();
+
+            if (!Objects.equals(leftNodeId, pending.crdId))
+                continue;
+
+            ClusterNode crd = coordinator();
+
+            if (crd == null) {
+                onAllServersLeft();
+
+                return;
+            }
+
+            pending.crdId = crd.id();
+
+            if (crd.isLocal())
+                start(e.getKey(), pending.req);
+        }
+    }
+
     /**
      * Initiates process coordinator.
      *
@@ -373,6 +432,7 @@ private void finishProcess(Process p) {
     /** Handles case when all server nodes have left the grid. */
     private void onAllServersLeft() {
         processes.clear();
+        pendingStarts.clear();
     }
 
     /** @return Cluster coordinator, {@code null} if failed to determine. */
@@ -412,6 +472,21 @@ private Process(UUID id) {
         }
     }
 
+    /** Start waiting for the coordinator to send the initial request. */
+    private class PendingStart {
+        /** Coordinator node id. */
+        private volatile UUID crdId;
+
+        /** Initial request. */
+        private final I req;
+
+        /** */
+        private PendingStart(UUID crdId, I req) {
+            this.crdId = crdId;
+            this.req = req;
+        }
+    }
+
     /** Defines distributed processes. */
     public enum DistributedProcessType {
         /** For test purposes only. */

modules/core/src/test/java/org/apache/ignite/internal/encryption/MasterKeyChangeTest.java

@@ -28,16 +28,19 @@
 import org.apache.ignite.cluster.ClusterState;
 import org.apache.ignite.configuration.CacheConfiguration;
 import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.DiscoverySpiTestListener;
 import org.apache.ignite.internal.IgniteEx;
 import org.apache.ignite.internal.IgniteInternalFuture;
 import org.apache.ignite.internal.TestRecordingCommunicationSpi;
 import org.apache.ignite.internal.managers.encryption.GenerateEncryptionKeyResponse;
 import org.apache.ignite.internal.processors.cache.DynamicCacheDescriptor;
 import org.apache.ignite.internal.processors.cache.persistence.GridCacheDatabaseSharedManager;
 import org.apache.ignite.internal.processors.cache.persistence.metastorage.MetaStorage;
+import org.apache.ignite.internal.util.distributed.InitMessage;
 import org.apache.ignite.internal.util.distributed.SingleNodeMessage;
 import org.apache.ignite.internal.util.typedef.T2;
 import org.apache.ignite.lang.IgniteFuture;
+import org.apache.ignite.spi.discovery.tcp.IgniteDiscoverySpiInternalListenerSupport;
 import org.apache.ignite.transactions.Transaction;
 import org.junit.Test;
 
@@ -422,6 +425,29 @@ private void checkNodeFailsDuringRotation(boolean stopCrd, boolean prepare) thro
         assertEquals(MASTER_KEY_NAME_2, aliveNode.encryption().getMasterKeyName());
     }
 
+    /** Coordinator fails after the prepare phase is finished and before it starts the perform phase. */
+    @Test
+    public void testCoordinatorFailsBetweenPrepareAndPerform() throws Exception {
+        T2<IgniteEx, IgniteEx> grids = startTestGrids(true);
+
+        DiscoverySpiTestListener lsnr = new DiscoverySpiTestListener();
+
+        ((IgniteDiscoverySpiInternalListenerSupport)grids.get1().configuration().getDiscoverySpi())
+            .setInternalListener(lsnr);
+
+        lsnr.blockCustomEvent(InitMessage.class);
+
+        IgniteFuture<Void> fut = grids.get2().encryption().changeMasterKey(MASTER_KEY_NAME_2);
+
+        lsnr.waitCustomEvent();
+
+        stopGrid(GRID_0, true);
+
+        fut.get(getTestTimeout());
+
+        assertEquals(MASTER_KEY_NAME_2, grids.get2().encryption().getMasterKeyName());
+    }
+
     /** @throws Exception If failed. */
     @Test
     public void testMultiByteMasterKeyNameWalRecovery() throws Exception {