IGNITE-28381 research v1

modules/core/src/main/java/org/apache/ignite/compute/ComputeTask.java

@@ -229,7 +229,7 @@
  * @param <T> Type of the task argument that is passed into {@link ComputeTask#map(List, Object)} method.
  * @param <R> Type of the task result returning from {@link ComputeTask#reduce(List)} method.
  */
-public interface ComputeTask<T, R> extends Serializable {
+public interface  ComputeTask<T, R> 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.

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();

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();
             }

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());

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;
     }
 

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<Object> 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;
+    }
+}

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 {
 }