IGNITE-19034 VisorSnapshotStatusTask returns empty result for restoring snapshot until snapshot check is completed

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

@@ -48,6 +48,7 @@
 import java.util.function.BooleanSupplier;
 import java.util.function.Consumer;
 import java.util.function.Function;
+import java.util.function.Supplier;
 import java.util.function.UnaryOperator;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
@@ -77,6 +78,7 @@
 import org.apache.ignite.internal.IgniteEx;
 import org.apache.ignite.internal.IgniteInternalFuture;
 import org.apache.ignite.internal.IgniteInterruptedCheckedException;
+import org.apache.ignite.internal.IgniteVersionUtils;
 import org.apache.ignite.internal.Order;
 import org.apache.ignite.internal.TestRecordingCommunicationSpi;
 import org.apache.ignite.internal.dto.IgniteDataTransferObject;
@@ -116,11 +118,16 @@
 import org.apache.ignite.internal.processors.cluster.GridClusterStateProcessor;
 import org.apache.ignite.internal.processors.datastreamer.DataStreamerRequest;
 import org.apache.ignite.internal.processors.metric.MetricRegistryImpl;
+import org.apache.ignite.internal.processors.nodevalidation.DiscoveryNodeValidationProcessor;
+import org.apache.ignite.internal.processors.rollingupgrade.RollingUpgradeProcessor;
+import org.apache.ignite.internal.processors.rollingupgrade.feature.IgniteCoreFeatureSet;
+import org.apache.ignite.internal.processors.rollingupgrade.feature.IgniteFeatureSet;
 import org.apache.ignite.internal.util.BasicRateLimiter;
 import org.apache.ignite.internal.util.GridConcurrentHashSet;
 import org.apache.ignite.internal.util.distributed.DistributedProcess;
 import org.apache.ignite.internal.util.distributed.SingleNodeMessage;
 import org.apache.ignite.internal.util.future.IgniteFinishedFutureImpl;
+import org.apache.ignite.internal.util.future.IgniteFutureImpl;
 import org.apache.ignite.internal.util.lang.GridAbsPredicate;
 import org.apache.ignite.internal.util.lang.GridFunc;
 import org.apache.ignite.internal.util.typedef.F;
@@ -134,7 +141,11 @@
 import org.apache.ignite.lang.IgnitePredicate;
 import org.apache.ignite.lang.IgniteUuid;
 import org.apache.ignite.metric.MetricRegistry;
+import org.apache.ignite.plugin.AbstractTestPluginProvider;
+import org.apache.ignite.plugin.PluginContext;
+import org.apache.ignite.plugin.PluginProvider;
 import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.spi.IgniteNodeValidationResult;
 import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
 import org.apache.ignite.spi.metric.LongMetric;
 import org.apache.ignite.spi.metric.Metric;
@@ -148,6 +159,7 @@
 import org.apache.ignite.transactions.TransactionRollbackException;
 import org.apache.ignite.transactions.TransactionTimeoutException;
 import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
 import org.junit.Assume;
 import org.junit.Test;
 
@@ -182,6 +194,7 @@
 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.internal.util.distributed.DistributedProcess.DistributedProcessType.CHECK_SNAPSHOT_PARTS;
 import static org.apache.ignite.internal.util.distributed.DistributedProcess.DistributedProcessType.RESTORE_CACHE_GROUP_SNAPSHOT_PREPARE;
 import static org.apache.ignite.testframework.GridTestUtils.assertContains;
 import static org.apache.ignite.testframework.GridTestUtils.assertNotContains;
@@ -233,20 +246,22 @@ public class GridCommandHandlerTest extends GridCommandHandlerClusterPerMethodAb
     /** */
     protected ListeningTestLogger listeningLog;
 
+    /** */
+    protected @Nullable Supplier<TcpCommunicationSpi> communicationSpiSupp;
+
+    /** */
+    protected @Nullable PluginProvider pluginProvider;
+
+    /** */
+    protected boolean walCompaction;
+
     /** {@inheritDoc} */
     @Override protected void beforeTest() throws Exception {
         super.beforeTest();
 
         initDiagnosticDir();
 
-        cleanDiagnosticDir();
-    }
-
-    /** {@inheritDoc} */
-    @Override protected void afterTest() throws Exception {
-        super.afterTest();
-
-        listeningLog = null;
+        cleanPersistenceDir();
     }
 
     /** {@inheritDoc} */
@@ -263,6 +278,14 @@ public class GridCommandHandlerTest extends GridCommandHandlerClusterPerMethodAb
         if (listeningLog != null)
             cfg.setGridLogger(listeningLog);
 
+        if (communicationSpiSupp != null)
+            cfg.setCommunicationSpi(communicationSpiSupp.get());
+
+        if (pluginProvider != null)
+            cfg.setPluginProviders(pluginProvider);
+
+        cfg.getDataStorageConfiguration().setWalCompactionEnabled(walCompaction);
+
         return cfg;
     }
 
@@ -3658,6 +3681,268 @@ public void testSnapshotRestoreCancelAndStatus() throws Exception {
         assertNull(ig.cache(DEFAULT_CACHE_NAME));
     }
 
+    /**
+     * Tests that snapshot metrics aren't empty when being restored snapshot waits for the check process.
+     * All nodes support the check status.
+     */
+    @Test
+    public void testRestoreSnapshotMetricsAtStartAllNodesSupport() throws Exception {
+        doTestRestoreSnapshotMetricsAtStart(null);
+    }
+
+    /**
+     * Tests that snapshot metrics aren't empty when being restored snapshot waits for the check process.
+     * One node doesn't support the check status.
+     */
+    @Test
+    public void testRestoreSnapshotMetricsAtStartOneNodeDoesntSupport() throws Exception {
+        doTestRestoreSnapshotMetricsAtStart(false);
+    }
+
+    /**
+     * Tests that snapshot metrics aren't empty when being restored snapshot waits for the check process.
+     * All nodes don't support the check status.
+     */
+    @Test
+    public void testRestoreSnapshotMetricsAtStartNoSupport() throws Exception {
+        doTestRestoreSnapshotMetricsAtStart(true);
+    }
+
+    /**
+     * @param allNodesNotSupporting Flag of how many nodes don't support the snapshot check status.
+     *                              If {@code null}, all nodes support.
+     *                              If {@code true}, none supports.
+     *                              If {@code false}, just one node doesn't support.
+     */
+    private void doTestRestoreSnapshotMetricsAtStart(@Nullable Boolean allNodesNotSupporting) throws Exception {
+        communicationSpiSupp = TestRecordingCommunicationSpi::new;
+
+        // Creates empty feature set unsupporting the snapshot check ststus if required.
+        pluginProvider = allNodesNotSupporting == null ? null : new AbstractTestPluginProvider() {
+            @Override public String name() {
+                return "Test Ignite features provider";
+            }
+
+            @Override public <T> @Nullable T createComponent(PluginContext ctx, Class<T> cls) {
+                if (!cls.equals(DiscoveryNodeValidationProcessor.class))
+                    return null;
+
+                boolean doNotSupport = allNodesNotSupporting
+                    || ctx.igniteConfiguration().getIgniteInstanceName().equals(getTestIgniteInstanceName(2));
+
+                return (T)new RollingUpgradeProcessor(
+                    ((IgniteEx)ctx.grid()).context(),
+                    doNotSupport ? new IgniteCoreFeatureSet(IgniteVersionUtils.VER, new IgniteFeatureSet()) : IgniteCoreFeatureSet.local()
+                ) {
+                    @Override public @Nullable IgniteNodeValidationResult validateNode(ClusterNode joiningNode) {
+                        return null;
+                    }
+                };
+            }
+        };
+
+        listeningLog = new ListeningTestLogger(log);
+
+        startGrids(3).cluster().state(ClusterState.ACTIVE);
+
+        createCacheAndPreload(grid(1), 1000);
+
+        IgniteSnapshotManager snpMgr = (IgniteSnapshotManager)grid(0).snapshot();
+
+        snpMgr.createSnapshot("test_snapshot").get(getTestTimeout());
+
+        grid(0).destroyCache(DEFAULT_CACHE_NAME);
+
+        awaitPartitionMapExchange();
+
+        TestRecordingCommunicationSpi cm1 = ((TestRecordingCommunicationSpi)grid(1).configuration().getCommunicationSpi());
+        TestRecordingCommunicationSpi cm2 = ((TestRecordingCommunicationSpi)grid(2).configuration().getCommunicationSpi());
+
+        // Block one of the process' first messages of snapshot restoring or snapshot checking.
+        F.asList(cm1, cm2).forEach(cm -> cm.blockMessages((node, msg) -> msg instanceof SingleNodeMessage<?> sm
+                && (sm.type() == CHECK_SNAPSHOT_PARTS.ordinal() || sm.type() == RESTORE_CACHE_GROUP_SNAPSHOT_PREPARE.ordinal())));
+
+        // Snapshot restoration should get paused at the preceeding snapshot check.
+        IgniteFutureImpl<Void> restoreFut = snpMgr.restoreSnapshot("test_snapshot", null, null, 0, true);
+
+        // Waiting for each node to send snapshot check single result.
+        for (var cm : F.asList(cm1, cm2)) {
+            assertTrue(waitForCondition(() -> cm.blockedMessages().stream().anyMatch(m ->
+                    m.ioMessage().message() instanceof SingleNodeMessage<?> sm && sm.type() == CHECK_SNAPSHOT_PARTS.ordinal()),
+                getTestTimeout()));
+        }
+
+        injectTestSystemOut();
+
+        LogListener logLsnr = null;
+
+        if (Boolean.TRUE.equals(allNodesNotSupporting)) {
+            logLsnr = LogListener.matches("The snapshot-check-aware status feature isn't enabled").build();
+
+            listeningLog.registerListener(logLsnr);
+        }
+        else if (Boolean.FALSE.equals(allNodesNotSupporting)) {
+            logLsnr = LogListener.matches("Node %s doesn't support the snapshot check status feature"
+                .formatted(grid(2).localNode().id())).build();
+
+            listeningLog.registerListener(logLsnr);
+        }
+
+        int code = execute("--snapshot", "status");
+
+        assertEquals("Unexpected exit code", EXIT_CODE_OK, code);
+
+        var out = testOut.toString();
+
+        if (allNodesNotSupporting == null) {
+            assertContains(log, out, "Check snapshot operation is in progress");
+            assertContains(log, out, "Snapshot name: test_snapshot");
+        }
+        else {
+            assertTrue(logLsnr.check(getTestTimeout()));
+            assertContains(log, out, "There is no create or restore snapshot operation in progress");
+        }
+
+        // Ensure that no snapshot restore started or finished.
+        assertFalse("Snapshot future has finished", restoreFut.isDone());
+
+        F.asList(cm1, cm2).forEach(cm -> {
+            assertTrue(cm.blockedMessages().stream().noneMatch(m ->
+                m.ioMessage().message() instanceof SingleNodeMessage<?> sm
+                    && sm.type() == RESTORE_CACHE_GROUP_SNAPSHOT_PREPARE.ordinal()));
+
+            cm.stopBlock();
+        });
+
+        restoreFut.get(getTestTimeout());
+    }
+
+    /** */
+    @Test
+    public void testOneSnapshotCheckStatus() throws Exception {
+        doTestSnapshotsChecksStatus(false, false);
+    }
+
+    /** */
+    @Test
+    public void testOneIncrementalSnapshotCheckStatus() throws Exception {
+        doTestSnapshotsChecksStatus(false, true);
+    }
+
+    /** */
+    @Test
+    public void testTwoSnapshotsChecksStatus() throws Exception {
+        doTestSnapshotsChecksStatus(true, false);
+    }
+
+    /** */
+    @Test
+    public void testTwoIncrementalsSnapshotsChecksStatus() throws Exception {
+        doTestSnapshotsChecksStatus(true, true);
+    }
+
+    /** */
+    private void doTestSnapshotsChecksStatus(boolean twoSnapshots, boolean incremental) throws Exception {
+        communicationSpiSupp = TestRecordingCommunicationSpi::new;
+
+        walCompaction = incremental;
+
+        startGrids(3).cluster().state(ClusterState.ACTIVE);
+
+        IgniteSnapshotManager snpMgr = (IgniteSnapshotManager)grid(0).snapshot();
+
+        createCacheAndPreload(grid(1), DEFAULT_CACHE_NAME, 1000, 32, null);
+        snpMgr.createSnapshot("testSnapshot0").get(getTestTimeout());
+
+        if (incremental) {
+            try (IgniteDataStreamer<Object, Object> streamer = grid(0).dataStreamer(DEFAULT_CACHE_NAME)) {
+                for (int i = 1000; i < 2000; i++)
+                    streamer.addData(i, i);
+            }
+
+            snpMgr.createIncrementalSnapshot("testSnapshot0").get(getTestTimeout());
+        }
+
+        if (twoSnapshots) {
+            createCacheAndPreload(grid(1), "cache2", 1000, 32, null);
+            snpMgr.createSnapshot("testSnapshot1").get(getTestTimeout());
+
+            if (incremental) {
+                try (IgniteDataStreamer<Object, Object> streamer = grid(0).dataStreamer("cache2")) {
+                    for (int i = 1000; i < 2000; i++)
+                        streamer.addData(i, i);
+                }
+
+                snpMgr.createIncrementalSnapshot("testSnapshot1").get(getTestTimeout());
+            }
+        }
+
+        grid(0).destroyCaches(twoSnapshots ? F.asList(DEFAULT_CACHE_NAME, "cache2") : F.asList(DEFAULT_CACHE_NAME));
+
+        awaitPartitionMapExchange();
+
+        TestRecordingCommunicationSpi cm1 = ((TestRecordingCommunicationSpi)grid(1).configuration().getCommunicationSpi());
+        TestRecordingCommunicationSpi cm2 = ((TestRecordingCommunicationSpi)grid(2).configuration().getCommunicationSpi());
+
+        F.asList(cm1, cm2).forEach(cm -> cm.blockMessages((node, msg) ->
+            msg instanceof SingleNodeMessage<?> sm && (sm.type() == CHECK_SNAPSHOT_PARTS.ordinal())));
+
+        var checkFut0 = runAsync(() -> incremental
+            ? execute("--snapshot", "check", "testSnapshot0", "--increment", "1")
+            : execute("--snapshot", "check", "testSnapshot0"));
+
+        var checkFut1 = twoSnapshots
+            ? runAsync(() -> incremental
+                             ? execute("--snapshot", "check", "testSnapshot1", "--increment", "1")
+                             : execute("--snapshot", "check", "testSnapshot1"))
+            : null;
+
+        // Waiting for the nodes each to send snapshot check single result.
+        for (var cm : F.asList(cm1, cm2)) {
+            assertTrue(waitForCondition(
+                () -> cm.blockedMessages().stream().filter(
+                    m -> m.ioMessage().message() instanceof SingleNodeMessage<?> sm
+                        && sm.type() == CHECK_SNAPSHOT_PARTS.ordinal()).count() == (twoSnapshots ? 2 : 1),
+                getTestTimeout()
+            ));
+        }
+
+        injectTestSystemOut();
+
+        assertEquals("Unexpected exit code", EXIT_CODE_OK, execute("--snapshot", "status"));
+
+        var out = testOut.toString();
+
+        if (log.isInfoEnabled())
+            log.info("Test out:" + U.nl() + out);
+
+        assertTrue(out.contains(twoSnapshots ? "Check snapshot operations are in progress" : "Check snapshot operation is in progress"));
+        assertTrue(out.contains("Snapshot name: testSnapshot0"));
+
+        if (incremental)
+            assertTrue(out.contains("Increment index: 1"));
+
+        if (twoSnapshots) {
+            assertTrue(out.contains("Snapshot name: testSnapshot1"));
+
+            if (incremental) {
+                // Number of 'Increment index: 1' entries.
+                var sum = Arrays.stream(out.split(U.nl()))
+                    .mapToInt(l -> (l.length() - l.replace("Increment index: 1", "").length()) / "Increment index: 1".length())
+                    .sum();
+
+                assertEquals(2, sum);
+            }
+        }
+
+        F.asList(cm1, cm2).forEach(TestRecordingCommunicationSpi::stopBlock);
+
+        checkFut0.get();
+
+        if (twoSnapshots)
+            checkFut1.get();
+    }
+
     /** @throws Exception If fails. */
     @Test
     public void testSnapshotStatusInMemory() throws Exception {