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..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 @@ -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 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 res, Map 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. */ 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 5df474911e095..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 @@ -28,6 +28,7 @@ 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; @@ -35,9 +36,11 @@ 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 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(getTestTimeout()); + + 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..82c125b5641b7 --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/util/distributed/DistributedProcessStartByCoordinatorTest.java @@ -0,0 +1,164 @@ +/* + * 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 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; +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; + } + + /** {@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 { + 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.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(1).name()).start(UUID.randomUUID(), new TestIntegerMessage(1)); + + lsnr.waitCustomEvent(); + + stopGrid(0); + + 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()); + } +} 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,