IGNITE-23655 Introduce CacheIdleVerifyCancelCommand

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

@@ -39,7 +39,10 @@
 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;
@@ -84,6 +87,7 @@
 import org.apache.ignite.internal.client.util.GridConcurrentHashSet;
 import org.apache.ignite.internal.management.cache.FindAndDeleteGarbageInPersistenceTaskResult;
 import org.apache.ignite.internal.management.cache.IdleVerifyDumpTask;
+import org.apache.ignite.internal.management.cache.IdleVerifyTaskV2;
 import org.apache.ignite.internal.management.cache.VerifyBackupPartitionsTaskV2;
 import org.apache.ignite.internal.management.tx.TxInfo;
 import org.apache.ignite.internal.management.tx.TxTaskResult;
@@ -98,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;
@@ -135,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;
@@ -173,6 +181,8 @@
 import static org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotRestoreProcess.SNAPSHOT_RESTORE_METRICS;
 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;
@@ -219,13 +229,16 @@ public class GridCommandHandlerTest extends GridCommandHandlerClusterPerMethodAb
     /** */
     protected static File customDiagnosticDir;
 
+    /** */
+    protected ListeningTestLogger listeningLog = new ListeningTestLogger(log);
+
     /** {@inheritDoc} */
     @Override protected void beforeTest() throws Exception {
         super.beforeTest();
 
-        initDiagnosticDir();
+        cleanPersistenceDir();
 
-        cleanDiagnosticDir();
+        initDiagnosticDir();
     }
 
     /** {@inheritDoc} */
@@ -235,6 +248,15 @@ public class GridCommandHandlerTest extends GridCommandHandlerClusterPerMethodAb
         cleanDiagnosticDir();
     }
 
+    /** {@inheritDoc} */
+    @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+        IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+        cfg.setGridLogger(listeningLog);
+
+        return cfg;
+    }
+
     /**
      * @throws IgniteCheckedException If failed.
      */
@@ -752,6 +774,173 @@ public void testIdleVerifyOnInactiveClusterWithPersistence() throws Exception {
         assertContains(log, testOut.toString(), "The check procedure has finished, no conflicts have been found.");
     }
 
+    /**
+     *
+     */
+    @Test
+    public void testIdleVerifyCancelCommandOnCheckpoint() throws Exception {
+        final int gridsCnt = 4;
+
+        IgniteEx srv = startGrids(gridsCnt);
+
+        srv.cluster().state(ACTIVE);
+
+        CountDownLatch beforeCancelLatch = new CountDownLatch(1);
+        CountDownLatch afterCancelLatch = new CountDownLatch(1);
+
+        GridCacheDatabaseSharedManager dbMgr = (GridCacheDatabaseSharedManager)grid(1).context().cache().context().database();
+
+        dbMgr.addCheckpointListener(new CheckpointListener() {
+            @Override public void beforeCheckpointBegin(Context ctx) {
+                if (ctx.progress().reason().equals("VerifyBackupPartitions"))
+                    beforeCancelLatch.countDown();
+            }
+
+            @Override public void afterCheckpointEnd(Context ctx) throws IgniteCheckedException {
+                if (ctx.progress().reason().equals("VerifyBackupPartitions")) {
+                    try {
+                        afterCancelLatch.await(30, TimeUnit.SECONDS);
+                    }
+                    catch (InterruptedException e) {
+                        throw new IgniteInterruptedCheckedException(e);
+                    }
+                }
+            }
+
+            @Override public void onMarkCheckpointBegin(Context ctx) {
+                // No-op.
+            }
+
+            @Override public void onCheckpointBegin(Context ctx) {
+                // No-op.
+            }
+        });
+
+        List<LogListener> listeners = registerIdleVerifyCancelListeners();
+
+        IgniteCache<Integer, Integer> cache = srv.createCache(new CacheConfiguration<Integer, Integer>(DEFAULT_CACHE_NAME).setBackups(3));
+
+        for (int i = 0; i < 100; i++)
+            cache.put(i, i);
+
+        cancelIdleVerifyAndCheck(beforeCancelLatch, afterCancelLatch, gridsCnt, listeners);
+    }
+
+
+    /**
+     *
+     */
+    @Test
+    public void testIdleVerifyTrackedForkJoinPool() throws Exception {
+        final int gridsCnt = 4;
+
+        IgniteEx srv = startGrids(gridsCnt);
+
+        srv.cluster().state(ACTIVE);
+
+        List<LogListener> listeners = registerIdleVerifyCancelListeners();
+
+        IgniteCache<Integer, Integer> cache = srv.createCache(new CacheConfiguration<Integer, Integer>(DEFAULT_CACHE_NAME).setBackups(3));
+
+        for (int i = 0; i < 100; i++)
+            cache.put(i, i);
+
+        CountDownLatch beforeCancelLatch = new CountDownLatch(1);
+        CountDownLatch afterCancelLatch = new CountDownLatch(1);
+
+        ForkJoinPool forkJoinPool = new ForkJoinPool() {
+            @Override public <T> ForkJoinTask<T> submit(Callable<T> task) {
+                beforeCancelLatch.countDown();
+
+                ForkJoinTask<T> submitted = super.submit(task);
+
+                try {
+                    afterCancelLatch.await();
+                }
+                catch (InterruptedException e) {
+                    throw new RuntimeException(e);
+                }
+
+                return submitted;
+            }
+        };
+
+        VerifyBackupPartitionsTaskV2.poolSupplier = () -> forkJoinPool;
+
+        cancelIdleVerifyAndCheck(beforeCancelLatch, afterCancelLatch, gridsCnt, listeners);
+    }
+
+    /**
+     * @param beforeCancelLatch Latch for await before cancel.
+     * @param afterCancelLatch Latch for await before cancel completes.
+     * @param gridsCnt Grids count.
+     * @param listeners Log listeners.
+     */
+    private void cancelIdleVerifyAndCheck(
+        CountDownLatch beforeCancelLatch,
+        CountDownLatch afterCancelLatch,
+        int gridsCnt,
+        List<LogListener> listeners
+    ) throws InterruptedException, IgniteCheckedException {
+        IgniteInternalFuture<Integer> idleVerifyFut = GridTestUtils.runAsync(() -> execute("--cache", "idle_verify"));
+
+        beforeCancelLatch.await(30, TimeUnit.SECONDS);
+
+        assertEquals(EXIT_CODE_OK, execute("--cache", "idle_verify", "--cancel"));
+
+        afterCancelLatch.countDown();
+
+        checkSystemViewsNoIdleVerify(gridsCnt);
+
+        idleVerifyFut.get(getTestTimeout());
+
+        for (LogListener listener : listeners)
+            assertTrue(listener.check());
+    }
+
+    /**
+     * @param gridsCnt Grids count.
+     */
+    private void checkSystemViewsNoIdleVerify(int gridsCnt) throws IgniteInterruptedCheckedException {
+        for (int i = 0; i < gridsCnt; i++) {
+            int finalI = i;
+
+            waitForCondition(() -> {
+                for (ComputeTaskView taskView : grid(finalI).context().systemView().<ComputeTaskView>view(TASKS_VIEW)) {
+                    if (IdleVerifyTaskV2.class.getName().equals(taskView.taskName()))
+                        return false;
+                }
+
+                return true;
+            }, 1000);
+
+            waitForCondition(() -> {
+                for (ComputeJobView jobView : grid(finalI).context().systemView().<ComputeJobView>view(JOBS_VIEW)) {
+                    if (IdleVerifyTaskV2.class.getName().equals(jobView.taskName()))
+                        return false;
+                }
+
+                return true;
+            }, 1000);
+        }
+    }
+
+    /**
+     *
+     */
+    private List<LogListener> registerIdleVerifyCancelListeners() {
+        List<LogListener> listeners = new ArrayList<>();
+
+        listeners.add(LogListener.matches("Idle verify was cancelled.").build());
+
+        listeners.add(LogListener.matches("Cancel request sent to VerifyBackupPartitionsJobV2.").build());
+
+        for (LogListener listener : listeners)
+            listeningLog.registerListener(listener);
+
+        return listeners;
+    }
+
     /**
      * Test deactivation works via control.sh
      *

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

@@ -0,0 +1,49 @@
+/*
+ * 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.
+ */
+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;
+    }
+}