You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by mm...@apache.org on 2021/02/26 08:29:06 UTC
[ignite] branch master updated: IGNITE-13725 Add the snapshot check
command (#8715)
This is an automated email from the ASF dual-hosted git repository.
mmuzaf pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new 7700feb IGNITE-13725 Add the snapshot check command (#8715)
7700feb is described below
commit 7700febe0d76b8406ae706196c773ea71e620c17
Author: Maxim Muzafarov <mm...@apache.org>
AuthorDate: Fri Feb 26 11:28:35 2021 +0300
IGNITE-13725 Add the snapshot check command (#8715)
---
.../internal/commandline/cache/IdleVerify.java | 44 ++-
.../commandline/snapshot/SnapshotCommand.java | 51 ++--
.../commandline/snapshot/SnapshotSubcommand.java | 25 +-
.../util/GridCommandHandlerAbstractTest.java | 7 +-
.../util/GridCommandHandlerClusterByClassTest.java | 32 +--
.../util/GridCommandHandlerIndexingTest.java | 2 +-
.../GridCommandHandlerInterruptCommandTest.java | 2 +-
.../apache/ignite/util/GridCommandHandlerTest.java | 36 ++-
.../cache/persistence/file/FilePageStore.java | 16 +-
.../persistence/file/FilePageStoreManager.java | 62 ++++
.../snapshot/IgniteSnapshotManager.java | 212 +++++++++++++-
.../snapshot/IgniteSnapshotVerifyException.java | 48 ++++
.../persistence/snapshot/SnapshotFutureTask.java | 6 +-
.../persistence/snapshot/SnapshotMetadata.java | 198 +++++++++++++
.../snapshot/SnapshotMetadataCollectorTask.java | 111 +++++++
.../snapshot/SnapshotPartitionsVerifyTask.java | 298 +++++++++++++++++++
.../cache/verify/IdleVerifyResultV2.java | 81 ++----
.../processors/cache/verify/IdleVerifyUtility.java | 62 ++--
.../verify/VerifyBackupPartitionsDumpTask.java | 8 +-
.../cache/verify/VerifyBackupPartitionsTask.java | 2 +-
.../cache/verify/VerifyBackupPartitionsTaskV2.java | 81 ++----
.../visor/snapshot/VisorSnapshotCheckTask.java | 60 ++++
.../GridCacheFastNodeLeftForTransactionTest.java | 2 +-
.../snapshot/IgniteClusterSnapshotCheckTest.java | 318 +++++++++++++++++++++
.../TxRollbackOnTimeoutOnePhaseCommitTest.java | 8 +-
.../apache/ignite/testframework/GridTestUtils.java | 11 -
.../junits/common/GridCommonAbstractTest.java | 4 +-
.../IgniteBasicWithPersistenceTestSuite.java | 2 +
...ridCommandHandlerClusterByClassTest_help.output | 6 +
...andHandlerClusterByClassWithSSLTest_help.output | 6 +
.../visor/verify/ValidateIndexesClosure.java | 9 +-
31 files changed, 1552 insertions(+), 258 deletions(-)
diff --git a/modules/control-utility/src/main/java/org/apache/ignite/internal/commandline/cache/IdleVerify.java b/modules/control-utility/src/main/java/org/apache/ignite/internal/commandline/cache/IdleVerify.java
index 29b3447..bcd4653 100644
--- a/modules/control-utility/src/main/java/org/apache/ignite/internal/commandline/cache/IdleVerify.java
+++ b/modules/control-utility/src/main/java/org/apache/ignite/internal/commandline/cache/IdleVerify.java
@@ -17,6 +17,11 @@
package org.apache.ignite.internal.commandline.cache;
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.PrintWriter;
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
@@ -26,6 +31,7 @@ import java.util.Set;
import java.util.logging.Logger;
import java.util.regex.Pattern;
import java.util.regex.PatternSyntaxException;
+import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
import org.apache.ignite.internal.IgniteNodeAttributes;
import org.apache.ignite.internal.client.GridClient;
@@ -41,6 +47,8 @@ import org.apache.ignite.internal.processors.cache.verify.IdleVerifyResultV2;
import org.apache.ignite.internal.processors.cache.verify.PartitionHashRecord;
import org.apache.ignite.internal.processors.cache.verify.PartitionKey;
import org.apache.ignite.internal.processors.cache.verify.VerifyBackupPartitionsTaskV2;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.internal.visor.verify.CacheFilterEnum;
import org.apache.ignite.internal.visor.verify.VisorIdleVerifyDumpTask;
import org.apache.ignite.internal.visor.verify.VisorIdleVerifyDumpTaskArg;
@@ -67,6 +75,12 @@ import static org.apache.ignite.internal.processors.cache.verify.VerifyBackupPar
*
*/
public class IdleVerify extends AbstractCommand<IdleVerify.Arguments> {
+ /** */
+ public static final String IDLE_VERIFY_FILE_PREFIX = "idle_verify-";
+
+ /** Time formatter for log file name. */
+ private static final DateTimeFormatter TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH-mm-ss_SSS");
+
/** {@inheritDoc} */
@Override public void printUsage(Logger logger) {
String CACHES = "cacheName1,...,cacheNameN";
@@ -322,8 +336,32 @@ public class IdleVerify extends AbstractCommand<IdleVerify.Arguments> {
IdleVerifyResultV2 res = executeTask(client, VisorIdleVerifyTaskV2.class, taskArg, clientCfg);
logParsedArgs(taskArg, System.out::print);
+ res.print(System.out::print, false);
+
+ if (F.isEmpty(res.exceptions()))
+ return;
- res.print(System.out::print);
+ try {
+ File f = new File(U.resolveWorkDirectory(U.defaultWorkDirectory(), "", false),
+ IDLE_VERIFY_FILE_PREFIX + LocalDateTime.now().format(TIME_FORMATTER) + ".txt");
+
+ try (PrintWriter pw = new PrintWriter(f)) {
+ res.print(pw::print, true);
+ pw.flush();
+
+ System.out.println("See log for additional information. " + f.getAbsolutePath());
+ }
+ catch (FileNotFoundException e) {
+ System.err.println("Can't write exceptions to file " + f.getAbsolutePath() + " " + e.getMessage());
+
+ e.printStackTrace();
+ }
+ }
+ catch (IgniteCheckedException e) {
+ System.err.println("Can't find work directory. " + e.getMessage());
+
+ e.printStackTrace();
+ }
}
/**
@@ -350,11 +388,11 @@ public class IdleVerify extends AbstractCommand<IdleVerify.Arguments> {
Map<PartitionKey, List<PartitionHashRecord>> conflicts = res.getConflicts();
if (conflicts.isEmpty()) {
- logger.info("idle_verify check has finished, no conflicts have been found.");
+ logger.info("The check procedure has finished, no conflicts have been found.");
logger.info("");
}
else {
- logger.info("idle_verify check has finished, found " + conflicts.size() + " conflict partitions.");
+ logger.info("The check procedure has finished, found " + conflicts.size() + " conflict partitions.");
logger.info("");
for (Map.Entry<PartitionKey, List<PartitionHashRecord>> entry : conflicts.entrySet()) {
diff --git a/modules/control-utility/src/main/java/org/apache/ignite/internal/commandline/snapshot/SnapshotCommand.java b/modules/control-utility/src/main/java/org/apache/ignite/internal/commandline/snapshot/SnapshotCommand.java
index 2f9597e..79010ee 100644
--- a/modules/control-utility/src/main/java/org/apache/ignite/internal/commandline/snapshot/SnapshotCommand.java
+++ b/modules/control-utility/src/main/java/org/apache/ignite/internal/commandline/snapshot/SnapshotCommand.java
@@ -25,14 +25,15 @@ import org.apache.ignite.internal.commandline.Command;
import org.apache.ignite.internal.commandline.CommandArgIterator;
import org.apache.ignite.internal.commandline.CommandLogger;
import org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotManager;
-import org.apache.ignite.internal.visor.snapshot.VisorSnapshotCancelTask;
-import org.apache.ignite.internal.visor.snapshot.VisorSnapshotCreateTask;
+import org.apache.ignite.internal.processors.cache.verify.IdleVerifyResultV2;
+import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.mxbean.SnapshotMXBean;
import static java.util.Collections.singletonMap;
import static org.apache.ignite.internal.commandline.CommandList.SNAPSHOT;
import static org.apache.ignite.internal.commandline.TaskExecutor.executeTaskByNameOnNode;
import static org.apache.ignite.internal.commandline.snapshot.SnapshotSubcommand.CANCEL;
+import static org.apache.ignite.internal.commandline.snapshot.SnapshotSubcommand.CHECK;
import static org.apache.ignite.internal.commandline.snapshot.SnapshotSubcommand.CREATE;
import static org.apache.ignite.internal.commandline.snapshot.SnapshotSubcommand.of;
@@ -44,21 +45,26 @@ import static org.apache.ignite.internal.commandline.snapshot.SnapshotSubcommand
*/
public class SnapshotCommand extends AbstractCommand<Object> {
/** Command argument. */
- private Object taskArgs;
+ private String snpName;
- /** Task name. */
- private String taskName;
+ /** Snapshot sub-command to execute. */
+ private SnapshotSubcommand cmd;
/** {@inheritDoc} */
@Override public Object execute(GridClientConfiguration clientCfg, Logger log) throws Exception {
try (GridClient client = Command.startClient(clientCfg)) {
- return executeTaskByNameOnNode(
+ Object res = executeTaskByNameOnNode(
client,
- taskName,
- taskArgs,
+ cmd.taskName(),
+ snpName,
null,
clientCfg
);
+
+ if (cmd == CHECK)
+ ((IdleVerifyResultV2)res).print(log::info, true);
+
+ return res;
}
catch (Throwable e) {
log.severe("Failed to perform operation.");
@@ -70,32 +76,16 @@ public class SnapshotCommand extends AbstractCommand<Object> {
/** {@inheritDoc} */
@Override public Object arg() {
- return taskArgs;
+ return snpName;
}
/** {@inheritDoc} */
@Override public void parseArguments(CommandArgIterator argIter) {
- SnapshotSubcommand cmd = of(argIter.nextArg("Expected snapshot action."));
-
- if (cmd == null)
- throw new IllegalArgumentException("Expected correct action.");
-
- switch (cmd) {
- case CREATE:
- taskName = VisorSnapshotCreateTask.class.getName();
- taskArgs = argIter.nextArg("Expected snapshot name.");
-
- break;
+ cmd = of(argIter.nextArg("Expected snapshot action."));
+ snpName = argIter.nextArg("Expected snapshot name.");
- case CANCEL:
- taskName = VisorSnapshotCancelTask.class.getName();
- taskArgs = argIter.nextArg("Expected snapshot name.");
-
- break;
-
- default:
- throw new IllegalArgumentException("Unknown snapshot sub-command: " + cmd);
- }
+ if (F.isEmpty(snpName))
+ throw new IllegalArgumentException("Expected snapshot name.");
}
/** {@inheritDoc} */
@@ -105,6 +95,9 @@ public class SnapshotCommand extends AbstractCommand<Object> {
Command.usage(log, "Cancel running snapshot:", SNAPSHOT, singletonMap("snapshot_name", "Snapshot name."),
CANCEL.toString(), "snapshot_name");
+
+ Command.usage(log, "Check snapshot:", SNAPSHOT, singletonMap("snapshot_name", "Snapshot name."),
+ CHECK.toString(), "snapshot_name");
}
/** {@inheritDoc} */
diff --git a/modules/control-utility/src/main/java/org/apache/ignite/internal/commandline/snapshot/SnapshotSubcommand.java b/modules/control-utility/src/main/java/org/apache/ignite/internal/commandline/snapshot/SnapshotSubcommand.java
index 0d365fa..d9d3a61 100644
--- a/modules/control-utility/src/main/java/org/apache/ignite/internal/commandline/snapshot/SnapshotSubcommand.java
+++ b/modules/control-utility/src/main/java/org/apache/ignite/internal/commandline/snapshot/SnapshotSubcommand.java
@@ -17,6 +17,9 @@
package org.apache.ignite.internal.commandline.snapshot;
+import org.apache.ignite.internal.visor.snapshot.VisorSnapshotCancelTask;
+import org.apache.ignite.internal.visor.snapshot.VisorSnapshotCheckTask;
+import org.apache.ignite.internal.visor.snapshot.VisorSnapshotCreateTask;
import org.jetbrains.annotations.Nullable;
/**
@@ -26,17 +29,24 @@ import org.jetbrains.annotations.Nullable;
*/
public enum SnapshotSubcommand {
/** Sub-command to create a cluster snapshot. */
- CREATE("create"),
+ CREATE("create", VisorSnapshotCreateTask.class.getName()),
/** Sub-command to cancel running snapshot. */
- CANCEL("cancel");
+ CANCEL("cancel", VisorSnapshotCancelTask.class.getName()),
+
+ /** Sub-command to check snapshot. */
+ CHECK("check", VisorSnapshotCheckTask.class.getName());
/** Sub-command name. */
private final String name;
+ /** Task class name to execute. */
+ private final String taskName;
+
/** @param name Snapshot sub-command name. */
- SnapshotSubcommand(String name) {
+ SnapshotSubcommand(String name, String taskName) {
this.name = name;
+ this.taskName = taskName;
}
/**
@@ -49,7 +59,14 @@ public enum SnapshotSubcommand {
return cmd;
}
- return null;
+ throw new IllegalArgumentException("Expected correct action: " + text);
+ }
+
+ /**
+ * @return Task class name to execute.
+ */
+ public String taskName() {
+ return taskName;
}
/** {@inheritDoc} */
diff --git a/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerAbstractTest.java b/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerAbstractTest.java
index 8dfbb37..14b48d1 100644
--- a/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerAbstractTest.java
+++ b/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerAbstractTest.java
@@ -19,6 +19,7 @@ package org.apache.ignite.util;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
+import java.io.File;
import java.io.InputStream;
import java.io.PrintStream;
import java.nio.file.DirectoryStream;
@@ -47,6 +48,7 @@ import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.TestRecordingCommunicationSpi;
import org.apache.ignite.internal.client.GridClientFactory;
import org.apache.ignite.internal.commandline.CommandHandler;
+import org.apache.ignite.internal.commandline.cache.IdleVerify;
import org.apache.ignite.internal.processors.cache.GridCacheFuture;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxPrepareFuture;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxPrepareFutureAdapter;
@@ -74,7 +76,6 @@ import static org.apache.ignite.configuration.EncryptionConfiguration.DFLT_REENC
import static org.apache.ignite.internal.encryption.AbstractEncryptionTest.KEYSTORE_PASSWORD;
import static org.apache.ignite.internal.encryption.AbstractEncryptionTest.KEYSTORE_PATH;
import static org.apache.ignite.internal.processors.cache.verify.VerifyBackupPartitionsDumpTask.IDLE_DUMP_FILE_PREFIX;
-import static org.apache.ignite.testframework.GridTestUtils.cleanIdleVerifyLogFiles;
import static org.apache.ignite.util.GridCommandHandlerTestUtils.addSslParams;
/**
@@ -164,7 +165,9 @@ public abstract class GridCommandHandlerAbstractTest extends GridCommonAbstractT
@Override protected void afterTestsStopped() throws Exception {
super.afterTestsStopped();
- cleanIdleVerifyLogFiles();
+ // Clean idle_verify log files.
+ for (File f : new File(".").listFiles(n -> n.getName().startsWith(IdleVerify.IDLE_VERIFY_FILE_PREFIX)))
+ U.delete(f);
GridClientFactory.stopAll(false);
}
diff --git a/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerClusterByClassTest.java b/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerClusterByClassTest.java
index a8dc398..fde7f72 100644
--- a/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerClusterByClassTest.java
+++ b/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerClusterByClassTest.java
@@ -529,9 +529,9 @@ public class GridCommandHandlerClusterByClassTest extends GridCommandHandlerClus
String zeroUpdateCntrs = new String(Files.readAllBytes(Paths.get(fileNameMatcher.group(1))));
- assertContains(log, zeroUpdateCntrs, "idle_verify check has finished, found " + emptyPartId + " partitions");
+ assertContains(log, zeroUpdateCntrs, "The check procedure has finished, found " + emptyPartId + " partitions");
assertContains(log, zeroUpdateCntrs, "1 partitions was skipped");
- assertContains(log, zeroUpdateCntrs, "idle_verify check has finished, no conflicts have been found.");
+ assertContains(log, zeroUpdateCntrs, "The check procedure has finished, no conflicts have been found.");
assertSort(emptyPartId, zeroUpdateCntrs);
@@ -549,9 +549,9 @@ public class GridCommandHandlerClusterByClassTest extends GridCommandHandlerClus
String nonZeroUpdateCntrs = new String(Files.readAllBytes(Paths.get(fileNameMatcher.group(1))));
- assertContains(log, nonZeroUpdateCntrs, "idle_verify check has finished, found " + 31 + " partitions");
+ assertContains(log, nonZeroUpdateCntrs, "The check procedure has finished, found " + 31 + " partitions");
assertContains(log, nonZeroUpdateCntrs, "1 partitions was skipped");
- assertContains(log, nonZeroUpdateCntrs, "idle_verify check has finished, no conflicts have been found.");
+ assertContains(log, nonZeroUpdateCntrs, "The check procedure has finished, no conflicts have been found.");
assertSort(31, zeroUpdateCntrs);
@@ -587,7 +587,7 @@ public class GridCommandHandlerClusterByClassTest extends GridCommandHandlerClus
if (fileNameMatcher.find()) {
String dumpWithZeros = new String(Files.readAllBytes(Paths.get(fileNameMatcher.group(1))));
- assertContains(log, dumpWithZeros, "idle_verify check has finished, found " + parts + " partitions");
+ assertContains(log, dumpWithZeros, "The check procedure has finished, found " + parts + " partitions");
assertContains(log, dumpWithZeros, "Partition: PartitionKeyV2 [grpId=1544803905, grpName=default, partId=0]");
assertContains(log, dumpWithZeros, "updateCntr=0, partitionState=OWNING, size=0, partHash=0");
assertContains(log, dumpWithZeros, "no conflicts have been found");
@@ -602,7 +602,7 @@ public class GridCommandHandlerClusterByClassTest extends GridCommandHandlerClus
if (fileNameMatcher.find()) {
String dumpWithoutZeros = new String(Files.readAllBytes(Paths.get(fileNameMatcher.group(1))));
- assertContains(log, dumpWithoutZeros, "idle_verify check has finished, found " + keysCount + " partitions");
+ assertContains(log, dumpWithoutZeros, "The check procedure has finished, found " + keysCount + " partitions");
assertContains(log, dumpWithoutZeros, (parts - keysCount) + " partitions was skipped");
assertContains(log, dumpWithoutZeros, "Partition: PartitionKeyV2 [grpId=1544803905, grpName=default, partId=");
@@ -650,37 +650,37 @@ public class GridCommandHandlerClusterByClassTest extends GridCommandHandlerClus
testCacheIdleVerifyMultipleCacheFilterOptionsCommon(
true,
- "idle_verify check has finished, found",
- "idle_verify task was executed with the following args: caches=[], excluded=[wrong.*], cacheFilter=[SYSTEM]",
+ "The check procedure has finished, found",
+ "The check procedure task was executed with the following args: caches=[], excluded=[wrong.*], cacheFilter=[SYSTEM]",
"--cache", "idle_verify", "--dump", "--cache-filter", "SYSTEM", "--exclude-caches", "wrong.*"
);
testCacheIdleVerifyMultipleCacheFilterOptionsCommon(
true,
- "idle_verify check has finished, found 96 partitions",
+ "The check procedure has finished, found 96 partitions",
null,
"--cache", "idle_verify", "--dump", "--exclude-caches", "wrong.*"
);
testCacheIdleVerifyMultipleCacheFilterOptionsCommon(
true,
- "idle_verify check has finished, found 32 partitions",
+ "The check procedure has finished, found 32 partitions",
null,
"--cache", "idle_verify", "--dump", "shared.*"
);
testCacheIdleVerifyMultipleCacheFilterOptionsCommon(
true,
- "idle_verify check has finished, found 160 partitions",
+ "The check procedure has finished, found 160 partitions",
null,
"--cache", "idle_verify", "--dump", "shared.*,wrong.*"
);
testCacheIdleVerifyMultipleCacheFilterOptionsCommon(
true,
- "idle_verify check has finished, found 160 partitions",
+ "The check procedure has finished, found 160 partitions",
null,
"--cache", "idle_verify", "--dump", "shared.*,wrong.*", "--cache-filter", "USER"
);
testCacheIdleVerifyMultipleCacheFilterOptionsCommon(
true,
- "idle_verify check has finished, found 160 partitions",
+ "The check procedure has finished, found 160 partitions",
null,
"--cache", "idle_verify", "--dump", "shared.*,wrong.*"
);
@@ -698,13 +698,13 @@ public class GridCommandHandlerClusterByClassTest extends GridCommandHandlerClus
);
testCacheIdleVerifyMultipleCacheFilterOptionsCommon(
true,
- "idle_verify check has finished, no conflicts have been found.",
+ "The check procedure has finished, no conflicts have been found.",
null,
"--cache", "idle_verify", "--exclude-caches", "wrong.*"
);
testCacheIdleVerifyMultipleCacheFilterOptionsCommon(
true,
- "idle_verify check has finished, no conflicts have been found.",
+ "The check procedure has finished, no conflicts have been found.",
null,
"--cache", "idle_verify", "--dump", "--cache-filter", "PERSISTENT"
);
@@ -1011,7 +1011,7 @@ public class GridCommandHandlerClusterByClassTest extends GridCommandHandlerClus
if (fileNameMatcher.find()) {
String dumpWithConflicts = new String(Files.readAllBytes(Paths.get(fileNameMatcher.group(1))));
- assertContains(log, dumpWithConflicts, "idle_verify check has finished, found 32 partitions");
+ assertContains(log, dumpWithConflicts, "The check procedure has finished, found 32 partitions");
assertContains(log, dumpWithConflicts, "default_third");
assertNotContains(log, dumpWithConflicts, "shared_grp");
}
diff --git a/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerIndexingTest.java b/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerIndexingTest.java
index fda02ea..504a3c1 100644
--- a/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerIndexingTest.java
+++ b/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerIndexingTest.java
@@ -263,7 +263,7 @@ public class GridCommandHandlerIndexingTest extends GridCommandHandlerClusterPer
assertEquals(EXIT_CODE_OK, execute("--cache", "validate_indexes", "--check-crc", CACHE_NAME));
assertContains(log, testOut.toString(), "issues found (listed above)");
- assertContains(log, testOut.toString(), "CRC validation failed");
+ assertContains(log, testOut.toString(), "CRC check of partition failed");
assertNotContains(log, testOut.toString(), "Runtime failure on bounds");
}
diff --git a/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerInterruptCommandTest.java b/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerInterruptCommandTest.java
index d750439..f27cbee 100644
--- a/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerInterruptCommandTest.java
+++ b/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerInterruptCommandTest.java
@@ -229,7 +229,7 @@ public class GridCommandHandlerInterruptCommandTest extends GridCommandHandlerAb
CountDownLatch startTaskLatch = waitForTaskEvent(ignite, IDLE_VERIFY_TASK_V2);
- LogListener lnsrValidationCancelled = LogListener.matches("Idle verify was cancelled.").build();
+ LogListener lnsrValidationCancelled = LogListener.matches("The check procedure was cancelled.").build();
lnsrLog.registerListener(lnsrValidationCancelled);
diff --git a/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerTest.java b/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerTest.java
index 4a2bfe1..f9e3ccb 100644
--- a/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerTest.java
+++ b/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerTest.java
@@ -97,6 +97,7 @@ import org.apache.ignite.internal.processors.cache.persistence.diagnostic.pagelo
import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxEntry;
import org.apache.ignite.internal.processors.cache.transactions.TransactionProxyImpl;
+import org.apache.ignite.internal.processors.cache.verify.IdleVerifyResultV2;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.processors.cache.warmup.BlockedWarmUpConfiguration;
import org.apache.ignite.internal.processors.cache.warmup.BlockedWarmUpStrategy;
@@ -155,6 +156,7 @@ import static org.apache.ignite.internal.commandline.encryption.EncryptionSubcom
import static org.apache.ignite.internal.encryption.AbstractEncryptionTest.MASTER_KEY_NAME_2;
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;
import static org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotManager.SNAPSHOT_METRICS;
import static org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotManager.resolveSnapshotWorkDirectory;
import static org.apache.ignite.internal.processors.cache.verify.IdleVerifyUtility.GRID_NOT_IDLE_MSG;
@@ -2032,7 +2034,7 @@ public class GridCommandHandlerTest extends GridCommandHandlerClusterPerMethodAb
String out = testOut.toString();
- assertContains(log, out, "idle_verify failed");
+ assertContains(log, out, "The check procedure failed");
assertContains(log, out, "See log for additional information.");
String logFileName = (out.split("See log for additional information. ")[1]).split(".txt")[0];
@@ -2136,7 +2138,7 @@ public class GridCommandHandlerTest extends GridCommandHandlerClusterPerMethodAb
String out = testOut.toString();
- assertContains(log, out, "idle_verify failed on 1 node.");
+ assertContains(log, out, "The check procedure failed on 1 node.");
assertContains(log, out, "See log for additional information.");
}
@@ -2156,8 +2158,8 @@ public class GridCommandHandlerTest extends GridCommandHandlerClusterPerMethodAb
String outputStr = testOut.toString();
- assertContains(log,outputStr, "idle_verify failed on 1 node.");
- assertContains(log, outputStr, "idle_verify check has finished, no conflicts have been found.");
+ assertContains(log,outputStr, "The check procedure failed on 1 node.");
+ assertContains(log, outputStr, "The check procedure has finished, no conflicts have been found.");
}
/** */
@@ -2219,7 +2221,7 @@ public class GridCommandHandlerTest extends GridCommandHandlerClusterPerMethodAb
if (fileNameMatcher.find()) {
String dumpWithConflicts = new String(Files.readAllBytes(Paths.get(fileNameMatcher.group(1))));
- assertContains(log, dumpWithConflicts, "Idle verify failed on nodes:");
+ assertContains(log, dumpWithConflicts, "The check procedure failed on nodes:");
assertContains(log, dumpWithConflicts, "Node ID: " + unstableNodeId);
}
@@ -3086,6 +3088,30 @@ public class GridCommandHandlerTest extends GridCommandHandlerClusterPerMethodAb
snpName -> assertEquals(EXIT_CODE_OK, execute(h, "--snapshot", "cancel", snpName)));
}
+ /** @throws Exception If fails. */
+ @Test
+ public void testCheckSnapshot() throws Exception {
+ String snpName = "snapshot_02052020";
+
+ IgniteEx ig = startGrid(0);
+ ig.cluster().state(ACTIVE);
+
+ createCacheAndPreload(ig, 1000);
+
+ snp(ig).createSnapshot(snpName)
+ .get();
+
+ CommandHandler h = new CommandHandler();
+
+ assertEquals(EXIT_CODE_OK, execute(h, "--snapshot", "check", snpName));
+
+ StringBuilder sb = new StringBuilder();
+
+ ((IdleVerifyResultV2)h.getLastOperationResult()).print(sb::append, true);
+
+ assertContains(log, sb.toString(), "The check procedure has finished, no conflicts have been found");
+ }
+
/**
* @throws Exception If failed.
*/
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FilePageStore.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FilePageStore.java
index 67a8c51..4a90d39 100755
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FilePageStore.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FilePageStore.java
@@ -118,7 +118,7 @@ public class FilePageStore implements PageStore {
private volatile int tag;
/** */
- private boolean skipCrc = IgniteSystemProperties.getBoolean(IGNITE_PDS_SKIP_CRC);
+ private final boolean skipCrc = IgniteSystemProperties.getBoolean(IGNITE_PDS_SKIP_CRC);
/** */
private final ReadWriteLock lock = new ReentrantReadWriteLock();
@@ -484,6 +484,18 @@ public class FilePageStore implements PageStore {
/** {@inheritDoc} */
@Override public boolean read(long pageId, ByteBuffer pageBuf, boolean keepCrc) throws IgniteCheckedException {
+ return read(pageId, pageBuf, !skipCrc, keepCrc);
+ }
+
+ /**
+ * @param pageId Page ID.
+ * @param pageBuf Page buffer to read into.
+ * @param checkCrc Check CRC on page.
+ * @param keepCrc By default reading zeroes CRC which was on file, but you can keep it in pageBuf if set keepCrc
+ * @return {@code true} if page has been read successfully, {@code false} if page hasn't been written yet.
+ * @throws IgniteCheckedException If reading failed (IO error occurred).
+ */
+ public boolean read(long pageId, ByteBuffer pageBuf, boolean checkCrc, boolean keepCrc) throws IgniteCheckedException {
init();
try {
@@ -512,7 +524,7 @@ public class FilePageStore implements PageStore {
pageBuf.position(0);
- if (!skipCrc) {
+ if (checkCrc) {
int curCrc32 = FastCrc.calcCrc(pageBuf, getCrcSize(pageId, pageBuf));
if ((savedCrc32 ^ curCrc32) != 0)
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FilePageStoreManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FilePageStoreManager.java
index 2e29a36..755dc6f 100755
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FilePageStoreManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FilePageStoreManager.java
@@ -1000,6 +1000,68 @@ public class FilePageStoreManager extends GridCacheSharedManagerAdapter implemen
}
/**
+ * @param dir Directory to check.
+ * @return Files that match cache or cache group pattern.
+ */
+ public static List<File> cacheDirectories(File dir) {
+ File[] files = dir.listFiles();
+
+ if (files == null)
+ return Collections.emptyList();
+
+ return Arrays.stream(dir.listFiles())
+ .sorted()
+ .filter(File::isDirectory)
+ .filter(f -> f.getName().startsWith(CACHE_DIR_PREFIX) || f.getName().startsWith(CACHE_GRP_DIR_PREFIX))
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * @param partFileName Partition file name.
+ * @return Partition id.
+ */
+ public static int partId(String partFileName) {
+ if (partFileName.equals(INDEX_FILE_NAME))
+ return PageIdAllocator.INDEX_PARTITION;
+
+ if (partFileName.startsWith(PART_FILE_PREFIX))
+ return Integer.parseInt(partFileName.substring(PART_FILE_PREFIX.length(), partFileName.indexOf('.')));
+
+ throw new IllegalStateException("Illegal partition file name: " + partFileName);
+ }
+
+ /**
+ * @param cacheDir Cache directory to check.
+ * @return List of cache partitions in given directory.
+ */
+ public static List<File> cachePartitionFiles(File cacheDir) {
+ File[] files = cacheDir.listFiles();
+
+ if (files == null)
+ return Collections.emptyList();
+
+ return Arrays.stream(files)
+ .filter(File::isFile)
+ .filter(f -> f.getName().startsWith(PART_FILE_PREFIX))
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * @param dir Cache directory on disk.
+ * @return Cache or cache group name.
+ */
+ public static String cacheGroupName(File dir) {
+ String name = dir.getName();
+
+ if (name.startsWith(CACHE_GRP_DIR_PREFIX))
+ return name.substring(CACHE_GRP_DIR_PREFIX.length());
+ else if (name.startsWith(CACHE_DIR_PREFIX))
+ return name.substring(CACHE_DIR_PREFIX.length());
+ else
+ throw new IgniteException("Directory doesn't match the cache or cache group prefix: " + dir);
+ }
+
+ /**
* @param grpDir Group directory.
* @param ccfgs Cache configurations.
* @throws IgniteCheckedException If failed.
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
index 621e765..d48985c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
@@ -17,9 +17,14 @@
package org.apache.ignite.internal.processors.cache.persistence.snapshot;
+import java.io.BufferedInputStream;
+import java.io.BufferedOutputStream;
import java.io.File;
+import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
import java.io.Serializable;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
@@ -30,6 +35,7 @@ import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.SimpleFileVisitor;
import java.nio.file.attribute.BasicFileAttributes;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
@@ -84,6 +90,7 @@ import org.apache.ignite.internal.processors.cache.persistence.metastorage.ReadW
import org.apache.ignite.internal.processors.cache.persistence.partstate.GroupPartitionId;
import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO;
import org.apache.ignite.internal.processors.cache.persistence.wal.crc.FastCrc;
+import org.apache.ignite.internal.processors.cache.verify.IdleVerifyResultV2;
import org.apache.ignite.internal.processors.cluster.DiscoveryDataClusterState;
import org.apache.ignite.internal.processors.marshaller.MappedName;
import org.apache.ignite.internal.processors.metric.MetricRegistry;
@@ -105,6 +112,8 @@ import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteCallable;
import org.apache.ignite.lang.IgniteFuture;
+import org.apache.ignite.marshaller.Marshaller;
+import org.apache.ignite.marshaller.MarshallerUtils;
import org.apache.ignite.resources.IgniteInstanceResource;
import org.apache.ignite.thread.IgniteThreadPoolExecutor;
import org.apache.ignite.thread.OomExceptionHandler;
@@ -130,9 +139,11 @@ import static org.apache.ignite.internal.pagemem.PageIdAllocator.MAX_PARTITION_I
import static org.apache.ignite.internal.processors.cache.binary.CacheObjectBinaryProcessorImpl.binaryWorkDir;
import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.INDEX_FILE_NAME;
import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.PART_FILE_TEMPLATE;
+import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.cacheDirectories;
import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.getPartitionFile;
import static org.apache.ignite.internal.processors.cache.persistence.filename.PdsConsistentIdProcessor.DB_DEFAULT_FOLDER;
import static org.apache.ignite.internal.processors.cache.persistence.partstate.GroupPartitionId.getTypeByPartId;
+import static org.apache.ignite.internal.processors.task.GridTaskThreadContextKey.TC_SKIP_AUTH;
import static org.apache.ignite.internal.util.IgniteUtils.isLocalNodeCoordinator;
import static org.apache.ignite.internal.util.distributed.DistributedProcess.DistributedProcessType.END_SNAPSHOT;
import static org.apache.ignite.internal.util.distributed.DistributedProcess.DistributedProcessType.START_SNAPSHOT;
@@ -176,6 +187,9 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
/** Snapshot metrics prefix. */
public static final String SNAPSHOT_METRICS = "snapshot";
+ /** Snapshot metafile extension. */
+ public static final String SNAPSHOT_METAFILE_EXT = ".smf";
+
/** Prefix for snapshot threads. */
private static final String SNAPSHOT_RUNNER_THREAD_PREFIX = "snapshot-runner";
@@ -211,6 +225,9 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
/** Check previously performed snapshot operation and delete uncompleted files if need. */
private final DistributedProcess<SnapshotOperationRequest, SnapshotOperationResponse> endSnpProc;
+ /** Marshaller. */
+ private final Marshaller marsh;
+
/** Resolved persistent data storage settings. */
private volatile PdsFolderSettings pdsSettings;
@@ -266,6 +283,8 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
endSnpProc = new DistributedProcess<>(ctx, END_SNAPSHOT, this::initLocalSnapshotEndStage,
this::processLocalSnapshotEndStageResult);
+
+ marsh = MarshallerUtils.jdkMarshaller(ctx.igniteInstanceName());
}
/**
@@ -526,21 +545,55 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
parts.put(grpId, null);
}
+ IgniteInternalFuture<Set<GroupPartitionId>> task0;
+
if (parts.isEmpty())
- return new GridFinishedFuture<>();
+ task0 = new GridFinishedFuture<>(Collections.emptySet());
+ else {
+ task0 = registerSnapshotTask(req.snpName,
+ req.srcNodeId,
+ parts,
+ locSndrFactory.apply(req.snpName));
+
+ clusterSnpReq = req;
+ }
- SnapshotFutureTask task0 = registerSnapshotTask(req.snpName,
- req.srcNodeId,
- parts,
- locSndrFactory.apply(req.snpName));
+ return task0.chain(fut -> {
+ if (fut.error() != null)
+ throw F.wrap(fut.error());
- clusterSnpReq = req;
+ try {
+ Set<String> blts = req.bltNodes.stream()
+ .map(n -> cctx.discovery().node(n).consistentId().toString())
+ .collect(Collectors.toSet());
+
+ File smf = new File(snapshotLocalDir(req.snpName), snapshotMetaFileName(cctx.localNode().consistentId().toString()));
+
+ if (smf.exists())
+ throw new GridClosureException(new IgniteException("Snapshot metafile must not exist: " + smf.getAbsolutePath()));
+
+ smf.getParentFile().mkdirs();
+
+ try (OutputStream out = new BufferedOutputStream(new FileOutputStream(smf))) {
+ U.marshal(marsh,
+ new SnapshotMetadata(req.rqId,
+ req.snpName,
+ cctx.localNode().consistentId().toString(),
+ pdsSettings.folderName(),
+ cctx.gridConfig().getDataStorageConfiguration().getPageSize(),
+ req.grpIds,
+ blts,
+ fut.result()),
+ out);
+
+ log.info("Snapshot metafile has been created: " + smf.getAbsolutePath());
+ }
- return task0.chain(fut -> {
- if (fut.error() == null)
return new SnapshotOperationResponse();
- else
- throw new GridClosureException(fut.error());
+ }
+ catch (IOException | IgniteCheckedException e) {
+ throw F.wrap(e);
+ }
});
}
@@ -738,6 +791,137 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
}
}
+ /**
+ * @param name Snapshot name.
+ * @return {@code true} if snapshot is OK.
+ */
+ public IgniteInternalFuture<IdleVerifyResultV2> checkSnapshot(String name) {
+ A.notNullOrEmpty(name, "Snapshot name cannot be null or empty.");
+ A.ensure(U.alphanumericUnderscore(name), "Snapshot name must satisfy the following name pattern: a-zA-Z0-9_");
+
+ GridKernalContext kctx0 = cctx.kernalContext();
+ GridFutureAdapter<IdleVerifyResultV2> res = new GridFutureAdapter<>();
+
+ kctx0.security().authorize(ADMIN_SNAPSHOT);
+
+ kctx0.task().setThreadContext(TC_SKIP_AUTH, true);
+ kctx0.task().execute(SnapshotMetadataCollectorTask.class, name)
+ .listen(f0 -> {
+ if (f0.error() == null) {
+ kctx0.task().setThreadContext(TC_SKIP_AUTH, true);
+ kctx0.task().execute(SnapshotPartitionsVerifyTask.class, f0.result())
+ .listen(f1 -> {
+ if (f1.error() == null)
+ res.onDone(f1.result());
+ else if (f1.error() instanceof IgniteSnapshotVerifyException)
+ res.onDone(new IdleVerifyResultV2(((IgniteSnapshotVerifyException)f1.error()).exceptions()));
+ else
+ res.onDone(f1.error());
+ });
+ }
+ else {
+ if (f0.error() instanceof IgniteSnapshotVerifyException)
+ res.onDone(new IdleVerifyResultV2(((IgniteSnapshotVerifyException)f0.error()).exceptions()));
+ else
+ res.onDone(f0.error());
+ }
+ });
+
+ return res;
+ }
+
+ /**
+ * @param snpName Snapshot name.
+ * @param folderName Directory name for cache group.
+ * @return The list of cache or cache group names in given snapshot on local node.
+ */
+ public List<File> snapshotCacheDirectories(String snpName, String folderName) {
+ File snpDir = snapshotLocalDir(snpName);
+
+ if (!snpDir.exists())
+ return Collections.emptyList();
+
+ return cacheDirectories(new File(snpDir, databaseRelativePath(folderName)));
+ }
+
+ /**
+ * @param snpName Snapshot name.
+ * @param consId Node consistent id to read medata for.
+ * @return Snapshot metadata instance.
+ */
+ public SnapshotMetadata readSnapshotMetadata(String snpName, String consId) {
+ return readSnapshotMetadata(new File(snapshotLocalDir(snpName), snapshotMetaFileName(consId)));
+ }
+
+ /**
+ * @param smf File denoting to snapshot metafile.
+ * @return Snapshot metadata instance.
+ */
+ private SnapshotMetadata readSnapshotMetadata(File smf) {
+ if (!smf.exists())
+ throw new IgniteException("Snapshot metafile cannot be read due to it doesn't exist: " + smf);
+
+ String smfName = smf.getName().substring(0, smf.getName().length() - SNAPSHOT_METAFILE_EXT.length());
+
+ try (InputStream in = new BufferedInputStream(new FileInputStream(smf))) {
+ SnapshotMetadata meta = marsh.unmarshal(in, U.resolveClassLoader(cctx.gridConfig()));
+
+ if (!U.maskForFileName(meta.consistentId()).equals(smfName))
+ throw new IgniteException("Error reading snapshot metadata [smfName=" + smfName + ", consId=" + U.maskForFileName(meta.consistentId()));
+
+ return meta;
+ }
+ catch (IgniteCheckedException | IOException e) {
+ throw new IgniteException("An error occurred during reading snapshot metadata file [file=" +
+ smf.getAbsolutePath() + "]", e);
+ }
+ }
+
+ /**
+ * @param snpName Snapshot name.
+ * @return List of snapshot metadata for the given snapshot name on local node.
+ * If snapshot has been taken from local node the snapshot metadata for given
+ * local node will be placed on the first place.
+ */
+ public List<SnapshotMetadata> readSnapshotMetadatas(String snpName) {
+ A.notNullOrEmpty(snpName, "Snapshot name cannot be null or empty.");
+ A.ensure(U.alphanumericUnderscore(snpName), "Snapshot name must satisfy the following name pattern: a-zA-Z0-9_");
+
+ File[] smfs = snapshotLocalDir(snpName).listFiles((dir, name) ->
+ name.toLowerCase().endsWith(SNAPSHOT_METAFILE_EXT));
+
+ if (smfs == null)
+ throw new IgniteException("Snapshot directory doesn't exists or an I/O error occurred during directory read.");
+
+ Map<String, SnapshotMetadata> metasMap = new HashMap<>();
+ SnapshotMetadata prev = null;
+
+ for (File smf : smfs) {
+ SnapshotMetadata curr = readSnapshotMetadata(smf);
+
+ if (prev != null && !prev.sameSnapshot(curr))
+ throw new IgniteException("Snapshot metadata files are from different snapshots [prev=" + prev + ", curr=" + curr);
+
+ metasMap.put(curr.consistentId(), curr);
+
+ prev = curr;
+ }
+
+ SnapshotMetadata currNodeSmf = metasMap.remove(cctx.localNode().consistentId().toString());
+
+ // Snapshot metadata for the local node must be first in the result map.
+ if (currNodeSmf == null)
+ return new ArrayList<>(metasMap.values());
+ else {
+ List<SnapshotMetadata> result = new ArrayList<>();
+
+ result.add(currNodeSmf);
+ result.addAll(metasMap.values());
+
+ return result;
+ }
+ }
+
/** {@inheritDoc} */
@Override public IgniteFuture<Void> createSnapshot(String name) {
A.notNullOrEmpty(name, "Snapshot name cannot be null or empty.");
@@ -922,6 +1106,14 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
}
/**
+ * @param consId Consistent node id.
+ * @return Snapshot metadata file name.
+ */
+ private static String snapshotMetaFileName(String consId) {
+ return U.maskForFileName(consId) + SNAPSHOT_METAFILE_EXT;
+ }
+
+ /**
* @param snpName Unique snapshot name.
* @param srcNodeId Node id which cause snapshot operation.
* @param parts Collection of pairs group and appropriate cache partition to be snapshot.
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotVerifyException.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotVerifyException.java
new file mode 100644
index 0000000..bcaea42
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotVerifyException.java
@@ -0,0 +1,48 @@
+/*
+ * 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.persistence.snapshot;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.cluster.ClusterNode;
+
+/**
+ * Compound snapshot verification exception from the nodes where the verification process executed.
+ */
+public class IgniteSnapshotVerifyException extends IgniteException {
+ /** Serial version UID. */
+ private static final long serialVersionUID = 0L;
+
+ /** Map of received exceptions. */
+ private final Map<ClusterNode, Exception> exs = new HashMap<>();
+
+ /**
+ * @param map Map of received exceptions.
+ */
+ public IgniteSnapshotVerifyException(Map<ClusterNode, ? extends Exception> map) {
+ exs.putAll(map);
+ }
+
+ /**
+ * @return Map of received exceptions.
+ */
+ public Map<ClusterNode, Exception> exceptions() {
+ return exs;
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotFutureTask.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotFutureTask.java
index cd4aaa1..89484f7 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotFutureTask.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotFutureTask.java
@@ -87,7 +87,7 @@ import static org.apache.ignite.internal.processors.cache.persistence.snapshot.I
/**
*
*/
-class SnapshotFutureTask extends GridFutureAdapter<Boolean> implements CheckpointListener {
+class SnapshotFutureTask extends GridFutureAdapter<Set<GroupPartitionId>> implements CheckpointListener {
/** Shared context. */
private final GridCacheSharedContext<?, ?> cctx;
@@ -268,7 +268,7 @@ class SnapshotFutureTask extends GridFutureAdapter<Boolean> implements Checkpoin
}
/** {@inheritDoc} */
- @Override public boolean onDone(@Nullable Boolean res, @Nullable Throwable err) {
+ @Override public boolean onDone(@Nullable Set<GroupPartitionId> res, @Nullable Throwable err) {
for (PageStoreSerialWriter writer : partDeltaWriters.values())
U.closeQuiet(writer);
@@ -620,7 +620,7 @@ class SnapshotFutureTask extends GridFutureAdapter<Boolean> implements Checkpoin
if (closeFut == null) {
Throwable err0 = err.get();
- closeFut = CompletableFuture.runAsync(() -> onDone(true, err0),
+ closeFut = CompletableFuture.runAsync(() -> onDone(partFileLengths.keySet(), err0),
cctx.kernalContext().getSystemExecutorService());
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotMetadata.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotMetadata.java
new file mode 100644
index 0000000..d2a8918
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotMetadata.java
@@ -0,0 +1,198 @@
+/*
+ * 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.persistence.snapshot;
+
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.UUID;
+import org.apache.ignite.internal.processors.cache.persistence.partstate.GroupPartitionId;
+import org.apache.ignite.internal.util.tostring.GridToStringInclude;
+import org.apache.ignite.internal.util.typedef.internal.S;
+
+/**
+ * Snapshot metadata file.
+ */
+public class SnapshotMetadata implements Serializable {
+ /** Serial version uid. */
+ private static final long serialVersionUID = 0L;
+
+ /** Unique snapshot request id. */
+ private final UUID rqId;
+
+ /** Snapshot name. */
+ private final String snpName;
+
+ /** Consistent id of a node to which this metadata relates. */
+ private final String consId;
+
+ /**
+ * Directory related to the current consistent node id on which partition files are stored.
+ * For some of the cases, consId doesn't equal the directory name.
+ */
+ private final String folderName;
+
+ /** Page size of stored snapshot data. */
+ private final int pageSize;
+
+ /** The list of cache groups ids which were included into snapshot. */
+ @GridToStringInclude
+ private final List<Integer> grpIds;
+
+ /** The set of affected by snapshot baseline nodes. */
+ @GridToStringInclude
+ private final Set<String> bltNodes;
+
+ /**
+ * Map of cache group partitions from which snapshot has been taken on the local node. This map can be empty
+ * since for instance, due to the node filter there is no cache data on node.
+ */
+ @GridToStringInclude
+ private final Map<Integer, Set<Integer>> locParts = new HashMap<>();
+
+ /**
+ * @param rqId Unique snapshot request id.
+ * @param snpName Snapshot name.
+ * @param consId Consistent id of a node to which this metadata relates.
+ * @param folderName Directory name which stores the data files.
+ * @param pageSize Page size of stored snapshot data.
+ * @param grpIds The list of cache groups ids which were included into snapshot.
+ * @param bltNodes The set of affected by snapshot baseline nodes.
+ */
+ public SnapshotMetadata(
+ UUID rqId,
+ String snpName,
+ String consId,
+ String folderName,
+ int pageSize,
+ List<Integer> grpIds,
+ Set<String> bltNodes,
+ Set<GroupPartitionId> pairs
+ ) {
+ this.rqId = rqId;
+ this.snpName = snpName;
+ this.consId = consId;
+ this.folderName = folderName;
+ this.pageSize = pageSize;
+ this.grpIds = grpIds;
+ this.bltNodes = bltNodes;
+
+ pairs.forEach(p ->
+ locParts.computeIfAbsent(p.getGroupId(), k -> new HashSet<>())
+ .add(p.getPartitionId()));
+ }
+
+ /**
+ * @return Unique snapshot request id.
+ */
+ public UUID requestId() {
+ return rqId;
+ }
+
+ /**
+ * @return Snapshot name.
+ */
+ public String snapshotName() {
+ return snpName;
+ }
+
+ /**
+ * @return Consistent id of a node to which this metadata relates.
+ */
+ public String consistentId() {
+ return consId;
+ }
+
+ /**
+ * @return Directory name which stores the data files.
+ */
+ public String folderName() {
+ return folderName;
+ }
+
+ /**
+ * @return Page size of stored snapshot data.
+ */
+ public int pageSize() {
+ return pageSize;
+ }
+
+ /**
+ * @return The list of cache groups ids which were included into snapshot.
+ */
+ public List<Integer> cacheGroupIds() {
+ return grpIds;
+ }
+
+ /**
+ * @return The set of affected by snapshot baseline nodes.
+ */
+ public Set<String> baselineNodes() {
+ return bltNodes;
+ }
+
+ /**
+ * @return Map of cache group partitions from which snapshot has been taken on local node.
+ */
+ public Map<Integer, Set<Integer>> partitions() {
+ return locParts;
+ }
+
+ /**
+ * @param compare Snapshot metadata to compare.
+ * @return {@code true} if given metadata belongs to the same snapshot.
+ */
+ public boolean sameSnapshot(SnapshotMetadata compare) {
+ return requestId().equals(compare.requestId()) &&
+ snapshotName().equals(compare.snapshotName()) &&
+ pageSize() == compare.pageSize() &&
+ Objects.equals(cacheGroupIds(), compare.cacheGroupIds()) &&
+ Objects.equals(baselineNodes(), compare.baselineNodes());
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean equals(Object o) {
+ if (this == o)
+ return true;
+
+ if (o == null || getClass() != o.getClass())
+ return false;
+
+ SnapshotMetadata meta = (SnapshotMetadata)o;
+
+ return rqId.equals(meta.rqId) &&
+ snpName.equals(meta.snpName) &&
+ consId.equals(meta.consId) &&
+ Objects.equals(grpIds, meta.grpIds) &&
+ Objects.equals(bltNodes, meta.bltNodes);
+ }
+
+ /** {@inheritDoc} */
+ @Override public int hashCode() {
+ return Objects.hash(rqId, snpName, consId, grpIds, bltNodes);
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(SnapshotMetadata.class, this);
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotMetadataCollectorTask.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotMetadataCollectorTask.java
new file mode 100644
index 0000000..8ddd00a
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotMetadataCollectorTask.java
@@ -0,0 +1,111 @@
+/*
+ * 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.persistence.snapshot;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.compute.ComputeJob;
+import org.apache.ignite.compute.ComputeJobAdapter;
+import org.apache.ignite.compute.ComputeJobResult;
+import org.apache.ignite.compute.ComputeJobResultPolicy;
+import org.apache.ignite.compute.ComputeTaskAdapter;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.processors.task.GridInternal;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.resources.IgniteInstanceResource;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/** Snapshot task to collect snapshot metadata from the baseline nodes for given snapshot name. */
+@GridInternal
+public class SnapshotMetadataCollectorTask
+ extends ComputeTaskAdapter<String, Map<ClusterNode, List<SnapshotMetadata>>> {
+ /** Serial version uid. */
+ private static final long serialVersionUID = 0L;
+
+ @Override public @NotNull Map<? extends ComputeJob, ClusterNode> map(
+ List<ClusterNode> subgrid,
+ @Nullable String snpName
+ ) throws IgniteException {
+ Map<ComputeJob, ClusterNode> map = U.newHashMap(subgrid.size());
+
+ for (ClusterNode node : subgrid) {
+ map.put(new ComputeJobAdapter(snpName) {
+ @IgniteInstanceResource
+ private transient IgniteEx ignite;
+
+ @Override public List<SnapshotMetadata> execute() throws IgniteException {
+ return ignite.context().cache().context().snapshotMgr()
+ .readSnapshotMetadatas(snpName);
+ }
+ }, node);
+ }
+
+ return map;
+ }
+
+ @Override public @Nullable Map<ClusterNode, List<SnapshotMetadata>> reduce(
+ List<ComputeJobResult> results
+ ) throws IgniteException {
+ Map<ClusterNode, List<SnapshotMetadata>> reduceRes = new HashMap<>();
+ Map<ClusterNode, Exception> exs = new HashMap<>();
+
+ SnapshotMetadata first = null;
+
+ for (ComputeJobResult res: results) {
+ if (res.getException() != null) {
+ exs.put(res.getNode(), res.getException());
+
+ continue;
+ }
+
+ List<SnapshotMetadata> metas = res.getData();
+
+ for (SnapshotMetadata meta : metas) {
+ if (first == null)
+ first = meta;
+
+ if (!first.sameSnapshot(meta)) {
+ exs.put(res.getNode(),
+ new IgniteException("An error occurred during comparing snapshot metadata from cluster nodes " +
+ "[first=" + first + ", meta=" + meta + ", nodeId=" + res.getNode().id() + ']'));
+
+ continue;
+ }
+
+ reduceRes.computeIfAbsent(res.getNode(), n -> new ArrayList<>())
+ .add(meta);
+ }
+ }
+
+ if (exs.isEmpty())
+ return reduceRes;
+ else
+ throw new IgniteSnapshotVerifyException(exs);
+ }
+
+ /** {@inheritDoc} */
+ @Override public ComputeJobResultPolicy result(ComputeJobResult res, List<ComputeJobResult> rcvd) throws IgniteException {
+ // Handle all exceptions during the `reduce` operation.
+ return ComputeJobResultPolicy.WAIT;
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotPartitionsVerifyTask.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotPartitionsVerifyTask.java
new file mode 100644
index 0000000..c6a8a72
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotPartitionsVerifyTask.java
@@ -0,0 +1,298 @@
+/*
+ * 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.persistence.snapshot;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.IgniteLogger;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.compute.ComputeJob;
+import org.apache.ignite.compute.ComputeJobAdapter;
+import org.apache.ignite.compute.ComputeJobResult;
+import org.apache.ignite.compute.ComputeJobResultPolicy;
+import org.apache.ignite.compute.ComputeTaskAdapter;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.pagemem.PageIdAllocator;
+import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState;
+import org.apache.ignite.internal.processors.cache.persistence.file.FilePageStore;
+import org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.PagePartitionMetaIO;
+import org.apache.ignite.internal.processors.cache.verify.IdleVerifyResultV2;
+import org.apache.ignite.internal.processors.cache.verify.PartitionHashRecordV2;
+import org.apache.ignite.internal.processors.cache.verify.PartitionKeyV2;
+import org.apache.ignite.internal.processors.cache.verify.VerifyBackupPartitionsTaskV2;
+import org.apache.ignite.internal.processors.task.GridInternal;
+import org.apache.ignite.internal.util.GridUnsafe;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.internal.CU;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.resources.IgniteInstanceResource;
+import org.apache.ignite.resources.LoggerResource;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState.OWNING;
+import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState.fromOrdinal;
+import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.cacheGroupName;
+import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.cachePartitionFiles;
+import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.partId;
+import static org.apache.ignite.internal.processors.cache.persistence.partstate.GroupPartitionId.getTypeByPartId;
+import static org.apache.ignite.internal.processors.cache.verify.IdleVerifyUtility.checkPartitionsPageCrcSum;
+
+/** */
+@GridInternal
+public class SnapshotPartitionsVerifyTask
+ extends ComputeTaskAdapter<Map<ClusterNode, List<SnapshotMetadata>>, IdleVerifyResultV2> {
+ /** Serial version uid. */
+ private static final long serialVersionUID = 0L;
+
+ /** Ignite instance. */
+ @IgniteInstanceResource
+ private IgniteEx ignite;
+
+ /** {@inheritDoc} */
+ @Override public @NotNull Map<? extends ComputeJob, ClusterNode> map(
+ List<ClusterNode> subgrid,
+ @Nullable Map<ClusterNode, List<SnapshotMetadata>> clusterMetas
+ ) throws IgniteException {
+ if (!subgrid.containsAll(clusterMetas.keySet())) {
+ throw new IgniteSnapshotVerifyException(F.asMap(ignite.localNode(),
+ new IgniteException("Some of Ignite nodes left the cluster during the snapshot verification " +
+ "[curr=" + F.viewReadOnly(subgrid, F.node2id()) +
+ ", init=" + F.viewReadOnly(clusterMetas.keySet(), F.node2id()) + ']')));
+ }
+
+ Map<ComputeJob, ClusterNode> jobs = new HashMap<>();
+ Set<SnapshotMetadata> allMetas = new HashSet<>();
+ clusterMetas.values().forEach(allMetas::addAll);
+
+ Set<String> missed = null;
+
+ for (SnapshotMetadata meta : allMetas) {
+ if (missed == null)
+ missed = new HashSet<>(meta.baselineNodes());
+
+ missed.remove(meta.consistentId());
+
+ if (missed.isEmpty())
+ break;
+ }
+
+ if (!missed.isEmpty()) {
+ throw new IgniteSnapshotVerifyException(F.asMap(ignite.localNode(),
+ new IgniteException("Some metadata is missing from the snapshot: " + missed)));
+ }
+
+ while (!allMetas.isEmpty()) {
+ for (Map.Entry<ClusterNode, List<SnapshotMetadata>> e : clusterMetas.entrySet()) {
+ SnapshotMetadata meta = F.find(e.getValue(), null, allMetas::remove);
+
+ if (meta == null)
+ continue;
+
+ jobs.put(new VisorVerifySnapshotPartitionsJob(meta.snapshotName(), meta.consistentId()), e.getKey());
+
+ if (allMetas.isEmpty())
+ break;
+ }
+ }
+
+ return jobs;
+ }
+
+ /** {@inheritDoc} */
+ @Override public @Nullable IdleVerifyResultV2 reduce(List<ComputeJobResult> results) throws IgniteException {
+ return VerifyBackupPartitionsTaskV2.reduce0(results);
+ }
+
+ /** {@inheritDoc} */
+ @Override public ComputeJobResultPolicy result(ComputeJobResult res, List<ComputeJobResult> rcvd) throws IgniteException {
+ // Handle all exceptions during the `reduce` operation.
+ return ComputeJobResultPolicy.WAIT;
+ }
+
+ /** Job that collects update counters of snapshot partitions on the node it executes. */
+ private static class VisorVerifySnapshotPartitionsJob extends ComputeJobAdapter {
+ /** Serial version uid. */
+ private static final long serialVersionUID = 0L;
+
+ /** Ignite instance. */
+ @IgniteInstanceResource
+ private IgniteEx ignite;
+
+ /** Injected logger. */
+ @LoggerResource
+ private IgniteLogger log;
+
+ /** Snapshot name to validate. */
+ private final String snpName;
+
+ /** Consistent snapshot metadata file name. */
+ private final String consId;
+
+ /**
+ * @param snpName Snapshot name to validate.
+ * @param consId Consistent snapshot metadata file name.
+ */
+ public VisorVerifySnapshotPartitionsJob(String snpName, String consId) {
+ this.snpName = snpName;
+ this.consId = consId;
+ }
+
+ @Override public Map<PartitionKeyV2, PartitionHashRecordV2> execute() throws IgniteException {
+ IgniteSnapshotManager snpMgr = ignite.context().cache().context().snapshotMgr();
+
+ if (log.isInfoEnabled()) {
+ log.info("Verify snapshot partitions procedure has been initiated " +
+ "[snpName=" + snpName + ", consId=" + consId + ']');
+ }
+
+ SnapshotMetadata meta = snpMgr.readSnapshotMetadata(snpName, consId);
+ Set<Integer> grps = new HashSet<>(meta.partitions().keySet());
+ Set<File> partFiles = new HashSet<>();
+
+ for (File dir : snpMgr.snapshotCacheDirectories(snpName, meta.folderName())) {
+ int grpId = CU.cacheId(cacheGroupName(dir));
+
+ if (!grps.remove(grpId))
+ continue;
+
+ Set<Integer> parts = new HashSet<>(meta.partitions().get(grpId));
+
+ for (File part : cachePartitionFiles(dir)) {
+ int partId = partId(part.getName());
+
+ if (!parts.remove(partId))
+ continue;
+
+ partFiles.add(part);
+ }
+
+ if (!parts.isEmpty()) {
+ throw new IgniteException("Snapshot data doesn't contain required cache group partition " +
+ "[grpId=" + grpId + ", snpName=" + snpName + ", consId=" + consId +
+ ", missed=" + parts + ", meta=" + meta + ']');
+ }
+ }
+
+ if (!grps.isEmpty()) {
+ throw new IgniteException("Snapshot data doesn't contain required cache groups " +
+ "[grps=" + grps + ", snpName=" + snpName + ", consId=" + consId +
+ ", meta=" + meta + ']');
+ }
+
+ Map<PartitionKeyV2, PartitionHashRecordV2> res = new HashMap<>();
+ ThreadLocal<ByteBuffer> buff = ThreadLocal.withInitial(() -> ByteBuffer.allocateDirect(meta.pageSize())
+ .order(ByteOrder.nativeOrder()));
+
+ try {
+ U.doInParallel(
+ ignite.context().getSystemExecutorService(),
+ partFiles,
+ part -> {
+ String grpName = cacheGroupName(part.getParentFile());
+ int grpId = CU.cacheId(grpName);
+ int partId = partId(part.getName());
+
+ FilePageStoreManager storeMgr = (FilePageStoreManager)ignite.context().cache().context().pageStore();
+
+ try {
+ try (FilePageStore pageStore = (FilePageStore)storeMgr.getPageStoreFactory(grpId, false)
+ .createPageStore(getTypeByPartId(partId),
+ part::toPath,
+ val -> {
+ })
+ ) {
+ ByteBuffer pageBuff = buff.get();
+ pageBuff.clear();
+ pageStore.read(0, pageBuff, true);
+
+ long pageAddr = GridUnsafe.bufferAddress(pageBuff);
+
+ PagePartitionMetaIO io = PageIO.getPageIO(pageBuff);
+ GridDhtPartitionState partState = fromOrdinal(io.getPartitionState(pageAddr));
+
+ if (partState != OWNING) {
+ throw new IgniteCheckedException("Snapshot partitions must be in the OWNING " +
+ "state only: " + partState);
+ }
+
+ long updateCntr = io.getUpdateCounter(pageAddr);
+ long size = io.getSize(pageAddr);
+
+ if (log.isDebugEnabled()) {
+ log.debug("Partition [grpId=" + grpId
+ + ", id=" + partId
+ + ", counter=" + updateCntr
+ + ", size=" + size + "]");
+ }
+
+ checkPartitionsPageCrcSum(() -> pageStore, partId, PageIdAllocator.FLAG_DATA);
+
+ // Snapshot partitions must always be in OWNING state.
+ // There is no `primary` partitions for snapshot.
+ res.computeIfAbsent(new PartitionKeyV2(grpId, partId, grpName),
+ key -> new PartitionHashRecordV2(key, false, consId,
+ 0, updateCntr, size, PartitionHashRecordV2.PartitionState.OWNING));
+ }
+ }
+ catch (IOException e) {
+ throw new IgniteCheckedException(e);
+ }
+
+ return null;
+ }
+ );
+ }
+ catch (IgniteCheckedException e) {
+ throw new IgniteException(e);
+ }
+
+ return res;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean equals(Object o) {
+ if (this == o)
+ return true;
+
+ if (o == null || getClass() != o.getClass())
+ return false;
+
+ VisorVerifySnapshotPartitionsJob job = (VisorVerifySnapshotPartitionsJob)o;
+
+ return snpName.equals(job.snpName) && consId.equals(job.consId);
+ }
+
+ /** {@inheritDoc} */
+ @Override public int hashCode() {
+ return Objects.hash(snpName, consId);
+ }
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/IdleVerifyResultV2.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/IdleVerifyResultV2.java
index 363266e..637306c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/IdleVerifyResultV2.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/IdleVerifyResultV2.java
@@ -16,38 +16,25 @@
*/
package org.apache.ignite.internal.processors.cache.verify;
-import java.io.File;
-import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.ObjectInput;
import java.io.ObjectOutput;
-import java.io.PrintWriter;
-import java.time.LocalDateTime;
-import java.time.format.DateTimeFormatter;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;
-import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.internal.visor.VisorDataTransferObject;
-import org.jetbrains.annotations.Nullable;
/**
* Encapsulates result of {@link VerifyBackupPartitionsTaskV2}.
*/
public class IdleVerifyResultV2 extends VisorDataTransferObject {
/** */
- public static final String IDLE_VERIFY_FILE_PREFIX = "idle_verify-";
-
- /** Time formatter for log file name. */
- private static final DateTimeFormatter TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH-mm-ss_SSS");
-
- /** */
private static final long serialVersionUID = 0L;
/** Counter conflicts. */
@@ -74,7 +61,7 @@ public class IdleVerifyResultV2 extends VisorDataTransferObject {
* @param cntrConflicts Counter conflicts.
* @param hashConflicts Hash conflicts.
* @param movingPartitions Moving partitions.
- * @param exceptions Occured exceptions.
+ * @param exceptions Occurred exceptions.
*/
public IdleVerifyResultV2(
Map<PartitionKeyV2, List<PartitionHashRecordV2>> cntrConflicts,
@@ -91,6 +78,13 @@ public class IdleVerifyResultV2 extends VisorDataTransferObject {
}
/**
+ * @param exceptions Occurred exceptions.
+ */
+ public IdleVerifyResultV2(Map<ClusterNode, Exception> exceptions) {
+ this(Collections.emptyMap(), Collections.emptyMap(), Collections.emptyMap(), Collections.emptyMap(), exceptions);
+ }
+
+ /**
* Default constructor for Externalizable.
*/
public IdleVerifyResultV2() {
@@ -153,7 +147,7 @@ public class IdleVerifyResultV2 extends VisorDataTransferObject {
}
/**
- * @return <code>true</code> if any conflicts were discovered during idle_verify check.
+ * @return {@code true} if any conflicts were discovered during the check.
*/
public boolean hasConflicts() {
return !F.isEmpty(hashConflicts()) || !F.isEmpty(counterConflicts());
@@ -167,49 +161,12 @@ public class IdleVerifyResultV2 extends VisorDataTransferObject {
}
/**
- * Print formatted result to given printer. If exceptions presented exception messages will be written to log file.
+ * Print formatted result to the given printer.
*
* @param printer Consumer for handle formatted result.
- * @return Path to log file if exceptions presented and {@code null} otherwise.
+ * @param printExceptionMessages {@code true} if exceptions must be included too.
*/
- public @Nullable String print(Consumer<String> printer) {
- print(printer, false);
-
- if (!F.isEmpty(exceptions)) {
- File wd = null;
-
- try {
- wd = U.resolveWorkDirectory(U.defaultWorkDirectory(), "", false);
- }
- catch (IgniteCheckedException e) {
- printer.accept("Can't find work directory. " + e.getMessage() + "\n");
-
- e.printStackTrace();
- }
-
- File f = new File(wd, IDLE_VERIFY_FILE_PREFIX + LocalDateTime.now().format(TIME_FORMATTER) + ".txt");
-
- try (PrintWriter pw = new PrintWriter(f)) {
- print(pw::write, true);
-
- pw.flush();
-
- printer.accept("See log for additional information. " + f.getAbsolutePath() + "\n");
-
- return f.getAbsolutePath();
- }
- catch (FileNotFoundException e) {
- printer.accept("Can't write exceptions to file " + f.getAbsolutePath() + " " + e.getMessage() + "\n");
-
- e.printStackTrace();
- }
- }
-
- return null;
- }
-
- /** */
- private void print(Consumer<String> printer, boolean printExceptionMessages) {
+ public void print(Consumer<String> printer, boolean printExceptionMessages) {
boolean noMatchingCaches = false;
boolean succeeded = true;
@@ -227,11 +184,11 @@ public class IdleVerifyResultV2 extends VisorDataTransferObject {
if (!F.isEmpty(exceptions)) {
int size = exceptions.size();
- printer.accept("idle_verify failed on " + size + " node" + (size == 1 ? "" : "s") + ".\n");
+ printer.accept("The check procedure failed on " + size + " node" + (size == 1 ? "" : "s") + ".\n");
}
if (!hasConflicts())
- printer.accept("idle_verify check has finished, no conflicts have been found.\n");
+ printer.accept("The check procedure has finished, no conflicts have been found.\n");
else
printConflicts(printer);
@@ -244,14 +201,14 @@ public class IdleVerifyResultV2 extends VisorDataTransferObject {
printSkippedPartitions(printer, lostPartitions(), "LOST");
}
else {
- printer.accept("\nidle_verify failed.\n");
+ printer.accept("\nThe check procedure failed.\n");
if (noMatchingCaches)
printer.accept("\nThere are no caches matching given filter options.\n");
}
if (!F.isEmpty(exceptions())) {
- printer.accept("\nIdle verify failed on nodes:\n");
+ printer.accept("\nThe check procedure failed on nodes:\n");
for (Map.Entry<ClusterNode, Exception> e : exceptions().entrySet()) {
ClusterNode n = e.getKey();
@@ -298,7 +255,7 @@ public class IdleVerifyResultV2 extends VisorDataTransferObject {
int cntrConflictsSize = counterConflicts().size();
int hashConflictsSize = hashConflicts().size();
- printer.accept("idle_verify check has finished, found " + (cntrConflictsSize + hashConflictsSize) +
+ printer.accept("The check procedure has finished, found " + (cntrConflictsSize + hashConflictsSize) +
" conflict partitions: [counterConflicts=" + cntrConflictsSize + ", hashConflicts=" +
hashConflictsSize + "]\n");
@@ -322,9 +279,9 @@ public class IdleVerifyResultV2 extends VisorDataTransferObject {
printer.accept("Partition instances: " + entry.getValue() + "\n");
}
-
- printer.accept("\n");
}
+
+ printer.accept("\n");
}
/** {@inheritDoc} */
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/IdleVerifyUtility.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/IdleVerifyUtility.java
index c2ba3b1..f38fd93 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/IdleVerifyUtility.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/IdleVerifyUtility.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.processors.cache.verify;
+import java.io.File;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.util.ArrayList;
@@ -25,7 +26,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
-import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteException;
import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.pagemem.PageIdAllocator;
import org.apache.ignite.internal.pagemem.PageIdUtils;
@@ -36,7 +37,7 @@ import org.apache.ignite.internal.processors.cache.distributed.dht.topology.Grid
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState;
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionTopology;
import org.apache.ignite.internal.processors.cache.persistence.file.FilePageStore;
-import org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager;
+import org.apache.ignite.internal.util.lang.IgniteThrowableSupplier;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.SB;
import org.apache.ignite.lang.IgniteInClosure;
@@ -45,6 +46,7 @@ import org.jetbrains.annotations.Nullable;
import static org.apache.ignite.internal.pagemem.PageIdAllocator.FLAG_AUX;
import static org.apache.ignite.internal.pagemem.PageIdAllocator.FLAG_DATA;
import static org.apache.ignite.internal.pagemem.PageIdAllocator.FLAG_IDX;
+import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.cacheGroupName;
/**
* Utility class for idle verify command.
@@ -55,52 +57,42 @@ public class IdleVerifyUtility {
"Cluster not idle. Modifications found in caches or groups: ";
/**
- * See {@link IdleVerifyUtility#checkPartitionsPageCrcSum(FilePageStore, CacheGroupContext, int, byte)}.
- */
- public static void checkPartitionsPageCrcSum(
- @Nullable FilePageStoreManager pageStoreMgr,
- CacheGroupContext grpCtx,
- int partId,
- byte pageType
- ) throws IgniteCheckedException, GridNotIdleException {
- if (!grpCtx.persistenceEnabled() || pageStoreMgr == null)
- return;
-
- FilePageStore pageStore = (FilePageStore)pageStoreMgr.getStore(grpCtx.groupId(), partId);
-
- checkPartitionsPageCrcSum(pageStore, grpCtx, partId, pageType);
- }
-
- /**
- * Checks CRC sum of pages with {@code pageType} page type stored in partiion with {@code partId} id and assosiated
- * with cache group. <br/> Method could be invoked only on idle cluster!
+ * Checks CRC sum of pages with {@code pageType} page type stored in partition with {@code partId} id
+ * and associated with cache group.
*
- * @param pageStore Page store.
- * @param grpCtx Passed cache group context.
+ * @param pageStoreSup Page store supplier.
* @param partId Partition id.
* @param pageType Page type. Possible types {@link PageIdAllocator#FLAG_DATA}, {@link PageIdAllocator#FLAG_IDX}
* and {@link PageIdAllocator#FLAG_AUX}.
- * @throws IgniteCheckedException If reading page failed.
- * @throws GridNotIdleException If cluster not idle.
*/
public static void checkPartitionsPageCrcSum(
- FilePageStore pageStore,
- CacheGroupContext grpCtx,
+ IgniteThrowableSupplier<FilePageStore> pageStoreSup,
int partId,
- @Deprecated byte pageType
- ) throws IgniteCheckedException, GridNotIdleException {
+ byte pageType
+ ) {
assert pageType == FLAG_DATA || pageType == FLAG_IDX || pageType == FLAG_AUX : pageType;
- long pageId = PageIdUtils.pageId(partId, (byte)0, 0);
+ FilePageStore pageStore = null;
+
+ try {
+ pageStore = pageStoreSup.get();
- ByteBuffer buf = ByteBuffer.allocateDirect(grpCtx.dataRegion().pageMemory().pageSize());
+ long pageId = PageIdUtils.pageId(partId, (byte)0, 0);
- buf.order(ByteOrder.nativeOrder());
+ ByteBuffer buf = ByteBuffer.allocateDirect(pageStore.getPageSize()).order(ByteOrder.nativeOrder());
- for (int pageNo = 0; pageNo < pageStore.pages(); pageId++, pageNo++) {
- buf.clear();
+ for (int pageNo = 0; pageNo < pageStore.pages(); pageId++, pageNo++) {
+ buf.clear();
+
+ pageStore.read(pageId, buf, true,true);
+ }
+ }
+ catch (Throwable e) {
+ String msg0 = "CRC check of partition failed [partId=" + partId +
+ ", grpName=" + (pageStore == null ? "" : cacheGroupName(new File(pageStore.getFileAbsolutePath()).getParentFile())) +
+ ", part=" + (pageStore == null ? "" : pageStore.getFileAbsolutePath()) + ']';
- pageStore.read(pageId, buf, true);
+ throw new IgniteException(msg0, e);
}
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/VerifyBackupPartitionsDumpTask.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/VerifyBackupPartitionsDumpTask.java
index 38cf342..b8d2512 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/VerifyBackupPartitionsDumpTask.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/VerifyBackupPartitionsDumpTask.java
@@ -220,14 +220,14 @@ public class VerifyBackupPartitionsDumpTask extends ComputeTaskAdapter<VisorIdle
int size = exceptions.size();
- writer.write("idle_verify failed on " + size + " node" + (size == 1 ? "" : "s") + ".\n");
+ writer.write("The check procedure failed on " + size + " node" + (size == 1 ? "" : "s") + ".\n");
if (noMatchingCaches)
writer.write("There are no caches matching given filter options.");
}
if (!partitions.isEmpty())
- writer.write("idle_verify check has finished, found " + partitions.size() + " partitions\n");
+ writer.write("The check procedure has finished, found " + partitions.size() + " partitions\n");
logParsedArgs(taskArg, writer::write);
@@ -245,7 +245,7 @@ public class VerifyBackupPartitionsDumpTask extends ComputeTaskAdapter<VisorIdle
writer.write("\n\n-----------------------------------\n\n");
- conflictRes.print(writer::write);
+ conflictRes.print(writer::write, true);
}
}
@@ -284,7 +284,7 @@ public class VerifyBackupPartitionsDumpTask extends ComputeTaskAdapter<VisorIdle
* @param logConsumer Logger.
*/
public static void logParsedArgs(VisorIdleVerifyTaskArg args, Consumer<String> logConsumer) {
- SB options = new SB("idle_verify task was executed with the following args: ");
+ SB options = new SB("The check procedure task was executed with the following args: ");
options
.a("caches=[")
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/VerifyBackupPartitionsTask.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/VerifyBackupPartitionsTask.java
index 67b1146..f14ab77 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/VerifyBackupPartitionsTask.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/VerifyBackupPartitionsTask.java
@@ -272,7 +272,7 @@ public class VerifyBackupPartitionsTask extends ComputeTaskAdapter<Set<String>,
if (U.currentTimeMillis() - lastProgressLogTs > 3 * 60 * 1000L) {
lastProgressLogTs = U.currentTimeMillis();
- log.warning("idle_verify is still running, processed " + completionCntr.get() + " of " +
+ log.warning("The check procedure is still running, processed " + completionCntr.get() + " of " +
partHashCalcFutures.size() + " local partitions");
}
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/VerifyBackupPartitionsTaskV2.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/VerifyBackupPartitionsTaskV2.java
index bc76877..08f5b2b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/VerifyBackupPartitionsTaskV2.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/verify/VerifyBackupPartitionsTaskV2.java
@@ -59,7 +59,6 @@ import org.apache.ignite.internal.processors.cache.verify.PartitionHashRecordV2.
import org.apache.ignite.internal.processors.task.GridInternal;
import org.apache.ignite.internal.util.lang.GridIterator;
import org.apache.ignite.internal.util.typedef.F;
-import org.apache.ignite.internal.util.typedef.internal.SB;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.internal.visor.verify.VisorIdleVerifyTaskArg;
import org.apache.ignite.lang.IgniteInClosure;
@@ -73,6 +72,7 @@ import static java.util.Collections.emptyMap;
import static org.apache.ignite.cache.CacheMode.LOCAL;
import static org.apache.ignite.internal.pagemem.PageIdAllocator.FLAG_DATA;
import static org.apache.ignite.internal.processors.cache.verify.IdleVerifyUtility.GRID_NOT_IDLE_MSG;
+import static org.apache.ignite.internal.processors.cache.verify.IdleVerifyUtility.checkPartitionsPageCrcSum;
/**
* Task for comparing update counters and checksums between primary and backup partitions of specified caches.
@@ -111,16 +111,7 @@ public class VerifyBackupPartitionsTaskV2 extends ComputeTaskAdapter<VisorIdleVe
/** {@inheritDoc} */
@Nullable @Override public IdleVerifyResultV2 reduce(List<ComputeJobResult> results) throws IgniteException {
- Map<PartitionKeyV2, List<PartitionHashRecordV2>> clusterHashes = new HashMap<>();
-
- Map<ClusterNode, Exception> exceptions = new HashMap<>();
-
- reduceResults(results, clusterHashes, exceptions);
-
- if (results.size() != exceptions.size())
- return checkConflicts(clusterHashes, exceptions);
- else
- return new IdleVerifyResultV2(emptyMap(), emptyMap(), emptyMap(), emptyMap(), exceptions);
+ return reduce0(results);
}
/** {@inheritDoc} */
@@ -149,7 +140,7 @@ public class VerifyBackupPartitionsTaskV2 extends ComputeTaskAdapter<VisorIdleVe
}
/** */
- private IdleVerifyResultV2 checkConflicts(
+ private static IdleVerifyResultV2 checkConflicts(
Map<PartitionKeyV2, List<PartitionHashRecordV2>> clusterHashes,
Map<ClusterNode, Exception> exceptions
) {
@@ -198,15 +189,17 @@ public class VerifyBackupPartitionsTaskV2 extends ComputeTaskAdapter<VisorIdleVe
return new IdleVerifyResultV2(updateCntrConflicts, hashConflicts, movingParts, lostParts, exceptions);
}
- /** */
- private void reduceResults(
- List<ComputeJobResult> results,
- Map<PartitionKeyV2, List<PartitionHashRecordV2>> clusterHashes,
- Map<ClusterNode, Exception> exceptions
- ) {
+ /**
+ * @param results Received results of broadcast remote requests.
+ * @return Idle verify job result constructed from results of remote executions.
+ */
+ public static IdleVerifyResultV2 reduce0(List<ComputeJobResult> results) {
+ Map<PartitionKeyV2, List<PartitionHashRecordV2>> clusterHashes = new HashMap<>();
+ Map<ClusterNode, Exception> ex = new HashMap<>();
+
for (ComputeJobResult res : results) {
if (res.getException() != null) {
- exceptions.put(res.getNode(), res.getException());
+ ex.put(res.getNode(), res.getException());
continue;
}
@@ -219,6 +212,11 @@ public class VerifyBackupPartitionsTaskV2 extends ComputeTaskAdapter<VisorIdleVe
records.add(e.getValue());
}
}
+
+ if (results.size() != ex.size())
+ return checkConflicts(clusterHashes, ex);
+ else
+ return new IdleVerifyResultV2(ex);
}
/**
@@ -560,8 +558,13 @@ public class VerifyBackupPartitionsTaskV2 extends ComputeTaskAdapter<VisorIdleVe
partSize = part.dataStore().fullSize();
- if (arg.checkCrc())
- checkPartitionCrc(grpCtx, part);
+ if (arg.checkCrc() && grpCtx.persistenceEnabled()) {
+ FilePageStoreManager pageStoreMgr =
+ (FilePageStoreManager)ignite.context().cache().context().pageStore();
+
+ checkPartitionsPageCrcSum(() -> (FilePageStore)pageStoreMgr.getStore(grpCtx.groupId(), part.id()),
+ part.id(), FLAG_DATA);
+ }
GridIterator<CacheDataRow> it = grpCtx.offheap().partitionIterator(part.id());
@@ -601,41 +604,5 @@ public class VerifyBackupPartitionsTaskV2 extends ComputeTaskAdapter<VisorIdleVe
return Collections.singletonMap(partKey, partRec);
}
-
- /**
- * Checks correct CRC sum for given partition and cache group.
- *
- * @param grpCtx Cache group context
- * @param part partition.
- */
- private void checkPartitionCrc(CacheGroupContext grpCtx, GridDhtLocalPartition part) {
- if (grpCtx.persistenceEnabled()) {
- FilePageStore pageStore = null;
-
- try {
- FilePageStoreManager pageStoreMgr =
- (FilePageStoreManager)ignite.context().cache().context().pageStore();
-
- if (pageStoreMgr == null)
- return;
-
- pageStore = (FilePageStore)pageStoreMgr.getStore(grpCtx.groupId(), part.id());
-
- IdleVerifyUtility.checkPartitionsPageCrcSum(pageStore, grpCtx, part.id(), FLAG_DATA);
- }
- catch (GridNotIdleException e) {
- throw e;
- }
- catch (Exception | AssertionError e) {
- String msg = new SB("CRC check of partition: ").a(part.id()).a(", for cache group \"")
- .a(grpCtx.cacheOrGroupName()).a("\" failed.")
- .a(pageStore != null ? " file: " + pageStore.getFileAbsolutePath() : "").toString();
-
- log.error(msg, e);
-
- throw new IgniteException(msg, e);
- }
- }
- }
}
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/visor/snapshot/VisorSnapshotCheckTask.java b/modules/core/src/main/java/org/apache/ignite/internal/visor/snapshot/VisorSnapshotCheckTask.java
new file mode 100644
index 0000000..7b0615d2
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/visor/snapshot/VisorSnapshotCheckTask.java
@@ -0,0 +1,60 @@
+/*
+ * 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.visor.snapshot;
+
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotManager;
+import org.apache.ignite.internal.processors.cache.verify.IdleVerifyResultV2;
+import org.apache.ignite.internal.processors.task.GridInternal;
+import org.apache.ignite.internal.util.future.IgniteFutureImpl;
+import org.apache.ignite.internal.visor.VisorJob;
+import org.apache.ignite.internal.visor.VisorOneNodeTask;
+
+/**
+ * @see IgniteSnapshotManager#checkSnapshot(String)
+ */
+@GridInternal
+public class VisorSnapshotCheckTask extends VisorOneNodeTask<String, IdleVerifyResultV2> {
+ /** Serial version uid. */
+ private static final long serialVersionUID = 0L;
+
+ /** {@inheritDoc} */
+ @Override protected VisorJob<String, IdleVerifyResultV2> job(String arg) {
+ return new VisorSnapshotCheckJob(arg, debug);
+ }
+
+ /** */
+ private static class VisorSnapshotCheckJob extends VisorJob<String, IdleVerifyResultV2> {
+ /** Serial version uid. */
+ private static final long serialVersionUID = 0L;
+
+ /**
+ * @param name Snapshot name.
+ * @param debug Flag indicating whether debug information should be printed into node log.
+ */
+ protected VisorSnapshotCheckJob(String name, boolean debug) {
+ super(name, debug);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IdleVerifyResultV2 run(String name) throws IgniteException {
+ return new IgniteFutureImpl<>(ignite.context().cache().context().snapshotMgr().checkSnapshot(name))
+ .get();
+ }
+ }
+}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/local/GridCacheFastNodeLeftForTransactionTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/local/GridCacheFastNodeLeftForTransactionTest.java
index 273dab3..b595517 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/local/GridCacheFastNodeLeftForTransactionTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/local/GridCacheFastNodeLeftForTransactionTest.java
@@ -241,7 +241,7 @@ public class GridCacheFastNodeLeftForTransactionTest extends GridCommonAbstractT
SB sb = new SB();
- idleVerifyResV2.print(sb::a);
+ idleVerifyResV2.print(sb::a, true);
assertContains(listeningLog, sb.toString(), "no conflicts have been found");
}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotCheckTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotCheckTest.java
new file mode 100644
index 0000000..d72042e
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotCheckTest.java
@@ -0,0 +1,318 @@
+/*
+ * 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.persistence.snapshot;
+
+import java.io.File;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.UUID;
+import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.internal.GridJobExecuteRequest;
+import org.apache.ignite.internal.GridTopic;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.managers.communication.GridMessageListener;
+import org.apache.ignite.internal.processors.cache.persistence.file.FilePageStore;
+import org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO;
+import org.apache.ignite.internal.processors.cache.persistence.tree.io.PagePartitionMetaIO;
+import org.apache.ignite.internal.processors.cache.persistence.wal.crc.IgniteDataIntegrityViolationException;
+import org.apache.ignite.internal.processors.cache.verify.IdleVerifyResultV2;
+import org.apache.ignite.internal.util.GridUnsafe;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.X;
+import org.apache.ignite.internal.util.typedef.internal.CU;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.junit.Test;
+
+import static org.apache.ignite.cluster.ClusterState.ACTIVE;
+import static org.apache.ignite.configuration.IgniteConfiguration.DFLT_SNAPSHOT_DIRECTORY;
+import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.cacheDirName;
+import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.getPartitionFileName;
+import static org.apache.ignite.internal.processors.cache.persistence.partstate.GroupPartitionId.getTypeByPartId;
+import static org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotManager.SNAPSHOT_METAFILE_EXT;
+import static org.apache.ignite.testframework.GridTestUtils.assertContains;
+import static org.apache.ignite.testframework.GridTestUtils.waitForCondition;
+
+/**
+ * Cluster-wide snapshot check procedure tests.
+ */
+public class IgniteClusterSnapshotCheckTest extends AbstractSnapshotSelfTest {
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCheck() throws Exception {
+ IgniteEx ignite = startGridsWithCache(3, dfltCacheCfg, CACHE_KEYS_RANGE);
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME)
+ .get();
+
+ IdleVerifyResultV2 res = snp(ignite).checkSnapshot(SNAPSHOT_NAME).get();
+
+ StringBuilder b = new StringBuilder();
+ res.print(b::append, true);
+
+ assertTrue(F.isEmpty(res.exceptions()));
+ assertPartitionsSame(res);
+ assertContains(log, b.toString(), "The check procedure has finished, no conflicts have been found");
+ }
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCheckMissedPart() throws Exception {
+ IgniteEx ignite = startGridsWithCache(3, dfltCacheCfg, CACHE_KEYS_RANGE);
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME)
+ .get();
+
+ Path part0 = U.searchFileRecursively(snp(ignite).snapshotLocalDir(SNAPSHOT_NAME).toPath(),
+ getPartitionFileName(0));
+
+ assertNotNull(part0);
+ assertTrue(part0.toString(), part0.toFile().exists());
+ assertTrue(part0.toFile().delete());
+
+ IdleVerifyResultV2 res = snp(ignite).checkSnapshot(SNAPSHOT_NAME).get();
+
+ StringBuilder b = new StringBuilder();
+ res.print(b::append, true);
+
+ assertFalse(F.isEmpty(res.exceptions()));
+ assertContains(log, b.toString(), "Snapshot data doesn't contain required cache group partition");
+ }
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCheckMissedGroup() throws Exception {
+ IgniteEx ignite = startGridsWithCache(3, dfltCacheCfg, CACHE_KEYS_RANGE);
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME)
+ .get();
+
+ Path dir = Files.walk(snp(ignite).snapshotLocalDir(SNAPSHOT_NAME).toPath())
+ .filter(d -> d.toFile().getName().equals(cacheDirName(dfltCacheCfg)))
+ .findFirst()
+ .orElseThrow(() -> new RuntimeException("Cache directory not found"));
+
+ assertTrue(dir.toString(), dir.toFile().exists());
+ assertTrue(U.delete(dir));
+
+ IdleVerifyResultV2 res = snp(ignite).checkSnapshot(SNAPSHOT_NAME).get();
+
+ StringBuilder b = new StringBuilder();
+ res.print(b::append, true);
+
+ assertFalse(F.isEmpty(res.exceptions()));
+ assertContains(log, b.toString(), "Snapshot data doesn't contain required cache groups");
+ }
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCheckMissedMeta() throws Exception {
+ IgniteEx ignite = startGridsWithCache(3, dfltCacheCfg, CACHE_KEYS_RANGE);
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME)
+ .get();
+
+ File[] smfs = snp(ignite).snapshotLocalDir(SNAPSHOT_NAME).listFiles((dir, name) ->
+ name.toLowerCase().endsWith(SNAPSHOT_METAFILE_EXT));
+
+ assertNotNull(smfs);
+ assertTrue(smfs[0].toString(), smfs[0].exists());
+ assertTrue(U.delete(smfs[0]));
+
+ IdleVerifyResultV2 res = snp(ignite).checkSnapshot(SNAPSHOT_NAME).get();
+
+ StringBuilder b = new StringBuilder();
+ res.print(b::append, true);
+
+ assertFalse(F.isEmpty(res.exceptions()));
+ assertContains(log, b.toString(), "Some metadata is missing from the snapshot");
+ }
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCheckWithNodeFilter() throws Exception {
+ IgniteEx ig0 = startGridsWithoutCache(3);
+
+ for (int i = 0; i < CACHE_KEYS_RANGE; i++) {
+ ig0.getOrCreateCache(txCacheConfig(new CacheConfiguration<Integer, Integer>(DEFAULT_CACHE_NAME))
+ .setNodeFilter(node -> node.consistentId().toString().endsWith("0"))).put(i, i);
+ }
+
+ ig0.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ IdleVerifyResultV2 res = snp(ig0).checkSnapshot(SNAPSHOT_NAME).get();
+
+ StringBuilder b = new StringBuilder();
+ res.print(b::append, true);
+
+ assertTrue(F.isEmpty(res.exceptions()));
+ assertPartitionsSame(res);
+ assertContains(log, b.toString(), "The check procedure has finished, no conflicts have been found");
+ }
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCheckPartitionCounters() throws Exception {
+ IgniteEx ignite = startGridsWithCache(3, dfltCacheCfg.
+ setAffinity(new RendezvousAffinityFunction(false, 1)),
+ CACHE_KEYS_RANGE);
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ Path part0 = U.searchFileRecursively(snp(ignite).snapshotLocalDir(SNAPSHOT_NAME).toPath(),
+ getPartitionFileName(0));
+
+ assertNotNull(part0);
+ assertTrue(part0.toString(), part0.toFile().exists());
+
+ try (FilePageStore pageStore = (FilePageStore)((FilePageStoreManager)ignite.context().cache().context().pageStore())
+ .getPageStoreFactory(CU.cacheId(dfltCacheCfg.getName()), false)
+ .createPageStore(getTypeByPartId(0),
+ () -> part0,
+ val -> {
+ })
+ ) {
+ ByteBuffer buff = ByteBuffer.allocateDirect(ignite.configuration().getDataStorageConfiguration().getPageSize())
+ .order(ByteOrder.nativeOrder());
+
+ buff.clear();
+ pageStore.read(0, buff, false);
+
+ PagePartitionMetaIO io = PageIO.getPageIO(buff);
+
+ long pageAddr = GridUnsafe.bufferAddress(buff);
+
+ io.setUpdateCounter(pageAddr, CACHE_KEYS_RANGE * 2);
+
+ pageStore.beginRecover();
+
+ buff.flip();
+ pageStore.write(PageIO.getPageId(buff), buff, 0, true);
+ pageStore.finishRecover();
+ }
+
+ IdleVerifyResultV2 res = snp(ignite).checkSnapshot(SNAPSHOT_NAME).get();
+
+ StringBuilder b = new StringBuilder();
+ res.print(b::append, true);
+
+ assertTrue(F.isEmpty(res.exceptions()));
+ assertContains(log, b.toString(), "The check procedure has finished, found 1 conflict partitions");
+ }
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCheckOtherCluster() throws Exception {
+ IgniteEx ig0 = startGridsWithCache(3, dfltCacheCfg.
+ setAffinity(new RendezvousAffinityFunction(false, 1)),
+ CACHE_KEYS_RANGE);
+
+ ig0.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+ stopAllGrids();
+
+ // Cleanup persistence directory except created snapshots.
+ Arrays.stream(new File(U.defaultWorkDirectory()).listFiles())
+ .filter(f -> !f.getName().equals(DFLT_SNAPSHOT_DIRECTORY))
+ .forEach(U::delete);
+
+ Set<UUID> assigns = new HashSet<>();
+
+ for (int i = 4; i < 7; i++) {
+ startGrid(optimize(getConfiguration(getTestIgniteInstanceName(i)).setCacheConfiguration()));
+
+ UUID locNodeId = grid(i).localNode().id();
+
+ grid(i).context().io().addMessageListener(GridTopic.TOPIC_JOB, new GridMessageListener() {
+ @Override public void onMessage(UUID nodeId, Object msg, byte plc) {
+ if (msg instanceof GridJobExecuteRequest) {
+ GridJobExecuteRequest msg0 = (GridJobExecuteRequest)msg;
+
+ if (msg0.getTaskName().contains(SnapshotPartitionsVerifyTask.class.getName()))
+ assigns.add(locNodeId);
+ }
+ }
+ });
+ }
+
+ IgniteEx ignite = grid(4);
+ ignite.cluster().baselineAutoAdjustEnabled(false);
+ ignite.cluster().state(ACTIVE);
+
+ IdleVerifyResultV2 res = snp(ignite).checkSnapshot(SNAPSHOT_NAME).get();
+
+ StringBuilder b = new StringBuilder();
+ res.print(b::append, true);
+
+ // GridJobExecuteRequest is not send to the local node.
+ assertTrue("Number of jobs must be equal to the cluster size (except local node): " + assigns,
+ waitForCondition(() -> assigns.size() == 2, 5_000L));
+
+ assertTrue(F.isEmpty(res.exceptions()));
+ assertPartitionsSame(res);
+ assertContains(log, b.toString(), "The check procedure has finished, no conflicts have been found");
+ }
+
+ /** @throws Exception If fails. */
+ @Test
+ public void testClusterSnapshotCheckCRCFail() throws Exception {
+ IgniteEx ignite = startGridsWithCache(3, dfltCacheCfg.
+ setAffinity(new RendezvousAffinityFunction(false, 1)), CACHE_KEYS_RANGE);
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ Path part0 = U.searchFileRecursively(snp(ignite).snapshotLocalDir(SNAPSHOT_NAME).toPath(),
+ getPartitionFileName(0));
+
+ try (FilePageStore pageStore = (FilePageStore)((FilePageStoreManager)ignite.context().cache().context().pageStore())
+ .getPageStoreFactory(CU.cacheId(dfltCacheCfg.getName()), false)
+ .createPageStore(getTypeByPartId(0),
+ () -> part0,
+ val -> {
+ })
+ ) {
+ ByteBuffer buff = ByteBuffer.allocateDirect(ignite.configuration().getDataStorageConfiguration().getPageSize())
+ .order(ByteOrder.nativeOrder());
+ pageStore.read(0, buff, false);
+
+ pageStore.beginRecover();
+
+ PageIO.setCrc(buff, 1);
+
+ buff.flip();
+ pageStore.write(PageIO.getPageId(buff), buff, 0, false);
+ pageStore.finishRecover();
+ }
+
+ IdleVerifyResultV2 res = snp(ignite).checkSnapshot(SNAPSHOT_NAME).get();
+
+ StringBuilder b = new StringBuilder();
+ res.print(b::append, true);
+
+ assertEquals(1, res.exceptions().size());
+ assertContains(log, b.toString(), "The check procedure failed on 1 node.");
+
+ Exception ex = res.exceptions().values().iterator().next();
+ assertTrue(X.hasCause(ex, IgniteDataIntegrityViolationException.class));
+ }
+}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackOnTimeoutOnePhaseCommitTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackOnTimeoutOnePhaseCommitTest.java
index c7a125a..4a6a1e7 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackOnTimeoutOnePhaseCommitTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackOnTimeoutOnePhaseCommitTest.java
@@ -195,13 +195,7 @@ public class TxRollbackOnTimeoutOnePhaseCommitTest extends GridCommonAbstractTes
IdleVerifyResultV2 res = idleVerify(client, DEFAULT_CACHE_NAME);
- if (res.hasConflicts()) {
- StringBuilder b = new StringBuilder();
-
- res.print(b::append);
-
- fail(b.toString());
- }
+ assertPartitionsSame(res);
checkFutures();
}
diff --git a/modules/core/src/test/java/org/apache/ignite/testframework/GridTestUtils.java b/modules/core/src/test/java/org/apache/ignite/testframework/GridTestUtils.java
index ea7b0ea..6d871f2 100644
--- a/modules/core/src/test/java/org/apache/ignite/testframework/GridTestUtils.java
+++ b/modules/core/src/test/java/org/apache/ignite/testframework/GridTestUtils.java
@@ -94,7 +94,6 @@ import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtCacheAdapter;
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionTopology;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearCacheAdapter;
-import org.apache.ignite.internal.processors.cache.verify.IdleVerifyResultV2;
import org.apache.ignite.internal.processors.odbc.ClientListenerProcessor;
import org.apache.ignite.internal.processors.port.GridPortRecord;
import org.apache.ignite.internal.util.GridBusyLock;
@@ -2375,16 +2374,6 @@ public final class GridTestUtils {
}
/**
- * Removes idle_verify log files created in tests.
- */
- public static void cleanIdleVerifyLogFiles() {
- File dir = new File(".");
-
- for (File f : dir.listFiles(n -> n.getName().startsWith(IdleVerifyResultV2.IDLE_VERIFY_FILE_PREFIX)))
- f.delete();
- }
-
- /**
* @param grid Node.
* @param grp Group name.
* @param name Object name.
diff --git a/modules/core/src/test/java/org/apache/ignite/testframework/junits/common/GridCommonAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/testframework/junits/common/GridCommonAbstractTest.java
index 8a35caa..e8a2aa7 100755
--- a/modules/core/src/test/java/org/apache/ignite/testframework/junits/common/GridCommonAbstractTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/testframework/junits/common/GridCommonAbstractTest.java
@@ -2378,11 +2378,11 @@ public abstract class GridCommonAbstractTest extends GridAbstractTest {
/**
* @param res Response.
*/
- protected void assertPartitionsSame(IdleVerifyResultV2 res) throws AssertionFailedError {
+ protected static void assertPartitionsSame(IdleVerifyResultV2 res) throws AssertionFailedError {
if (res.hasConflicts()) {
StringBuilder b = new StringBuilder();
- res.print(b::append);
+ res.print(b::append, true);
fail(b.toString());
}
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteBasicWithPersistenceTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteBasicWithPersistenceTestSuite.java
index 25a8f8c..fd51913 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteBasicWithPersistenceTestSuite.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteBasicWithPersistenceTestSuite.java
@@ -40,6 +40,7 @@ import org.apache.ignite.internal.encryption.MasterKeyChangeTest;
import org.apache.ignite.internal.processors.cache.persistence.CheckpointReadLockFailureTest;
import org.apache.ignite.internal.processors.cache.persistence.CommonPoolStarvationCheckpointTest;
import org.apache.ignite.internal.processors.cache.persistence.SingleNodePersistenceSslTest;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteClusterSnapshotCheckTest;
import org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteClusterSnapshotSelfTest;
import org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotMXBeanTest;
import org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotManagerSelfTest;
@@ -94,6 +95,7 @@ import org.junit.runners.Suite;
IgniteSnapshotManagerSelfTest.class,
IgniteClusterSnapshotSelfTest.class,
+ IgniteClusterSnapshotCheckTest.class,
IgniteSnapshotMXBeanTest.class,
IgniteClusterIdTagTest.class,
diff --git a/modules/core/src/test/resources/org.apache.ignite.util/GridCommandHandlerClusterByClassTest_help.output b/modules/core/src/test/resources/org.apache.ignite.util/GridCommandHandlerClusterByClassTest_help.output
index 47da50b..75dea27 100644
--- a/modules/core/src/test/resources/org.apache.ignite.util/GridCommandHandlerClusterByClassTest_help.output
+++ b/modules/core/src/test/resources/org.apache.ignite.util/GridCommandHandlerClusterByClassTest_help.output
@@ -149,6 +149,12 @@ This utility can do the following commands:
Parameters:
snapshot_name - Snapshot name.
+ Check snapshot:
+ control.(sh|bat) --snapshot check snapshot_name
+
+ Parameters:
+ snapshot_name - Snapshot name.
+
Change cluster tag to new value:
control.(sh|bat) --change-tag newTagValue [--yes]
diff --git a/modules/core/src/test/resources/org.apache.ignite.util/GridCommandHandlerClusterByClassWithSSLTest_help.output b/modules/core/src/test/resources/org.apache.ignite.util/GridCommandHandlerClusterByClassWithSSLTest_help.output
index 47da50b..75dea27 100644
--- a/modules/core/src/test/resources/org.apache.ignite.util/GridCommandHandlerClusterByClassWithSSLTest_help.output
+++ b/modules/core/src/test/resources/org.apache.ignite.util/GridCommandHandlerClusterByClassWithSSLTest_help.output
@@ -149,6 +149,12 @@ This utility can do the following commands:
Parameters:
snapshot_name - Snapshot name.
+ Check snapshot:
+ control.(sh|bat) --snapshot check snapshot_name
+
+ Parameters:
+ snapshot_name - Snapshot name.
+
Change cluster tag to new value:
control.(sh|bat) --change-tag newTagValue [--yes]
diff --git a/modules/indexing/src/main/java/org/apache/ignite/internal/visor/verify/ValidateIndexesClosure.java b/modules/indexing/src/main/java/org/apache/ignite/internal/visor/verify/ValidateIndexesClosure.java
index 64c17eb..8c1dd93 100644
--- a/modules/indexing/src/main/java/org/apache/ignite/internal/visor/verify/ValidateIndexesClosure.java
+++ b/modules/indexing/src/main/java/org/apache/ignite/internal/visor/verify/ValidateIndexesClosure.java
@@ -53,6 +53,7 @@ import org.apache.ignite.internal.processors.cache.mvcc.MvccSnapshot;
import org.apache.ignite.internal.processors.cache.mvcc.MvccUtils;
import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
import org.apache.ignite.internal.processors.cache.persistence.IgniteCacheDatabaseSharedManager;
+import org.apache.ignite.internal.processors.cache.persistence.file.FilePageStore;
import org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager;
import org.apache.ignite.internal.processors.cache.persistence.tree.BPlusTree;
import org.apache.ignite.internal.processors.cache.persistence.tree.CorruptedTreeException;
@@ -96,6 +97,7 @@ import static org.apache.ignite.internal.pagemem.PageIdAllocator.FLAG_IDX;
import static org.apache.ignite.internal.pagemem.PageIdAllocator.INDEX_PARTITION;
import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState.OWNING;
import static org.apache.ignite.internal.processors.cache.verify.IdleVerifyUtility.GRID_NOT_IDLE_MSG;
+import static org.apache.ignite.internal.processors.cache.verify.IdleVerifyUtility.checkPartitionsPageCrcSum;
import static org.apache.ignite.internal.processors.cache.verify.IdleVerifyUtility.compareUpdateCounters;
import static org.apache.ignite.internal.processors.cache.verify.IdleVerifyUtility.formatUpdateCountersDiff;
import static org.apache.ignite.internal.processors.cache.verify.IdleVerifyUtility.getUpdateCountersSnapshot;
@@ -488,12 +490,15 @@ public class ValidateIndexesClosure implements IgniteCallable<VisorValidateIndex
IgniteInClosure<Integer> idleChecker
) {
GridKernalContext ctx = ignite.context();
- GridCacheSharedContext cctx = ctx.cache().context();
+ GridCacheSharedContext<?, ?> cctx = ctx.cache().context();
try {
FilePageStoreManager pageStoreMgr = (FilePageStoreManager)cctx.pageStore();
- IdleVerifyUtility.checkPartitionsPageCrcSum(pageStoreMgr, gctx, INDEX_PARTITION, FLAG_IDX);
+ if (pageStoreMgr != null && gctx.persistenceEnabled()) {
+ checkPartitionsPageCrcSum(() -> (FilePageStore)pageStoreMgr.getStore(gctx.groupId(), INDEX_PARTITION),
+ INDEX_PARTITION, FLAG_IDX);
+ }
idleChecker.apply(gctx.groupId());