You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ti...@apache.org on 2025/09/03 09:01:32 UTC
(ignite) branch master updated: IGNITE-22662 Snapshot check bases on DistributedProcess (#11897)
This is an automated email from the ASF dual-hosted git repository.
timoninmaxim 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 8716b826888 IGNITE-22662 Snapshot check bases on DistributedProcess (#11897)
8716b826888 is described below
commit 8716b826888fb9c40e77d5b893d3475bad6298fd
Author: Maksim Timonin <ti...@apache.org>
AuthorDate: Wed Sep 3 14:01:26 2025 +0500
IGNITE-22662 Snapshot check bases on DistributedProcess (#11897)
Co-authored-by: Vladimir Steshin <vl...@gmail.com>
---
.../ignite/common/ComputeTaskPermissionsTest.java | 4 +-
.../apache/ignite/util/GridCommandHandlerTest.java | 9 +-
.../snapshot/AbstractSnapshotOperationRequest.java | 119 ++++
.../snapshot/AbstractSnapshotVerificationTask.java | 184 -----
.../snapshot/IgniteSnapshotManager.java | 85 +--
.../IncrementalSnapshotVerificationTask.java | 130 ++--
.../IncrementalSnapshotVerificationTaskResult.java | 26 +-
.../persistence/snapshot/SnapshotCheckProcess.java | 721 +++++++++++++++++++
.../snapshot/SnapshotCheckProcessRequest.java | 107 +++
.../persistence/snapshot/SnapshotChecker.java | 148 ++++
.../snapshot/SnapshotHandlerRestoreTask.java | 114 +--
.../snapshot/SnapshotMetadataVerificationTask.java | 142 ++--
.../SnapshotMetadataVerificationTaskArg.java | 111 ---
.../SnapshotMetadataVerificationTaskResult.java | 75 --
.../snapshot/SnapshotOperationRequest.java | 73 +-
.../snapshot/SnapshotPartitionsVerifyHandler.java | 6 +-
.../snapshot/SnapshotPartitionsVerifyTask.java | 132 ----
.../snapshot/SnapshotPartitionsVerifyTaskArg.java | 134 ----
.../SnapshotPartitionsVerifyTaskResult.java | 2 +-
.../snapshot/SnapshotRestoreProcess.java | 6 +-
.../util/distributed/DistributedProcess.java | 12 +-
.../main/resources/META-INF/classnames.properties | 8 -
.../snapshot/AbstractSnapshotSelfTest.java | 28 +-
.../snapshot/EncryptedSnapshotTest.java | 4 +-
.../snapshot/IgniteClusterSnapshotCheckTest.java | 788 ++++++++++++++++++++-
.../IgniteClusterSnapshotRestoreSelfTest.java | 4 +-
.../snapshot/IgniteSnapshotMXBeanTest.java | 7 +-
27 files changed, 2114 insertions(+), 1065 deletions(-)
diff --git a/modules/clients/src/test/java/org/apache/ignite/common/ComputeTaskPermissionsTest.java b/modules/clients/src/test/java/org/apache/ignite/common/ComputeTaskPermissionsTest.java
index efe5f7a4bac..390478bef79 100644
--- a/modules/clients/src/test/java/org/apache/ignite/common/ComputeTaskPermissionsTest.java
+++ b/modules/clients/src/test/java/org/apache/ignite/common/ComputeTaskPermissionsTest.java
@@ -48,7 +48,7 @@ import org.apache.ignite.configuration.ConnectorConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.configuration.ThinClientConfiguration;
import org.apache.ignite.internal.IgniteEx;
-import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotMetadataVerificationTask;
+import org.apache.ignite.internal.management.cache.VerifyBackupPartitionsTask;
import org.apache.ignite.internal.processors.security.AbstractSecurityTest;
import org.apache.ignite.internal.processors.security.OperationSecurityContext;
import org.apache.ignite.internal.processors.security.PublicAccessJob;
@@ -108,7 +108,7 @@ public class ComputeTaskPermissionsTest extends AbstractSecurityTest {
private static final IgniteClosure SYSTEM_CLOSURE = new ToStringClosure<>();
/** */
- private static final ComputeTask SYSTEM_TASK = new SnapshotMetadataVerificationTask();
+ private static final ComputeTask SYSTEM_TASK = new VerifyBackupPartitionsTask();
/** */
private static final AtomicInteger EXECUTED_TASK_CNTR = new AtomicInteger();
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 eb41f36f23f..9121b2d50b0 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
@@ -187,6 +187,7 @@ import static org.apache.ignite.internal.processors.cache.verify.IdleVerifyUtili
import static org.apache.ignite.internal.processors.diagnostic.DiagnosticProcessor.DEFAULT_TARGET_FOLDER;
import static org.apache.ignite.internal.processors.job.GridJobProcessor.JOBS_VIEW;
import static org.apache.ignite.internal.processors.task.GridTaskProcessor.TASKS_VIEW;
+import static org.apache.ignite.internal.util.distributed.DistributedProcess.DistributedProcessType.RESTORE_CACHE_GROUP_SNAPSHOT_PREPARE;
import static org.apache.ignite.testframework.GridTestUtils.assertContains;
import static org.apache.ignite.testframework.GridTestUtils.assertNotContains;
import static org.apache.ignite.testframework.GridTestUtils.runAsync;
@@ -759,7 +760,7 @@ public class GridCommandHandlerTest extends GridCommandHandlerClusterPerMethodAb
final String newTag = "new_tag";
Ignite ignite = startGrids(2);
-
+
startClientGrid("client");
assertFalse(ignite.cluster().state().active());
@@ -3740,7 +3741,8 @@ public class GridCommandHandlerTest extends GridCommandHandlerClusterPerMethodAb
awaitPartitionMapExchange();
- spi.blockMessages((node, msg) -> msg instanceof SingleNodeMessage);
+ spi.blockMessages((node, msg) -> msg instanceof SingleNodeMessage
+ && ((SingleNodeMessage)msg).type() == RESTORE_CACHE_GROUP_SNAPSHOT_PREPARE.ordinal());
fut = srv.snapshot().restoreSnapshot(snapshotName, F.asList(DEFAULT_CACHE_NAME));
@@ -3759,7 +3761,8 @@ public class GridCommandHandlerTest extends GridCommandHandlerClusterPerMethodAb
awaitPartitionMapExchange();
- spi.blockMessages((node, msg) -> msg instanceof SingleNodeMessage);
+ spi.blockMessages((node, msg) -> msg instanceof SingleNodeMessage
+ && ((SingleNodeMessage)msg).type() == RESTORE_CACHE_GROUP_SNAPSHOT_PREPARE.ordinal());
fut = srv.snapshot().restoreSnapshot(snapshotName, F.asList(DEFAULT_CACHE_NAME), 1);
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/AbstractSnapshotOperationRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/AbstractSnapshotOperationRequest.java
new file mode 100644
index 00000000000..2e0a5ce64a3
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/AbstractSnapshotOperationRequest.java
@@ -0,0 +1,119 @@
+/*
+ * 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.Collection;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.UUID;
+import org.apache.ignite.internal.util.distributed.DistributedProcess;
+import org.apache.ignite.internal.util.tostring.GridToStringInclude;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Snapshot operation start request for {@link DistributedProcess} initiate message.
+ */
+abstract class AbstractSnapshotOperationRequest implements Serializable {
+ /** Serial version uid. */
+ private static final long serialVersionUID = 0L;
+
+ /** Request ID. */
+ @GridToStringInclude
+ private final UUID reqId;
+
+ /** Snapshot name. */
+ @GridToStringInclude
+ private final String snpName;
+
+ /** Snapshot directory path. */
+ @GridToStringInclude
+ private final String snpPath;
+
+ /** Collection of cache group names. */
+ @GridToStringInclude
+ private final Collection<String> grps;
+
+ /** Start time. */
+ @GridToStringInclude
+ private final long startTime;
+
+ /** IDs of the nodes that must be alive to complete the operation. */
+ @GridToStringInclude
+ private final Set<UUID> nodes;
+
+ /**
+ * @param reqId Request ID.
+ * @param snpName Snapshot name.
+ * @param snpPath Snapshot directory path.
+ * @param grps Collection of cache group names.
+ * @param incIdx Incremental snapshot index.
+ * @param nodes IDs of the nodes that must be alive to complete the operation.
+ */
+ protected AbstractSnapshotOperationRequest(
+ UUID reqId,
+ String snpName,
+ String snpPath,
+ @Nullable Collection<String> grps,
+ int incIdx,
+ Collection<UUID> nodes
+ ) {
+ this.reqId = reqId;
+ this.snpName = snpName;
+ this.grps = grps;
+ this.snpPath = snpPath;
+ this.nodes = new HashSet<>(nodes);
+ this.startTime = System.currentTimeMillis();
+ }
+
+ /** @return Request ID. */
+ @Nullable public UUID requestId() {
+ return reqId;
+ }
+
+ /** @return Snapshot name. */
+ public String snapshotName() {
+ return snpName;
+ }
+
+ /** @return Snapshot directory path. */
+ public String snapshotPath() {
+ return snpPath;
+ }
+
+ /** @return List of cache group names. */
+ public @Nullable Collection<String> groups() {
+ return grps;
+ }
+
+ /** @return Start time. */
+ public long startTime() {
+ return startTime;
+ }
+
+ /** @return IDs of the nodes that must be alive to complete the operation. */
+ public Set<UUID> nodes() {
+ return nodes;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(AbstractSnapshotOperationRequest.class, this);
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/AbstractSnapshotVerificationTask.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/AbstractSnapshotVerificationTask.java
deleted file mode 100644
index 2751f861529..00000000000
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/AbstractSnapshotVerificationTask.java
+++ /dev/null
@@ -1,184 +0,0 @@
-/*
- * 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.Collection;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-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.processors.cache.persistence.filename.SnapshotFileTree;
-import org.apache.ignite.internal.util.typedef.F;
-import org.apache.ignite.resources.IgniteInstanceResource;
-import org.apache.ignite.resources.LoggerResource;
-import org.jetbrains.annotations.Nullable;
-
-import static org.apache.ignite.internal.util.lang.ClusterNodeFunc.node2id;
-
-/**
- * The task for checking the consistency of snapshots in the cluster.
- */
-public abstract class AbstractSnapshotVerificationTask extends
- ComputeTaskAdapter<SnapshotPartitionsVerifyTaskArg, SnapshotPartitionsVerifyTaskResult> {
- /** Serial version uid. */
- private static final long serialVersionUID = 0L;
-
- /** Map of snapshot metadata information found on each cluster node. */
- protected final Map<ClusterNode, List<SnapshotMetadata>> metas = new HashMap<>();
-
- /** Ignite instance. */
- @IgniteInstanceResource
- protected IgniteEx ignite;
-
- /** Injected logger. */
- @LoggerResource
- protected IgniteLogger log;
-
- /** {@inheritDoc} */
- @Override public Map<ComputeJob, ClusterNode> map(List<ClusterNode> subgrid, SnapshotPartitionsVerifyTaskArg arg) {
- Map<ClusterNode, List<SnapshotMetadata>> clusterMetas = arg.clusterMetadata();
-
- 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, node2id()) +
- ", init=" + F.viewReadOnly(clusterMetas.keySet(), node2id()) + ']')));
- }
-
- Map<ComputeJob, ClusterNode> jobs = new HashMap<>();
- Set<SnapshotMetadata> allMetas = new HashSet<>();
- clusterMetas.values().forEach(allMetas::addAll);
- metas.putAll(clusterMetas);
-
- 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(createJob(meta.snapshotName(), meta.folderName(), meta.consistentId(), arg), e.getKey());
-
- if (allMetas.isEmpty())
- break;
- }
- }
-
- return jobs;
- }
-
- /** {@inheritDoc} */
- @Override public ComputeJobResultPolicy result(ComputeJobResult res, List<ComputeJobResult> rcvd) throws IgniteException {
- // Handle all exceptions during the `reduce` operation.
- return ComputeJobResultPolicy.WAIT;
- }
-
- /**
- * @param name Snapshot name.
- * @param folderName Folder name for snapshot.
- * @param consId Consistent id of the related node.
- * @param args Check snapshot parameters.
- *
- * @return Compute job.
- */
- protected abstract AbstractSnapshotVerificationJob createJob(
- String name,
- String folderName,
- String consId,
- SnapshotPartitionsVerifyTaskArg args
- );
-
- /** */
- protected abstract static class AbstractSnapshotVerificationJob extends ComputeJobAdapter {
- /** Serial version uid. */
- private static final long serialVersionUID = 0L;
-
- /** Ignite instance. */
- @IgniteInstanceResource
- protected IgniteEx ignite;
-
- /** Injected logger. */
- @LoggerResource
- protected IgniteLogger log;
-
- /** Snapshot name. */
- protected final String snpName;
-
- /** Snapshot directory path. */
- @Nullable protected final String snpPath;
-
- /** Folder name for snapshot. */
- protected final String folderName;
-
- /** Consistent id of the snapshot data. */
- protected final String consId;
-
- /** Set of cache groups to be checked in the snapshot. {@code Null} or empty to check everything. */
- @Nullable protected final Collection<String> rqGrps;
-
- /** If {@code true}, calculates and compares partition hashes. Otherwise, only basic snapshot validation is launched. */
- protected final boolean check;
-
- /** Snapshot file tree. */
- protected transient SnapshotFileTree sft;
-
- /**
- * @param snpName Snapshot name.
- * @param snpPath Snapshot directory path.
- * @param folderName Folder name for snapshot.
- * @param consId Consistent id of the related node.
- * @param rqGrps Set of cache groups to be checked in the snapshot. {@code Null} or empty to check everything.
- * @param check If {@code true}, calculates and compares partition hashes. Otherwise, only basic snapshot validation is launched.
- */
- protected AbstractSnapshotVerificationJob(
- String snpName,
- @Nullable String snpPath,
- String folderName,
- String consId,
- @Nullable Collection<String> rqGrps,
- boolean check
- ) {
- this.snpName = snpName;
- this.snpPath = snpPath;
- this.folderName = folderName;
- this.consId = consId;
- this.rqGrps = rqGrps;
- this.check = check;
- }
-
- /** {@inheritDoc} */
- @Override public Object execute() throws IgniteException {
- sft = new SnapshotFileTree(ignite.context(), snpName, snpPath, folderName, consId);
-
- return execute0();
- }
-
- /** Exectues actual job. */
- protected abstract Object execute0();
- }
-}
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 c42c302ad42..857e75bc610 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
@@ -339,6 +339,9 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
/** Take snapshot operation procedure. */
private final DistributedProcess<SnapshotOperationRequest, SnapshotOperationResponse> startSnpProc;
+ /** Snapshot validation distributed process. */
+ private final SnapshotCheckProcess checkSnpProc;
+
/** Check previously performed snapshot operation and delete uncompleted files if we need. */
private final DistributedProcess<SnapshotOperationRequest, SnapshotOperationResponse> endSnpProc;
@@ -439,6 +442,8 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
restoreCacheGrpProc = new SnapshotRestoreProcess(ctx, locBuff);
+ checkSnpProc = new SnapshotCheckProcess(ctx);
+
// Manage remote snapshots.
snpRmtMgr = new SequentialRemoteSnapshotManager();
}
@@ -651,7 +656,10 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
try {
snpRmtMgr.stop();
- restoreCacheGrpProc.interrupt(new NodeStoppingException("Node is stopping."));
+ IgniteCheckedException stopErr = new NodeStoppingException("Node is stopping.");
+
+ restoreCacheGrpProc.interrupt(stopErr);
+ checkSnpProc.interrupt(stopErr);
// Try stop all snapshot processing if not yet.
for (AbstractSnapshotFutureTask<?> sctx : locSnpTasks.values())
@@ -1394,6 +1402,13 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
}
}
+ /**
+ * @return {@code True} if checking of a snapshot with specified name is in progress.
+ */
+ public boolean isSnapshotChecking(String snpName) {
+ return checkSnpProc.isSnapshotChecking(snpName);
+ }
+
/**
* Sets the streamer warning flag to current snapshot process if it is active.
*/
@@ -1710,75 +1725,25 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
"Collection of cache groups names cannot contain null elements.");
A.ensure(!includeCustomHandlers || incIdx < 1, "Snapshot handlers aren't supported for incremental snapshot.");
- GridFutureAdapter<SnapshotPartitionsVerifyTaskResult> res = new GridFutureAdapter<>();
-
if (log.isInfoEnabled()) {
log.info("The check snapshot procedure started [snpName=" + name + ", snpPath=" + snpPath +
- ", incIdx=" + incIdx + ", grps=" + grps + ']');
+ ", incIdx=" + incIdx + ", grps=" + grps + ", validateParts=" + check + ']');
}
- GridKernalContext kctx0 = cctx.kernalContext();
-
- Collection<ClusterNode> bltNodes = F.view(cctx.discovery().serverNodes(AffinityTopologyVersion.NONE),
- (node) -> CU.baselineNode(node, kctx0.state().clusterState()));
-
- Collection<Integer> grpIds = grps == null ? Collections.emptySet() : F.viewReadOnly(grps, CU::cacheId);
-
- SnapshotMetadataVerificationTaskArg taskArg = new SnapshotMetadataVerificationTaskArg(name, snpPath, incIdx, grpIds);
-
- kctx0.task().execute(
- SnapshotMetadataVerificationTask.class,
- taskArg,
- options(bltNodes)
- ).listen(f0 -> {
- SnapshotMetadataVerificationTaskResult metasRes = f0.result();
+ IgniteInternalFuture<SnapshotPartitionsVerifyTaskResult> res = checkSnpProc.start(
+ name, snpPath, grps, check, incIdx, includeCustomHandlers);
- if (f0.error() == null && F.isEmpty(metasRes.exceptions())) {
- Map<ClusterNode, List<SnapshotMetadata>> metas = metasRes.meta();
+ res.listen(lsnr -> {
+ if (log.isInfoEnabled()) {
+ Throwable err = lsnr.error();
- Class<? extends AbstractSnapshotVerificationTask> cls;
+ boolean success = err == null;
- if (includeCustomHandlers)
- cls = SnapshotHandlerRestoreTask.class;
- else
- cls = incIdx > 0 ? IncrementalSnapshotVerificationTask.class : SnapshotPartitionsVerifyTask.class;
-
- kctx0.task().execute(
- cls,
- new SnapshotPartitionsVerifyTaskArg(grps, metas, snpPath, incIdx, check),
- options(new ArrayList<>(metas.keySet()))
- ).listen(f1 -> {
- if (f1.error() == null)
- res.onDone(f1.result());
- else if (f1.error() instanceof IgniteSnapshotVerifyException) {
- IdleVerifyResult idleRes = IdleVerifyResult.builder()
- .exceptions(((IgniteSnapshotVerifyException)f1.error()).exceptions()).build();
-
- res.onDone(new SnapshotPartitionsVerifyTaskResult(metas, idleRes));
- }
- else
- res.onDone(f1.error());
- });
- }
- else {
- if (f0.error() == null)
- res.onDone(new IgniteSnapshotVerifyException(metasRes.exceptions()));
- else if (f0.error() instanceof IgniteSnapshotVerifyException) {
- IdleVerifyResult idleRes = IdleVerifyResult.builder()
- .exceptions(((IgniteSnapshotVerifyException)f0.error()).exceptions()).build();
-
- res.onDone(new SnapshotPartitionsVerifyTaskResult(null, idleRes));
- }
- else
- res.onDone(f0.error());
+ log.info("The check snapshot procedure finished [snpName=" + name + ", success=" + success + ", snpPath=" + snpPath
+ + ", incIdx=" + incIdx + ", grps=" + grps + (success ? "" : ", err=" + err.getMessage()) + ']');
}
});
- if (log.isInfoEnabled()) {
- res.listen(() -> log.info("The check snapshot procedure finished [snpName=" + name +
- ", snpPath=" + snpPath + ", incIdx=" + incIdx + ", grps=" + grps + ']'));
- }
-
return res;
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IncrementalSnapshotVerificationTask.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IncrementalSnapshotVerificationTask.java
index d4febb6abd9..27bf90db3f2 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IncrementalSnapshotVerificationTask.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IncrementalSnapshotVerificationTask.java
@@ -25,7 +25,6 @@ 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.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.LongAdder;
@@ -34,114 +33,110 @@ import java.util.function.Function;
import java.util.stream.Collectors;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
+import org.apache.ignite.IgniteLogger;
import org.apache.ignite.cache.CacheAtomicityMode;
-import org.apache.ignite.compute.ComputeJobResult;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.management.cache.IdleVerifyResult;
import org.apache.ignite.internal.management.cache.PartitionKey;
import org.apache.ignite.internal.processors.cache.GridCacheOperation;
import org.apache.ignite.internal.processors.cache.GridLocalConfigManager;
import org.apache.ignite.internal.processors.cache.StoredCacheData;
+import org.apache.ignite.internal.processors.cache.persistence.filename.SnapshotFileTree;
import org.apache.ignite.internal.processors.cache.verify.IdleVerifyUtility.VerifyPartitionContext;
import org.apache.ignite.internal.processors.cache.verify.PartitionHashRecord;
import org.apache.ignite.internal.processors.cache.verify.TransactionsHashRecord;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.processors.cluster.BaselineTopology;
-import org.apache.ignite.internal.processors.task.GridInternal;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.transactions.TransactionState;
-import org.jetbrains.annotations.Nullable;
import static org.apache.ignite.internal.managers.discovery.ConsistentIdMapper.ALL_NODES;
/** */
-@GridInternal
-public class IncrementalSnapshotVerificationTask extends AbstractSnapshotVerificationTask {
- /** Serial version uid. */
- private static final long serialVersionUID = 0L;
-
- /** {@inheritDoc} */
- @Override public SnapshotPartitionsVerifyTaskResult reduce(List<ComputeJobResult> results) throws IgniteException {
- IdleVerifyResult.Builder bldr = IdleVerifyResult.builder();
-
- for (ComputeJobResult nodeRes: results) {
- if (nodeRes.getException() != null) {
- bldr.addException(nodeRes.getNode(), nodeRes.getException());
+public class IncrementalSnapshotVerificationTask {
+ /** */
+ private final VerifyIncrementalSnapshotJob job;
- continue;
- }
+ /** */
+ private final IgniteLogger log;
- IncrementalSnapshotVerificationTaskResult res = nodeRes.getData();
+ /** */
+ public IncrementalSnapshotVerificationTask(IgniteEx ignite, IgniteLogger log, SnapshotFileTree sft, int incrementalIdx) {
+ job = new VerifyIncrementalSnapshotJob(ignite, log, sft, incrementalIdx);
+ this.log = log;
+ }
- if (!F.isEmpty(res.exceptions())) {
- bldr.addException(nodeRes.getNode(), F.first(res.exceptions()));
+ /** */
+ public IdleVerifyResult reduce(Map<ClusterNode, IncrementalSnapshotVerificationTaskResult> results) throws IgniteException {
+ IdleVerifyResult.Builder bldr = IdleVerifyResult.builder();
- continue;
- }
+ for (Map.Entry<ClusterNode, IncrementalSnapshotVerificationTaskResult> nodeRes: results.entrySet()) {
+ IncrementalSnapshotVerificationTaskResult res = nodeRes.getValue();
if (!F.isEmpty(res.partiallyCommittedTxs()))
- bldr.addPartiallyCommited(nodeRes.getNode(), res.partiallyCommittedTxs());
+ bldr.addPartiallyCommited(nodeRes.getKey(), res.partiallyCommittedTxs());
bldr.addPartitionHashes(res.partHashRes());
if (log.isDebugEnabled())
- log.debug("Handle VerifyIncrementalSnapshotJob result [node=" + nodeRes.getNode() + ", taskRes=" + res + ']');
+ log.debug("Handle VerifyIncrementalSnapshotJob result [node=" + nodeRes.getKey() + ", taskRes=" + res + ']');
- bldr.addIncrementalHashRecords(nodeRes.getNode(), res.txHashRes());
+ bldr.addIncrementalHashRecords(nodeRes.getKey(), res.txHashRes());
}
- return new SnapshotPartitionsVerifyTaskResult(metas, bldr.build());
+ return bldr.build();
}
- /** {@inheritDoc} */
- @Override protected VerifyIncrementalSnapshotJob createJob(
- String name,
- String folderName,
- String consId,
- SnapshotPartitionsVerifyTaskArg args
- ) {
- return new VerifyIncrementalSnapshotJob(name, args.snapshotPath(), args.incrementIndex(), folderName, consId);
+ /** */
+ public IncrementalSnapshotVerificationTaskResult execute() {
+ return job.execute0();
}
/** */
- private static class VerifyIncrementalSnapshotJob extends AbstractSnapshotVerificationJob {
- /** Serial version uid. */
- private static final long serialVersionUID = 0L;
+ private static class VerifyIncrementalSnapshotJob {
+ /** */
+ private final IgniteEx ignite;
- /** Incremental snapshot index. */
+ /** */
+ private final IgniteLogger log;
+
+ /** */
+ private final SnapshotFileTree sft;
+
+ /** */
private final int incIdx;
/** */
private LongAdder procEntriesCnt;
/**
- * @param snpName Snapshot name.
- * @param snpPath Snapshot directory path.
- * @param folderName Folder name for snapshot.
+ * @param ignite Ignite instance.
+ * @param log Ignite logger.
+ * @param sft Snapshot file tree
* @param incIdx Incremental snapshot index.
- * @param folderName Folder name for snapshot.
- * @param consId Consistent id of the related node.
*/
- public VerifyIncrementalSnapshotJob(
- String snpName,
- @Nullable String snpPath,
- int incIdx,
- String folderName,
- String consId
+ private VerifyIncrementalSnapshotJob(
+ IgniteEx ignite,
+ IgniteLogger log,
+ SnapshotFileTree sft,
+ int incIdx
) {
- super(snpName, snpPath, folderName, consId, null, true);
-
+ this.ignite = ignite;
+ this.log = log;
+ this.sft = sft;
this.incIdx = incIdx;
}
/**
* @return Map containing calculated transactions hash for every remote node in the cluster.
*/
- @Override public IncrementalSnapshotVerificationTaskResult execute0() throws IgniteException {
+ public IncrementalSnapshotVerificationTaskResult execute0() throws IgniteException {
try {
if (log.isInfoEnabled()) {
log.info("Verify incremental snapshot procedure has been initiated " +
- "[snpName=" + snpName + ", incrementIndex=" + incIdx + ", consId=" + consId + ']');
+ "[snpName=" + sft.name() + ", incrementIndex=" + incIdx + ", consId=" + sft.consistentId() + ']');
}
if (incIdx <= 0)
@@ -174,7 +169,7 @@ public class IncrementalSnapshotVerificationTask extends AbstractSnapshotVerific
}
};
- short locNodeId = cstIdsMap.get(consId);
+ short locNodeId = cstIdsMap.get(sft.consistentId());
Set<GridCacheVersion> activeDhtTxs = new HashSet<>();
Map<GridCacheVersion, Set<Short>> txPrimParticipatingNodes = new HashMap<>();
@@ -292,7 +287,7 @@ public class IncrementalSnapshotVerificationTask extends AbstractSnapshotVerific
Map<Object, TransactionsHashRecord> txHashRes = nodesTxHash.entrySet().stream()
.map(e -> new TransactionsHashRecord(
- consId,
+ sft.consistentId(),
blt.compactIdMapping().get(e.getKey()),
e.getValue().hash
))
@@ -307,7 +302,7 @@ public class IncrementalSnapshotVerificationTask extends AbstractSnapshotVerific
e -> new PartitionHashRecord(
e.getKey(),
false,
- consId,
+ sft.consistentId(),
null,
0,
null,
@@ -317,7 +312,7 @@ public class IncrementalSnapshotVerificationTask extends AbstractSnapshotVerific
if (log.isInfoEnabled()) {
log.info("Verify incremental snapshot procedure finished " +
- "[snpName=" + snpName + ", incrementIndex=" + incIdx + ", consId=" + consId +
+ "[snpName=" + sft.name() + ", incrementIndex=" + incIdx + ", consId=" + sft.consistentId() +
", txCnt=" + procTxCnt.sum() + ", dataEntries=" + procEntriesCnt.sum() +
", walSegments=" + procSegCnt.get() + ']');
}
@@ -355,25 +350,6 @@ public class IncrementalSnapshotVerificationTask extends AbstractSnapshotVerific
.filter(data -> data.config().getAtomicityMode() == CacheAtomicityMode.TRANSACTIONAL)
.collect(Collectors.toMap(StoredCacheData::cacheId, Function.identity()));
}
-
- /** {@inheritDoc} */
- @Override public boolean equals(Object o) {
- if (this == o)
- return true;
-
- if (o == null || getClass() != o.getClass())
- return false;
-
- VerifyIncrementalSnapshotJob job = (VerifyIncrementalSnapshotJob)o;
-
- return snpName.equals(job.snpName) && Objects.equals(incIdx, job.incIdx) && Objects.equals(snpPath, job.snpPath)
- && consId.equals(job.consId);
- }
-
- /** {@inheritDoc} */
- @Override public int hashCode() {
- return Objects.hash(snpName, incIdx, snpPath, consId);
- }
}
/** Holder for calculated hashes. */
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IncrementalSnapshotVerificationTaskResult.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IncrementalSnapshotVerificationTaskResult.java
index ee6539525d5..986f0b9dfa8 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IncrementalSnapshotVerificationTaskResult.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IncrementalSnapshotVerificationTaskResult.java
@@ -17,21 +17,17 @@
package org.apache.ignite.internal.processors.cache.persistence.snapshot;
-import java.io.IOException;
-import java.io.ObjectInput;
-import java.io.ObjectOutput;
+import java.io.Serializable;
import java.util.Collection;
import java.util.Map;
-import org.apache.ignite.internal.dto.IgniteDataTransferObject;
import org.apache.ignite.internal.management.cache.PartitionKey;
import org.apache.ignite.internal.pagemem.wal.record.DataEntry;
import org.apache.ignite.internal.processors.cache.verify.PartitionHashRecord;
import org.apache.ignite.internal.processors.cache.verify.TransactionsHashRecord;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
-import org.apache.ignite.internal.util.typedef.internal.U;
-/** Represents single job result for {@link IncrementalSnapshotVerificationTask}. */
-class IncrementalSnapshotVerificationTaskResult extends IgniteDataTransferObject {
+/** */
+public class IncrementalSnapshotVerificationTaskResult implements Serializable {
/** */
private static final long serialVersionUID = 0L;
@@ -87,20 +83,4 @@ class IncrementalSnapshotVerificationTaskResult extends IgniteDataTransferObject
public Collection<Exception> exceptions() {
return exceptions;
}
-
- /** {@inheritDoc} */
- @Override protected void writeExternalData(ObjectOutput out) throws IOException {
- U.writeMap(out, txHashRes);
- U.writeMap(out, partHashRes);
- U.writeCollection(out, partiallyCommittedTxs);
- U.writeCollection(out, exceptions);
- }
-
- /** {@inheritDoc} */
- @Override protected void readExternalData(ObjectInput in) throws IOException, ClassNotFoundException {
- txHashRes = U.readMap(in);
- partHashRes = U.readMap(in);
- partiallyCommittedTxs = U.readCollection(in);
- exceptions = U.readCollection(in);
- }
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotCheckProcess.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotCheckProcess.java
new file mode 100644
index 00000000000..02cc9783050
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotCheckProcess.java
@@ -0,0 +1,721 @@
+/*
+ * 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.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.IgniteLogger;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.internal.GridKernalContext;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.NodeStoppingException;
+import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
+import org.apache.ignite.internal.management.cache.IdleVerifyResult;
+import org.apache.ignite.internal.management.cache.PartitionKey;
+import org.apache.ignite.internal.processors.cache.persistence.filename.SnapshotFileTree;
+import org.apache.ignite.internal.processors.cache.verify.PartitionHashRecord;
+import org.apache.ignite.internal.util.distributed.DistributedProcess;
+import org.apache.ignite.internal.util.future.GridFinishedFuture;
+import org.apache.ignite.internal.util.future.GridFutureAdapter;
+import org.apache.ignite.internal.util.tostring.GridToStringInclude;
+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.S;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.jetbrains.annotations.Nullable;
+
+import static org.apache.ignite.internal.util.distributed.DistributedProcess.DistributedProcessType.CHECK_SNAPSHOT_METAS;
+import static org.apache.ignite.internal.util.distributed.DistributedProcess.DistributedProcessType.CHECK_SNAPSHOT_PARTS;
+import static org.apache.ignite.internal.util.lang.ClusterNodeFunc.node2id;
+
+/** Distributed process of snapshot checking. */
+public class SnapshotCheckProcess {
+ /** */
+ private final IgniteLogger log;
+
+ /** */
+ private final GridKernalContext kctx;
+
+ /** Snapshot checker. */
+ private final SnapshotChecker snpChecker;
+
+ /** Operation contexts by name. */
+ private final Map<String, SnapshotCheckContext> contexts = new ConcurrentHashMap<>();
+
+ /** Cluster-wide operation futures per snapshot called from current node. */
+ private final Map<UUID, GridFutureAdapter<SnapshotPartitionsVerifyTaskResult>> clusterOpFuts = new ConcurrentHashMap<>();
+
+ /** Check metas first phase subprocess. */
+ private final DistributedProcess<SnapshotCheckProcessRequest, SnapshotCheckResponse> phase1CheckMetas;
+
+ /** Partition hashes second phase subprocess. */
+ private final DistributedProcess<SnapshotCheckProcessRequest, SnapshotCheckResponse> phase2PartsHashes;
+
+ /** Stop node lock. */
+ private boolean nodeStopping;
+
+ /** */
+ public SnapshotCheckProcess(GridKernalContext kctx) {
+ this.kctx = kctx;
+
+ snpChecker = new SnapshotChecker(kctx);
+
+ log = kctx.log(getClass());
+
+ phase1CheckMetas = new DistributedProcess<>(kctx, CHECK_SNAPSHOT_METAS, this::prepareAndCheckMetas,
+ this::reducePreparationAndMetasCheck);
+
+ phase2PartsHashes = new DistributedProcess<>(kctx, CHECK_SNAPSHOT_PARTS, this::validateParts,
+ this::reduceValidatePartsAndFinish);
+ }
+
+ /**
+ * Stops all the processes with the passed exception.
+ *
+ * @param err The interrupt reason.
+ */
+ void interrupt(Throwable err) {
+ // Prevents starting new processes in #prepareAndCheckMetas.
+ synchronized (contexts) {
+ nodeStopping = true;
+ }
+
+ clusterOpFuts.forEach((reqId, fut) -> fut.onDone(err));
+ }
+
+ /** Phase 2 and process finish. */
+ private IgniteInternalFuture<?> reduceValidatePartsAndFinish(
+ UUID reqId,
+ Map<UUID, SnapshotCheckResponse> results,
+ Map<UUID, Throwable> errors
+ ) {
+ SnapshotCheckContext ctx = context(null, reqId);
+
+ if (ctx == null)
+ return new GridFinishedFuture<>();
+
+ try {
+ contexts.remove(ctx.req.snapshotName());
+
+ GridFutureAdapter<SnapshotPartitionsVerifyTaskResult> clusterOpFut = clusterOpFuts.get(reqId);
+
+ if (clusterOpFut == null)
+ return new GridFinishedFuture<>();
+
+ if (F.isEmpty(errors)) {
+ ClusterTopologyCheckedException ex = checkNodeLeft(ctx.req.nodes(), results.keySet());
+
+ if (ex != null) {
+ clusterOpFut.onDone(ex);
+
+ return new GridFinishedFuture<>();
+ }
+ }
+
+ if (ctx.req.incrementalIndex() > 0) {
+ SnapshotFileTree sft = ctx.locFileTree == null || kctx.config().getConsistentId() == null
+ ? null
+ : ctx.locFileTree.get(kctx.config().getConsistentId().toString());
+
+ reduceIncrementalResults(sft, ctx.req.incrementalIndex(), ctx.req.nodes(), ctx.clusterMetas, results, errors, clusterOpFut);
+ }
+ else if (ctx.req.allRestoreHandlers())
+ reduceCustomHandlersResults(ctx, results, errors, clusterOpFut);
+ else
+ reducePartitionsHashesResults(ctx.clusterMetas, results, errors, clusterOpFut);
+
+ return new GridFinishedFuture<>();
+ }
+ finally {
+ if (log.isInfoEnabled())
+ log.info("Finished snapshot validation [req=" + ctx.req + ']');
+ }
+ }
+
+ /** */
+ private void reduceIncrementalResults(
+ SnapshotFileTree sft,
+ int incIdx,
+ Set<UUID> requiredNodes,
+ Map<ClusterNode, List<SnapshotMetadata>> clusterMetas,
+ Map<UUID, SnapshotCheckResponse> results,
+ Map<UUID, Throwable> errors,
+ GridFutureAdapter<SnapshotPartitionsVerifyTaskResult> fut
+ ) {
+ Map<ClusterNode, IncrementalSnapshotVerificationTaskResult> perNodeResults = new HashMap<>(results.size(), 1.0f);
+
+ for (Map.Entry<UUID, SnapshotCheckResponse> resEntry : results.entrySet()) {
+ UUID nodeId = resEntry.getKey();
+
+ SnapshotCheckResponse incResp = resEntry.getValue();
+
+ if (incResp == null || !requiredNodes.contains(nodeId))
+ continue;
+
+ perNodeResults.put(kctx.cluster().get().node(nodeId), incResp.result());
+
+ if (F.isEmpty(incResp.exceptions()))
+ continue;
+
+ errors.putIfAbsent(nodeId, asException(F.firstValue(incResp.exceptions())));
+ }
+
+ IdleVerifyResult chkRes = snpChecker.reduceIncrementalResults(sft, incIdx, perNodeResults, mapErrors(errors));
+
+ fut.onDone(new SnapshotPartitionsVerifyTaskResult(clusterMetas, chkRes));
+ }
+
+ /** */
+ private void reduceCustomHandlersResults(
+ SnapshotCheckContext ctx,
+ Map<UUID, SnapshotCheckResponse> results,
+ Map<UUID, Throwable> errors,
+ GridFutureAdapter<SnapshotPartitionsVerifyTaskResult> fut
+ ) {
+ try {
+ if (!errors.isEmpty())
+ throw F.firstValue(errors);
+
+ // Check responses: checking node -> snapshot part's consistent id -> handler name -> handler result.
+ Map<ClusterNode, Map<Object, Map<String, SnapshotHandlerResult<?>>>> reduced = new HashMap<>(results.size(), 1.0f);
+
+ for (Map.Entry<UUID, SnapshotCheckResponse> respEntry : results.entrySet()) {
+ SnapshotCheckResponse nodeResp = respEntry.getValue();
+
+ if (nodeResp == null)
+ continue;
+
+ if (!F.isEmpty(nodeResp.exceptions()))
+ throw F.firstValue(nodeResp.exceptions());
+
+ UUID nodeId = respEntry.getKey();
+
+ Map<String, Map<String, SnapshotHandlerResult<Object>>> cstHndRes = nodeResp.result();
+
+ cstHndRes.forEach((consId, respPerConsIdMap) -> {
+ // Reduced map of the handlers results per snapshot part's consistent id for certain node.
+ Map<Object, Map<String, SnapshotHandlerResult<?>>> nodePerConsIdResultMap
+ = reduced.computeIfAbsent(kctx.cluster().get().node(nodeId), n -> new HashMap<>());
+
+ respPerConsIdMap.forEach((hndId, hndRes) ->
+ nodePerConsIdResultMap.computeIfAbsent(consId, cstId -> new HashMap<>()).put(hndId, hndRes));
+ });
+ }
+
+ snpChecker.checkCustomHandlersResults(ctx.req.snapshotName(), reduced);
+
+ fut.onDone(new SnapshotPartitionsVerifyTaskResult(ctx.clusterMetas, null));
+ }
+ catch (Throwable err) {
+ fut.onDone(err);
+ }
+ }
+
+ /** */
+ private void reducePartitionsHashesResults(
+ Map<ClusterNode, List<SnapshotMetadata>> clusterMetas,
+ Map<UUID, SnapshotCheckResponse> results,
+ Map<UUID, Throwable> errors,
+ GridFutureAdapter<SnapshotPartitionsVerifyTaskResult> fut
+ ) {
+ IdleVerifyResult.Builder bldr = IdleVerifyResult.builder();
+
+ Map<ClusterNode, Exception> errors0 = mapErrors(errors);
+
+ if (!results.isEmpty()) {
+ if (!errors0.isEmpty())
+ bldr.exceptions(errors0);
+
+ for (Map.Entry<UUID, SnapshotCheckResponse> respEntry : results.entrySet()) {
+ SnapshotCheckResponse resp = respEntry.getValue();
+
+ if (resp == null)
+ continue;
+
+ if (!F.isEmpty(resp.exceptions())) {
+ ClusterNode node = kctx.cluster().get().node(respEntry.getKey());
+
+ bldr.addException(node, asException(F.firstValue(resp.exceptions())));
+ }
+
+ Map<String, Map<PartitionKey, PartitionHashRecord>> partsHashesRes = resp.result();
+
+ partsHashesRes.forEach((consId, partsPerConsId) -> bldr.addPartitionHashes(partsPerConsId));
+ }
+
+ fut.onDone(new SnapshotPartitionsVerifyTaskResult(clusterMetas, bldr.build()));
+ }
+ else
+ fut.onDone(new IgniteSnapshotVerifyException(errors0));
+ }
+
+ /** Phase 2 beginning. */
+ private IgniteInternalFuture<SnapshotCheckResponse> validateParts(SnapshotCheckProcessRequest req) {
+ if (!req.nodes().contains(kctx.localNodeId()))
+ return new GridFinishedFuture<>();
+
+ SnapshotCheckContext ctx = context(req.snapshotName(), req.requestId());
+
+ assert ctx != null;
+
+ if (F.isEmpty(ctx.metas))
+ return new GridFinishedFuture<>();
+
+ GridFutureAdapter<SnapshotCheckResponse> phaseFut = new GridFutureAdapter<>();
+
+ CompletableFuture<SnapshotCheckResponse> workingFut;
+
+ if (req.incrementalIndex() > 0) {
+ assert !req.allRestoreHandlers() : "Snapshot handlers aren't supported for incremental snapshot.";
+
+ workingFut = incrementalFuture(ctx);
+ }
+ else if (req.allRestoreHandlers())
+ workingFut = allHandlersFuture(ctx);
+ else
+ workingFut = partitionsHashesFuture(ctx);
+
+ workingFut.whenComplete((res, err) -> {
+ if (err != null)
+ phaseFut.onDone(err);
+ else
+ phaseFut.onDone(res);
+ });
+
+ return phaseFut;
+ }
+
+ /** @return A composed future of increment checks for each consistent id regarding {@link SnapshotCheckContext#metas}. */
+ private CompletableFuture<SnapshotCheckResponse> incrementalFuture(SnapshotCheckContext ctx) {
+ // Incremental snapshots do not support working on other topology. Only single meta and snapshot part can be processed.
+ SnapshotMetadata meta = ctx.metas.get(0);
+
+ CompletableFuture<SnapshotCheckResponse> resFut = new CompletableFuture<>();
+
+ CompletableFuture<IncrementalSnapshotVerificationTaskResult> workingFut = snpChecker.checkIncrementalSnapshot(
+ ctx.locFileTree.get(meta.consistentId()), ctx.req.incrementalIndex());
+
+ workingFut.whenComplete((res, err) -> {
+ if (err != null)
+ resFut.completeExceptionally(err);
+ else
+ resFut.complete(new SnapshotCheckResponse(res, null));
+ });
+
+ return resFut;
+ }
+
+ /** @return A composed future of partitions checks for each consistent id regarding {@link SnapshotCheckContext#metas}. */
+ private CompletableFuture<SnapshotCheckResponse> partitionsHashesFuture(SnapshotCheckContext ctx) {
+ // Per metas result: consistent id -> check results per partition key.
+ Map<String, Map<PartitionKey, PartitionHashRecord>> perMetaResults = new ConcurrentHashMap<>(ctx.metas.size(), 1.0f);
+ // Per consistent id.
+ Map<String, Throwable> exceptions = new ConcurrentHashMap<>(ctx.metas.size(), 1.0f);
+ CompletableFuture<SnapshotCheckResponse> composedFut = new CompletableFuture<>();
+ AtomicInteger metasProcessed = new AtomicInteger(ctx.metas.size());
+
+ for (SnapshotMetadata meta : ctx.metas) {
+ CompletableFuture<Map<PartitionKey, PartitionHashRecord>> metaFut = snpChecker.checkPartitions(
+ meta,
+ ctx.locFileTree.get(meta.consistentId()),
+ ctx.req.groups(),
+ false,
+ ctx.req.fullCheck()
+ );
+
+ metaFut.whenComplete((res, err) -> {
+ if (err != null)
+ exceptions.put(meta.consistentId(), err);
+ else if (!F.isEmpty(res))
+ perMetaResults.put(meta.consistentId(), res);
+
+ if (metasProcessed.decrementAndGet() == 0)
+ composedFut.complete(new SnapshotCheckResponse(perMetaResults, exceptions));
+ });
+ }
+
+ return composedFut;
+ }
+
+ /**
+ * @return A composed future of all the snapshot handlers for each consistent id regarding {@link SnapshotCheckContext#metas}.
+ * @see IgniteSnapshotManager#handlers()
+ */
+ private CompletableFuture<SnapshotCheckResponse> allHandlersFuture(SnapshotCheckContext ctx) {
+ // Per metas result: snapshot part's consistent id -> check result per handler name.
+ Map<String, Map<String, SnapshotHandlerResult<Object>>> perMetaResults = new ConcurrentHashMap<>(ctx.metas.size(), 1.0f);
+ // Per consistent id.
+ Map<String, Throwable> exceptions = new ConcurrentHashMap<>(ctx.metas.size(), 1.0f);
+ CompletableFuture<SnapshotCheckResponse> composedFut = new CompletableFuture<>();
+ AtomicInteger metasProcessed = new AtomicInteger(ctx.metas.size());
+
+ for (SnapshotMetadata meta : ctx.metas) {
+ CompletableFuture<Map<String, SnapshotHandlerResult<Object>>> metaFut = snpChecker.invokeCustomHandlers(meta,
+ ctx.locFileTree.get(meta.consistentId()), ctx.req.groups(), ctx.req.fullCheck());
+
+ metaFut.whenComplete((res, err) -> {
+ if (err != null)
+ exceptions.put(meta.consistentId(), err);
+ else if (!F.isEmpty(res))
+ perMetaResults.put(meta.consistentId(), res);
+
+ if (metasProcessed.decrementAndGet() == 0)
+ composedFut.complete(new SnapshotCheckResponse(perMetaResults, exceptions));
+ });
+ }
+
+ return composedFut;
+ }
+
+ /** */
+ private Map<ClusterNode, Exception> mapErrors(Map<UUID, Throwable> errors) {
+ return errors.entrySet().stream().collect(Collectors.toMap(e -> kctx.cluster().get().node(e.getKey()),
+ e -> asException(e.getValue())));
+ }
+
+ /** */
+ private static Exception asException(Throwable th) {
+ return th instanceof Exception ? (Exception)th : new IgniteException(th);
+ }
+
+ /**
+ * @param snpName Snapshot name. If {@code null}, ignored.
+ * @param reqId If {@code ctxId} is {@code null}, is used to find the operation context.
+ * @return Current snapshot checking context by {@code snpName} or {@code reqId}.
+ */
+ private @Nullable SnapshotCheckContext context(@Nullable String snpName, UUID reqId) {
+ return snpName == null
+ ? contexts.values().stream().filter(ctx0 -> ctx0.req.requestId().equals(reqId)).findFirst().orElse(null)
+ : contexts.get(snpName);
+ }
+
+ /** Phase 1 beginning: prepare, collect and check local metas. */
+ private IgniteInternalFuture<SnapshotCheckResponse> prepareAndCheckMetas(SnapshotCheckProcessRequest req) {
+ if (!req.nodes().contains(kctx.localNodeId()) && clusterOpFuts.get(req.requestId()) == null)
+ return new GridFinishedFuture<>();
+
+ SnapshotCheckContext ctx;
+
+ synchronized (contexts) {
+ if (nodeStopping)
+ return new GridFinishedFuture<>(new NodeStoppingException("The node is stopping: " + kctx.localNodeId()));
+
+ ctx = contexts.computeIfAbsent(req.snapshotName(), snpName -> new SnapshotCheckContext(req));
+ }
+
+ if (!ctx.req.requestId().equals(req.requestId())) {
+ return new GridFinishedFuture<>(new IllegalStateException("Validation of snapshot '" + req.snapshotName()
+ + "' has already started [ctx=" + ctx + ']'));
+ }
+
+ // Excludes non-baseline initiator.
+ if (!baseline(kctx.localNodeId()))
+ return new GridFinishedFuture<>();
+
+ Collection<Integer> grpIds = F.isEmpty(req.groups()) ? null : F.viewReadOnly(req.groups(), CU::cacheId);
+
+ GridFutureAdapter<SnapshotCheckResponse> phaseFut = new GridFutureAdapter<>();
+
+ snpChecker.checkLocalMetas(
+ new SnapshotFileTree(kctx, req.snapshotName(), req.snapshotPath()),
+ req.incrementalIndex(),
+ grpIds
+ ).whenComplete((locMetas, err) -> {
+ if (err != null)
+ phaseFut.onDone(err);
+ else
+ phaseFut.onDone(new SnapshotCheckResponse(locMetas, null));
+ });
+
+ return phaseFut;
+ }
+
+ /** Phase 1 end. */
+ private void reducePreparationAndMetasCheck(
+ UUID reqId,
+ Map<UUID, SnapshotCheckResponse> results,
+ Map<UUID, Throwable> errors
+ ) {
+ SnapshotCheckContext ctx = context(null, reqId);
+
+ // The context is not stored in the case of concurrent check of the same snapshot but the operation future is registered.
+ GridFutureAdapter<SnapshotPartitionsVerifyTaskResult> clusterOpFut = clusterOpFuts.get(reqId);
+
+ try {
+ if (!errors.isEmpty())
+ throw new IgniteSnapshotVerifyException(mapErrors(errors));
+
+ if (ctx == null) {
+ assert clusterOpFut == null;
+
+ return;
+ }
+
+ ClusterTopologyCheckedException ex;
+
+ if ((ex = checkNodeLeft(ctx.req.nodes(), results.keySet())) != null)
+ throw ex;
+
+ Map<ClusterNode, List<SnapshotMetadata>> metas = new HashMap<>(results.size(), 1.0f);
+
+ results.forEach((nodeId, nodeRes) -> {
+ // A node might be not required. It gives null result. But a required node might have invalid empty result
+ // which must be validated.
+ if (ctx.req.nodes().contains(nodeId) && baseline(nodeId) && !F.isEmpty((Collection<?>)nodeRes.result())) {
+ assert nodeRes != null;
+
+ metas.put(kctx.cluster().get().node(nodeId), nodeRes.result());
+ }
+ });
+
+ Map<ClusterNode, Exception> metasCheckErrors = snpChecker.reduceMetasResults(
+ new SnapshotFileTree(kctx, ctx.req.snapshotName(), ctx.req.snapshotPath()), metas);
+
+ if (!metasCheckErrors.isEmpty())
+ throw new IgniteSnapshotVerifyException(metasCheckErrors);
+
+ // If the topology is lesser that the snapshot's, we have to check partitions of another nodes.
+ ctx.metas = assingMetas(metas);
+
+ if (!F.isEmpty(ctx.metas)) {
+ ctx.locFileTree = new HashMap<>(ctx.metas.size(), 1.0f);
+
+ for (SnapshotMetadata metaToProc : ctx.metas) {
+ SnapshotFileTree sft = new SnapshotFileTree(kctx, ctx.req.snapshotName(), ctx.req.snapshotPath(),
+ metaToProc.folderName(), metaToProc.consistentId());
+
+ ctx.locFileTree.put(metaToProc.consistentId(), sft);
+ }
+ }
+
+ if (clusterOpFut != null)
+ ctx.clusterMetas = metas;
+
+ if (U.isLocalNodeCoordinator(kctx.discovery()))
+ phase2PartsHashes.start(reqId, ctx.req);
+ }
+ catch (Throwable th) {
+ if (ctx != null)
+ contexts.remove(ctx.req.snapshotName());
+
+ if (clusterOpFut != null)
+ clusterOpFut.onDone(th);
+ }
+ }
+
+ /**
+ * Assigns snapshot metadatas to process. A snapshot can be checked on a smaller topology compared to the original one.
+ * In this case, some node has to check not only own meta and partitions.
+ *
+ * @return Metadatas to process on current node.
+ */
+ private @Nullable List<SnapshotMetadata> assingMetas(Map<ClusterNode, List<SnapshotMetadata>> clusterMetas) {
+ ClusterNode locNode = kctx.cluster().get().localNode();
+ List<SnapshotMetadata> locMetas = clusterMetas.get(locNode);
+
+ if (F.isEmpty(locMetas))
+ return null;
+
+ Set<String> onlineNodesConstIdsStr = new HashSet<>(clusterMetas.size());
+ // The nodes are sorted with lesser order.
+ Map<String, Collection<ClusterNode>> metasPerRespondedNodes = new HashMap<>();
+
+ clusterMetas.forEach((node, nodeMetas) -> {
+ if (!F.isEmpty(nodeMetas)) {
+ onlineNodesConstIdsStr.add(node.consistentId().toString());
+
+ nodeMetas.forEach(nodeMeta -> metasPerRespondedNodes.computeIfAbsent(nodeMeta.consistentId(),
+ m -> new TreeSet<>(Comparator.comparingLong(ClusterNode::order))).add(node));
+ }
+ });
+
+ String locNodeConsIdStr = locNode.consistentId().toString();
+ List<SnapshotMetadata> metasToProc = new ArrayList<>();
+
+ for (SnapshotMetadata meta : locMetas) {
+ if (meta.consistentId().equals(locNodeConsIdStr)) {
+ assert !metasToProc.contains(meta) : "Local snapshot metadata is already assigned to process";
+
+ metasToProc.add(meta);
+
+ continue;
+ }
+
+ if (!onlineNodesConstIdsStr.contains(meta.consistentId())
+ && F.first(metasPerRespondedNodes.get(meta.consistentId())).id().equals(kctx.localNodeId()))
+ metasToProc.add(meta);
+ }
+
+ return metasToProc;
+ }
+
+ /**
+ * Starts the snapshot validation process.
+ *
+ * @param snpName Snapshot name.
+ * @param snpPath Snapshot directory path.
+ * @param grpNames Collection of cache group names.
+ * @param fullCheck If {@code true}, additionally calculates partition hashes. Otherwise, checks only snapshot integrity
+ * and partition counters.
+ * @param incIdx Incremental snapshot index. If not positive, snapshot is not considered as incremental.
+ * @param allRestoreHandlers If {@code true}, all the registered {@link IgniteSnapshotManager#handlers()} of type
+ * {@link SnapshotHandlerType#RESTORE} are invoked. Otherwise, only snapshot metadatas and partition
+ * hashes are validated.
+ */
+ public IgniteInternalFuture<SnapshotPartitionsVerifyTaskResult> start(
+ String snpName,
+ @Nullable String snpPath,
+ @Nullable Collection<String> grpNames,
+ boolean fullCheck,
+ int incIdx,
+ boolean allRestoreHandlers
+ ) {
+ assert !F.isEmpty(snpName);
+
+ UUID reqId = UUID.randomUUID();
+
+ Set<UUID> requiredNodes = new HashSet<>(F.viewReadOnly(kctx.discovery().discoCache().aliveBaselineNodes(), node2id()));
+
+ SnapshotCheckProcessRequest req = new SnapshotCheckProcessRequest(
+ reqId,
+ requiredNodes,
+ snpName,
+ snpPath,
+ grpNames == null ? null : new HashSet<>(grpNames),
+ fullCheck,
+ incIdx,
+ allRestoreHandlers
+ );
+
+ GridFutureAdapter<SnapshotPartitionsVerifyTaskResult> clusterOpFut = new GridFutureAdapter<>();
+
+ clusterOpFut.listen(fut -> clusterOpFuts.remove(reqId));
+
+ clusterOpFuts.put(reqId, clusterOpFut);
+
+ phase1CheckMetas.start(req.requestId(), req);
+
+ return clusterOpFut;
+ }
+
+ /** @return {@code True} if snapshot with specified name is checking. */
+ boolean isSnapshotChecking(String snpName) {
+ return contexts.get(snpName) != null;
+ }
+
+ /** @return {@code True} if node with the provided id is in the cluster and is a baseline node. {@code False} otherwise. */
+ private boolean baseline(UUID nodeId) {
+ ClusterNode node = kctx.cluster().get().node(nodeId);
+
+ return node != null && CU.baselineNode(node, kctx.state().clusterState());
+ }
+
+ /**
+ * @param reqNodes Set of required topology nodes.
+ * @param respNodes Set of responded topology nodes.
+ * @return Error, if no response was received from a required topology node.
+ */
+ private static @Nullable ClusterTopologyCheckedException checkNodeLeft(Collection<UUID> reqNodes, Set<UUID> respNodes) {
+ if (!respNodes.containsAll(reqNodes)) {
+ Set<UUID> leftNodes = new HashSet<>(reqNodes);
+
+ leftNodes.removeAll(respNodes);
+
+ return new ClusterTopologyCheckedException("Snapshot validation stopped. " +
+ "Required node has left the cluster [nodeId=" + leftNodes + ']');
+ }
+
+ return null;
+ }
+
+ /** Operation context. */
+ private static final class SnapshotCheckContext {
+ /** Request. */
+ @GridToStringInclude
+ private final SnapshotCheckProcessRequest req;
+
+ /**
+ * Metadatas to process on this node. Also indicates the snapshot parts to check on this node.
+ * @see #partitionsHashesFuture(SnapshotCheckContext)
+ */
+ @Nullable private List<SnapshotMetadata> metas;
+
+ /** Map of snapshot pathes per consistent id for {@link #metas}. */
+ @GridToStringInclude
+ @Nullable private Map<String, SnapshotFileTree> locFileTree;
+
+ /** All the snapshot metadatas. */
+ @Nullable private Map<ClusterNode, List<SnapshotMetadata>> clusterMetas;
+
+ /** Creates operation context. */
+ private SnapshotCheckContext(SnapshotCheckProcessRequest req) {
+ this.req = req;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(SnapshotCheckContext.class, this);
+ }
+ }
+
+ /** A DTO to transfer node's results for the both phases. */
+ private static final class SnapshotCheckResponse implements Serializable {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** The result. Is usually a collection or a map of hashes, metas, etc. */
+ private final Object result;
+
+ /** Exceptions per consistent id. */
+ @Nullable private final Map<String, Throwable> exceptions;
+
+ /** */
+ private SnapshotCheckResponse(Object result, @Nullable Map<String, Throwable> exceptions) {
+ assert result instanceof Serializable : "Snapshot check result is not serializable.";
+ assert exceptions == null || exceptions instanceof Serializable : "Snapshot check exceptions aren't serializable.";
+
+ this.result = result;
+ this.exceptions = exceptions == null ? null : Collections.unmodifiableMap(exceptions);
+ }
+
+ /** @return Exceptions per snapshot part's consistent id. */
+ private @Nullable Map<String, Throwable> exceptions() {
+ return exceptions;
+ }
+
+ /** @return Certain phase's and process' result. */
+ private <T> T result() {
+ return (T)result;
+ }
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotCheckProcessRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotCheckProcessRequest.java
new file mode 100644
index 00000000000..3fa731fc0e4
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotCheckProcessRequest.java
@@ -0,0 +1,107 @@
+/*
+ * 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.Collection;
+import java.util.UUID;
+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.jetbrains.annotations.Nullable;
+
+/**
+ * Snapshot full check (validation) distributed process request.
+ *
+ * @see SnapshotCheckProcess
+ */
+public class SnapshotCheckProcessRequest extends AbstractSnapshotOperationRequest {
+ /** Serial version uid. */
+ private static final long serialVersionUID = 0L;
+
+ /** If {@code true}, additionally calculates partition hashes. Otherwise, checks only snapshot integrity and partition counters. */
+ @GridToStringInclude
+ private final boolean fullCheck;
+
+ /**
+ * If {@code true}, all the registered {@link IgniteSnapshotManager#handlers()} of type {@link SnapshotHandlerType#RESTORE}
+ * are invoked. Otherwise, only snapshot metadatas and partition hashes are validated.
+ */
+ @GridToStringInclude
+ private final boolean allRestoreHandlers;
+
+ /** Incremental snapshot index. If not positive, snapshot is not considered as incremental. */
+ @GridToStringInclude
+ private final int incIdx;
+
+ /**
+ * Creates snapshot check process request.
+ *
+ * @param reqId Request ID.
+ * @param snpName Snapshot name.
+ * @param nodes Baseline node IDs that must be alive to complete the operation..
+ * @param snpPath Snapshot directory path.
+ * @param grps List of cache group names.
+ * @param fullCheck If {@code true}, additionally calculates partition hashes. Otherwise, checks only snapshot integrity
+ * and partition counters.
+ * @param incIdx Incremental snapshot index. If not positive, snapshot is not considered as incremental.
+ * @param allRestoreHandlers If {@code true}, all the registered {@link IgniteSnapshotManager#handlers()} of type
+ * {@link SnapshotHandlerType#RESTORE} are invoked. Otherwise, only snapshot metadatas and
+ * partition hashes are validated.
+ */
+ SnapshotCheckProcessRequest(
+ UUID reqId,
+ Collection<UUID> nodes,
+ String snpName,
+ String snpPath,
+ @Nullable Collection<String> grps,
+ boolean fullCheck,
+ int incIdx,
+ boolean allRestoreHandlers
+ ) {
+ super(reqId, snpName, snpPath, grps, 0, nodes);
+
+ assert !F.isEmpty(nodes);
+
+ this.fullCheck = fullCheck;
+ this.allRestoreHandlers = allRestoreHandlers;
+ this.incIdx = incIdx;
+ }
+
+ /**
+ * If {@code true}, all the registered {@link IgniteSnapshotManager#handlers()} of type {@link SnapshotHandlerType#RESTORE}
+ * are invoked. Otherwise, only snapshot metadatas and partition hashes are checked.
+ */
+ public boolean allRestoreHandlers() {
+ return allRestoreHandlers;
+ }
+
+ /** If {@code true}, additionally calculates partition hashes. Otherwise, checks only snapshot integrity and partition counters. */
+ public boolean fullCheck() {
+ return fullCheck;
+ }
+
+ /** @return Incremental snapshot index. If not positive, snapshot is not considered as incremental. */
+ public int incrementalIndex() {
+ return incIdx;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(SnapshotCheckProcessRequest.class, this, super.toString());
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotChecker.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotChecker.java
new file mode 100644
index 00000000000..2ad7244ee52
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotChecker.java
@@ -0,0 +1,148 @@
+/*
+ * 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.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+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.internal.GridKernalContext;
+import org.apache.ignite.internal.management.cache.IdleVerifyResult;
+import org.apache.ignite.internal.management.cache.PartitionKey;
+import org.apache.ignite.internal.processors.cache.persistence.filename.SnapshotFileTree;
+import org.apache.ignite.internal.processors.cache.verify.PartitionHashRecord;
+import org.jetbrains.annotations.Nullable;
+
+/** */
+public class SnapshotChecker {
+ /** */
+ private final IgniteLogger log;
+
+ /** */
+ private final GridKernalContext kctx;
+
+ /** */
+ private final ExecutorService executor;
+
+ /** */
+ public SnapshotChecker(GridKernalContext kctx) {
+ this.kctx = kctx;
+
+ executor = kctx.pools().getSnapshotExecutorService();
+
+ log = kctx.log(getClass());
+ }
+
+ /** Launches local metas checking. */
+ public CompletableFuture<List<SnapshotMetadata>> checkLocalMetas(
+ SnapshotFileTree sft,
+ int incIdx,
+ @Nullable Collection<Integer> grpIds
+ ) {
+ return CompletableFuture.supplyAsync(() ->
+ new SnapshotMetadataVerificationTask(kctx.grid(), log, sft, incIdx, grpIds).execute(), executor);
+ }
+
+ /** */
+ public CompletableFuture<IncrementalSnapshotVerificationTaskResult> checkIncrementalSnapshot(
+ SnapshotFileTree sft,
+ int incIdx
+ ) {
+ assert incIdx > 0;
+
+ return CompletableFuture.supplyAsync(
+ () -> new IncrementalSnapshotVerificationTask(kctx.grid(), log, sft, incIdx).execute(),
+ executor
+ );
+ }
+
+ /** */
+ public IdleVerifyResult reduceIncrementalResults(
+ SnapshotFileTree sft,
+ int incIdx,
+ Map<ClusterNode, IncrementalSnapshotVerificationTaskResult> results,
+ Map<ClusterNode, Exception> operationErrors
+ ) {
+ if (!operationErrors.isEmpty())
+ return IdleVerifyResult.builder().exceptions(operationErrors).build();
+
+ return new IncrementalSnapshotVerificationTask(kctx.grid(), log, sft, incIdx).reduce(results);
+ }
+
+ /** */
+ public Map<ClusterNode, Exception> reduceMetasResults(SnapshotFileTree sft, Map<ClusterNode, List<SnapshotMetadata>> metas) {
+ return new SnapshotMetadataVerificationTask(kctx.grid(), log, sft, 0, null).reduce(metas);
+ }
+
+ /**
+ * Calls all the registered custom validaton handlers.
+ *
+ * @see IgniteSnapshotManager#handlers()
+ */
+ public CompletableFuture<Map<String, SnapshotHandlerResult<Object>>> invokeCustomHandlers(
+ SnapshotMetadata meta,
+ SnapshotFileTree sft,
+ @Nullable Collection<String> grps,
+ boolean check
+ ) {
+ // The handlers use or may use the same snapshot pool. If it is configured with 1 thread, launching waiting task in
+ // the same pool might block it.
+ return CompletableFuture.supplyAsync(() ->
+ new SnapshotHandlerRestoreTask(kctx.grid(), log, sft, grps, check).execute()
+ );
+ }
+
+ /** Launches local partitions checking. */
+ public CompletableFuture<Map<PartitionKey, PartitionHashRecord>> checkPartitions(
+ SnapshotMetadata meta,
+ SnapshotFileTree sft,
+ @Nullable Collection<String> grps,
+ boolean forCreation,
+ boolean checkParts
+ ) {
+ // Await in the default executor to avoid blocking the snapshot executor if it has just one thread.
+ return CompletableFuture.supplyAsync(() -> {
+ SnapshotHandlerContext hctx = new SnapshotHandlerContext(
+ meta, grps, kctx.cluster().get().localNode(), sft, false, checkParts);
+
+ try {
+ return new SnapshotPartitionsVerifyHandler(kctx.cache().context()).invoke(hctx);
+ }
+ catch (IgniteCheckedException e) {
+ throw new IgniteException(e);
+ }
+ }, executor);
+ }
+
+ /**
+ * Checks results of all the snapshot validation handlres.
+ * @param snpName Snapshot name.
+ * @param results Results: checking node -> snapshot part's consistend id -> custom handler name -> handler result.
+ * @see #invokeCustomHandlers(SnapshotMetadata, SnapshotFileTree, Collection, boolean)
+ */
+ public void checkCustomHandlersResults(
+ String snpName,
+ Map<ClusterNode, Map<Object, Map<String, SnapshotHandlerResult<?>>>> results
+ ) {
+ new SnapshotHandlerRestoreTask(kctx.grid(), log, null, null, true).reduce(snpName, results);
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotHandlerRestoreTask.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotHandlerRestoreTask.java
index 773f0de14c5..9a26cd94120 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotHandlerRestoreTask.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotHandlerRestoreTask.java
@@ -26,53 +26,69 @@ import java.util.Map;
import java.util.UUID;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
-import org.apache.ignite.compute.ComputeJobResult;
-import org.apache.ignite.internal.util.typedef.F;
-import org.jetbrains.annotations.Nullable;
+import org.apache.ignite.IgniteLogger;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.processors.cache.persistence.filename.SnapshotFileTree;
/**
* Snapshot restore operation handling task.
*/
-public class SnapshotHandlerRestoreTask extends AbstractSnapshotVerificationTask {
- /** Serial version uid. */
- private static final long serialVersionUID = 0L;
-
- /** {@inheritDoc} */
- @Override protected SnapshotHandlerRestoreJob createJob(
- String name,
- String folderName,
- String consId,
- SnapshotPartitionsVerifyTaskArg args
+public class SnapshotHandlerRestoreTask {
+ /** */
+ private final IgniteEx ignite;
+
+ /** */
+ private final IgniteLogger log;
+
+ /** */
+ private final SnapshotHandlerRestoreJob job;
+
+ /** */
+ SnapshotHandlerRestoreTask(
+ IgniteEx ignite,
+ IgniteLogger log,
+ SnapshotFileTree sft,
+ Collection<String> grps,
+ boolean check
) {
- return new SnapshotHandlerRestoreJob(name, args.snapshotPath(), folderName, consId, args.cacheGroupNames(), args.check());
+ job = new SnapshotHandlerRestoreJob(ignite, sft, grps, check);
+ this.ignite = ignite;
+ this.log = log;
}
- /** {@inheritDoc} */
- @SuppressWarnings("rawtypes")
- @Nullable @Override public SnapshotPartitionsVerifyTaskResult reduce(List<ComputeJobResult> results) {
+ /** */
+ public Map<String, SnapshotHandlerResult<Object>> execute() {
+ return job.execute0();
+ }
+
+ /** */
+ public void reduce(
+ String snapshotName,
+ Map<ClusterNode, Map<Object, Map<String, SnapshotHandlerResult<?>>>> results
+ ) {
Map<String, List<SnapshotHandlerResult<?>>> clusterResults = new HashMap<>();
Collection<UUID> execNodes = new ArrayList<>(results.size());
- for (ComputeJobResult res : results) {
- if (res.getException() != null)
- throw res.getException();
+ // Checking node -> Map by consistend id.
+ for (Map.Entry<ClusterNode, Map<Object, Map<String, SnapshotHandlerResult<?>>>> nodeRes : results.entrySet()) {
+ // Consistent id -> Map by handler name.
+ for (Map.Entry<Object, Map<String, SnapshotHandlerResult<?>>> res : nodeRes.getValue().entrySet()) {
+ // Depending on the job mapping, we can get several different results from one node.
+ execNodes.add(nodeRes.getKey().id());
- // Depending on the job mapping, we can get several different results from one node.
- execNodes.add(res.getNode().id());
+ Map<String, SnapshotHandlerResult<?>> nodeDataMap = res.getValue();
- Map<String, SnapshotHandlerResult> nodeDataMap = res.getData();
+ assert nodeDataMap != null : "At least the default snapshot restore handler should have been executed ";
- assert nodeDataMap != null : "At least the default snapshot restore handler should have been executed ";
+ for (Map.Entry<String, SnapshotHandlerResult<?>> entry : nodeDataMap.entrySet()) {
+ String hndName = entry.getKey();
- for (Map.Entry<String, SnapshotHandlerResult> entry : nodeDataMap.entrySet()) {
- String hndName = entry.getKey();
-
- clusterResults.computeIfAbsent(hndName, v -> new ArrayList<>()).add(entry.getValue());
+ clusterResults.computeIfAbsent(hndName, v -> new ArrayList<>()).add(entry.getValue());
+ }
}
}
- String snapshotName = F.first(F.first(metas.values())).snapshotName();
-
try {
ignite.context().cache().context().snapshotMgr().handlers().completeAll(
SnapshotHandlerType.RESTORE, snapshotName, clusterResults, execNodes, wrns -> {});
@@ -82,36 +98,40 @@ public class SnapshotHandlerRestoreTask extends AbstractSnapshotVerificationTask
throw new IgniteException(e);
}
-
- return new SnapshotPartitionsVerifyTaskResult(metas, null);
}
/** Invokes all {@link SnapshotHandlerType#RESTORE} handlers locally. */
- private static class SnapshotHandlerRestoreJob extends AbstractSnapshotVerificationJob {
- /** Serial version uid. */
- private static final long serialVersionUID = 0L;
+ private static class SnapshotHandlerRestoreJob {
+ /** */
+ private final IgniteEx ignite;
+
+ /** */
+ private final SnapshotFileTree sft;
+
+ /** */
+ private final Collection<String> rqGrps;
+
+ /** */
+ private final boolean check;
/**
- * @param snpName Snapshot name.
- * @param snpPath Snapshot directory path.
- * @param folderName Folder name for snapshot.
- * @param consId Consistent id of the related node.
* @param grps Cache group names.
* @param check If {@code true} check snapshot before restore.
*/
- public SnapshotHandlerRestoreJob(
- String snpName,
- @Nullable String snpPath,
- String folderName,
- String consId,
+ SnapshotHandlerRestoreJob(
+ IgniteEx ignite,
+ SnapshotFileTree sft,
Collection<String> grps,
boolean check
) {
- super(snpName, snpPath, folderName, consId, grps, check);
+ this.ignite = ignite;
+ this.sft = sft;
+ this.rqGrps = grps;
+ this.check = check;
}
- /** {@inheritDoc} */
- @Override public Map<String, SnapshotHandlerResult<Object>> execute0() {
+ /** */
+ public Map<String, SnapshotHandlerResult<Object>> execute0() {
try {
IgniteSnapshotManager snpMgr = ignite.context().cache().context().snapshotMgr();
SnapshotMetadata meta = snpMgr.readSnapshotMetadata(sft.meta());
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotMetadataVerificationTask.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotMetadataVerificationTask.java
index 0636a6e1015..a00f07e8a96 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotMetadataVerificationTask.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotMetadataVerificationTask.java
@@ -18,7 +18,6 @@
package org.apache.ignite.internal.processors.cache.persistence.snapshot;
import java.io.IOException;
-import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
@@ -32,98 +31,89 @@ 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.processors.cache.GridCacheSharedContext;
import org.apache.ignite.internal.processors.cache.persistence.filename.SnapshotFileTree;
import org.apache.ignite.internal.processors.cache.persistence.filename.SnapshotFileTree.IncrementalSnapshotFileTree;
import org.apache.ignite.internal.processors.cache.persistence.wal.FileDescriptor;
import org.apache.ignite.internal.processors.cache.persistence.wal.reader.IgniteWalIteratorFactory;
-import org.apache.ignite.internal.processors.task.GridInternal;
import org.apache.ignite.internal.util.typedef.F;
-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;
/** Snapshot task to verify snapshot metadata on the baseline nodes for given snapshot name. */
-@GridInternal
-public class SnapshotMetadataVerificationTask
- extends ComputeTaskAdapter<SnapshotMetadataVerificationTaskArg, SnapshotMetadataVerificationTaskResult> {
- /** Serial version uid. */
- private static final long serialVersionUID = 0L;
-
+public class SnapshotMetadataVerificationTask {
/** */
- private SnapshotMetadataVerificationTaskArg arg;
+ private final MetadataVerificationJob job;
/** */
- @IgniteInstanceResource
- private transient IgniteEx ignite;
-
- /** {@inheritDoc} */
- @Override public @NotNull Map<? extends ComputeJob, ClusterNode> map(
- List<ClusterNode> subgrid,
- SnapshotMetadataVerificationTaskArg arg
- ) throws IgniteException {
- this.arg = arg;
-
- Map<ComputeJob, ClusterNode> map = U.newHashMap(subgrid.size());
-
- for (ClusterNode node : subgrid)
- map.put(new MetadataVerificationJob(arg), node);
+ public SnapshotMetadataVerificationTask(
+ IgniteEx ignite,
+ IgniteLogger log,
+ SnapshotFileTree sft,
+ int incrementIdx,
+ Collection<Integer> grpIds
+ ) {
+ job = new MetadataVerificationJob(ignite, log, sft, incrementIdx, grpIds);
+ }
- return map;
+ /** */
+ public List<SnapshotMetadata> execute() {
+ return job.execute();
}
/** Job that verifies snapshot on an Ignite node. */
- private static class MetadataVerificationJob extends ComputeJobAdapter {
+ private static class MetadataVerificationJob {
+ /** */
+ private final IgniteEx ignite;
+
/** */
- private static final long serialVersionUID = 0L;
+ private final IgniteLogger log;
/** */
- @IgniteInstanceResource
- private transient IgniteEx ignite;
+ private final SnapshotFileTree sft;
/** */
- @LoggerResource
- private transient IgniteLogger log;
+ private final int incrementIdx;
/** */
- private final SnapshotMetadataVerificationTaskArg arg;
+ private final Collection<Integer> grpIds;
/** */
- public MetadataVerificationJob(SnapshotMetadataVerificationTaskArg arg) {
- this.arg = arg;
+ private MetadataVerificationJob(
+ IgniteEx ignite,
+ IgniteLogger log,
+ SnapshotFileTree sft,
+ int incrementIdx,
+ Collection<Integer> grpIds
+ ) {
+ this.ignite = ignite;
+ this.sft = sft;
+ this.incrementIdx = incrementIdx;
+ this.grpIds = grpIds;
+ this.log = log;
}
- /** {@inheritDoc} */
- @Override public List<SnapshotMetadata> execute() {
+ /** */
+ public List<SnapshotMetadata> execute() {
IgniteSnapshotManager snpMgr = ignite.context().cache().context().snapshotMgr();
- SnapshotFileTree sft = new SnapshotFileTree(ignite.context(), arg.snapshotName(), arg.snapshotPath());
-
List<SnapshotMetadata> snpMeta = snpMgr.readSnapshotMetadatas(sft);
for (SnapshotMetadata meta : snpMeta)
checkMeta(meta);
- if (arg.incrementIndex() > 0) {
+ if (incrementIdx > 0) {
List<SnapshotMetadata> metas = snpMeta.stream()
.filter(m -> m.consistentId().equals(sft.consistentId()))
.collect(Collectors.toList());
if (metas.size() != 1) {
throw new IgniteException("Failed to find single snapshot metafile on local node [locNodeId="
- + ignite.localNode().consistentId() + ", metas=" + snpMeta + ", snpName=" + arg.snapshotName()
- + ", snpPath=" + arg.snapshotPath() + "]. Incremental snapshots requires exactly one meta file " +
+ + ignite.localNode().consistentId() + ", metas=" + snpMeta + ", snpName=" + sft.name()
+ + ", snpPath=" + sft.root() + "]. Incremental snapshots requires exactly one meta file " +
"per node because they don't support restoring on a different topology.");
}
- checkIncrementalSnapshots(metas.get(0), sft, arg.incrementIndex());
+ checkIncrementalSnapshots(metas.get(0), sft, incrementIdx);
}
return snpMeta;
@@ -145,7 +135,7 @@ public class SnapshotMetadataVerificationTask
"key digest. To restore this snapshot, start Ignite with the same master key.");
}
- Collection<Integer> grpIds = new HashSet<>(F.isEmpty(arg.grpIds()) ? meta.cacheGroupIds() : arg.grpIds());
+ Collection<Integer> grpIds = new HashSet<>(F.isEmpty(this.grpIds) ? meta.cacheGroupIds() : this.grpIds);
if (meta.hasCompressedGroups() && grpIds.stream().anyMatch(meta::isGroupWithCompression)) {
try {
@@ -168,7 +158,7 @@ public class SnapshotMetadataVerificationTask
if (!grpIds.isEmpty() && !new HashSet<>(meta.cacheGroupIds()).containsAll(grpIds)) {
throw new IllegalArgumentException("Cache group(s) was not found in the snapshot [groups=" + grpIds +
- ", snapshot=" + arg.snapshotName() + ']');
+ ", snapshot=" + sft.name() + ']');
}
}
@@ -187,7 +177,7 @@ public class SnapshotMetadataVerificationTask
if (!ift.root().exists()) {
throw new IllegalArgumentException("No incremental snapshot found " +
- "[snpName=" + arg.snapshotName() + ", snpPath=" + arg.snapshotPath() + ", incrementIndex=" + inc + ']');
+ "[snpName=" + sft.name() + ", snpPath=" + sft.root() + ", incrementIndex=" + inc + ']');
}
IncrementalSnapshotMetadata incMeta = snpMgr.readIncrementalSnapshotMetadata(ift.meta());
@@ -246,22 +236,15 @@ public class SnapshotMetadataVerificationTask
}
}
- /** {@inheritDoc} */
- @Override public SnapshotMetadataVerificationTaskResult reduce(List<ComputeJobResult> results) throws IgniteException {
- Map<ClusterNode, List<SnapshotMetadata>> reduceRes = new HashMap<>();
+ /** */
+ public Map<ClusterNode, Exception> reduce(Map<ClusterNode, List<SnapshotMetadata>> results) throws IgniteException {
Map<ClusterNode, Exception> exs = new HashMap<>();
SnapshotMetadata first = null;
Set<String> baselineMetasLeft = Collections.emptySet();
- for (ComputeJobResult res : results) {
- if (res.getException() != null) {
- exs.put(res.getNode(), res.getException());
-
- continue;
- }
-
- List<SnapshotMetadata> metas = res.getData();
+ for (Map.Entry<ClusterNode, List<SnapshotMetadata>> res : results.entrySet()) {
+ List<SnapshotMetadata> metas = res.getValue();
for (SnapshotMetadata meta : metas) {
if (first == null) {
@@ -273,40 +256,25 @@ public class SnapshotMetadataVerificationTask
baselineMetasLeft.remove(meta.consistentId());
if (!first.sameSnapshot(meta)) {
- exs.put(res.getNode(),
+ exs.put(res.getKey(),
new IgniteException("An error occurred during comparing snapshot metadata from cluster nodes " +
- "[first=" + first + ", meta=" + meta + ", nodeId=" + res.getNode().id() + ']'));
-
- continue;
+ "[first=" + first + ", meta=" + meta + ", nodeId=" + res.getKey().id() + ']'));
}
-
- reduceRes.computeIfAbsent(res.getNode(), n -> new ArrayList<>()).add(meta);
}
}
if (first == null && exs.isEmpty()) {
- assert !results.isEmpty();
-
- for (ComputeJobResult res : results) {
- Exception e = new IllegalArgumentException("Snapshot does not exists [snapshot=" + arg.snapshotName()
- + (arg.snapshotPath() != null ? ", baseDir=" + arg.snapshotPath() : "") + ", consistentId="
- + res.getNode().consistentId() + ']');
+ SnapshotFileTree sft = job.sft;
- exs.put(res.getNode(), e);
- }
+ throw new IllegalArgumentException("Snapshot does not exists [snapshot=" + sft.name()
+ + ", baseDir=" + sft.root() + ", consistentId=" + sft.consistentId() + ']');
}
if (!F.isEmpty(baselineMetasLeft) && F.isEmpty(exs)) {
- exs.put(ignite.localNode(), new IgniteException("No snapshot metadatas found for the baseline nodes " +
- "with consistent ids: " + String.join(", ", baselineMetasLeft)));
+ throw new IgniteException("No snapshot metadatas found for the baseline nodes " +
+ "with consistent ids: " + String.join(", ", baselineMetasLeft));
}
- return new SnapshotMetadataVerificationTaskResult(reduceRes, exs);
- }
-
- /** {@inheritDoc} */
- @Override public ComputeJobResultPolicy result(ComputeJobResult res, List<ComputeJobResult> rcvd) throws IgniteException {
- // Handle all exceptions during the `reduce` operation.
- return ComputeJobResultPolicy.WAIT;
+ return exs;
}
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotMetadataVerificationTaskArg.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotMetadataVerificationTaskArg.java
deleted file mode 100644
index 83df7ba55c0..00000000000
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotMetadataVerificationTaskArg.java
+++ /dev/null
@@ -1,111 +0,0 @@
-/*
- * 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.IOException;
-import java.io.ObjectInput;
-import java.io.ObjectOutput;
-import java.util.Collection;
-import org.apache.ignite.internal.dto.IgniteDataTransferObject;
-import org.apache.ignite.internal.util.typedef.internal.S;
-import org.apache.ignite.internal.util.typedef.internal.U;
-import org.jetbrains.annotations.Nullable;
-
-/**
- * Input parameters for checking snapshot metadata.
- */
-public class SnapshotMetadataVerificationTaskArg extends IgniteDataTransferObject {
- /** Serial version UID. */
- private static final long serialVersionUID = 0L;
-
- /** Snapshot name. */
- private String snpName;
-
- /** Snapshot directory path. */
- private String snpPath;
-
- /** Incremental snapshot index. */
- private int incIdx;
-
- /** Cache group ids. */
- @Nullable private Collection<Integer> grpIds;
-
- /** Default constructor. */
- public SnapshotMetadataVerificationTaskArg() {
- // No-op.
- }
-
- /**
- * @param snpName Snapshot name.
- * @param snpPath Snapshot directory path.
- */
- public SnapshotMetadataVerificationTaskArg(String snpName, @Nullable String snpPath, int incIdx, Collection<Integer> grpIds) {
- this.snpName = snpName;
- this.snpPath = snpPath;
- this.incIdx = incIdx;
- this.grpIds = grpIds;
- }
-
- /**
- * @return Snapshot name.
- */
- public String snapshotName() {
- return snpName;
- }
-
- /**
- * @return Snapshot directory path.
- */
- public String snapshotPath() {
- return snpPath;
- }
-
- /**
- * @return Incremental snapshot index.
- */
- public int incrementIndex() {
- return incIdx;
- }
-
- /**
- * @return Cache group ids.
- */
- @Nullable public Collection<Integer> grpIds() {
- return grpIds;
- }
-
- /** {@inheritDoc} */
- @Override protected void writeExternalData(ObjectOutput out) throws IOException {
- U.writeString(out, snpName);
- U.writeString(out, snpPath);
- out.writeInt(incIdx);
- U.writeCollection(out, grpIds);
- }
-
- /** {@inheritDoc} */
- @Override protected void readExternalData(ObjectInput in) throws IOException, ClassNotFoundException {
- snpName = U.readString(in);
- snpPath = U.readString(in);
- incIdx = in.readInt();
- grpIds = U.readCollection(in);
- }
-
- /** {@inheritDoc} */
- @Override public String toString() {
- return S.toString(SnapshotMetadataVerificationTaskArg.class, this);
- }
-}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotMetadataVerificationTaskResult.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotMetadataVerificationTaskResult.java
deleted file mode 100644
index d8f3a76164e..00000000000
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotMetadataVerificationTaskResult.java
+++ /dev/null
@@ -1,75 +0,0 @@
-/*
- * 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.IOException;
-import java.io.ObjectInput;
-import java.io.ObjectOutput;
-import java.util.Collections;
-import java.util.List;
-import java.util.Map;
-import org.apache.ignite.cluster.ClusterNode;
-import org.apache.ignite.internal.dto.IgniteDataTransferObject;
-import org.apache.ignite.internal.util.typedef.internal.U;
-
-/** */
-public class SnapshotMetadataVerificationTaskResult extends IgniteDataTransferObject {
- /** */
- private static final long serialVersionUID = 0L;
-
- /** Full snapshot metadata. */
- private Map<ClusterNode, List<SnapshotMetadata>> meta;
-
- /** Errors happened during snapshot metadata verification. */
- private Map<ClusterNode, Exception> exceptions;
-
- /** */
- public SnapshotMetadataVerificationTaskResult(
- Map<ClusterNode, List<SnapshotMetadata>> meta,
- Map<ClusterNode, Exception> exceptions
- ) {
- this.meta = Collections.unmodifiableMap(meta);
- this.exceptions = Collections.unmodifiableMap(exceptions);
- }
-
- /** */
- public SnapshotMetadataVerificationTaskResult() {
- }
-
- /** @return Errors happened during snapshot metadata verification. */
- public Map<ClusterNode, Exception> exceptions() {
- return Collections.unmodifiableMap(exceptions);
- }
-
- /** @return Full snapshot metadata. */
- public Map<ClusterNode, List<SnapshotMetadata>> meta() {
- return Collections.unmodifiableMap(meta);
- }
-
- /** {@inheritDoc} */
- @Override protected void writeExternalData(ObjectOutput out) throws IOException {
- U.writeMap(out, meta);
- U.writeMap(out, exceptions);
- }
-
- /** {@inheritDoc} */
- @Override protected void readExternalData(ObjectInput in) throws IOException, ClassNotFoundException {
- meta = U.readMap(in);
- exceptions = U.readMap(in);
- }
-}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotOperationRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotOperationRequest.java
index f21ffd4b00b..e1b01b25611 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotOperationRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotOperationRequest.java
@@ -17,7 +17,6 @@
package org.apache.ignite.internal.processors.cache.persistence.snapshot;
-import java.io.Serializable;
import java.util.Collection;
import java.util.List;
import java.util.Set;
@@ -26,35 +25,16 @@ import org.apache.ignite.internal.processors.cache.persistence.filename.Snapshot
import org.apache.ignite.internal.util.distributed.DistributedProcess;
import org.apache.ignite.internal.util.distributed.DistributedProcess.DistributedProcessType;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
-import org.apache.ignite.internal.util.tostring.GridToStringInclude;
import org.apache.ignite.internal.util.typedef.internal.S;
-import org.apache.ignite.internal.util.typedef.internal.U;
import org.jetbrains.annotations.Nullable;
/**
* Snapshot operation start request for {@link DistributedProcess} initiate message.
*/
-public class SnapshotOperationRequest implements Serializable {
+public class SnapshotOperationRequest extends AbstractSnapshotOperationRequest {
/** Serial version uid. */
private static final long serialVersionUID = 0L;
- /** Request ID. */
- private final UUID reqId;
-
- /** Snapshot name. */
- private final String snpName;
-
- /** Snapshot directory path. */
- private final String snpPath;
-
- /** Baseline node IDs that must be alive to complete the operation. */
- @GridToStringInclude
- private final Set<UUID> nodes;
-
- /** List of cache group names. */
- @GridToStringInclude
- private final Collection<String> grps;
-
/** Operational node ID. */
private final UUID opNodeId;
@@ -84,9 +64,6 @@ public class SnapshotOperationRequest implements Serializable {
/** Flag indicating that the {@link DistributedProcessType#START_SNAPSHOT} phase has completed. */
private transient volatile boolean startStageEnded;
- /** Operation start time. */
- private final long startTime;
-
/** If {@code true} then incremental snapshot requested. */
private final boolean incremental;
@@ -133,54 +110,15 @@ public class SnapshotOperationRequest implements Serializable {
boolean compress,
boolean encrypt
) {
- this.reqId = reqId;
+ super(reqId, snpName, snpPath, grps, incIdx, nodes);
+
this.opNodeId = opNodeId;
- this.snpName = snpName;
- this.grps = grps;
- this.nodes = nodes;
- this.snpPath = snpPath;
this.incremental = incremental;
this.incIdx = incIdx;
this.onlyPrimary = onlyPrimary;
this.dump = dump;
this.compress = compress;
this.encrypt = encrypt;
- startTime = U.currentTimeMillis();
- }
-
- /**
- * @return Request ID.
- */
- public UUID requestId() {
- return reqId;
- }
-
- /**
- * @return Snapshot name.
- */
- public String snapshotName() {
- return snpName;
- }
-
- /**
- * @return Snapshot directory path.
- */
- public String snapshotPath() {
- return snpPath;
- }
-
- /**
- * @return List of cache group names.
- */
- public @Nullable Collection<String> groups() {
- return grps;
- }
-
- /**
- * @return Baseline node IDs that must be alive to complete the operation.
- */
- public Set<UUID> nodes() {
- return nodes;
}
/**
@@ -234,11 +172,6 @@ public class SnapshotOperationRequest implements Serializable {
return encrypt;
}
- /** @return Start time. */
- public long startTime() {
- return startTime;
- }
-
/**
* @return Flag indicating that the {@link DistributedProcessType#START_SNAPSHOT} phase has completed.
*/
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotPartitionsVerifyHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotPartitionsVerifyHandler.java
index b78606bccf5..61ac8c0f265 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotPartitionsVerifyHandler.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotPartitionsVerifyHandler.java
@@ -172,12 +172,12 @@ public class SnapshotPartitionsVerifyHandler implements SnapshotHandler<Map<Part
return meta.dump()
? checkDumpFiles(opCtx, partFiles)
- : checkSnapshotFiles(opCtx, grpDirs, meta, partFiles, isPunchHoleEnabled(opCtx, grpDirs.keySet()));
+ : checkSnapshotFiles(opCtx.snapshotFileTree(), grpDirs, meta, partFiles, isPunchHoleEnabled(opCtx, grpDirs.keySet()));
}
/** */
private Map<PartitionKey, PartitionHashRecord> checkSnapshotFiles(
- SnapshotHandlerContext opCtx,
+ SnapshotFileTree sft,
Map<Integer, List<File>> grpDirs,
SnapshotMetadata meta,
Set<File> partFiles,
@@ -192,7 +192,7 @@ public class SnapshotPartitionsVerifyHandler implements SnapshotHandler<Map<Part
GridKernalContext snpCtx = new StandaloneGridKernalContext(
log,
cctx.kernalContext().compress(),
- new NodeFileTree(opCtx.snapshotFileTree().root(), meta.folderName())
+ new NodeFileTree(sft.root(), meta.folderName())
);
FilePageStoreManager storeMgr = (FilePageStoreManager)cctx.pageStore();
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
deleted file mode 100644
index 3aae970b0e7..00000000000
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotPartitionsVerifyTask.java
+++ /dev/null
@@ -1,132 +0,0 @@
-/*
- * 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.IOException;
-import java.util.Collection;
-import java.util.List;
-import java.util.Map;
-import java.util.Objects;
-import org.apache.ignite.IgniteCheckedException;
-import org.apache.ignite.IgniteException;
-import org.apache.ignite.compute.ComputeJobResult;
-import org.apache.ignite.internal.management.cache.PartitionKey;
-import org.apache.ignite.internal.management.cache.VerifyBackupPartitionsTask;
-import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
-import org.apache.ignite.internal.processors.cache.verify.PartitionHashRecord;
-import org.apache.ignite.internal.processors.task.GridInternal;
-import org.jetbrains.annotations.Nullable;
-
-import static org.apache.ignite.internal.management.cache.VerifyBackupPartitionsTask.reduce0;
-
-/**
- * Task for checking snapshot partitions consistency the same way as {@link VerifyBackupPartitionsTask} does.
- * Since a snapshot partitions already stored apart on disk the is no requirement for a cluster upcoming updates
- * to be hold on.
- */
-@GridInternal
-public class SnapshotPartitionsVerifyTask extends AbstractSnapshotVerificationTask {
- /** Serial version uid. */
- private static final long serialVersionUID = 0L;
-
- /** {@inheritDoc} */
- @Override protected VerifySnapshotPartitionsJob createJob(
- String name,
- String folderName,
- String consId,
- SnapshotPartitionsVerifyTaskArg args
- ) {
- return new VerifySnapshotPartitionsJob(name, args.snapshotPath(), folderName, consId, args.cacheGroupNames(), args.check());
- }
-
- /** {@inheritDoc} */
- @Override public @Nullable SnapshotPartitionsVerifyTaskResult reduce(List<ComputeJobResult> results) throws IgniteException {
- return new SnapshotPartitionsVerifyTaskResult(metas, reduce0(results));
- }
-
- /** Job that collects update counters of snapshot partitions on the node it executes. */
- private static class VerifySnapshotPartitionsJob extends AbstractSnapshotVerificationJob {
- /** Serial version uid. */
- private static final long serialVersionUID = 0L;
-
- /**
- * @param snpName Snapshot name to validate.
- * @param snpPath Snapshot directory path.
- * @param folderName Folder name for snapshot.
- * @param consId Consistent id of the related node.
- * @param rqGrps Set of cache groups to be checked in the snapshot or {@code empty} to check everything.
- * @param check If {@code true} check snapshot before restore.
- */
- public VerifySnapshotPartitionsJob(
- String snpName,
- @Nullable String snpPath,
- String folderName,
- String consId,
- Collection<String> rqGrps,
- boolean check
- ) {
- super(snpName, snpPath, folderName, consId, rqGrps, check);
- }
-
- /** {@inheritDoc} */
- @Override public Map<PartitionKey, PartitionHashRecord> execute0() throws IgniteException {
- GridCacheSharedContext<?, ?> cctx = ignite.context().cache().context();
-
- if (log.isInfoEnabled()) {
- log.info("Verify snapshot partitions procedure has been initiated " +
- "[snpName=" + snpName + ", consId=" + consId + ']');
- }
-
- try {
- SnapshotMetadata meta = cctx.snapshotMgr().readSnapshotMetadata(sft.meta());
-
- return new SnapshotPartitionsVerifyHandler(cctx)
- .invoke(new SnapshotHandlerContext(meta, rqGrps, ignite.localNode(), sft, false, check));
- }
- catch (IgniteCheckedException | IOException e) {
- throw new IgniteException(e);
- }
- finally {
- if (log.isInfoEnabled()) {
- log.info("Verify snapshot partitions procedure has been finished " +
- "[snpName=" + snpName + ", consId=" + consId + ']');
- }
- }
- }
-
- /** {@inheritDoc} */
- @Override public boolean equals(Object o) {
- if (this == o)
- return true;
-
- if (o == null || getClass() != o.getClass())
- return false;
-
- VerifySnapshotPartitionsJob job = (VerifySnapshotPartitionsJob)o;
-
- return snpName.equals(job.snpName) && consId.equals(job.consId) &&
- Objects.equals(rqGrps, job.rqGrps) && Objects.equals(snpPath, job.snpPath) &&
- Objects.equals(check, job.check);
- }
-
- /** {@inheritDoc} */
- @Override public int hashCode() {
- return Objects.hash(snpName, consId, rqGrps, snpPath, check);
- }
- }
-}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotPartitionsVerifyTaskArg.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotPartitionsVerifyTaskArg.java
deleted file mode 100644
index 1b21c1aea58..00000000000
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotPartitionsVerifyTaskArg.java
+++ /dev/null
@@ -1,134 +0,0 @@
-/*
- * 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.IOException;
-import java.io.ObjectInput;
-import java.io.ObjectOutput;
-import java.util.Collection;
-import java.util.List;
-import java.util.Map;
-import org.apache.ignite.cluster.ClusterNode;
-import org.apache.ignite.internal.dto.IgniteDataTransferObject;
-import org.apache.ignite.internal.util.typedef.internal.S;
-import org.apache.ignite.internal.util.typedef.internal.U;
-import org.jetbrains.annotations.Nullable;
-
-/**
- * Input parameters for checking snapshot partitions consistency task.
- */
-public class SnapshotPartitionsVerifyTaskArg extends IgniteDataTransferObject {
- /** Serial version UID. */
- private static final long serialVersionUID = 0L;
-
- /** Cache group names to be verified. */
- @Nullable private Collection<String> grpNames;
-
- /** The map of distribution of snapshot metadata pieces across the cluster. */
- private Map<ClusterNode, List<SnapshotMetadata>> clusterMetas;
-
- /** Snapshot directory path. */
- @Nullable private String snpPath;
-
- /** If {@code true} check snapshot integrity. */
- private boolean check;
-
- /** Incremental snapshot index. */
- private int incIdx;
-
- /** Default constructor. */
- public SnapshotPartitionsVerifyTaskArg() {
- // No-op.
- }
-
- /**
- * @param grpNames Cache group names to be verified.
- * @param clusterMetas The map of distribution of snapshot metadata pieces across the cluster.
- * @param snpPath Snapshot directory path.
- * @param incIdx Incremental snapshot index.
- * @param check If {@code true} check snapshot integrity.
- */
- public SnapshotPartitionsVerifyTaskArg(
- @Nullable Collection<String> grpNames,
- Map<ClusterNode, List<SnapshotMetadata>> clusterMetas,
- @Nullable String snpPath,
- int incIdx,
- boolean check
- ) {
- this.grpNames = grpNames;
- this.clusterMetas = clusterMetas;
- this.snpPath = snpPath;
- this.incIdx = incIdx;
- this.check = check;
- }
-
- /**
- * @return Cache group names to be verified.
- */
- @Nullable public Collection<String> cacheGroupNames() {
- return grpNames;
- }
-
- /**
- * @return The map of distribution of snapshot metadata pieces across the cluster.
- */
- public Map<ClusterNode, List<SnapshotMetadata>> clusterMetadata() {
- return clusterMetas;
- }
-
- /**
- * @return Snapshot directory path.
- */
- @Nullable public String snapshotPath() {
- return snpPath;
- }
-
- /**
- * @return Incremental snapshot index.
- */
- public int incrementIndex() {
- return incIdx;
- }
-
- /** @return If {@code true} check snapshot integrity. */
- public boolean check() {
- return check;
- }
-
- /** {@inheritDoc} */
- @Override protected void writeExternalData(ObjectOutput out) throws IOException {
- U.writeCollection(out, grpNames);
- U.writeMap(out, clusterMetas);
- U.writeString(out, snpPath);
- out.writeBoolean(check);
- out.writeInt(incIdx);
- }
-
- /** {@inheritDoc} */
- @Override protected void readExternalData(ObjectInput in) throws IOException, ClassNotFoundException {
- grpNames = U.readCollection(in);
- clusterMetas = U.readMap(in);
- snpPath = U.readString(in);
- check = in.readBoolean();
- incIdx = in.readInt();
- }
-
- /** {@inheritDoc} */
- @Override public String toString() {
- return S.toString(SnapshotPartitionsVerifyTaskArg.class, this);
- }
-}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotPartitionsVerifyTaskResult.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotPartitionsVerifyTaskResult.java
index 68033ce19dc..6f64ebe9841 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotPartitionsVerifyTaskResult.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotPartitionsVerifyTaskResult.java
@@ -57,7 +57,7 @@ public class SnapshotPartitionsVerifyTaskResult extends IgniteDataTransferObject
* @param metas Map of snapshot metadata information found on each cluster node.
* @param idleRes Result of cluster nodes partitions comparison.
*/
- public SnapshotPartitionsVerifyTaskResult(
+ SnapshotPartitionsVerifyTaskResult(
Map<ClusterNode, List<SnapshotMetadata>> metas,
@Nullable IdleVerifyResult idleRes
) {
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotRestoreProcess.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotRestoreProcess.java
index 137a1e0734a..8b77b0c845a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotRestoreProcess.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotRestoreProcess.java
@@ -1611,10 +1611,10 @@ public class SnapshotRestoreProcess {
/**
* @param reqNodes Set of required topology nodes.
- * @param respNodes Set of responding topology nodes.
- * @return Error, if no response was received from the required topology node.
+ * @param respNodes Set of responded topology nodes.
+ * @return Error, if no response was received from a required topology node.
*/
- private Exception checkNodeLeft(Collection<UUID> reqNodes, Set<UUID> respNodes) {
+ private static ClusterTopologyCheckedException checkNodeLeft(Collection<UUID> reqNodes, Set<UUID> respNodes) {
if (!respNodes.containsAll(reqNodes)) {
Set<UUID> leftNodes = new HashSet<>(reqNodes);
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/distributed/DistributedProcess.java b/modules/core/src/main/java/org/apache/ignite/internal/util/distributed/DistributedProcess.java
index ee5a99931df..1c36eb825f1 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/distributed/DistributedProcess.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/distributed/DistributedProcess.java
@@ -489,6 +489,16 @@ public class DistributedProcess<I extends Serializable, R extends Serializable>
/**
* Incremental snapshot restore start phase.
*/
- RESTORE_INCREMENTAL_SNAPSHOT_START
+ RESTORE_INCREMENTAL_SNAPSHOT_START,
+
+ /**
+ * Snapshot metadatas check.
+ */
+ CHECK_SNAPSHOT_METAS,
+
+ /**
+ * Snapshot partitions validation.
+ */
+ CHECK_SNAPSHOT_PARTS
}
}
diff --git a/modules/core/src/main/resources/META-INF/classnames.properties b/modules/core/src/main/resources/META-INF/classnames.properties
index 071bc6eb437..319e2991b35 100644
--- a/modules/core/src/main/resources/META-INF/classnames.properties
+++ b/modules/core/src/main/resources/META-INF/classnames.properties
@@ -1328,16 +1328,8 @@ org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotHandler
org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotHandlerResult
org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotHandlerType
org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotMetadata
-org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotMetadataVerificationTask
-org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotMetadataVerificationTask$MetadataVerificationJob
-org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotMetadataVerificationTaskArg
-org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotMetadataVerificationTaskResult
org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotOperation
org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotOperationRequest
-org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotPartitionsVerifyTask
-org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotPartitionsVerifyTask$VerifySnapshotPartitionsJob
-org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotPartitionsVerifyTaskArg
-org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotPartitionsVerifyTaskResult
org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotRestoreProcess$SnapshotRestoreOperationResponse
org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotRestoreStatusTask
org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotRestoreStatusTask$1
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/AbstractSnapshotSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/AbstractSnapshotSelfTest.java
index b524fe341bc..4239c53ed78 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/AbstractSnapshotSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/AbstractSnapshotSelfTest.java
@@ -937,14 +937,28 @@ public abstract class AbstractSnapshotSelfTest extends GridCommonAbstractTest {
blockPred = pred;
}
- /** Unblock and send previously saved discovery custom messages */
+ /** Reassigns blocking predicate and releases current blocked queue. */
+ public synchronized void blockNextAndRelease(IgnitePredicate<DiscoveryCustomMessage> pred) {
+ blockPred = pred;
+
+ releaseBlocked();
+ }
+
+ /** Unblock and send previously saved discovery custom messages. */
public synchronized void unblock() {
blockPred = null;
+ releaseBlocked();
+ }
+
+ /** Releases the blocked messages. */
+ private void releaseBlocked() {
+ List<DiscoverySpiCustomMessage> blocked = new CopyOnWriteArrayList<>(this.blocked);
+
+ this.blocked.clear();
+
for (DiscoverySpiCustomMessage msg : blocked)
sendCustomEvent(msg);
-
- blocked.clear();
}
/**
@@ -954,6 +968,14 @@ public abstract class AbstractSnapshotSelfTest extends GridCommonAbstractTest {
public void waitBlocked(long timeout) throws IgniteInterruptedCheckedException {
GridTestUtils.waitForCondition(() -> !blocked.isEmpty(), timeout);
}
+
+ /**
+ * @param timeout Timeout to wait blocking messages.
+ * @throws IgniteInterruptedCheckedException If interrupted.
+ */
+ public void waitBlockedSize(int size, long timeout) throws IgniteInterruptedCheckedException {
+ GridTestUtils.waitForCondition(() -> blocked.size() == size, timeout);
+ }
}
/** */
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/EncryptedSnapshotTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/EncryptedSnapshotTest.java
index 13b97637eb2..67b6e82c469 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/EncryptedSnapshotTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/EncryptedSnapshotTest.java
@@ -43,6 +43,7 @@ import org.junit.runners.Parameterized;
import static org.apache.ignite.cluster.ClusterState.ACTIVE;
import static org.apache.ignite.internal.util.distributed.DistributedProcess.DistributedProcessType.CACHE_GROUP_KEY_CHANGE_PREPARE;
import static org.apache.ignite.internal.util.distributed.DistributedProcess.DistributedProcessType.MASTER_KEY_CHANGE_PREPARE;
+import static org.apache.ignite.internal.util.distributed.DistributedProcess.DistributedProcessType.RESTORE_CACHE_GROUP_SNAPSHOT_PREPARE;
import static org.apache.ignite.testframework.GridTestUtils.assertThrowsAnyCause;
/**
@@ -389,7 +390,8 @@ public class EncryptedSnapshotTest extends AbstractSnapshotSelfTest {
ensureCacheAbsent(dfltCacheCfg);
- spi0.block((msg) -> msg instanceof FullMessage && ((FullMessage<?>)msg).error().isEmpty());
+ spi0.block((msg) -> msg instanceof FullMessage && ((FullMessage<?>)msg).error().isEmpty()
+ && ((FullMessage<?>)msg).type() == RESTORE_CACHE_GROUP_SNAPSHOT_PREPARE.ordinal());
fut = grid(1).snapshot().restoreSnapshot(SNAPSHOT_NAME, Collections.singletonList(dfltCacheCfg.getName()));
}
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
index 2a5a3d023b7..a5140c89f73 100644
--- 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
@@ -36,20 +36,28 @@ import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
import javax.cache.expiry.CreatedExpiryPolicy;
import javax.cache.expiry.Duration;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteDataStreamer;
import org.apache.ignite.IgniteException;
import org.apache.ignite.binary.BinaryObject;
import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
+import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.compute.ComputeJobResult;
import org.apache.ignite.configuration.CacheConfiguration;
-import org.apache.ignite.internal.GridJobExecuteRequest;
+import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.GridTopic;
import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.NodeStoppingException;
import org.apache.ignite.internal.binary.BinaryContext;
import org.apache.ignite.internal.binary.BinaryUtils;
+import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.management.cache.CacheFilterEnum;
import org.apache.ignite.internal.management.cache.CacheIdleVerifyCommandArg;
import org.apache.ignite.internal.management.cache.IdleVerifyResult;
@@ -75,13 +83,18 @@ import org.apache.ignite.internal.processors.cache.verify.PartitionHashRecord;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.processors.compress.CompressionProcessor;
import org.apache.ignite.internal.util.GridUnsafe;
+import org.apache.ignite.internal.util.distributed.DistributedProcess;
+import org.apache.ignite.internal.util.distributed.FullMessage;
+import org.apache.ignite.internal.util.distributed.SingleNodeMessage;
+import org.apache.ignite.internal.util.future.IgniteFutureImpl;
import org.apache.ignite.internal.util.lang.GridIterator;
import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.G;
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.apache.ignite.lang.IgniteFuture;
import org.jetbrains.annotations.Nullable;
-import org.junit.Assume;
import org.junit.Before;
import org.junit.Test;
@@ -90,10 +103,14 @@ import static org.apache.ignite.internal.processors.affinity.AffinityTopologyVer
import static org.apache.ignite.internal.processors.cache.GridCacheUtils.TTL_ETERNAL;
import static org.apache.ignite.internal.processors.cache.persistence.partstate.GroupPartitionId.getTypeByPartId;
import static org.apache.ignite.internal.processors.dr.GridDrType.DR_NONE;
+import static org.apache.ignite.internal.util.distributed.DistributedProcess.DistributedProcessType.CHECK_SNAPSHOT_METAS;
+import static org.apache.ignite.internal.util.distributed.DistributedProcess.DistributedProcessType.CHECK_SNAPSHOT_PARTS;
+import static org.apache.ignite.internal.util.distributed.DistributedProcess.DistributedProcessType.RESTORE_CACHE_GROUP_SNAPSHOT_START;
import static org.apache.ignite.testframework.GridTestUtils.assertContains;
import static org.apache.ignite.testframework.GridTestUtils.assertNotContains;
import static org.apache.ignite.testframework.GridTestUtils.assertThrowsAnyCause;
import static org.apache.ignite.testframework.GridTestUtils.waitForCondition;
+import static org.junit.Assume.assumeFalse;
/**
* Cluster-wide snapshot check procedure tests.
@@ -222,7 +239,7 @@ public class IgniteClusterSnapshotCheckTest extends AbstractSnapshotSelfTest {
/** @throws Exception If fails. */
@Test
public void testClusterSnapshotCheckPartitionCounters() throws Exception {
- Assume.assumeFalse("One copy of partiton created in only primary mode", onlyPrimary);
+ assumeFalse("One copy of partiton created in only primary mode", onlyPrimary);
IgniteEx ignite = startGridsWithCache(3, dfltCacheCfg.
setAffinity(new RendezvousAffinityFunction(false, 1)),
@@ -313,17 +330,18 @@ public class IgniteClusterSnapshotCheckTest extends AbstractSnapshotSelfTest {
Set<UUID> assigns = Collections.newSetFromMap(new ConcurrentHashMap<>());
for (int i = 4; i < 7; i++) {
- startGrid(optimize(getConfiguration(getTestIgniteInstanceName(i)).setCacheConfiguration()));
+ IgniteEx grid = startGrid(optimize(getConfiguration(getTestIgniteInstanceName(i)).setCacheConfiguration()));
- UUID locNodeId = grid(i).localNode().id();
+ if (!U.isLocalNodeCoordinator(grid.context().discovery()))
+ continue;
- grid(i).context().io().addMessageListener(GridTopic.TOPIC_JOB, new GridMessageListener() {
+ grid.context().io().addMessageListener(GridTopic.TOPIC_DISTRIBUTED_PROCESS, new GridMessageListener() {
@Override public void onMessage(UUID nodeId, Object msg, byte plc) {
- if (msg instanceof GridJobExecuteRequest) {
- GridJobExecuteRequest msg0 = (GridJobExecuteRequest)msg;
+ if (msg instanceof SingleNodeMessage) {
+ SingleNodeMessage<?> msg0 = (SingleNodeMessage<?>)msg;
- if (msg0.getTaskName().contains(SnapshotPartitionsVerifyTask.class.getName()))
- assigns.add(locNodeId);
+ if (msg0.type() == CHECK_SNAPSHOT_PARTS.ordinal())
+ assigns.add(nodeId);
}
}
});
@@ -339,8 +357,8 @@ public class IgniteClusterSnapshotCheckTest extends AbstractSnapshotSelfTest {
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 + ", count: "
- + assigns.size(), waitForCondition(() -> assigns.size() == 2, 5_000L));
+ assertTrue("Number of distributed process single messages must be equal to the cluster size: "
+ + assigns + ", count: " + assigns.size(), waitForCondition(() -> assigns.size() == 2, 5_000L));
assertTrue(F.isEmpty(res.exceptions()));
assertPartitionsSame(res);
@@ -376,7 +394,7 @@ public class IgniteClusterSnapshotCheckTest extends AbstractSnapshotSelfTest {
/** @throws Exception If fails. */
@Test
public void testClusterSnapshotCheckFailsOnPartitionDataDiffers() throws Exception {
- Assume.assumeFalse("One copy of partiton created in only primary mode", onlyPrimary);
+ assumeFalse("One copy of partiton created in only primary mode", onlyPrimary);
CacheConfiguration<Integer, Value> ccfg = txCacheConfig(new CacheConfiguration<Integer, Value>(DEFAULT_CACHE_NAME))
.setAffinity(new RendezvousAffinityFunction(false, 1));
@@ -493,17 +511,7 @@ public class IgniteClusterSnapshotCheckTest extends AbstractSnapshotSelfTest {
IdleVerifyResult idleVerifyRes = ignite.compute().execute(new TestVisorBackupPartitionsTask(), arg);
- IdleVerifyResult snpVerifyRes = ignite.compute().execute(
- new TestSnapshotPartitionsVerifyTask(),
- new SnapshotPartitionsVerifyTaskArg(
- new HashSet<>(),
- Collections.singletonMap(ignite.cluster().localNode(),
- Collections.singletonList(snp(ignite).readSnapshotMetadata(snapshotFileTree(ignite, SNAPSHOT_NAME).meta()))),
- null,
- 0,
- true
- )
- ).idleVerifyResult();
+ IdleVerifyResult snpVerifyRes = snp(ignite).checkSnapshot(SNAPSHOT_NAME, null).get().idleVerifyResult();
Map<PartitionKey, List<PartitionHashRecord>> idleVerifyHashes = jobResults.get(TestVisorBackupPartitionsTask.class);
Map<PartitionKey, List<PartitionHashRecord>> snpCheckHashes = jobResults.get(TestVisorBackupPartitionsTask.class);
@@ -544,6 +552,48 @@ public class IgniteClusterSnapshotCheckTest extends AbstractSnapshotSelfTest {
assertNotContains(log, b.toString(), "Failed to read page (CRC validation failed)");
}
+ /** */
+ @Test
+ public void testCheckFromLesserTopology() throws Exception {
+ // {@link #corruptPartitionFile} incorrectly affects an encrypted partition.
+ assumeFalse(encryption);
+
+ int srvCnt = 3;
+ IdleVerifyResult chkRes;
+
+ IgniteEx client = startGridsWithSnapshot(srvCnt, CACHE_KEYS_RANGE, true, true);
+
+ for (int i = 1; i <= srvCnt; ++i) {
+ chkRes = snp(client).checkSnapshot(SNAPSHOT_NAME, null).get(getTestTimeout()).idleVerifyResult();
+
+ assertTrue(chkRes.exceptions().isEmpty());
+ assertFalse(chkRes.hasConflicts());
+
+ if (i == srvCnt)
+ break;
+
+ stopGrid(i);
+ }
+
+ for (int i = 1; i < srvCnt; ++i)
+ startGrid(i);
+
+ // Now ensure that a bad partition is detected.
+ corruptPartitionFile(grid(1), SNAPSHOT_NAME, dfltCacheCfg, 3);
+
+ for (int i = 1; i <= srvCnt; ++i) {
+ chkRes = snp(client).checkSnapshot(SNAPSHOT_NAME, null).get(getTestTimeout()).idleVerifyResult();
+
+ assertFalse(chkRes.exceptions().isEmpty());
+ assertTrue(X.hasCause(F.first(chkRes.exceptions().values()), IgniteDataIntegrityViolationException.class));
+
+ if (i == srvCnt)
+ break;
+
+ stopGrid(i);
+ }
+ }
+
/** @throws Exception If fails. */
@Test
public void testClusterSnapshotCheckWithTwoCachesCheckTwoCaches() throws Exception {
@@ -584,6 +634,669 @@ public class IgniteClusterSnapshotCheckTest extends AbstractSnapshotSelfTest {
assertTrue("Threads created: " + createdThreads, createdThreads < iterations);
}
+ /** Tests that concurrent snapshot full checks are declined for the same snapshot. */
+ @Test
+ public void testConcurrentTheSameSnpFullChecksDeclined() throws Exception {
+ // 0 - coordinator; 0,1,2 - baselines; 3 - non-baseline; 4,5 - clients.
+ prepareGridsAndSnapshot(4, 3, 2, false);
+
+ for (int i = 0; i < G.allGrids().size(); ++i) {
+ for (int j = 1; j < G.allGrids().size() - 1; ++j) {
+ int i0 = i;
+ int j0 = j;
+
+ doTestConcurrentSnpCheckOperations(
+ () -> new IgniteFutureImpl<>(snp(grid(i0)).checkSnapshot(SNAPSHOT_NAME, null)),
+ () -> new IgniteFutureImpl<>(snp(grid(j0)).checkSnapshot(SNAPSHOT_NAME, null)),
+ CHECK_SNAPSHOT_METAS,
+ CHECK_SNAPSHOT_PARTS,
+ true,
+ false,
+ null,
+ null
+ );
+ }
+ }
+ }
+
+ /** Tests that concurrent full checks of normal and incremental the same snapshot are declined . */
+ @Test
+ public void testConcurrentTheSameSnpFullAndIncrementalChecksDeclined() throws Exception {
+ assumeFalse(encryption);
+
+ // 0 - coordinator; 0,1 - baselines; 2 - non-baseline; 3,4 - clients.
+ prepareGridsAndSnapshot(3, 2, 2, false);
+
+ snp(grid(3)).createIncrementalSnapshot(SNAPSHOT_NAME).get();
+
+ for (int i = 0; i < G.allGrids().size(); ++i) {
+ for (int j = 1; j < G.allGrids().size() - 1; ++j) {
+ int i0 = i;
+ int j0 = j;
+
+ doTestConcurrentSnpCheckOperations(
+ () -> new IgniteFutureImpl<>(snp(grid(i0)).checkSnapshot(SNAPSHOT_NAME, null)),
+ () -> new IgniteFutureImpl<>(snp(grid(j0)).checkSnapshot(SNAPSHOT_NAME, null, 1)),
+ CHECK_SNAPSHOT_METAS,
+ CHECK_SNAPSHOT_PARTS,
+ true,
+ false,
+ null,
+ null
+ );
+ }
+ }
+ }
+
+ /** Tests that concurrent full checks of the same incremental snapshot are declined. */
+ @Test
+ public void testConcurrentTheSameIncrementalFullChecksDeclined() throws Exception {
+ assumeFalse(encryption);
+
+ // 0 - coordinator; 0,1 - baselines; 2 - non-baseline; 3,4 - clients.
+ prepareGridsAndSnapshot(3, 2, 2, false);
+
+ snp(grid(3)).createIncrementalSnapshot(SNAPSHOT_NAME).get();
+
+ for (int i = 0; i < G.allGrids().size(); ++i) {
+ for (int j = 1; j < G.allGrids().size() - 1; ++j) {
+ int i0 = i;
+ int j0 = j;
+
+ doTestConcurrentSnpCheckOperations(
+ () -> new IgniteFutureImpl<>(snp(grid(i0)).checkSnapshot(SNAPSHOT_NAME, null, 1)),
+ () -> new IgniteFutureImpl<>(snp(grid(j0)).checkSnapshot(SNAPSHOT_NAME, null, 1)),
+ CHECK_SNAPSHOT_METAS,
+ CHECK_SNAPSHOT_PARTS,
+ true,
+ false,
+ null,
+ null
+ );
+ }
+ }
+ }
+
+ /** Tests that concurrent checks of different incremental snapshots are declined. */
+ @Test
+ public void testConcurrentDifferentIncrementalFullChecksDeclined() throws Exception {
+ assumeFalse(encryption);
+
+ // 0 - coordinator; 0,1 - baselines; 2 - non-baseline; 3,4 - clients.
+ prepareGridsAndSnapshot(3, 2, 2, false);
+
+ snp(grid(3)).createIncrementalSnapshot(SNAPSHOT_NAME).get();
+ snp(grid(3)).createIncrementalSnapshot(SNAPSHOT_NAME).get();
+
+ for (int i = 0; i < G.allGrids().size(); ++i) {
+ for (int j = 1; j < G.allGrids().size() - 1; ++j) {
+ int i0 = i;
+ int j0 = j;
+
+ doTestConcurrentSnpCheckOperations(
+ () -> new IgniteFutureImpl<>(snp(grid(i0)).checkSnapshot(SNAPSHOT_NAME, null, 1)),
+ () -> new IgniteFutureImpl<>(snp(grid(j0)).checkSnapshot(SNAPSHOT_NAME, null, 2)),
+ CHECK_SNAPSHOT_METAS,
+ CHECK_SNAPSHOT_PARTS,
+ true,
+ false,
+ null,
+ null
+ );
+ }
+ }
+ }
+
+ /** Tests that concurrent full restoration of a normal snapshot and check of an incremental are declined. */
+ @Test
+ public void testConcurrentTheSameSnpIncrementalCheckAndFullRestoreDeclined() throws Exception {
+ assumeFalse(encryption);
+
+ // 0 - coordinator; 0,1 - baselines; 2 - non-baseline; 3,4 - clients.
+ prepareGridsAndSnapshot(3, 2, 2, false);
+
+ snp(grid(3)).createIncrementalSnapshot(SNAPSHOT_NAME).get();
+
+ grid(0).destroyCache(DEFAULT_CACHE_NAME);
+
+ awaitPartitionMapExchange();
+
+ // Snapshot restoration is disallowed from client nodes.
+ for (int i = 0; i < 3; ++i) {
+ for (int j = 1; j < G.allGrids().size() - 1; ++j) {
+ int i0 = i;
+ int j0 = j;
+
+ doTestConcurrentSnpCheckOperations(
+ () -> snp(grid(i0)).restoreSnapshot(SNAPSHOT_NAME, null, null, 0, true),
+ () -> new IgniteFutureImpl<>(snp(grid(j0)).checkSnapshot(SNAPSHOT_NAME, null, 1)),
+ CHECK_SNAPSHOT_METAS,
+ CHECK_SNAPSHOT_PARTS,
+ true,
+ false,
+ null,
+ () -> grid(0).destroyCache(DEFAULT_CACHE_NAME)
+ );
+ }
+ }
+ }
+
+ /** Tests that concurrent snapshot full checks are allowed for different snapshots. */
+ @Test
+ public void testConcurrentDifferentSnpFullChecksAllowed() throws Exception {
+ // 0 - coordinator; 0,1 - baselines; 2 - non-baseline; 3,4 - clients.
+ prepareGridsAndSnapshot(3, 2, 2, false);
+
+ snp(grid(3)).createSnapshot(SNAPSHOT_NAME + '2').get();
+
+ for (int i = 0; i < G.allGrids().size(); ++i) {
+ for (int j = 1; j < G.allGrids().size() - 1; ++j) {
+ int i0 = i;
+ int j0 = j;
+
+ doTestConcurrentSnpCheckOperations(
+ () -> new IgniteFutureImpl<>(snp(grid(i0)).checkSnapshot(SNAPSHOT_NAME, null)),
+ () -> new IgniteFutureImpl<>(snp(grid(j0)).checkSnapshot(SNAPSHOT_NAME + '2', null)),
+ CHECK_SNAPSHOT_METAS,
+ CHECK_SNAPSHOT_PARTS,
+ false,
+ true,
+ null,
+ null
+ );
+ }
+ }
+ }
+
+ /** Tests that concurrent snapshot full check and restoration (without full checking) are allowed for different snapshots. */
+ @Test
+ public void testConcurrentDifferentSnpFullCheckAndRestorationAllowed() throws Exception {
+ prepareGridsAndSnapshot(3, 2, 2, false);
+
+ snp(grid(3)).createSnapshot(SNAPSHOT_NAME + '2').get();
+
+ grid(0).destroyCache(DEFAULT_CACHE_NAME);
+
+ awaitPartitionMapExchange();
+
+ for (int i = 0; i < G.allGrids().size(); ++i) {
+ // Snapshot restoration is disallowed from client nodes.
+ for (int j = 1; j < 3; ++j) {
+ int i0 = i;
+ int j0 = j;
+
+ doTestConcurrentSnpCheckOperations(
+ () -> new IgniteFutureImpl<>(snp(grid(i0)).checkSnapshot(SNAPSHOT_NAME, null)),
+ () -> snp(grid(j0)).restoreSnapshot(SNAPSHOT_NAME + '2', null),
+ CHECK_SNAPSHOT_PARTS,
+ RESTORE_CACHE_GROUP_SNAPSHOT_START,
+ false,
+ false,
+ null,
+ () -> grid(0).destroyCache(DEFAULT_CACHE_NAME)
+ );
+ }
+ }
+ }
+
+ /** Tests concurrent snapshot full check and full restoration (with checking) are allowed for different snapshots. */
+ @Test
+ public void testConcurrentDifferentSnpCheckAndFullRestorationAllowed() throws Exception {
+ prepareGridsAndSnapshot(3, 2, 2, false);
+
+ snp(grid(0)).createSnapshot(SNAPSHOT_NAME + '2').get();
+
+ grid(0).destroyCache(DEFAULT_CACHE_NAME);
+
+ awaitPartitionMapExchange();
+
+ doTestConcurrentSnpCheckOperations(
+ () -> new IgniteFutureImpl<>(snp(grid(0)).checkSnapshot(SNAPSHOT_NAME, null)),
+ () -> snp(grid(1)).restoreSnapshot(SNAPSHOT_NAME + '2', null, null, 0, true),
+ CHECK_SNAPSHOT_METAS,
+ CHECK_SNAPSHOT_PARTS,
+ false,
+ true,
+ null,
+ () -> grid(0).destroyCache(DEFAULT_CACHE_NAME)
+ );
+ }
+
+ /** Tests that concurrent snapshot full restoration (with checking) is declined when the same snapshot is being fully checked. */
+ @Test
+ public void testConcurrentFullCheckAndFullRestoreDeclined() throws Exception {
+ prepareGridsAndSnapshot(3, 2, 2, false);
+
+ doTestConcurrentSnpCheckOperations(
+ () -> new IgniteFutureImpl<>(snp(grid(0)).checkSnapshot(SNAPSHOT_NAME, null)),
+ () -> snp(grid(1)).restoreSnapshot(SNAPSHOT_NAME, null, null, 0, true),
+ CHECK_SNAPSHOT_METAS,
+ CHECK_SNAPSHOT_PARTS,
+ true,
+ false,
+ null,
+ () -> grid(0).destroyCache(DEFAULT_CACHE_NAME)
+ );
+ }
+
+ /** Tests that concurrent snapshot full check is declined when the same snapshot is being fully restored (checked). */
+ @Test
+ public void testConcurrentTheSameSnpFullCheckWhenFullyRestoringDeclined() throws Exception {
+ prepareGridsAndSnapshot(3, 2, 2, true);
+
+ doTestConcurrentSnpCheckOperations(
+ () -> snp(grid(0)).restoreSnapshot(SNAPSHOT_NAME, null, null, 0, true),
+ () -> new IgniteFutureImpl<>(snp(grid(1)).checkSnapshot(SNAPSHOT_NAME, null)),
+ CHECK_SNAPSHOT_METAS,
+ CHECK_SNAPSHOT_PARTS,
+ true,
+ false,
+ null,
+ () -> grid(0).destroyCache(DEFAULT_CACHE_NAME)
+ );
+ }
+
+ /** Tests that concurrent full check and restoration (without full checking) of the same snapshot are declined. */
+ @Test
+ public void testConcurrentTheSameSnpFullCheckAndRestoreDeclined() throws Exception {
+ prepareGridsAndSnapshot(3, 2, 2, true);
+
+ doTestConcurrentSnpCheckOperations(
+ () -> new IgniteFutureImpl<>(snp(grid(0)).checkSnapshot(SNAPSHOT_NAME, null)),
+ () -> snp(grid(1)).restoreSnapshot(SNAPSHOT_NAME, null),
+ CHECK_SNAPSHOT_METAS,
+ CHECK_SNAPSHOT_METAS,
+ true,
+ false,
+ null,
+ () -> grid(0).destroyCache(DEFAULT_CACHE_NAME)
+ );
+ }
+
+ /** Tests that snapshot full check doesn't affect a snapshot creation. */
+ @Test
+ public void testConcurrentDifferentSnpCheckAndCreateAllowed() throws Exception {
+ prepareGridsAndSnapshot(3, 2, 2, false);
+
+ File dir = snapshotFileTree(grid(0), SNAPSHOT_NAME + "_2").root();
+
+ doTestConcurrentSnpCheckOperations(
+ () -> new IgniteFutureImpl<>(snp(grid(0)).checkSnapshot(SNAPSHOT_NAME, null)),
+ () -> snp(grid(1)).createSnapshot(SNAPSHOT_NAME + "_2", null, false, false),
+ CHECK_SNAPSHOT_METAS,
+ null,
+ false,
+ false,
+ () -> U.delete(dir),
+ () -> U.delete(dir)
+ );
+ }
+
+ /** Tests snapshot checking processes a baseline node leave. */
+ @Test
+ public void testBaselineLeavesDuringSnapshotChecking() throws Exception {
+ prepareGridsAndSnapshot(6, 5, 1, false);
+
+ Set<Integer> stopped = new HashSet<>();
+
+ // Snapshot checking started from the coordinator.
+ doTestNodeStopsDuringSnapshotChecking(0, 4, stopped);
+
+ // Snapshot checking started from non-baseline.
+ doTestNodeStopsDuringSnapshotChecking(5, 3, stopped);
+
+ // Snapshot checking started from a client.
+ doTestNodeStopsDuringSnapshotChecking(6, 2, stopped);
+
+ // The same baseline leaves.
+ doTestNodeStopsDuringSnapshotChecking(1, 1, stopped);
+ }
+
+ /** Tests snapshot checking processes a client node leave. */
+ @Test
+ public void testClientLeavesDuringSnapshotChecking() throws Exception {
+ prepareGridsAndSnapshot(3, 2, 6, false);
+
+ Set<Integer> stopped = new HashSet<>();
+
+ // Snapshot checking started from a baseline.
+ doTestNodeStopsDuringSnapshotChecking(1, 8, stopped);
+
+ // Snapshot checking started from a non-baseline.
+ doTestNodeStopsDuringSnapshotChecking(2, 7, stopped);
+
+ // Snapshot checking started from the coordinator.
+ doTestNodeStopsDuringSnapshotChecking(0, 6, stopped);
+
+ // Snapshot checking started from other client.
+ doTestNodeStopsDuringSnapshotChecking(4, 5, stopped);
+
+ // Snapshot checking started from the same client.
+ doTestNodeStopsDuringSnapshotChecking(4, 4, stopped);
+ }
+
+ /** Tests snapshot checking processes a non-baseline node leave. */
+ @Test
+ public void testNonBaselineServerLeavesDuringSnapshotChecking() throws Exception {
+ prepareGridsAndSnapshot(7, 2, 1, false);
+
+ Set<Integer> stopped = new HashSet<>();
+
+ // Snapshot checking started from a sever node.
+ doTestNodeStopsDuringSnapshotChecking(1, 6, stopped);
+
+ // Snapshot checking started from a client node.
+ doTestNodeStopsDuringSnapshotChecking(7, 5, stopped);
+
+ // Snapshot checking started from another non-baseline.
+ doTestNodeStopsDuringSnapshotChecking(3, 4, stopped);
+
+ // Snapshot checking started from coordinator.
+ doTestNodeStopsDuringSnapshotChecking(0, 3, stopped);
+
+ // Snapshot checking started from the same non-baseline.
+ doTestNodeStopsDuringSnapshotChecking(2, 2, stopped);
+ }
+
+ /** Tests snapshot checking process continues when a new baseline node leaves. */
+ @Test
+ public void testNewBaselineServerLeavesDuringSnapshotChecking() throws Exception {
+ prepareGridsAndSnapshot(3, 2, 1, false);
+
+ int grids = G.allGrids().size();
+
+ discoSpi(grid(0)).block(msg -> msg instanceof FullMessage && ((FullMessage<?>)msg).type() == CHECK_SNAPSHOT_METAS.ordinal());
+
+ IgniteInternalFuture<?> fut = snp(grid(3)).checkSnapshot(SNAPSHOT_NAME, null, null, false, 0, true);
+
+ discoSpi(grid(0)).waitBlocked(getTestTimeout());
+
+ grid(0).cluster().setBaselineTopology(Stream.of(grid(0).localNode(), grid(1).localNode(), grid(2).localNode())
+ .collect(Collectors.toList()));
+
+ stopGrid(2);
+
+ assertTrue(waitForCondition(() -> {
+ for (int i = 0; i < grids; ++i) {
+ if (i != 2 && grid(i).cluster().nodes().size() != grids - 1)
+ return false;
+ }
+
+ return true;
+ }, getTestTimeout()));
+
+ discoSpi(grid(0)).unblock();
+
+ fut.get(getTestTimeout());
+ }
+
+ /** Tests snapshot checking process stops when the coorditator leaves. */
+ @Test
+ public void testCoordinatorLeavesDuringSnapshotChecking() throws Exception {
+ prepareGridsAndSnapshot(5, 4, 1, false);
+
+ Set<Integer> stopped = new HashSet<>();
+
+ // Coordinator leaves when snapshot started from a server node.
+ doTestNodeStopsDuringSnapshotChecking(4, 0, stopped);
+
+ // Coordinator leaves when snapshot started from a client node.
+ assertTrue(U.isLocalNodeCoordinator(grid(1).context().discovery()));
+
+ doTestNodeStopsDuringSnapshotChecking(5, 1, stopped);
+
+ // Coordinator leaves when snapshot started from it.
+ assertTrue(U.isLocalNodeCoordinator(grid(2).context().discovery()));
+
+ doTestNodeStopsDuringSnapshotChecking(2, 2, stopped);
+ }
+
+ /** */
+ private void prepareGridsAndSnapshot(int servers, int baseLineCnt, int clients, boolean removeTheCache) throws Exception {
+ assert baseLineCnt > 0 && baseLineCnt <= servers;
+
+ IgniteEx ignite = null;
+
+ for (int i = 0; i < servers + clients; ++i) {
+ IgniteConfiguration cfg = getConfiguration(getTestIgniteInstanceName(i));
+
+ cfg.setDiscoverySpi(new BlockingCustomMessageDiscoverySpi());
+
+ if (i >= servers)
+ cfg.setClientMode(true);
+
+ ignite = startGrid(cfg);
+
+ if (i == baseLineCnt - 1) {
+ ignite.cluster().state(ACTIVE);
+
+ ignite.cluster().setBaselineTopology(ignite.cluster().topologyVersion());
+ }
+ }
+
+ try (IgniteDataStreamer<Integer, Integer> ds = grid(0).dataStreamer(DEFAULT_CACHE_NAME)) {
+ for (int i = 0; i < 100; ++i)
+ ds.addData(i, i);
+ }
+
+ ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
+
+ if (removeTheCache)
+ ignite.destroyCache(DEFAULT_CACHE_NAME);
+ }
+
+ /**
+ * Tests concurrent snapshot operations related to the snapshot checking.
+ *
+ * @param originatorOp First snapshot operation on an originator node.
+ * @param trierOp Second concurrent snapshot operation on a trier node.
+ * @param firstDelay First distributed process full message of {@code originatorOp} to delay on the coordinator
+ * to launch {@code trierOp}.
+ * @param secondDelay Second distributed process full message of {@code originatorOp} to delay on the coordinator
+ * to launch {@code trierOp} again.
+ * @param expectFailure If {@code true}, the 'snapshot-check-is-in-progress' error is excepted during excution of
+ * {@code trierOp}. Otherwise, {@code trierOp} must successfully finish.
+ * @param waitForBothFirstDelays If {@code true}, {@code firstDelay} are awaited for both concurrend operations before proceed.
+ * @param step2preparation If not {@code null}, is executed before delaying and waiting for {@code secondDelay}.
+ * @param cleaner If not {@code null}, is executed at the end.
+ */
+ private void doTestConcurrentSnpCheckOperations(
+ Supplier<IgniteFuture<?>> originatorOp,
+ Supplier<IgniteFuture<?>> trierOp,
+ DistributedProcess.DistributedProcessType firstDelay,
+ @Nullable DistributedProcess.DistributedProcessType secondDelay,
+ boolean expectFailure,
+ boolean waitForBothFirstDelays,
+ @Nullable Runnable step2preparation,
+ @Nullable Runnable cleaner
+ ) throws Exception {
+ try {
+ AtomicBoolean firstDelayed = new AtomicBoolean();
+
+ // Block any matching if the operation is the same. Otherwise, block only first.
+ discoSpi(grid(0)).block(
+ msg -> msg instanceof FullMessage
+ && ((FullMessage<?>)msg).type() == firstDelay.ordinal()
+ && (waitForBothFirstDelays || firstDelayed.compareAndSet(false, true)));
+
+ IgniteFuture<?> fut = originatorOp.get();
+
+ discoSpi(grid(0)).waitBlocked(getTestTimeout());
+
+ IgniteFuture<?> fut2 = trierOp.get();
+
+ if (expectFailure) {
+ assertThrowsAnyCause(
+ log,
+ fut2::get,
+ IllegalStateException.class,
+ "Validation of snapshot '" + SNAPSHOT_NAME + "' has already started"
+ );
+
+ if (secondDelay == null) {
+ discoSpi(grid(0)).unblock();
+
+ fut.get(getTestTimeout());
+
+ return;
+ }
+
+ discoSpi(grid(0)).blockNextAndRelease(msg -> msg instanceof FullMessage
+ && ((FullMessage<?>)msg).type() == secondDelay.ordinal());
+
+ discoSpi(grid(0)).waitBlocked(getTestTimeout());
+
+ if (step2preparation != null)
+ step2preparation.run();
+
+ assertThrowsAnyCause(
+ log,
+ fut2::get,
+ IllegalStateException.class,
+ "Validation of snapshot '" + SNAPSHOT_NAME + "' has already started"
+ );
+
+ discoSpi(grid(0)).unblock();
+
+ fut.get(getTestTimeout());
+ }
+ else {
+ if (waitForBothFirstDelays) {
+ discoSpi(grid(0)).waitBlockedSize(2, getTestTimeout());
+
+ if (secondDelay != null) {
+ discoSpi(grid(0)).blockNextAndRelease(msg -> msg instanceof FullMessage
+ && ((FullMessage<?>)msg).type() == secondDelay.ordinal());
+
+ discoSpi(grid(0)).waitBlockedSize(2, getTestTimeout());
+ }
+ }
+ else {
+ if (secondDelay != null) {
+ discoSpi(grid(0)).blockNextAndRelease(msg -> msg instanceof FullMessage
+ && ((FullMessage<?>)msg).type() == secondDelay.ordinal());
+
+ discoSpi(grid(0)).waitBlocked(getTestTimeout());
+ }
+ else
+ fut2.get();
+ }
+
+ discoSpi(grid(0)).unblock();
+
+ fut2.get();
+
+ fut.get();
+ }
+ }
+ finally {
+ discoSpi(grid(0)).unblock();
+
+ if (cleaner != null)
+ cleaner.run();
+
+ awaitPartitionMapExchange();
+ }
+ }
+
+ /** */
+ private void doTestNodeStopsDuringSnapshotChecking(int originatorIdx, int nodeToStopIdx, Set<Integer> stopped) throws Exception {
+ int grids = G.allGrids().size();
+
+ ClusterNode leaving = grid(nodeToStopIdx).cluster().localNode();
+
+ boolean requiredLeft = grid(nodeToStopIdx).cluster().currentBaselineTopology()
+ .stream()
+ .anyMatch(bl -> bl.consistentId().equals(leaving.consistentId()));
+
+ int coordIdx = -1;
+
+ for (int i = 0; i < grids; ++i) {
+ if (stopped.contains(i))
+ continue;
+
+ final IgniteEx g = grid(i);
+
+ if (U.isLocalNodeCoordinator(g.context().discovery())) {
+ coordIdx = i;
+
+ break;
+ }
+ }
+
+ try {
+ discoSpi(grid(coordIdx)).block(msg -> msg instanceof FullMessage
+ && ((FullMessage<?>)msg).type() == CHECK_SNAPSHOT_METAS.ordinal());
+
+ IgniteInternalFuture<?> fut = snp(grid(originatorIdx)).checkSnapshot(SNAPSHOT_NAME, null, null, false, 0, true);
+
+ discoSpi(grid(coordIdx)).waitBlocked(getTestTimeout());
+
+ stopGrid(nodeToStopIdx, true);
+
+ stopped.add(nodeToStopIdx);
+
+ assertTrue(waitForCondition(() -> {
+ for (int i = 0; i < grids; ++i) {
+ if (!stopped.contains(i) && grid(i).cluster().nodes().size() != grids - 1)
+ return false;
+ }
+
+ return true;
+ }, getTestTimeout()));
+
+ if (nodeToStopIdx != coordIdx)
+ discoSpi(grid(coordIdx)).unblock();
+
+ if (originatorIdx == nodeToStopIdx) {
+ assertThrowsAnyCause(
+ null,
+ () -> fut.get(getTestTimeout()),
+ NodeStoppingException.class,
+ "Node is stopping"
+ );
+
+ return;
+ }
+
+ if (requiredLeft) {
+ assertThrowsAnyCause(
+ null,
+ () -> fut.get(getTestTimeout()),
+ ClusterTopologyCheckedException.class,
+ "Snapshot validation stopped. Required node has left the cluster"
+ );
+ }
+ else
+ fut.get(getTestTimeout());
+ }
+ finally {
+ if (nodeToStopIdx != coordIdx)
+ discoSpi(grid(coordIdx)).unblock();
+ }
+
+ waitClusterFinishedSnapshotCheck(stopped);
+
+ if (requiredLeft) {
+ int chkAgainIdx = -1;
+
+ for (int i = 0; i < grids; ++i) {
+ if (stopped.contains(i))
+ continue;
+
+ chkAgainIdx = i;
+
+ break;
+ }
+
+ assert chkAgainIdx >= 0;
+
+ snp(grid(chkAgainIdx)).checkSnapshot(SNAPSHOT_NAME, null, null, false, 0, true).get(getTestTimeout());
+
+ waitClusterFinishedSnapshotCheck(stopped);
+ }
+ }
+
/** */
@Test
public void testClusterSnapshotCheckWithExpiring() throws Exception {
@@ -610,6 +1323,21 @@ public class IgniteClusterSnapshotCheckTest extends AbstractSnapshotSelfTest {
assertFalse(res.idleVerifyResult().hasConflicts());
}
+ /** */
+ private void waitClusterFinishedSnapshotCheck(Set<Integer> stopped) throws Exception {
+ for (int i = 0; i < G.allGrids().size(); ++i) {
+ if (stopped.contains(i))
+ continue;
+
+ final IgniteEx g = grid(i);
+
+ assertTrue(
+ "Node " + i + " hasn't stopped the check",
+ waitForCondition(() -> !snp(g).isSnapshotChecking(SNAPSHOT_NAME), 10_000)
+ );
+ }
+ }
+
/**
* @param cls Class of running task.
* @param results Results of compute.
@@ -702,16 +1430,4 @@ public class IgniteClusterSnapshotCheckTest extends AbstractSnapshotSelfTest {
return res;
}
}
-
- /** Test compute task to collect partition data hashes when the snapshot check procedure ends. */
- private class TestSnapshotPartitionsVerifyTask extends SnapshotPartitionsVerifyTask {
- /** {@inheritDoc} */
- @Override public @Nullable SnapshotPartitionsVerifyTaskResult reduce(List<ComputeJobResult> results) throws IgniteException {
- SnapshotPartitionsVerifyTaskResult res = super.reduce(results);
-
- saveHashes(TestSnapshotPartitionsVerifyTask.class, results);
-
- return res;
- }
- }
}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotRestoreSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotRestoreSelfTest.java
index 4c3fa6fa54c..409c1fb13cb 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotRestoreSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotRestoreSelfTest.java
@@ -284,8 +284,8 @@ public class IgniteClusterSnapshotRestoreSelfTest extends IgniteClusterSnapshotR
}
catch (Exception e) {
assertTrue("Unexpected exception: " + Throwables.getStackTraceAsString(e),
- X.hasCause(e, "The previous snapshot restore operation was not completed.",
- IgniteCheckedException.class, IgniteException.class));
+ X.hasCause(e, "The previous snapshot restore operation was not completed", IgniteException.class)
+ || X.hasCause(e, "has already started", IgniteException.class));
failCnt.incrementAndGet();
}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotMXBeanTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotMXBeanTest.java
index 91b88a45461..d1a804ada61 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotMXBeanTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotMXBeanTest.java
@@ -40,6 +40,7 @@ import org.junit.Test;
import static org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotManager.SNAPSHOT_METRICS;
import static org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotRestoreProcess.SNAPSHOT_RESTORE_METRICS;
import static org.apache.ignite.internal.util.distributed.DistributedProcess.DistributedProcessType.RESTORE_CACHE_GROUP_SNAPSHOT_PREPARE;
+import static org.apache.ignite.internal.util.distributed.DistributedProcess.DistributedProcessType.RESTORE_CACHE_GROUP_SNAPSHOT_START;
import static org.apache.ignite.testframework.GridTestUtils.assertContains;
import static org.apache.ignite.testframework.GridTestUtils.assertThrowsAnyCause;
import static org.apache.ignite.testframework.GridTestUtils.waitForCondition;
@@ -262,7 +263,8 @@ public class IgniteSnapshotMXBeanTest extends AbstractSnapshotSelfTest {
awaitPartitionMapExchange();
- spi.blockMessages((node, msg) -> msg instanceof SingleNodeMessage);
+ spi.blockMessages((node, msg) -> msg instanceof SingleNodeMessage
+ && ((SingleNodeMessage)msg).type() == RESTORE_CACHE_GROUP_SNAPSHOT_PREPARE.ordinal());
fut = srv.snapshot().restoreSnapshot(SNAPSHOT_NAME, F.asList(DEFAULT_CACHE_NAME));
@@ -283,7 +285,8 @@ public class IgniteSnapshotMXBeanTest extends AbstractSnapshotSelfTest {
awaitPartitionMapExchange();
- spi.blockMessages((node, msg) -> msg instanceof SingleNodeMessage);
+ spi.blockMessages((node, msg) -> msg instanceof SingleNodeMessage
+ && ((SingleNodeMessage)msg).type() == RESTORE_CACHE_GROUP_SNAPSHOT_START.ordinal());
fut = srv.snapshot().restoreSnapshot(SNAPSHOT_NAME, F.asList(DEFAULT_CACHE_NAME), 1);