From 69628ec829e8c49dbf1661ea708837929f6191a3 Mon Sep 17 00:00:00 2001 From: NSAmelchev Date: Wed, 7 Oct 2026 10:25:51 +0300 Subject: [PATCH 1/4] IGNITE-29114 Fix chained distributed processes hang when the coordinator leaves between phases --- .../encryption/GridEncryptionManager.java | 4 +- .../encryption/GroupKeyChangeProcess.java | 5 +- .../snapshot/SnapshotCheckProcess.java | 3 +- .../snapshot/SnapshotRestoreProcess.java | 21 ++--- .../RollingUpgradeProcessor.java | 4 +- .../util/distributed/DistributedProcess.java | 70 ++++++++++++++ .../encryption/MasterKeyChangeTest.java | 26 ++++++ ...tributedProcessStartByCoordinatorTest.java | 93 +++++++++++++++++++ .../testsuites/IgniteUtilSelfTestSuite.java | 2 + 9 files changed, 205 insertions(+), 23 deletions(-) create mode 100644 modules/core/src/test/java/org/apache/ignite/internal/util/distributed/DistributedProcessStartByCoordinatorTest.java diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/encryption/GridEncryptionManager.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/encryption/GridEncryptionManager.java index 9989036b6104a..980f62c99a193 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/managers/encryption/GridEncryptionManager.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/encryption/GridEncryptionManager.java @@ -1537,8 +1537,8 @@ private void finishPrepareMasterKeyChange(UUID id, Map res, Map res, Map completeFuture(id, err, fut); } - else if (U.isLocalNodeCoordinator(ctx.discovery())) - performGKChangeProc.start(id, req); + else + performGKChangeProc.startByCoordinator(id, req); } /** diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotCheckProcess.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotCheckProcess.java index 533f3e1a44d60..dd7ec0ef2917d 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotCheckProcess.java +++ b/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) { diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotRestoreProcess.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotRestoreProcess.java index 77f56f4d4cc7f..f2cdc160094b1 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotRestoreProcess.java +++ b/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 res, Map res, Map 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 res, Map res, Map grids = startTestGrids(true); + + DiscoverySpiTestListener lsnr = new DiscoverySpiTestListener(); + + ((IgniteDiscoverySpiInternalListenerSupport)grids.get1().configuration().getDiscoverySpi()) + .setInternalListener(lsnr); + + lsnr.blockCustomEvent(InitMessage.class); + + IgniteFuture fut = grids.get2().encryption().changeMasterKey(MASTER_KEY_NAME_2); + + lsnr.waitCustomEvent(); + + stopGrid(GRID_0, true); + + fut.get(); + + assertEquals(MASTER_KEY_NAME_2, grids.get2().encryption().getMasterKeyName()); + } + /** @throws Exception If failed. */ @Test public void testMultiByteMasterKeyNameWalRecovery() throws Exception { diff --git a/modules/core/src/test/java/org/apache/ignite/internal/util/distributed/DistributedProcessStartByCoordinatorTest.java b/modules/core/src/test/java/org/apache/ignite/internal/util/distributed/DistributedProcessStartByCoordinatorTest.java new file mode 100644 index 0000000000000..f1b50c4a0d0e5 --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/util/distributed/DistributedProcessStartByCoordinatorTest.java @@ -0,0 +1,93 @@ +/* + * 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.util.distributed; + +import java.util.HashMap; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.CountDownLatch; +import org.apache.ignite.Ignite; +import org.apache.ignite.configuration.IgniteConfiguration; +import org.apache.ignite.internal.DiscoverySpiTestListener; +import org.apache.ignite.internal.IgniteEx; +import org.apache.ignite.internal.util.future.GridFinishedFuture; +import org.apache.ignite.internal.util.typedef.G; +import org.apache.ignite.spi.MessagesPluginProvider; +import org.apache.ignite.spi.discovery.tcp.IgniteDiscoverySpiInternalListenerSupport; +import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; +import org.junit.Test; + +import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static org.apache.ignite.internal.util.distributed.DistributedProcess.DistributedProcessType.TEST_PROCESS; + +/** + * Tests {@link DistributedProcess#startByCoordinator} in case of coordinator node left. + */ +public class DistributedProcessStartByCoordinatorTest extends GridCommonAbstractTest { + /** Nodes count. */ + private static final int NODES_CNT = 3; + + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception { + IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName); + + cfg.setPluginProviders(new MessagesPluginProvider(TestIntegerMessage.class)); + + return cfg; + } + + /** Coordinator fails after the first process is finished and before it sends the initial request of the next one. */ + @Test + public void testCoordinatorFailsBetweenProcesses() throws Exception { + startGrids(NODES_CNT); + + DiscoverySpiTestListener lsnr = new DiscoverySpiTestListener(); + + ((IgniteDiscoverySpiInternalListenerSupport)grid(0).configuration().getDiscoverySpi()).setInternalListener(lsnr); + + lsnr.blockCustomEvent(InitMessage.class); + + CountDownLatch finishLatch = new CountDownLatch(NODES_CNT - 1); + + // The first process starts the next one with the same id, the request value is the process number. + Map> procByNode = new HashMap<>(); + + for (Ignite grid : G.allGrids()) { + procByNode.put(grid.name(), new DistributedProcess<>( + ((IgniteEx)grid).context(), + TEST_PROCESS, + (ignored, req) -> new GridFinishedFuture<>(req), + (id, res, err) -> { + if (res.values().stream().allMatch(msg -> msg.value() == 1)) + procByNode.get(grid.name()).startByCoordinator(id, new TestIntegerMessage(2)); + else if (res.size() == NODES_CNT - 1 && res.values().stream().allMatch(msg -> msg.value() == 2)) + finishLatch.countDown(); + else + fail("Unexpected process result [res=" + res + ", err=" + err + ']'); + })); + } + + procByNode.get(grid(1).name()).start(UUID.randomUUID(), new TestIntegerMessage(1)); + + lsnr.waitCustomEvent(); + + stopGrid(0); + + assertTrue(finishLatch.await(getTestTimeout(), MILLISECONDS)); + } +} diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteUtilSelfTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteUtilSelfTestSuite.java index ad2129d4d2e2a..c998accb81f19 100644 --- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteUtilSelfTestSuite.java +++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteUtilSelfTestSuite.java @@ -37,6 +37,7 @@ import org.apache.ignite.internal.util.distributed.DistributedProcessCoordinatorLeftTest; import org.apache.ignite.internal.util.distributed.DistributedProcessErrorHandlingTest; import org.apache.ignite.internal.util.distributed.DistributedProcessResultMarshallingTest; +import org.apache.ignite.internal.util.distributed.DistributedProcessStartByCoordinatorTest; import org.apache.ignite.internal.util.nio.GridNioDelimitedBufferSelfTest; import org.apache.ignite.internal.util.nio.GridNioSelfTest; import org.apache.ignite.internal.util.nio.GridNioServerTest; @@ -147,6 +148,7 @@ DistributedProcessErrorHandlingTest.class, DistributedProcessCoordinatorLeftTest.class, + DistributedProcessStartByCoordinatorTest.class, DistributedProcessClientAwaitTest.class, DistributedProcessResultMarshallingTest.class, From 247ce710697315878c50bd2f557c18b2d3f95e6f Mon Sep 17 00:00:00 2001 From: NSAmelchev Date: Fri, 9 Oct 2026 22:43:47 +0300 Subject: [PATCH 2/4] review fixes --- .../util/distributed/DistributedProcess.java | 43 +++++++++++-------- 1 file changed, 24 insertions(+), 19 deletions(-) diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/distributed/DistributedProcess.java b/modules/core/src/main/java/org/apache/ignite/internal/util/distributed/DistributedProcess.java index 139e5befedde7..53c327d52b872 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/util/distributed/DistributedProcess.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/util/distributed/DistributedProcess.java @@ -223,25 +223,7 @@ public DistributedProcess( ctx.event().addDiscoveryEventListener((evt, discoCache) -> { UUID leftNodeId = evt.eventNode().id(); - for (Map.Entry 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); - } + reassignPendingStarts(leftNodeId); for (Process p : processes.values()) { p.initFut.listen(() -> { @@ -319,6 +301,29 @@ public void startByCoordinator(UUID id, I req) { 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 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. * From b38ef879a66f9b6e6ef911c1f0262096b207d880 Mon Sep 17 00:00:00 2001 From: NSAmelchev Date: Sun, 11 Oct 2026 20:10:38 +0300 Subject: [PATCH 3/4] add test --- ...tributedProcessStartByCoordinatorTest.java | 73 ++++++++++++++++++- 1 file changed, 72 insertions(+), 1 deletion(-) diff --git a/modules/core/src/test/java/org/apache/ignite/internal/util/distributed/DistributedProcessStartByCoordinatorTest.java b/modules/core/src/test/java/org/apache/ignite/internal/util/distributed/DistributedProcessStartByCoordinatorTest.java index f1b50c4a0d0e5..82c125b5641b7 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/util/distributed/DistributedProcessStartByCoordinatorTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/util/distributed/DistributedProcessStartByCoordinatorTest.java @@ -21,12 +21,15 @@ import java.util.Map; import java.util.UUID; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicInteger; import org.apache.ignite.Ignite; import org.apache.ignite.configuration.IgniteConfiguration; import org.apache.ignite.internal.DiscoverySpiTestListener; import org.apache.ignite.internal.IgniteEx; import org.apache.ignite.internal.util.future.GridFinishedFuture; +import org.apache.ignite.internal.util.future.GridFutureAdapter; import org.apache.ignite.internal.util.typedef.G; +import org.apache.ignite.internal.util.typedef.internal.U; import org.apache.ignite.spi.MessagesPluginProvider; import org.apache.ignite.spi.discovery.tcp.IgniteDiscoverySpiInternalListenerSupport; import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; @@ -51,6 +54,11 @@ public class DistributedProcessStartByCoordinatorTest extends GridCommonAbstract return cfg; } + /** {@inheritDoc} */ + @Override protected void afterTest() { + stopAllGrids(); + } + /** Coordinator fails after the first process is finished and before it sends the initial request of the next one. */ @Test public void testCoordinatorFailsBetweenProcesses() throws Exception { @@ -73,7 +81,7 @@ public void testCoordinatorFailsBetweenProcesses() throws Exception { TEST_PROCESS, (ignored, req) -> new GridFinishedFuture<>(req), (id, res, err) -> { - if (res.values().stream().allMatch(msg -> msg.value() == 1)) + if (res.size() == NODES_CNT && res.values().stream().allMatch(msg -> msg.value() == 1)) procByNode.get(grid.name()).startByCoordinator(id, new TestIntegerMessage(2)); else if (res.size() == NODES_CNT - 1 && res.values().stream().allMatch(msg -> msg.value() == 2)) finishLatch.countDown(); @@ -90,4 +98,67 @@ else if (res.size() == NODES_CNT - 1 && res.values().stream().allMatch(msg -> ms assertTrue(finishLatch.await(getTestTimeout(), MILLISECONDS)); } + + /** + * Coordinator fails after the initial request of the next process is delivered, so the next coordinator must not + * send it again. + */ + @Test + public void testCoordinatorFailsAfterNextProcessStarted() throws Exception { + startGrids(NODES_CNT); + + // Initial requests sent by the next coordinator. + AtomicInteger initMsgs = new AtomicInteger(); + + ((IgniteDiscoverySpiInternalListenerSupport)grid(1).configuration().getDiscoverySpi()).setInternalListener( + (spi, log, msg) -> { + if (U.unwrapCustomMessage(msg) instanceof InitMessage) + initMsgs.incrementAndGet(); + + return true; + }); + + CountDownLatch nextProcStarted = new CountDownLatch(NODES_CNT); + CountDownLatch finishLatch = new CountDownLatch(NODES_CNT - 1); + + // The next process is in progress while the coordinator leaves, so it finishes on the new coordinator after + // the node left is handled there and the counter is final. + GridFutureAdapter nextProcRes = new GridFutureAdapter<>(); + + Map> procByNode = new HashMap<>(); + + for (Ignite grid : G.allGrids()) { + procByNode.put(grid.name(), new DistributedProcess<>( + ((IgniteEx)grid).context(), + TEST_PROCESS, + (ignored, req) -> { + if (req.value() == 1) + return new GridFinishedFuture<>(req); + + nextProcStarted.countDown(); + + return nextProcRes; + }, + (id, res, err) -> { + if (res.size() == NODES_CNT && res.values().stream().allMatch(msg -> msg.value() == 1)) + procByNode.get(grid.name()).startByCoordinator(id, new TestIntegerMessage(2)); + else if (res.size() == NODES_CNT - 1 && res.values().stream().allMatch(msg -> msg.value() == 2)) + finishLatch.countDown(); + else + fail("Unexpected process result [res=" + res + ", err=" + err + ']'); + })); + } + + procByNode.get(grid(0).name()).start(UUID.randomUUID(), new TestIntegerMessage(1)); + + assertTrue(nextProcStarted.await(getTestTimeout(), MILLISECONDS)); + + stopGrid(0); + + nextProcRes.onDone(new TestIntegerMessage(2)); + + assertTrue(finishLatch.await(getTestTimeout(), MILLISECONDS)); + + assertEquals(0, initMsgs.get()); + } } From 1517208de957aca1006b1dbaccf53af535f815d6 Mon Sep 17 00:00:00 2001 From: NSAmelchev Date: Mon, 12 Oct 2026 00:23:17 +0300 Subject: [PATCH 4/4] review fixes --- .../cache/persistence/snapshot/SnapshotRestoreProcess.java | 2 +- .../apache/ignite/internal/encryption/MasterKeyChangeTest.java | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotRestoreProcess.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotRestoreProcess.java index f2cdc160094b1..43db339d103fe 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotRestoreProcess.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotRestoreProcess.java @@ -1332,7 +1332,7 @@ private IgniteInternalFuture 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; diff --git a/modules/core/src/test/java/org/apache/ignite/internal/encryption/MasterKeyChangeTest.java b/modules/core/src/test/java/org/apache/ignite/internal/encryption/MasterKeyChangeTest.java index b49b03510b239..0266d91058f1e 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/encryption/MasterKeyChangeTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/encryption/MasterKeyChangeTest.java @@ -443,7 +443,7 @@ public void testCoordinatorFailsBetweenPrepareAndPerform() throws Exception { stopGrid(GRID_0, true); - fut.get(); + fut.get(getTestTimeout()); assertEquals(MASTER_KEY_NAME_2, grids.get2().encryption().getMasterKeyName()); }