From 8ce75896bc8981259f4d455f11b0357ab87c30c0 Mon Sep 17 00:00:00 2001 From: Steshin Vladimir Date: Thu, 27 Aug 2026 10:50:10 +0300 Subject: [PATCH 1/5] research --- .../GridNearAtomicSingleUpdateFuture.java | 11 +- ...CacheOperationRemappingOnNodeStopTest.java | 109 ++++++++++++++++++ 2 files changed, 115 insertions(+), 5 deletions(-) create mode 100644 modules/core/src/test/java/org/apache/ignite/AtomicCacheOperationRemappingOnNodeStopTest.java 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..19f4e50abd518 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,11 +240,12 @@ else if (rcvAll) boolean remapKey = res.remapTopologyVersion() != null; if (remapKey) { - assert !req.topologyVersion().equals(res.remapTopologyVersion()); - - assert remapTopVer == null : remapTopVer; - - remapTopVer = res.remapTopologyVersion(); + if (remapTopVer == null) + remapTopVer = res.remapTopologyVersion(); + else if (!remapTopVer.equals(res.remapTopologyVersion())) { + onPrimaryError(req, res); + return; + } } else if (res.error() != null) onPrimaryError(req, res); diff --git a/modules/core/src/test/java/org/apache/ignite/AtomicCacheOperationRemappingOnNodeStopTest.java b/modules/core/src/test/java/org/apache/ignite/AtomicCacheOperationRemappingOnNodeStopTest.java new file mode 100644 index 0000000000000..e88fbcd15fa48 --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/AtomicCacheOperationRemappingOnNodeStopTest.java @@ -0,0 +1,109 @@ +/* + * 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; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicInteger; +import org.apache.ignite.configuration.IgniteConfiguration; +import org.apache.ignite.failure.StopNodeOrHaltFailureHandler; +import org.apache.ignite.internal.IgniteEx; +import org.apache.ignite.internal.IgniteInternalFuture; +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.testframework.GridTestUtils; +import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; +import org.junit.Test; + +import static java.util.concurrent.TimeUnit.MILLISECONDS; + +/** */ +public class AtomicCacheOperationRemappingOnNodeStopTest extends GridCommonAbstractTest { + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration() throws Exception { + return super.getConfiguration().setFailureHandler(new StopNodeOrHaltFailureHandler()); + } + + /** */ + @Test + public void test() throws Exception { + startGrids(3); + + grid(0).createCache(DEFAULT_CACHE_NAME); + + IgniteEx firstNode = grid(1); + + CountDownLatch zeroNodePmeFinishedLatch = new CountDownLatch(1); + CountDownLatch firstNodePmeStartedLatch = new CountDownLatch(1); + CountDownLatch firstNodePmeUnblockedLatch = new CountDownLatch(1); + CountDownLatch firstNodePmeFinishedLatch = new CountDownLatch(1); + CountDownLatch firstNodeStopUnblockedLatch = new CountDownLatch(1); + + grid(0).context().cache().context().exchange().registerExchangeAwareComponent(new PartitionsExchangeAware() { + @Override public void onDoneAfterTopologyUnlock(GridDhtPartitionsExchangeFuture fut) { + zeroNodePmeFinishedLatch.countDown(); + } + }); + + firstNode.context().cache().context().exchange().registerExchangeAwareComponent(new PartitionsExchangeAware() { + @Override public void onInitBeforeTopologyLock(GridDhtPartitionsExchangeFuture fut) { + try { + firstNodePmeStartedLatch.countDown(); + + firstNodePmeUnblockedLatch.await(getTestTimeout(), MILLISECONDS); + } + catch (InterruptedException e) { + throw new RuntimeException(e); + } + } + }); + + stopGrid(2); + + firstNodePmeStartedLatch.await(getTestTimeout(), MILLISECONDS); + + firstNode.context().cache().context().exchange().lastTopologyFuture().listen(() -> { + try { + firstNodePmeFinishedLatch.countDown(); + + firstNodeStopUnblockedLatch.await(getTestTimeout(), MILLISECONDS); + } + catch (InterruptedException e) { + throw new RuntimeException(e); + } + }); + + IgniteInternalFuture firstNodeStopFut = GridTestUtils.runAsync(() -> stopGrid(1)); + + firstNodePmeUnblockedLatch.countDown(); + + firstNodePmeFinishedLatch.await(getTestTimeout(), MILLISECONDS); + + zeroNodePmeFinishedLatch.await(getTestTimeout(), MILLISECONDS); + + grid(0).cache(DEFAULT_CACHE_NAME).put(keyForNode(grid(0).affinity(DEFAULT_CACHE_NAME), new AtomicInteger(), firstNode.localNode()), "test-val"); + + firstNodeStopUnblockedLatch.countDown(); + + firstNodeStopFut.get(getTestTimeout(), MILLISECONDS); + } + + /** {@inheritDoc} */ + @Override protected long getTestTimeout() { + return 20_000; + } +} From 0ef6aab3c12971c5aeb6f7d1d03c3119eb2401ba Mon Sep 17 00:00:00 2001 From: Steshin Vladimir Date: Thu, 27 Aug 2026 21:44:28 +0300 Subject: [PATCH 2/5] research --- .../dht/atomic/GridDhtAtomicCache.java | 12 +- .../GridNearAtomicSingleUpdateFuture.java | 16 +- .../atomic/GridNearAtomicUpdateResponse.java | 4 +- ...CacheOperationRemappingOnNodeStopTest.java | 109 ------------- ...CacheOperationRemappingOnNodeStopTest.java | 147 ++++++++++++++++++ .../IgniteCacheAffinityRunTestSuite.java | 4 +- 6 files changed, 171 insertions(+), 121 deletions(-) delete mode 100644 modules/core/src/test/java/org/apache/ignite/AtomicCacheOperationRemappingOnNodeStopTest.java create mode 100644 modules/core/src/test/java/org/apache/ignite/internal/processors/cache/AtomicCacheOperationRemappingOnNodeStopTest.java 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..8e29f87655423 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. - res.remapTopologyVersion(top.lastTopologyChangeVersion()); + else { + // Should remap all keys. If current node is stopping, suppose. + res.remapTopologyVersion(ctx.kernalContext().isStopping() ? AffinityTopologyVersion.NONE + : top.lastTopologyChangeVersion()); + } } finally { top.readUnlock(); @@ -1932,7 +1934,9 @@ private void updateAllAsyncInternal0( if (log.isDebugEnabled()) log.debug("Caught invalid partition exception for cache entry (will remap update request): " + req); - res.remapTopologyVersion(ctx.topology().lastTopologyChangeVersion()); + // If current node is stopping, send 'unknown', unawared topology. + res.remapTopologyVersion(ctx.kernalContext().isStopping() ? AffinityTopologyVersion.NONE + : ctx.topology().lastTopologyChangeVersion()); } catch (Throwable e) { // At least RuntimeException can be thrown by the code above when GridCacheContext is cleaned and there is 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 19f4e50abd518..11ad1d2416826 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,12 +240,18 @@ else if (rcvAll) boolean remapKey = res.remapTopologyVersion() != null; if (remapKey) { - if (remapTopVer == null) - remapTopVer = res.remapTopologyVersion(); - else if (!remapTopVer.equals(res.remapTopologyVersion())) { - onPrimaryError(req, res); - return; + assert !req.topologyVersion().equals(res.remapTopologyVersion()) + : "Update response holds the same remap-to topology version"; + assert remapTopVer == null : "Current remap-to version is not null: " + remapTopVer; + + // Remote node is stopping and doesn't care of topology anymore. + if (res.topologyVersion().equals(AffinityTopologyVersion.NONE)) { + // Suppose topology would change on the `+1` version. if not, we'll remap again. + remapTopVer = new AffinityTopologyVersion(req.topologyVersion().topologyVersion() + 1, + req.topVer.minorTopologyVersion()); } + else + remapTopVer = res.remapTopologyVersion(); } else if (res.error() != null) onPrimaryError(req, res); 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/AtomicCacheOperationRemappingOnNodeStopTest.java b/modules/core/src/test/java/org/apache/ignite/AtomicCacheOperationRemappingOnNodeStopTest.java deleted file mode 100644 index e88fbcd15fa48..0000000000000 --- a/modules/core/src/test/java/org/apache/ignite/AtomicCacheOperationRemappingOnNodeStopTest.java +++ /dev/null @@ -1,109 +0,0 @@ -/* - * 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; - -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.atomic.AtomicInteger; -import org.apache.ignite.configuration.IgniteConfiguration; -import org.apache.ignite.failure.StopNodeOrHaltFailureHandler; -import org.apache.ignite.internal.IgniteEx; -import org.apache.ignite.internal.IgniteInternalFuture; -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.testframework.GridTestUtils; -import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; -import org.junit.Test; - -import static java.util.concurrent.TimeUnit.MILLISECONDS; - -/** */ -public class AtomicCacheOperationRemappingOnNodeStopTest extends GridCommonAbstractTest { - /** {@inheritDoc} */ - @Override protected IgniteConfiguration getConfiguration() throws Exception { - return super.getConfiguration().setFailureHandler(new StopNodeOrHaltFailureHandler()); - } - - /** */ - @Test - public void test() throws Exception { - startGrids(3); - - grid(0).createCache(DEFAULT_CACHE_NAME); - - IgniteEx firstNode = grid(1); - - CountDownLatch zeroNodePmeFinishedLatch = new CountDownLatch(1); - CountDownLatch firstNodePmeStartedLatch = new CountDownLatch(1); - CountDownLatch firstNodePmeUnblockedLatch = new CountDownLatch(1); - CountDownLatch firstNodePmeFinishedLatch = new CountDownLatch(1); - CountDownLatch firstNodeStopUnblockedLatch = new CountDownLatch(1); - - grid(0).context().cache().context().exchange().registerExchangeAwareComponent(new PartitionsExchangeAware() { - @Override public void onDoneAfterTopologyUnlock(GridDhtPartitionsExchangeFuture fut) { - zeroNodePmeFinishedLatch.countDown(); - } - }); - - firstNode.context().cache().context().exchange().registerExchangeAwareComponent(new PartitionsExchangeAware() { - @Override public void onInitBeforeTopologyLock(GridDhtPartitionsExchangeFuture fut) { - try { - firstNodePmeStartedLatch.countDown(); - - firstNodePmeUnblockedLatch.await(getTestTimeout(), MILLISECONDS); - } - catch (InterruptedException e) { - throw new RuntimeException(e); - } - } - }); - - stopGrid(2); - - firstNodePmeStartedLatch.await(getTestTimeout(), MILLISECONDS); - - firstNode.context().cache().context().exchange().lastTopologyFuture().listen(() -> { - try { - firstNodePmeFinishedLatch.countDown(); - - firstNodeStopUnblockedLatch.await(getTestTimeout(), MILLISECONDS); - } - catch (InterruptedException e) { - throw new RuntimeException(e); - } - }); - - IgniteInternalFuture firstNodeStopFut = GridTestUtils.runAsync(() -> stopGrid(1)); - - firstNodePmeUnblockedLatch.countDown(); - - firstNodePmeFinishedLatch.await(getTestTimeout(), MILLISECONDS); - - zeroNodePmeFinishedLatch.await(getTestTimeout(), MILLISECONDS); - - grid(0).cache(DEFAULT_CACHE_NAME).put(keyForNode(grid(0).affinity(DEFAULT_CACHE_NAME), new AtomicInteger(), firstNode.localNode()), "test-val"); - - firstNodeStopUnblockedLatch.countDown(); - - firstNodeStopFut.get(getTestTimeout(), MILLISECONDS); - } - - /** {@inheritDoc} */ - @Override protected long getTestTimeout() { - return 20_000; - } -} diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/AtomicCacheOperationRemappingOnNodeStopTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/AtomicCacheOperationRemappingOnNodeStopTest.java new file mode 100644 index 0000000000000..9db4bc0b6e66e --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/AtomicCacheOperationRemappingOnNodeStopTest.java @@ -0,0 +1,147 @@ +/* + * 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.Collections; +import java.util.UUID; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicInteger; +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.testframework.GridTestUtils; +import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; +import org.junit.Test; + +import static java.util.concurrent.TimeUnit.MILLISECONDS; + +/** */ +public class AtomicCacheOperationRemappingOnNodeStopTest extends GridCommonAbstractTest { + /** {@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(3); + + node0.createCache(DEFAULT_CACHE_NAME); + + 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()); + } + + /** {@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..0a45c697528d4 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.AtomicCacheOperationRemappingOnNodeStopTest; 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, + AtomicCacheOperationRemappingOnNodeStopTest.class }) public class IgniteCacheAffinityRunTestSuite { } From 922be254f0c76d27b31cbe821dc105a1132df5d1 Mon Sep 17 00:00:00 2001 From: Steshin Vladimir Date: Fri, 28 Aug 2026 00:15:51 +0300 Subject: [PATCH 3/5] fixes --- .../dht/atomic/GridDhtAtomicCache.java | 2 +- .../atomic/GridNearAtomicUpdateFuture.java | 3 +- ...CacheOperationRemappingOnNodeStopTest.java | 34 +++++++++++++++++++ 3 files changed, 37 insertions(+), 2 deletions(-) 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 8e29f87655423..980455ad0f8c7 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 @@ -1861,7 +1861,7 @@ private void updateAllAsyncInternal0( expiry = updDhtRes.expiryPolicy(); } else { - // Should remap all keys. If current node is stopping, suppose. + // Should remap all keys. // If current node is stopping, send 'unknown', unawared topology. res.remapTopologyVersion(ctx.kernalContext().isStopping() ? AffinityTopologyVersion.NONE : top.lastTopologyChangeVersion()); } 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..d818824b77d93 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,8 @@ else if (rcvAll) assert req.topologyVersion().equals(topVer) : req; if (res.remapTopologyVersion() != null) { - assert !req.topologyVersion().equals(res.remapTopologyVersion()); + assert !req.topologyVersion().equals(res.remapTopologyVersion()) + : "Update response holds the same remap-to topology version"; if (remapKeys == null) remapKeys = U.newHashSet(req.size()); diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/AtomicCacheOperationRemappingOnNodeStopTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/AtomicCacheOperationRemappingOnNodeStopTest.java index 9db4bc0b6e66e..a8faaa9915e83 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/AtomicCacheOperationRemappingOnNodeStopTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/AtomicCacheOperationRemappingOnNodeStopTest.java @@ -17,10 +17,13 @@ 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; @@ -30,14 +33,43 @@ 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; /** */ +@RunWith(Parameterized.class) public class AtomicCacheOperationRemappingOnNodeStopTest 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()); @@ -61,6 +93,8 @@ private void doTest(boolean putAll) throws Exception { node0.createCache(DEFAULT_CACHE_NAME); + awaitPartitionMapExchange(); + IgniteEx node1 = grid(1); CountDownLatch node0PmeFinishedLatch = new CountDownLatch(1); From 9159725ba1c14412586b8a29293738b96ba6239d Mon Sep 17 00:00:00 2001 From: Steshin Vladimir Date: Fri, 28 Aug 2026 00:46:32 +0300 Subject: [PATCH 4/5] fixes --- .../dht/atomic/GridDhtAtomicCache.java | 10 ++++------ .../atomic/GridNearAtomicSingleUpdateFuture.java | 16 ++++++---------- .../dht/atomic/GridNearAtomicUpdateFuture.java | 7 +++++-- ...micCacheOperationRemappingOnNodeStopTest.java | 9 ++++++++- 4 files changed, 23 insertions(+), 19 deletions(-) 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 980455ad0f8c7..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 @@ -1861,9 +1861,9 @@ private void updateAllAsyncInternal0( expiry = updDhtRes.expiryPolicy(); } else { - // Should remap all keys. // If current node is stopping, send 'unknown', unawared topology. - res.remapTopologyVersion(ctx.kernalContext().isStopping() ? AffinityTopologyVersion.NONE - : top.lastTopologyChangeVersion()); + // 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 { @@ -1934,9 +1934,7 @@ private void updateAllAsyncInternal0( if (log.isDebugEnabled()) log.debug("Caught invalid partition exception for cache entry (will remap update request): " + req); - // If current node is stopping, send 'unknown', unawared topology. - res.remapTopologyVersion(ctx.kernalContext().isStopping() ? AffinityTopologyVersion.NONE - : ctx.topology().lastTopologyChangeVersion()); + res.remapTopologyVersion(ctx.topology().lastTopologyChangeVersion()); } catch (Throwable e) { // At least RuntimeException can be thrown by the code above when GridCacheContext is cleaned and there is 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 11ad1d2416826..3cdbb9c3fb02c 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,18 +240,14 @@ else if (rcvAll) boolean remapKey = res.remapTopologyVersion() != null; if (remapKey) { - assert !req.topologyVersion().equals(res.remapTopologyVersion()) - : "Update response holds the same remap-to topology version"; + // Remote topology might be the same even if the remapping responded with. A 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 + : "Update response holds the same or lesser remap-to topology version"; assert remapTopVer == null : "Current remap-to version is not null: " + remapTopVer; - // Remote node is stopping and doesn't care of topology anymore. - if (res.topologyVersion().equals(AffinityTopologyVersion.NONE)) { - // Suppose topology would change on the `+1` version. if not, we'll remap again. - remapTopVer = new AffinityTopologyVersion(req.topologyVersion().topologyVersion() + 1, - req.topVer.minorTopologyVersion()); - } - else - remapTopVer = res.remapTopologyVersion(); + remapTopVer = res.remapTopologyVersion(); } else if (res.error() != null) onPrimaryError(req, res); 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 d818824b77d93..e316b20fa345f 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,8 +400,11 @@ else if (rcvAll) assert req.topologyVersion().equals(topVer) : req; if (res.remapTopologyVersion() != null) { - assert !req.topologyVersion().equals(res.remapTopologyVersion()) - : "Update response holds the same remap-to topology version"; + // Remote topology might be the same even if the remapping responded with. A 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 + : "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/test/java/org/apache/ignite/internal/processors/cache/AtomicCacheOperationRemappingOnNodeStopTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/AtomicCacheOperationRemappingOnNodeStopTest.java index a8faaa9915e83..aebf77dfc5c76 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/AtomicCacheOperationRemappingOnNodeStopTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/AtomicCacheOperationRemappingOnNodeStopTest.java @@ -41,6 +41,7 @@ import org.junit.runners.Parameterized; import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static org.apache.ignite.testframework.GridTestUtils.waitForCondition; /** */ @RunWith(Parameterized.class) @@ -89,7 +90,7 @@ public void testPutAll() throws Exception { /** */ private void doTest(boolean putAll) throws Exception { - IgniteEx node0 = startGrids(3); + IgniteEx node0 = startGrids(4); node0.createCache(DEFAULT_CACHE_NAME); @@ -172,6 +173,12 @@ private void doTest(boolean putAll) throws Exception { 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} */ From 3add80ce3ec646c1adb104e2dd52a43a0fd1a9fa Mon Sep 17 00:00:00 2001 From: Steshin Vladimir Date: Fri, 28 Aug 2026 13:31:00 +0300 Subject: [PATCH 5/5] minor manual review fixes. --- .../src/main/java/org/apache/ignite/compute/ComputeTask.java | 2 +- .../dht/atomic/GridNearAtomicSingleUpdateFuture.java | 4 ++-- .../distributed/dht/atomic/GridNearAtomicUpdateFuture.java | 4 ++-- ...eStopTest.java => CacheUpdateRemappingOnNodeStopTest.java} | 2 +- .../ignite/testsuites/IgniteCacheAffinityRunTestSuite.java | 4 ++-- 5 files changed, 8 insertions(+), 8 deletions(-) rename modules/core/src/test/java/org/apache/ignite/internal/processors/cache/{AtomicCacheOperationRemappingOnNodeStopTest.java => CacheUpdateRemappingOnNodeStopTest.java} (98%) 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/GridNearAtomicSingleUpdateFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicSingleUpdateFuture.java index 3cdbb9c3fb02c..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,11 +240,11 @@ else if (rcvAll) boolean remapKey = res.remapTopologyVersion() != null; if (remapKey) { - // Remote topology might be the same even if the remapping responded with. A remote node may respond + // 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 - : "Update response holds the same or lesser remap-to topology version"; + : "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 e316b20fa345f..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,11 +400,11 @@ else if (rcvAll) assert req.topologyVersion().equals(topVer) : req; if (res.remapTopologyVersion() != null) { - // Remote topology might be the same even if the remapping responded with. A remote node may respond + // 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 - : "Update response holds the same or lesser remap-to topology version"; + : "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/test/java/org/apache/ignite/internal/processors/cache/AtomicCacheOperationRemappingOnNodeStopTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheUpdateRemappingOnNodeStopTest.java similarity index 98% rename from modules/core/src/test/java/org/apache/ignite/internal/processors/cache/AtomicCacheOperationRemappingOnNodeStopTest.java rename to modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheUpdateRemappingOnNodeStopTest.java index aebf77dfc5c76..3fd3ceb2c68c9 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/AtomicCacheOperationRemappingOnNodeStopTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheUpdateRemappingOnNodeStopTest.java @@ -45,7 +45,7 @@ /** */ @RunWith(Parameterized.class) -public class AtomicCacheOperationRemappingOnNodeStopTest extends GridCommonAbstractTest { +public class CacheUpdateRemappingOnNodeStopTest extends GridCommonAbstractTest { /** */ @Parameterized.Parameter public CacheAtomicityMode atomicityMode; 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 0a45c697528d4..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,7 +17,7 @@ package org.apache.ignite.testsuites; -import org.apache.ignite.internal.processors.cache.AtomicCacheOperationRemappingOnNodeStopTest; +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; @@ -39,7 +39,7 @@ IgniteBaselineLockPartitionOnAffinityRunAtomicCacheTest.class, IgniteBaselineLockPartitionOnAffinityRunTxCacheTest.class, IgniteCacheLockPartitionOnAffinityRunTxCacheOpTest.class, - AtomicCacheOperationRemappingOnNodeStopTest.class + CacheUpdateRemappingOnNodeStopTest.class }) public class IgniteCacheAffinityRunTestSuite { }