IGNITE-23655 Introduce CacheIdleVerifyCancelCommand (#11673)

Co-authored-by: Nikolay Izhikov <nizhikov@apache.org>

Author: vladnovoren

modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerTest.java

@@ -34,18 +34,23 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Random;
 import java.util.Set;
 import java.util.TreeMap;
 import java.util.TreeSet;
 import java.util.UUID;
+import java.util.concurrent.Callable;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.ForkJoinTask;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.LongAdder;
+import java.util.function.BiConsumer;
 import java.util.function.BooleanSupplier;
 import java.util.function.Function;
 import java.util.function.UnaryOperator;
@@ -97,6 +102,7 @@
 import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxLocal;
 import org.apache.ignite.internal.processors.cache.persistence.CheckpointState;
 import org.apache.ignite.internal.processors.cache.persistence.GridCacheDatabaseSharedManager;
+import org.apache.ignite.internal.processors.cache.persistence.checkpoint.CheckpointListener;
 import org.apache.ignite.internal.processors.cache.persistence.db.IgniteCacheGroupsWithRestartsTest;
 import org.apache.ignite.internal.processors.cache.persistence.diagnostic.pagelocktracker.dumpprocessors.ToFileDumpProcessor;
 import org.apache.ignite.internal.processors.cache.persistence.file.FileIO;
@@ -134,7 +140,10 @@
 import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
 import org.apache.ignite.spi.metric.LongMetric;
 import org.apache.ignite.spi.metric.Metric;
+import org.apache.ignite.spi.systemview.view.ComputeJobView;
+import org.apache.ignite.spi.systemview.view.ComputeTaskView;
 import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.ListeningTestLogger;
 import org.apache.ignite.testframework.LogListener;
 import org.apache.ignite.testframework.junits.WithSystemProperty;
 import org.apache.ignite.transactions.Transaction;
@@ -162,6 +171,9 @@
 import static org.apache.ignite.internal.commandline.CommandHandler.EXIT_CODE_OK;
 import static org.apache.ignite.internal.commandline.CommandHandler.EXIT_CODE_UNEXPECTED_ERROR;
 import static org.apache.ignite.internal.encryption.AbstractEncryptionTest.MASTER_KEY_NAME_2;
+import static org.apache.ignite.internal.management.cache.CacheIdleVerifyCancelTask.TASKS_TO_CANCEL;
+import static org.apache.ignite.internal.management.cache.VerifyBackupPartitionsTaskV2.CACL_PART_HASH_ERR_MSG;
+import static org.apache.ignite.internal.management.cache.VerifyBackupPartitionsTaskV2.CP_REASON;
 import static org.apache.ignite.internal.processors.cache.persistence.GridCacheDatabaseSharedManager.IGNITE_PDS_SKIP_CHECKPOINT_ON_NODE_STOP;
 import static org.apache.ignite.internal.processors.cache.persistence.snapshot.AbstractSnapshotSelfTest.doSnapshotCancellationTest;
 import static org.apache.ignite.internal.processors.cache.persistence.snapshot.AbstractSnapshotSelfTest.snp;
@@ -170,8 +182,11 @@
 import static org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotManager.SNAPSHOT_TRANSFER_RATE_DMS_KEY;
 import static org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotManager.resolveSnapshotWorkDirectory;
 import static org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotRestoreProcess.SNAPSHOT_RESTORE_METRICS;
+import static org.apache.ignite.internal.processors.cache.verify.IdleVerifyUtility.CRC_CHECK_ERR_MSG;
 import static org.apache.ignite.internal.processors.cache.verify.IdleVerifyUtility.GRID_NOT_IDLE_MSG;
 import static org.apache.ignite.internal.processors.diagnostic.DiagnosticProcessor.DEFAULT_TARGET_FOLDER;
+import static org.apache.ignite.internal.processors.job.GridJobProcessor.JOBS_VIEW;
+import static org.apache.ignite.internal.processors.task.GridTaskProcessor.TASKS_VIEW;
 import static org.apache.ignite.testframework.GridTestUtils.assertContains;
 import static org.apache.ignite.testframework.GridTestUtils.assertNotContains;
 import static org.apache.ignite.testframework.GridTestUtils.assertThrows;
@@ -218,6 +233,9 @@ public class GridCommandHandlerTest extends GridCommandHandlerClusterPerMethodAb
     /** */
     protected static File customDiagnosticDir;
 
+    /** */
+    protected ListeningTestLogger listeningLog;
+
     /** {@inheritDoc} */
     @Override protected void beforeTest() throws Exception {
         super.beforeTest();
@@ -227,13 +245,30 @@ public class GridCommandHandlerTest extends GridCommandHandlerClusterPerMethodAb
         cleanDiagnosticDir();
     }
 
+    /** {@inheritDoc} */
+    @Override protected void afterTest() throws Exception {
+        super.afterTest();
+
+        listeningLog = null;
+    }
+
     /** {@inheritDoc} */
     @Override protected void cleanPersistenceDir() throws Exception {
         super.cleanPersistenceDir();
 
         cleanDiagnosticDir();
     }
 
+    /** {@inheritDoc} */
+    @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+        IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+        if (listeningLog != null)
+            cfg.setGridLogger(listeningLog);
+
+        return cfg;
+    }
+
     /**
      * @throws IgniteCheckedException If failed.
      */
@@ -747,6 +782,199 @@ public void testIdleVerifyOnInactiveClusterWithPersistence() throws Exception {
         assertContains(log, testOut.toString(), "The check procedure has finished, no conflicts have been found.");
     }
 
+    /** */
+    @Test
+    public void testIdleVerifyCancelOnCheckpoint() throws Exception {
+        doTestCancelIdleVerify((beforeCancelLatch, afterCancelLatch) -> {
+            G.allGrids().forEach(grid -> {
+                if (grid.configuration().isClientMode())
+                    return;
+
+                GridCacheDatabaseSharedManager dbMgr =
+                    (GridCacheDatabaseSharedManager)((IgniteEx)grid).context().cache().context().database();
+
+                dbMgr.addCheckpointListener(new CheckpointListener() {
+                    @Override public void beforeCheckpointBegin(Context ctx) {
+                        if (Objects.equals(ctx.progress().reason(), CP_REASON))
+                            beforeCancelLatch.countDown();
+                    }
+
+                    @Override public void afterCheckpointEnd(Context ctx) throws IgniteCheckedException {
+                        if (Objects.equals(ctx.progress().reason(), CP_REASON)) {
+                            try {
+                                assertTrue(afterCancelLatch.await(getTestTimeout(), TimeUnit.MILLISECONDS));
+                            }
+                            catch (InterruptedException e) {
+                                throw new IgniteInterruptedCheckedException(e);
+                            }
+                        }
+                    }
+
+                    @Override public void onMarkCheckpointBegin(Context ctx) {
+                        // No-op.
+                    }
+
+                    @Override public void onCheckpointBegin(Context ctx) {
+                        // No-op.
+                    }
+                });
+            });
+        }, false);
+    }
+
+    /** */
+    @Test
+    public void testIdleVerifyCancelBeforeCalcPartitionHashStarted() throws Exception {
+        doTestCancelIdleVerify((beforeCancelLatch, afterCancelLatch) -> {
+            ForkJoinPool pool = new ForkJoinPool() {
+                @Override public <T> ForkJoinTask<T> submit(Callable<T> task) {
+                    beforeCancelLatch.countDown();
+
+                    ForkJoinTask<T> submitted = super.submit(task);
+
+                    try {
+                        assertTrue(afterCancelLatch.await(getTestTimeout(), TimeUnit.MILLISECONDS));
+                    }
+                    catch (InterruptedException e) {
+                        throw new RuntimeException(e);
+                    }
+
+                    return submitted;
+                }
+            };
+
+            VerifyBackupPartitionsTaskV2.poolSupplier = () -> pool;
+        }, false);
+    }
+
+    /** */
+    @Test
+    public void testIdleVerifyCancelWhileCalcPartitionHashRunning() throws Exception {
+        for (boolean checkCrc : new boolean[] {false, true}) {
+            // Can't place assert inside pool, because exceptions from task ignored.
+            AtomicBoolean interruptedOnCancel = new AtomicBoolean(true);
+            AtomicBoolean eCatched = new AtomicBoolean(false);
+
+            doTestCancelIdleVerify((beforeCancelLatch, afterCancelLatch) -> {
+                ForkJoinPool pool = new ForkJoinPool() {
+                    @Override public <T> ForkJoinTask<T> submit(Callable<T> task) {
+                        return super.submit(new Callable<T>() {
+                            @Override public T call() throws Exception {
+                                beforeCancelLatch.countDown();
+
+                                try {
+                                    assertTrue(afterCancelLatch.await(getTestTimeout(), TimeUnit.MILLISECONDS));
+                                }
+                                catch (InterruptedException ignored) {
+                                    interruptedOnCancel.set(false);
+                                }
+
+                                try {
+                                    // Call must fail.
+                                    T res = task.call();
+
+                                    interruptedOnCancel.set(false);
+
+                                    return res;
+                                }
+                                catch (IgniteException e) {
+                                    if (!e.getMessage().startsWith(checkCrc ? CRC_CHECK_ERR_MSG : CACL_PART_HASH_ERR_MSG))
+                                        interruptedOnCancel.set(false);
+
+                                    eCatched.set(true);
+
+                                    throw e;
+                                }
+                                catch (Throwable e) {
+                                    interruptedOnCancel.set(false);
+
+                                    throw e;
+                                }
+                            }
+                        });
+                    }
+                };
+
+                VerifyBackupPartitionsTaskV2.poolSupplier = () -> pool;
+            }, checkCrc);
+
+            assertTrue("All tasks must be cancelled", interruptedOnCancel.get());
+            assertTrue("Task must fail with expected exception", eCatched.get());
+
+            eCatched.set(false);
+        }
+    }
+
+    /**
+     * Wrapper for tests for idle verify cancel command.
+     *
+     * @param prepare Prepares the test using beforeCancelLatch and afterCancelLatch.
+     * @param checkCrc If {@code true} then run idle verify with --check-crc argument.
+    */
+    private void doTestCancelIdleVerify(BiConsumer<CountDownLatch, CountDownLatch> prepare, boolean checkCrc) throws Exception {
+        final int gridsCnt = 4;
+
+        if (G.allGrids().isEmpty()) {
+            listeningLog = new ListeningTestLogger(log);
+
+            IgniteEx srv = startGrids(gridsCnt);
+
+            srv.cluster().state(ACTIVE);
+
+            IgniteCache<Integer, Integer> cache = srv.getOrCreateCache(new CacheConfiguration<Integer, Integer>(DEFAULT_CACHE_NAME)
+                .setBackups(3)
+                .setAffinity(new RendezvousAffinityFunction().setPartitions(3)));
+
+            for (int part = 0; part < 3; part++) {
+                for (Integer key : partitionKeys(cache, part, 3, 0)) {
+                    cache.put(key, key);
+                }
+            }
+        }
+
+        CountDownLatch beforeCancelLatch = new CountDownLatch(1);
+        CountDownLatch afterCancelLatch = new CountDownLatch(1);
+
+        prepare.accept(beforeCancelLatch, afterCancelLatch);
+
+        LogListener lsnr = LogListener.matches("Idle verify was cancelled.").build();
+
+        listeningLog.registerListener(lsnr);
+
+        IgniteInternalFuture<Integer> idleVerifyFut = GridTestUtils.runAsync(() -> {
+            if (checkCrc)
+                execute("--cache", "idle_verify", "--check-crc");
+            else
+                execute("--cache", "idle_verify");
+        });
+
+        assertTrue(beforeCancelLatch.await(getTestTimeout(), TimeUnit.MILLISECONDS));
+
+        assertEquals(EXIT_CODE_OK, execute("--cache", "idle_verify", "--cancel"));
+
+        afterCancelLatch.countDown();
+
+        assertTrue(waitForCondition(() -> {
+            for (int i = 0; i < gridsCnt; i++) {
+                for (ComputeTaskView taskView : grid(i).context().systemView().<ComputeTaskView>view(TASKS_VIEW)) {
+                    if (TASKS_TO_CANCEL.contains(taskView.taskName()))
+                        return false;
+                }
+
+                for (ComputeJobView jobView : grid(i).context().systemView().<ComputeJobView>view(JOBS_VIEW)) {
+                    if (TASKS_TO_CANCEL.contains(jobView.taskName()))
+                        return false;
+                }
+            }
+
+            return true;
+        }, getTestTimeout()));
+
+        idleVerifyFut.get(getTestTimeout(), TimeUnit.MILLISECONDS);
+
+        assertTrue(lsnr.check());
+    }
+
     /**
      * Test deactivation works via control.sh
      *
@@ -2252,7 +2480,7 @@ public void testCacheIdleVerifyDumpCrcWithCorruptedPartition() throws Exception
         String outputStr = testOut.toString();
 
         assertContains(log, outputStr, "The check procedure failed on 1 node.");
-        assertContains(log, outputStr, "CRC check of partition failed");
+        assertContains(log, outputStr, CRC_CHECK_ERR_MSG);
     }
 
     /** */

modules/core/src/main/java/org/apache/ignite/internal/management/cache/CacheIdleVerifyCancelCommand.java

@@ -0,0 +1,51 @@
+/*
+ * 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.management.cache;
+
+import java.util.Collection;
+
+import org.apache.ignite.internal.client.GridClientNode;
+import org.apache.ignite.internal.management.api.ComputeCommand;
+import org.apache.ignite.internal.management.api.NoArg;
+
+/**
+ * Cancels idle_verify command.
+ *
+ * @see CacheIdleVerifyCommand
+ */
+public class CacheIdleVerifyCancelCommand implements ComputeCommand<NoArg, Void> {
+    /** {@inheritDoc} */
+    @Override public Class<CacheIdleVerifyCancelTask> taskClass() {
+        return CacheIdleVerifyCancelTask.class;
+    }
+
+    /** {@inheritDoc} */
+    @Override public String description() {
+        return "Cancels idle_verify command";
+    }
+
+    /** {@inheritDoc} */
+    @Override public Class<NoArg> argClass() {
+        return NoArg.class;
+    }
+
+    /** {@inheritDoc} */
+    @Override public Collection<GridClientNode> nodes(Collection<GridClientNode> nodes, NoArg arg) {
+        return nodes;
+    }
+}