diff --git a/modules/core/src/main/java/org/apache/ignite/compute/ComputeTask.java b/modules/core/src/main/java/org/apache/ignite/compute/ComputeTask.java index 031323d349d5d..af9fb95d780e3 100644 --- a/modules/core/src/main/java/org/apache/ignite/compute/ComputeTask.java +++ b/modules/core/src/main/java/org/apache/ignite/compute/ComputeTask.java @@ -229,7 +229,7 @@ * @param Type of the task argument that is passed into {@link ComputeTask#map(List, Object)} method. * @param Type of the task result returning from {@link ComputeTask#reduce(List)} method. */ -public interface ComputeTask extends Serializable { +public interface ComputeTask extends Serializable { /** * This method is called to map or split grid task into multiple grid jobs. This is the * first method that gets called when task execution starts. diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java index 496f579a33fcf..8eda2a49b59de 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java @@ -1860,9 +1860,11 @@ private void updateAllAsyncInternal0( deleted = updDhtRes.deleted(); expiry = updDhtRes.expiryPolicy(); } - else - // Should remap all keys. + else { + // Should remap all keys. If current node is stopping, topology might be the same as + // in the request because topology version might not be accepted/updated when node is stopping. res.remapTopologyVersion(top.lastTopologyChangeVersion()); + } } finally { top.readUnlock(); diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicSingleUpdateFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicSingleUpdateFuture.java index 94b8db2443068..3af109ceee89d 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicSingleUpdateFuture.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicSingleUpdateFuture.java @@ -240,9 +240,12 @@ else if (rcvAll) boolean remapKey = res.remapTopologyVersion() != null; if (remapKey) { - assert !req.topologyVersion().equals(res.remapTopologyVersion()); - - assert remapTopVer == null : remapTopVer; + // Remote topology might be the same even if the remapping responded with. Remote node may respond + // when stopping. But if is stopping, might not accept and update topology version and would use some last + // kept version. + assert req.topologyVersion().compareTo(res.remapTopologyVersion()) <= 0 + : "Near atomic update response holds the same or lesser remap-to topology version"; + assert remapTopVer == null : "Current remap-to version is not null: " + remapTopVer; remapTopVer = res.remapTopologyVersion(); } diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateFuture.java index 94fdc0c8ec0e8..af5bd7e07bc2b 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateFuture.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateFuture.java @@ -400,7 +400,11 @@ else if (rcvAll) assert req.topologyVersion().equals(topVer) : req; if (res.remapTopologyVersion() != null) { - assert !req.topologyVersion().equals(res.remapTopologyVersion()); + // Remote topology might be the same even if the remapping responded with. Remote node may respond + // when stopping. But if is stopping, might not accept and update topology version and would use some last + // kept version. + assert req.topologyVersion().compareTo(res.remapTopologyVersion()) <= 0 + : "Near atomic update response holds the same or lesser remap-to topology version"; if (remapKeys == null) remapKeys = U.newHashSet(req.size()); diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateResponse.java index 3d9ab417c80a2..8e260b9f6c6bc 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateResponse.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateResponse.java @@ -62,7 +62,7 @@ public class GridNearAtomicUpdateResponse extends GridCacheIdMessage implements /** */ @Order(3) - AffinityTopologyVersion remapTopVer; + @Nullable AffinityTopologyVersion remapTopVer; /** Data for near cache update. */ @Order(4) @@ -206,7 +206,7 @@ public void remapTopologyVersion(AffinityTopologyVersion remapTopVer) { /** * @return Topology version if update should be remapped. */ - @Nullable public AffinityTopologyVersion remapTopologyVersion() { + public @Nullable AffinityTopologyVersion remapTopologyVersion() { return remapTopVer; } diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheUpdateRemappingOnNodeStopTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheUpdateRemappingOnNodeStopTest.java new file mode 100644 index 0000000000000..3fd3ceb2c68c9 --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheUpdateRemappingOnNodeStopTest.java @@ -0,0 +1,188 @@ +/* + * 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.processors.cache; + +import java.util.Collection; +import java.util.Collections; +import java.util.UUID; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicInteger; +import org.apache.ignite.cache.CacheAtomicityMode; +import org.apache.ignite.cache.CacheWriteSynchronizationMode; +import org.apache.ignite.configuration.IgniteConfiguration; +import org.apache.ignite.failure.StopNodeOrHaltFailureHandler; +import org.apache.ignite.internal.GridTopic; +import org.apache.ignite.internal.IgniteEx; +import org.apache.ignite.internal.IgniteInternalFuture; +import org.apache.ignite.internal.managers.communication.GridMessageListener; +import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridNearAtomicUpdateResponse; +import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsExchangeFuture; +import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.PartitionsExchangeAware; +import org.apache.ignite.internal.util.typedef.F; +import org.apache.ignite.testframework.GridTestUtils; +import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; + +import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static org.apache.ignite.testframework.GridTestUtils.waitForCondition; + +/** */ +@RunWith(Parameterized.class) +public class CacheUpdateRemappingOnNodeStopTest extends GridCommonAbstractTest { + /** */ + @Parameterized.Parameter + public CacheAtomicityMode atomicityMode; + + /** */ + @Parameterized.Parameter(1) + public CacheWriteSynchronizationMode writeSyncMode; + + /** */ + @Parameterized.Parameters(name = "cacheAtomicity={0}, writeSync={1}") + public static Collection params() { + return GridTestUtils.cartesianProduct( + F.asList(CacheAtomicityMode.ATOMIC, CacheAtomicityMode.TRANSACTIONAL), + F.asList(CacheWriteSynchronizationMode.PRIMARY_SYNC, CacheWriteSynchronizationMode.FULL_SYNC, + CacheWriteSynchronizationMode.FULL_ASYNC) + ); + } + + /** {@inheritDoc} */ + @Override protected void afterTest() throws Exception { + super.afterTest(); + + stopAllGrids(); + } + + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration() throws Exception { + return super.getConfiguration().setFailureHandler(new StopNodeOrHaltFailureHandler()); + } + + /** */ + @Test + public void testPut() throws Exception { + doTest(false); + } + + /** */ + @Test + public void testPutAll() throws Exception { + doTest(false); + } + + /** */ + private void doTest(boolean putAll) throws Exception { + IgniteEx node0 = startGrids(4); + + node0.createCache(DEFAULT_CACHE_NAME); + + awaitPartitionMapExchange(); + + IgniteEx node1 = grid(1); + + CountDownLatch node0PmeFinishedLatch = new CountDownLatch(1); + CountDownLatch node1PmeStartedLatch = new CountDownLatch(1); + CountDownLatch node1PmeProceedLatch = new CountDownLatch(1); + CountDownLatch node1PmeFinishedLatch = new CountDownLatch(1); + CountDownLatch node1StoppageBockedLatch = new CountDownLatch(1); + + node0.context().cache().context().exchange().registerExchangeAwareComponent(new PartitionsExchangeAware() { + @Override public void onDoneAfterTopologyUnlock(GridDhtPartitionsExchangeFuture fut) { + node0PmeFinishedLatch.countDown(); + } + }); + + node1.context().cache().context().exchange().registerExchangeAwareComponent(new PartitionsExchangeAware() { + @Override public void onInitBeforeTopologyLock(GridDhtPartitionsExchangeFuture fut) { + try { + node1PmeStartedLatch.countDown(); + + node1PmeProceedLatch.await(getTestTimeout(), MILLISECONDS); + } + catch (InterruptedException e) { + throw new RuntimeException(e); + } + } + }); + + stopGrid(2); + + // Node 1 started PME but isn't proceeding. + node1PmeStartedLatch.await(getTestTimeout(), MILLISECONDS); + + node1.context().cache().context().exchange().lastTopologyFuture().listen(() -> { + try { + node1PmeFinishedLatch.countDown(); + + node1StoppageBockedLatch.await(getTestTimeout(), MILLISECONDS); + } + catch (InterruptedException e) { + throw new RuntimeException(e); + } + }); + + assertEquals(0, node0.cache(DEFAULT_CACHE_NAME).size()); + + IgniteInternalFuture node1StopFut = GridTestUtils.runAsync(() -> stopGrid(1)); + + // Node 1 proceeds PME. + node1PmeProceedLatch.countDown(); + + node1PmeFinishedLatch.await(getTestTimeout(), MILLISECONDS); + + node0PmeFinishedLatch.await(getTestTimeout(), MILLISECONDS); + + node0.context().io().addMessageListener(GridTopic.TOPIC_CACHE, new GridMessageListener() { + @Override public void onMessage(UUID nodeId, Object msg, byte plc) { + if (msg instanceof GridNearAtomicUpdateResponse && nodeId.equals(node1.localNode().id())) + node1StoppageBockedLatch.countDown(); + } + }); + + if (putAll) { + node0.cache(DEFAULT_CACHE_NAME).putAll(Collections.singletonMap( + keyForNode(node0.affinity(DEFAULT_CACHE_NAME), new AtomicInteger(), node1.localNode()), + "test-val" + )); + } + else { + node0.cache(DEFAULT_CACHE_NAME).put( + keyForNode(node0.affinity(DEFAULT_CACHE_NAME), new AtomicInteger(), node1.localNode()), + "test-val" + ); + } + + node1StopFut.get(getTestTimeout(), MILLISECONDS); + + assertEquals(1, node0.cache(DEFAULT_CACHE_NAME).size()); + assertTrue(waitForCondition(() -> 1 == grid(3).cache(DEFAULT_CACHE_NAME).size(), getTestTimeout())); + + var vervifyRes = idleVerify(node0); + + assertFalse(vervifyRes.hasConflicts()); + assertTrue(F.isEmpty(vervifyRes.exceptions())); + } + + /** {@inheritDoc} */ + @Override protected long getTestTimeout() { + return 20_000; + } +} diff --git a/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheAffinityRunTestSuite.java b/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheAffinityRunTestSuite.java index 2b1d3044cc47c..abcb201e6792d 100644 --- a/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheAffinityRunTestSuite.java +++ b/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheAffinityRunTestSuite.java @@ -17,6 +17,7 @@ package org.apache.ignite.testsuites; +import org.apache.ignite.internal.processors.cache.CacheUpdateRemappingOnNodeStopTest; import org.apache.ignite.internal.processors.cache.IgniteCacheLockPartitionOnAffinityRunAtomicCacheOpTest; import org.apache.ignite.internal.processors.cache.IgniteCacheLockPartitionOnAffinityRunTest; import org.apache.ignite.internal.processors.cache.IgniteCacheLockPartitionOnAffinityRunTxCacheOpTest; @@ -37,7 +38,8 @@ IgniteCacheLockPartitionOnAffinityRunAtomicCacheOpTest.class, IgniteBaselineLockPartitionOnAffinityRunAtomicCacheTest.class, IgniteBaselineLockPartitionOnAffinityRunTxCacheTest.class, - IgniteCacheLockPartitionOnAffinityRunTxCacheOpTest.class + IgniteCacheLockPartitionOnAffinityRunTxCacheOpTest.class, + CacheUpdateRemappingOnNodeStopTest.class }) public class IgniteCacheAffinityRunTestSuite { }