diff --git a/modules/compress/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotCompressionBasicTest.java b/modules/compress/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotCompressionBasicTest.java index 08a4acbb1cdf2..0af75e51d29dd 100644 --- a/modules/compress/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotCompressionBasicTest.java +++ b/modules/compress/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotCompressionBasicTest.java @@ -49,7 +49,7 @@ import org.apache.ignite.configuration.IgniteConfiguration; import org.apache.ignite.internal.GridKernalContextImpl; import org.apache.ignite.internal.IgniteEx; -import org.apache.ignite.internal.management.cache.IdleVerifyResultV2; +import org.apache.ignite.internal.management.cache.IdleVerifyResult; import org.apache.ignite.internal.processors.cache.persistence.file.FileIO; import org.apache.ignite.internal.processors.cache.persistence.file.RandomAccessFileIO; import org.apache.ignite.internal.processors.compress.CompressionProcessor; @@ -336,7 +336,7 @@ protected void createTestSnapshot() throws Exception { for (String snpName : Arrays.asList(SNAPSHOT_WITH_HOLES, SNAPSHOT_WITHOUT_HOLES)) { snp(ignite).createSnapshot(snpName, null, false, onlyPrimary).get(TIMEOUT); - IdleVerifyResultV2 res = ignite.context().cache().context().snapshotMgr().checkSnapshot(snpName, null) + IdleVerifyResult res = ignite.context().cache().context().snapshotMgr().checkSnapshot(snpName, null) .get().idleVerifyResult(); StringBuilder b = new StringBuilder(); diff --git a/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerCheckIncrementalSnapshotTest.java b/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerCheckIncrementalSnapshotTest.java index cbee596138c41..21fe7d4091304 100644 --- a/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerCheckIncrementalSnapshotTest.java +++ b/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerCheckIncrementalSnapshotTest.java @@ -29,7 +29,7 @@ import org.apache.ignite.internal.GridKernalContext; import org.apache.ignite.internal.IgniteEx; import org.apache.ignite.internal.TestRecordingCommunicationSpi; -import org.apache.ignite.internal.management.cache.IdleVerifyResultV2; +import org.apache.ignite.internal.management.cache.IdleVerifyResult; import org.apache.ignite.internal.pagemem.wal.IgniteWriteAheadLogManager; import org.apache.ignite.internal.pagemem.wal.record.DataRecord; import org.apache.ignite.internal.pagemem.wal.record.TxRecord; @@ -378,7 +378,7 @@ public void atomicCachesAreSkippedDuringTheCheck() throws Exception { grid(0).snapshot().restoreSnapshot(atomicSnp, null, 1).get(); - IdleVerifyResultV2 idleVerRes = idleVerify(grid(0), CACHE, atomicCache); + IdleVerifyResult idleVerRes = idleVerify(grid(0), CACHE, atomicCache); idleVerRes.print(System.out::println, true); 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 ae6d68b2bb866..024cb3f154e28 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 @@ -84,7 +84,7 @@ import org.apache.ignite.internal.client.util.GridConcurrentHashSet; import org.apache.ignite.internal.management.cache.FindAndDeleteGarbageInPersistenceTaskResult; import org.apache.ignite.internal.management.cache.IdleVerifyDumpTask; -import org.apache.ignite.internal.management.cache.VerifyBackupPartitionsTaskV2; +import org.apache.ignite.internal.management.cache.VerifyBackupPartitionsTask; import org.apache.ignite.internal.management.tx.TxInfo; import org.apache.ignite.internal.management.tx.TxTaskResult; import org.apache.ignite.internal.managers.communication.GridIoMessage; @@ -743,7 +743,7 @@ public void testIdleVerifyOnInactiveClusterWithPersistence() throws Exception { assertEquals(EXIT_CODE_UNEXPECTED_ERROR, execute("--cache", "idle_verify")); - assertContains(log, testOut.toString(), VerifyBackupPartitionsTaskV2.IDLE_VERIFY_ON_INACTIVE_CLUSTER_ERROR_MESSAGE); + assertContains(log, testOut.toString(), VerifyBackupPartitionsTask.IDLE_VERIFY_ON_INACTIVE_CLUSTER_ERROR_MESSAGE); assertContains(log, testOut.toString(), "Failed to perform operation"); srv.cluster().state(ACTIVE); diff --git a/modules/core/src/main/java/org/apache/ignite/internal/management/cache/CacheFilterEnum.java b/modules/core/src/main/java/org/apache/ignite/internal/management/cache/CacheFilterEnum.java index 08ace4012f79d..db362918c65c7 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/management/cache/CacheFilterEnum.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/management/cache/CacheFilterEnum.java @@ -22,7 +22,7 @@ /** * Represents a type of cache(s) that can be used for comparing update counters and checksums between primary and backup partitions. *
- * @see VerifyBackupPartitionsTaskV2 + * @see VerifyBackupPartitionsTask */ public enum CacheFilterEnum { /** Default - user only, or all caches specified by name. */ diff --git a/modules/core/src/main/java/org/apache/ignite/internal/management/cache/CacheIdleVerifyCommand.java b/modules/core/src/main/java/org/apache/ignite/internal/management/cache/CacheIdleVerifyCommand.java index 867d37b1628fd..4dd5971729723 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/management/cache/CacheIdleVerifyCommand.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/management/cache/CacheIdleVerifyCommand.java @@ -36,8 +36,8 @@ /** Checks consistency of primary and backup partitions assuming no concurrent updates are happening in the cluster. */ @CliSubcommandsWithPrefix public class CacheIdleVerifyCommand - extends CommandRegistryImpl - implements ComputeCommand { + extends CommandRegistryImpl + implements ComputeCommand { /** */ public static final String IDLE_VERIFY_FILE_PREFIX = "idle_verify-"; @@ -72,7 +72,7 @@ public CacheIdleVerifyCommand() { } /** {@inheritDoc} */ - @Override public void printResult(CacheIdleVerifyCommandArg arg, IdleVerifyResultV2 res, Consumer printer) { + @Override public void printResult(CacheIdleVerifyCommandArg arg, IdleVerifyResult res, Consumer printer) { logParsedArgs(arg, printer); StringBuilder sb = new StringBuilder(); diff --git a/modules/core/src/main/java/org/apache/ignite/internal/management/cache/IdleVerifyException.java b/modules/core/src/main/java/org/apache/ignite/internal/management/cache/IdleVerifyException.java index 5f4b430b9a5de..e54fe3ad2a0b9 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/management/cache/IdleVerifyException.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/management/cache/IdleVerifyException.java @@ -23,7 +23,7 @@ import org.apache.ignite.internal.util.typedef.F; /** - * This exception is used to collect exceptions occured in {@link VerifyBackupPartitionsTaskV2} execution. + * This exception is used to collect exceptions occured in {@link VerifyBackupPartitionsTask} execution. */ public class IdleVerifyException extends IgniteException { /** */ diff --git a/modules/core/src/main/java/org/apache/ignite/internal/management/cache/IdleVerifyResultV2.java b/modules/core/src/main/java/org/apache/ignite/internal/management/cache/IdleVerifyResult.java similarity index 75% rename from modules/core/src/main/java/org/apache/ignite/internal/management/cache/IdleVerifyResultV2.java rename to modules/core/src/main/java/org/apache/ignite/internal/management/cache/IdleVerifyResult.java index f6dfb7afb7a83..0451a6ab60d38 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/management/cache/IdleVerifyResultV2.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/management/cache/IdleVerifyResult.java @@ -45,9 +45,9 @@ import static org.apache.ignite.internal.util.IgniteUtils.nl; /** - * Encapsulates result of {@link VerifyBackupPartitionsTaskV2}. + * Encapsulates result of {@link VerifyBackupPartitionsTask}. */ -public class IdleVerifyResultV2 extends VisorDataTransferObject { +public class IdleVerifyResult extends VisorDataTransferObject { /** */ private static final long serialVersionUID = 0L; @@ -82,36 +82,24 @@ public class IdleVerifyResultV2 extends VisorDataTransferObject { /** * Default constructor for Externalizable. */ - public IdleVerifyResultV2() { + public IdleVerifyResult() { + // No-op. } /** - * @param exceptions Occurred exceptions. + * @see Builder */ - public IdleVerifyResultV2(Map exceptions) { + private IdleVerifyResult(Map exceptions) { this.exceptions = exceptions; } /** - * @param txHashConflicts Transaction hashes conflicts. + * @see Builder */ - public IdleVerifyResultV2( + private IdleVerifyResult( Map> clusterHashes, @Nullable List> txHashConflicts, - @Nullable Map> partiallyCommittedTxs - ) { - this(clusterHashes, Collections.emptyMap()); - - this.txHashConflicts = txHashConflicts; - this.partiallyCommittedTxs = partiallyCommittedTxs; - } - - /** - * @param clusterHashes Map of cluster partition hashes. - * @param exceptions Exceptions on each cluster node. - */ - public IdleVerifyResultV2( - Map> clusterHashes, + @Nullable Map> partiallyCommittedTxs, Map exceptions ) { for (Map.Entry> e : clusterHashes.entrySet()) { @@ -151,6 +139,8 @@ public IdleVerifyResultV2( } this.exceptions = exceptions; + this.txHashConflicts = txHashConflicts; + this.partiallyCommittedTxs = partiallyCommittedTxs; } /** {@inheritDoc} */ @@ -390,7 +380,7 @@ private void printConflicts(Consumer printer) { if (o == null || getClass() != o.getClass()) return false; - IdleVerifyResultV2 v2 = (IdleVerifyResultV2)o; + IdleVerifyResult v2 = (IdleVerifyResult)o; return Objects.equals(cntrConflicts, v2.cntrConflicts) && Objects.equals(hashConflicts, v2.hashConflicts) && Objects.equals(movingPartitions, v2.movingPartitions) && Objects.equals(lostPartitions, v2.lostPartitions) && @@ -412,6 +402,135 @@ private void printConflicts(Consumer printer) { /** {@inheritDoc} */ @Override public String toString() { - return S.toString(IdleVerifyResultV2.class, this); + return S.toString(IdleVerifyResult.class, this); + } + + /** @return A fresh result builder. */ + public static Builder builder() { + return new Builder(); + } + + /** Builder of {@link IdleVerifyResult}. Is not thread-safe. */ + public static final class Builder { + /** */ + private @Nullable Map> partHashes; + + /** */ + private @Nullable List> txHashConflicts; + + /** */ + private @Nullable Map> partiallyCommittedTxs; + + /** */ + private Map exceptions; + + /** */ + private Builder() { + // No-op. + } + + /** Build the final result. */ + public IdleVerifyResult build() { + if (partHashes == null) + partHashes = Collections.emptyMap(); + + if (exceptions == null) + exceptions = Collections.emptyMap(); + + return new IdleVerifyResult(partHashes, txHashConflicts, partiallyCommittedTxs, exceptions); + } + + /** Stores an exception if none is assigned for {@code node}. */ + public Builder addException(ClusterNode node, Exception e) { + assert e != null; + + if (exceptions == null) + exceptions = new HashMap<>(); + + exceptions.putIfAbsent(node, e); + + return this; + } + + /** Sets the exceptions. */ + public Builder exceptions(Map exceptions) { + assert this.exceptions == null : "Exceptions are already set."; + assert exceptions != null; + + this.exceptions = exceptions; + + return this; + } + + /** Stores a collection of partition hashes for partition key {@code key}. */ + private Builder addPartitionHashes(PartitionKeyV2 key, Collection newHashes) { + if (partHashes == null) + partHashes = new HashMap<>(); + + partHashes.compute(key, (key0, hashes0) -> { + if (hashes0 == null) + hashes0 = new ArrayList<>(); + + hashes0.addAll(newHashes); + + return hashes0; + }); + + return this; + } + + /** Stores a partition hashes map. */ + public void addPartitionHashes(Map newHashes) { + newHashes.forEach((key, hash) -> addPartitionHashes(key, Collections.singletonList(hash))); + } + + /** Stores a single partition hash for partition key {@code key}. */ + public Builder addPartitionHash(PartitionKeyV2 key, PartitionHashRecordV2 newHash) { + addPartitionHashes(key, Collections.singletonList(newHash)); + + return this; + } + + /** Sets partition hashes. */ + public Builder partitionHashes(Map> partHashes) { + assert F.isEmpty(this.partHashes) : "Partition hashes are already set."; + assert partHashes != null; + + this.partHashes = partHashes; + + return this; + } + + /** Adds transaction conflicts. */ + public Builder addTxConflicts(List newTxConflicts) { + if (txHashConflicts == null) + txHashConflicts = new ArrayList<>(); + + txHashConflicts.add(newTxConflicts); + + return this; + } + + /** Adds partially commited transactions. */ + public Builder addPartiallyCommited(ClusterNode node, Collection newVerisons) { + if (partiallyCommittedTxs == null) + partiallyCommittedTxs = new HashMap<>(); + + partiallyCommittedTxs.compute(node, (node0, versions0) -> { + if (versions0 == null) + versions0 = new ArrayList<>(); + + versions0.addAll(newVerisons); + + return versions0; + }); + + return this; + } + + /** @return {@code True} if any error is stopre. {@code False} otherwise. */ + public boolean hasErrors() { + return !F.isEmpty(exceptions); + } } } diff --git a/modules/core/src/main/java/org/apache/ignite/internal/management/cache/IdleVerifyTaskV2.java b/modules/core/src/main/java/org/apache/ignite/internal/management/cache/IdleVerifyTaskV2.java index 1e03a864e0ecf..250757df90ea8 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/management/cache/IdleVerifyTaskV2.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/management/cache/IdleVerifyTaskV2.java @@ -26,15 +26,15 @@ * Task to verify checksums of backup partitions. */ @GridInternal -public class IdleVerifyTaskV2 extends VisorOneNodeTask { +public class IdleVerifyTaskV2 extends VisorOneNodeTask { /** */ private static final long serialVersionUID = 0L; /** {@inheritDoc} */ - @Override protected VisorJob job(CacheIdleVerifyCommandArg arg) { + @Override protected VisorJob job(CacheIdleVerifyCommandArg arg) { if (!ignite.context().state().publicApiActiveState(true)) - throw new IgniteException(VerifyBackupPartitionsTaskV2.IDLE_VERIFY_ON_INACTIVE_CLUSTER_ERROR_MESSAGE); + throw new IgniteException(VerifyBackupPartitionsTask.IDLE_VERIFY_ON_INACTIVE_CLUSTER_ERROR_MESSAGE); - return new IdleVerifyJob<>(arg, debug, VerifyBackupPartitionsTaskV2.class); + return new IdleVerifyJob<>(arg, debug, VerifyBackupPartitionsTask.class); } } diff --git a/modules/core/src/main/java/org/apache/ignite/internal/management/cache/NoMatchingCachesException.java b/modules/core/src/main/java/org/apache/ignite/internal/management/cache/NoMatchingCachesException.java index 323c75992f441..4aa9bf269d737 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/management/cache/NoMatchingCachesException.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/management/cache/NoMatchingCachesException.java @@ -19,7 +19,7 @@ import org.apache.ignite.IgniteException; /** - * Runtime exception that can be thrown in {@link VerifyBackupPartitionsTaskV2} when no caches matching given + * Runtime exception that can be thrown in {@link VerifyBackupPartitionsTask} when no caches matching given * filter options can be found. */ public class NoMatchingCachesException extends IgniteException { diff --git a/modules/core/src/main/java/org/apache/ignite/internal/management/cache/VerifyBackupPartitionsDumpTask.java b/modules/core/src/main/java/org/apache/ignite/internal/management/cache/VerifyBackupPartitionsDumpTask.java index 7ac73ad9e88e8..8e4711116d2d0 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/management/cache/VerifyBackupPartitionsDumpTask.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/management/cache/VerifyBackupPartitionsDumpTask.java @@ -64,7 +64,7 @@ public class VerifyBackupPartitionsDumpTask extends ComputeTaskAdapter> partitions, - IdleVerifyResultV2 conflictRes, + IdleVerifyResult conflictRes, int skippedRecords ) throws IgniteException { String wd = ignite.configuration().getWorkDirectory(); @@ -204,7 +204,7 @@ private String writeHashes( /** */ private void writeResult( Map> partitions, - IdleVerifyResultV2 conflictRes, + IdleVerifyResult conflictRes, int skippedRecords, PrintWriter writer ) { diff --git a/modules/core/src/main/java/org/apache/ignite/internal/management/cache/VerifyBackupPartitionsTaskV2.java b/modules/core/src/main/java/org/apache/ignite/internal/management/cache/VerifyBackupPartitionsTask.java similarity index 96% rename from modules/core/src/main/java/org/apache/ignite/internal/management/cache/VerifyBackupPartitionsTaskV2.java rename to modules/core/src/main/java/org/apache/ignite/internal/management/cache/VerifyBackupPartitionsTask.java index 048b39020b7a5..8777823695081 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/management/cache/VerifyBackupPartitionsTaskV2.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/management/cache/VerifyBackupPartitionsTask.java @@ -52,7 +52,6 @@ import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtLocalPartition; import org.apache.ignite.internal.processors.cache.persistence.file.FilePageStore; import org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager; -import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotChecker; import org.apache.ignite.internal.processors.cache.verify.GridNotIdleException; import org.apache.ignite.internal.processors.cache.verify.PartitionHashRecordV2; import org.apache.ignite.internal.processors.task.GridInternal; @@ -76,13 +75,13 @@ *
* Argument: Set of cache names, 'null' will trigger verification for all caches. *
- * Result: {@link IdleVerifyResultV2} with conflict partitions. + * Result: {@link IdleVerifyResult} with conflict partitions. *
* Works properly only on idle cluster - there may be false positive conflict reports if data in cluster is being * concurrently updated. */ @GridInternal -public class VerifyBackupPartitionsTaskV2 extends ComputeTaskAdapter { +public class VerifyBackupPartitionsTask extends ComputeTaskAdapter { /** First version of Ignite that is capable of executing Idle Verify V2. */ public static final IgniteProductVersion V2_SINCE_VER = IgniteProductVersion.fromString("2.5.3"); @@ -111,7 +110,7 @@ public class VerifyBackupPartitionsTaskV2 extends ComputeTaskAdapter results) throws IgniteException { + @Nullable @Override public IdleVerifyResult reduce(List results) throws IgniteException { return reduce0(results); } @@ -144,21 +143,20 @@ public class VerifyBackupPartitionsTaskV2 extends ComputeTaskAdapter results) { - Map ex = new HashMap<>(); - Map> hashes = new HashMap<>(); + public static IdleVerifyResult reduce0(List results) { + IdleVerifyResult.Builder bldr = IdleVerifyResult.builder(); for (ComputeJobResult res : results) { if (res.getException() != null) { - ex.put(res.getNode(), res.getException()); + bldr.addException(res.getNode(), res.getException()); continue; } - hashes.put(res.getNode(), res.getData()); + bldr.addPartitionHashes(res.getData()); } - return SnapshotChecker.reduceHashesResults(hashes, ex); + return bldr.build(); } /** 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 f83ce0e595021..0e5225029edbd 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 @@ -99,7 +99,7 @@ import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException; import org.apache.ignite.internal.cluster.DistributedConfigurationUtils; import org.apache.ignite.internal.events.DiscoveryCustomEvent; -import org.apache.ignite.internal.management.cache.IdleVerifyResultV2; +import org.apache.ignite.internal.management.cache.IdleVerifyResult; import org.apache.ignite.internal.managers.communication.GridIoManager; import org.apache.ignite.internal.managers.communication.GridMessageListener; import org.apache.ignite.internal.managers.communication.TransmissionCancelledException; @@ -1840,7 +1840,7 @@ public IgniteFuture cancelLocalRestoreTask(String name) { * @param name Snapshot name. * @param snpPath Snapshot directory path. * @return Future with the result of execution snapshot partitions verify task, which besides calculating partition - * hashes of {@link IdleVerifyResultV2} also contains the snapshot metadata distribution across the cluster. + * hashes of {@link IdleVerifyResult} also contains the snapshot metadata distribution across the cluster. */ public IgniteInternalFuture checkSnapshot(String name, @Nullable String snpPath) { return checkSnapshot(name, snpPath, -1); @@ -1853,7 +1853,7 @@ public IgniteInternalFuture checkSnapshot(String * @param snpPath Snapshot directory path. * @param incIdx Incremental snapshot index. * @return Future with the result of execution snapshot partitions verify task, which besides calculating partition - * hashes of {@link IdleVerifyResultV2} also contains the snapshot metadata distribution across the cluster. + * hashes of {@link IdleVerifyResult} also contains the snapshot metadata distribution across the cluster. */ public IgniteInternalFuture checkSnapshot(String name, @Nullable String snpPath, int incIdx) { A.notNullOrEmpty(name, "Snapshot name cannot be null or empty."); @@ -1884,7 +1884,7 @@ public IgniteInternalFuture checkSnapshot(String * @param incIdx Incremental snapshot index. * @param check If {@code true} check snapshot integrity. * @return Future with the result of execution snapshot partitions verify task, which besides calculating partition - * hashes of {@link IdleVerifyResultV2} also contains the snapshot metadata distribution across the cluster. + * hashes of {@link IdleVerifyResult} also contains the snapshot metadata distribution across the cluster. */ public IgniteInternalFuture checkSnapshot( String name, 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 index 130d21c79f21f..ca58e1126e350 100644 --- 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 @@ -18,16 +18,20 @@ 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.function.Function; +import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; import org.apache.ignite.IgniteException; import org.apache.ignite.IgniteLogger; @@ -37,7 +41,7 @@ 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.IdleVerifyResultV2; +import org.apache.ignite.internal.management.cache.IdleVerifyResult; import org.apache.ignite.internal.management.cache.PartitionKeyV2; import org.apache.ignite.internal.processors.cache.verify.PartitionHashRecordV2; import org.apache.ignite.internal.util.distributed.DistributedProcess; @@ -146,50 +150,136 @@ private IgniteInternalFuture reduceValidatePartsAndFinish( if (clusterOpFut == null) return new GridFinishedFuture<>(); - assert results.values().stream().noneMatch(res -> res != null && res.metas != null); + if (ctx.req.incrementalIndex() > 0) + reduceIncrementalResults(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<>(); + } + + /** */ + private void reduceIncrementalResults( + Set requiredNodes, + Map> clusterMetas, + Map results, + Map errors, + GridFutureAdapter fut + ) { SnapshotChecker checker = kctx.cache().context().snapshotMgr().checker(); - if (ctx.req.incrementalIndex() > 0) { - IdleVerifyResultV2 chkRes = checker.reduceIncrementalResults( - mapResults(results, ctx.req.nodes(), SnapshotCheckResponse::incrementalResult), - mapErrors(errors) - ); + Map> reduced = new HashMap<>(); - clusterOpFut.onDone(new SnapshotPartitionsVerifyResult(ctx.clusterMetas, chkRes)); + for (Map.Entry resEntry : results.entrySet()) { + UUID nodeId = resEntry.getKey(); + + SnapshotCheckResponse incResp = resEntry.getValue(); + + if (incResp == null || !requiredNodes.contains(nodeId)) + continue; + + ClusterNode node = kctx.cluster().get().node(nodeId); + + Map incRes = incResp.result(); + + incRes.forEach((consId, res) -> reduced.computeIfAbsent(node, nid -> new ArrayList<>()).add(res)); + + if (F.isEmpty(incResp.exceptions())) + continue; + + errors.putIfAbsent(nodeId, asException(F.firstValue(incResp.exceptions()))); } - else if (ctx.req.allRestoreHandlers()) { - try { - if (!errors.isEmpty()) - throw F.firstValue(errors); - Map>> cstRes = mapResults(results, ctx.req.nodes(), - SnapshotCheckResponse::customHandlersResults); + IdleVerifyResult chkRes = checker.reduceIncrementalResults(reduced, mapErrors(errors)); + + fut.onDone(new SnapshotPartitionsVerifyResult(clusterMetas, chkRes)); + } + + /** */ + private void reduceCustomHandlersResults( + SnapshotCheckContext ctx, + Map results, + Map errors, + GridFutureAdapter fut + ) { + try { + if (!errors.isEmpty()) + throw F.firstValue(errors); - checker.checkCustomHandlersResults(ctx.req.snapshotName(), cstRes); + SnapshotChecker snpChecker = kctx.cache().context().snapshotMgr().checker(); - clusterOpFut.onDone(new SnapshotPartitionsVerifyResult(ctx.clusterMetas, null)); - } - catch (Throwable err) { - clusterOpFut.onDone(err); + // Check responses: checking node -> snapshot part's consistent id -> handler name -> handler result. + Map>>> reduced = new HashMap<>(); + + for (Map.Entry 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>> cstHndRes = nodeResp.result(); + + cstHndRes.forEach((consId, respPerConsIdMap) -> { + // Reduced map of the handlers results per snapshot part's consistent id for certain node. + Map>> 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 SnapshotPartitionsVerifyResult(ctx.clusterMetas, null)); + } + catch (Throwable err) { + fut.onDone(err); } - else { - Map errors0 = mapErrors(errors); + } + + /** */ + private void reducePartitionsHashesResults( + Map> clusterMetas, + Map results, + Map errors, + GridFutureAdapter fut + ) { + IdleVerifyResult.Builder bldr = IdleVerifyResult.builder(); + + Map errors0 = mapErrors(errors); - if (!results.isEmpty()) { - Map> results0 = mapResults(results, ctx.req.nodes(), - SnapshotCheckResponse::partsHashes); + if (!results.isEmpty()) { + bldr.exceptions(mapErrors(errors)); - IdleVerifyResultV2 chkRes = SnapshotChecker.reduceHashesResults(results0, errors0); + for (Map.Entry respEntry : results.entrySet()) { + SnapshotCheckResponse resp = respEntry.getValue(); - clusterOpFut.onDone(new SnapshotPartitionsVerifyResult(ctx.clusterMetas, chkRes)); + 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> partsHashesRes = resp.result(); + + partsHashesRes.forEach((consId, partsPerConsId) -> bldr.addPartitionHashes(partsPerConsId)); } - else - clusterOpFut.onDone(new IgniteSnapshotVerifyException(errors0)); - } - return new GridFinishedFuture<>(); + fut.onDone(new SnapshotPartitionsVerifyResult(clusterMetas, bldr.build())); + } + else + fut.onDone(new IgniteSnapshotVerifyException(errors0)); } /** Phase 2 beginning. */ @@ -201,60 +291,138 @@ private IgniteInternalFuture validateParts(SnapshotCheckP assert ctx != null; - if (ctx.locMeta == null) + if (F.isEmpty(ctx.metas)) return new GridFinishedFuture<>(); - IgniteSnapshotManager snpMgr = kctx.cache().context().snapshotMgr(); - GridFutureAdapter phaseFut = ctx.phaseFuture(); // Might be already finished by asynchronous leave of a required node. if (!phaseFut.isDone()) { - CompletableFuture workingFut; + CompletableFuture workingFut; if (req.incrementalIndex() > 0) { assert !req.allRestoreHandlers() : "Snapshot handlers aren't supported for incremental snapshot."; - workingFut = snpMgr.checker().checkIncrementalSnapshot(req.snapshotName(), req.snapshotPath(), req.incrementalIndex()); - } - else { - workingFut = req.allRestoreHandlers() - ? snpMgr.checker().invokeCustomHandlers(ctx.locMeta, req.snapshotPath(), req.groups(), true) - : snpMgr.checker().checkPartitions(ctx.locMeta, snpMgr.snapshotLocalDir(req.snapshotName(), req.snapshotPath()), - req.groups(), false, req.fullCheck(), false); + 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 { - if (req.incrementalIndex() > 0) - phaseFut.onDone(new SnapshotCheckResponse((IncrementalSnapshotCheckResult)res)); - else - phaseFut.onDone(new SnapshotCheckResponse((Map)res)); - } + else + phaseFut.onDone(res); }); } return phaseFut; } + /** @return A composed future of increment checks for each consistent id regarding {@link SnapshotCheckContext#metas}. */ + private CompletableFuture incrementalFuture(SnapshotCheckContext ctx) { + SnapshotChecker snpChecker = kctx.cache().context().snapshotMgr().checker(); + // Per metas result: consistent id -> check result. + Map perMetaResults = new ConcurrentHashMap<>(ctx.metas.size(), 1.0f); + // Per consistent id. + Map exceptions = new ConcurrentHashMap<>(ctx.metas.size(), 1.0f); + AtomicInteger metasProcessed = new AtomicInteger(ctx.metas.size()); + CompletableFuture composedFut = new CompletableFuture<>(); + + for (SnapshotMetadata meta : ctx.metas) { + CompletableFuture workingFut = snpChecker.checkIncrementalSnapshot(ctx.req.snapshotName(), + ctx.req.snapshotPath(), ctx.req.incrementalIndex()); + + workingFut.whenComplete((res, err) -> { + if (err != null) + exceptions.put(meta.consistentId(), err); + else + perMetaResults.put(meta.consistentId(), res); + + if (metasProcessed.decrementAndGet() == 0) + composedFut.complete( new SnapshotCheckResponse(perMetaResults, exceptions)); + }); + } + + return composedFut; + } + + /** @return A composed future of partitions checks for each consistent id regarding {@link SnapshotCheckContext#metas}. */ + private CompletableFuture partitionsHashesFuture(SnapshotCheckContext ctx) { + // Per metas result: consistent id -> check results per partition key. + Map> perMetaResults = new ConcurrentHashMap<>(ctx.metas.size(), 1.0f); + // Per consistent id. + Map exceptions = new ConcurrentHashMap<>(ctx.metas.size(), 1.0f); + CompletableFuture composedFut = new CompletableFuture<>(); + AtomicInteger metasProcessed = new AtomicInteger(ctx.metas.size()); + IgniteSnapshotManager snpMgr = kctx.cache().context().snapshotMgr(); + + for (SnapshotMetadata meta : ctx.metas) { + CompletableFuture> metaFut = snpMgr.checker().checkPartitions( + meta, + snpMgr.snapshotLocalDir(ctx.req.snapshotName(), ctx.req.snapshotPath()), + ctx.req.groups(), + false, + ctx.req.fullCheck(), + false + ); + + 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 allHandlersFuture(SnapshotCheckContext ctx) { + SnapshotChecker snpChecker = kctx.cache().context().snapshotMgr().checker(); + // Per metas result: snapshot part's consistent id -> check result per handler name. + Map>> perMetaResults = new ConcurrentHashMap<>(ctx.metas.size(), 1.0f); + // Per consistent id. + Map exceptions = new ConcurrentHashMap<>(ctx.metas.size(), 1.0f); + CompletableFuture composedFut = new CompletableFuture<>(); + AtomicInteger metasProcessed = new AtomicInteger(ctx.metas.size()); + + for (SnapshotMetadata meta : ctx.metas) { + CompletableFuture>> metaFut = snpChecker.invokeCustomHandlers(meta, + ctx.req.snapshotPath(), ctx.req.groups(), true); + + 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 mapErrors(Map errors) { - return errors.entrySet().stream() - .collect(Collectors.toMap(e -> kctx.cluster().get().node(e.getKey()), - e -> e.getValue() instanceof Exception ? (Exception)e.getValue() : new IgniteException(e.getValue()))); + return errors.entrySet().stream().collect(Collectors.toMap(e -> kctx.cluster().get().node(e.getKey()), + e -> asException(e.getValue()))); } /** */ - private Map mapResults( - Map results, - Set requiredNodes, - Function resExtractor - ) { - return results.entrySet().stream() - .filter(e -> requiredNodes.contains(e.getKey()) && e.getValue() != null) - .collect(Collectors.toMap(e -> kctx.cluster().get().node(e.getKey()), e -> resExtractor.apply(e.getValue()))); + private static Exception asException(Throwable th) { + return th instanceof Exception ? (Exception)th : new IgniteException(th); } /** @@ -309,7 +477,7 @@ private IgniteInternalFuture prepareAndCheckMetas(Snapsho if (err != null) phaseFut.onDone(err); else - phaseFut.onDone(new SnapshotCheckResponse(locMetas)); + phaseFut.onDone(new SnapshotCheckResponse(locMetas, null)); }); } @@ -345,10 +513,10 @@ private void reducePreparationAndMetasCheck( 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) && !nodeRes.metas.isEmpty()) { - assert nodeRes != null && nodeRes.partsResults == null; + if (ctx.req.nodes().contains(nodeId) && baseline(nodeId) && !F.isEmpty((Collection)nodeRes.result())) { + assert nodeRes != null; - metas.put(kctx.cluster().get().node(nodeId), nodeRes.metas); + metas.put(kctx.cluster().get().node(nodeId), nodeRes.result()); } }); @@ -358,9 +526,8 @@ private void reducePreparationAndMetasCheck( if (!metasCheck.isEmpty()) throw new IgniteSnapshotVerifyException(metasCheck); - List locMetas = metas.get(kctx.cluster().get().localNode()); - - ctx.locMeta = F.isEmpty(locMetas) ? null : locMetas.get(0); + // If the topology is lesser that the snapshot's, we have to check another partitions parts. + ctx.metas = assingMetas(metas); if (clusterOpFut != null) ctx.clusterMetas = metas; @@ -381,6 +548,52 @@ private void reducePreparationAndMetasCheck( } } + /** + * 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 assingMetas(Map> clusterMetas) { + ClusterNode locNode = kctx.cluster().get().localNode(); + List locMetas = clusterMetas.get(locNode); + + if (F.isEmpty(locMetas)) + return null; + + Set onlineNodesConstIdsStr = new HashSet<>(clusterMetas.size()); + // The nodes are sorted with lesser order. + Map> 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 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. * @@ -408,7 +621,7 @@ public IgniteInternalFuture start( Set requiredNodes = new HashSet<>(F.viewReadOnly(kctx.discovery().discoCache().aliveBaselineNodes(), F.node2id())); - // Initiator is also a required node. It collects the final oparation result. + // Initiator is also a required node. It collects the final operation result. requiredNodes.add(kctx.localNodeId()); SnapshotCheckProcessRequest req = new SnapshotCheckProcessRequest( @@ -453,8 +666,11 @@ private static final class SnapshotCheckContext { /** Current process' future. Listens error, stop requests, etc. */ private final GridFutureAdapter locProcFut = new GridFutureAdapter<>(); - /** Local snapshot metadata. */ - @Nullable private SnapshotMetadata locMeta; + /** + * Metadatas to process on this node. Also indicates the snapshot parts to check on this node. + * @see #partitionsHashesFuture(SnapshotCheckContext) + */ + @Nullable private List metas; /** All the snapshot metadatas. */ @Nullable private Map> clusterMetas; @@ -474,70 +690,34 @@ private GridFutureAdapter phaseFuture() { } } - /** A DTO used to transfer nodes' results for the both phases. */ + /** A DTO to transfer node's results for the both phases. */ private static final class SnapshotCheckResponse implements Serializable { - /** Serial version uid. */ + /** */ private static final long serialVersionUID = 0L; - /** @see #metas() */ - @Nullable private final List metas; - - /** - * @see #partsHashes() - * @see #customHandlersResults() - */ - @Nullable private final Map partsResults; - - /** @see #incrementalResult() */ - @Nullable private final IncrementalSnapshotCheckResult incRes; + /** The result. Is usually a collection or a map of hashes, metast, etc. */ + private final Object result; - /** Ctor for the phase 1. */ - private SnapshotCheckResponse(@Nullable List metas) { - this.metas = metas; - this.partsResults = null; - this.incRes = null; - } + /** Exceptions per snapshot part's consistent id. */ + @Nullable private final Map exceptions; - /** Ctor for the phase 2 for normal snapshot. */ - private SnapshotCheckResponse(Map partsResults) { - this.metas = null; - this.partsResults = partsResults; - this.incRes = null; - } + /** */ + private SnapshotCheckResponse(Object result, @Nullable Map exceptions) { + assert result instanceof Serializable : "Snapshot check result is not serializable."; + assert exceptions == null || exceptions instanceof Serializable : "Snapshot check exceptions aren't serializable."; - /** Ctor for the phase 2 for incremental snapshot. */ - private SnapshotCheckResponse(IncrementalSnapshotCheckResult incRes) { - this.metas = null; - this.partsResults = null; - this.incRes = incRes; + this.result = result; + this.exceptions = exceptions == null ? null : Collections.unmodifiableMap(exceptions); } - /** Metas for the phase 1. Is always {@code null} for the phase 2. */ - @Nullable private List metas() { - return metas; - } - - /** - * Node's partition hashes for the phase 2. Is always {@code null} for the phase 1 or in case of incremental - * snapshot. - */ - private @Nullable Map partsHashes() { - return (Map)partsResults; - } - - /** - * Results of the custom handlers for the phase 2. Is always {@code null} for the phase 1 or in case of incremental - * snapshot. - * - * @see IgniteSnapshotManager#handlers() - */ - private @Nullable Map> customHandlersResults() { - return (Map>)partsResults; + /** @return Exceptions per snapshot part's consistent id. */ + private @Nullable Map exceptions() { + return exceptions; } - /** Incremental snapshot result for the phase 2. Is always {@code null} for the phase 1 or in case of normal snapshot. */ - private @Nullable IncrementalSnapshotCheckResult incrementalResult() { - return incRes; + /** @return Certain phase's and process' result. */ + private T result() { + return (T)result; } } } 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 index 994a362221f46..5a154182cd9b6 100644 --- 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 @@ -49,7 +49,7 @@ import org.apache.ignite.cluster.ClusterNode; import org.apache.ignite.dump.DumpEntry; import org.apache.ignite.internal.GridKernalContext; -import org.apache.ignite.internal.management.cache.IdleVerifyResultV2; +import org.apache.ignite.internal.management.cache.IdleVerifyResult; import org.apache.ignite.internal.management.cache.PartitionKeyV2; import org.apache.ignite.internal.managers.encryption.EncryptionCacheKeyProvider; import org.apache.ignite.internal.managers.encryption.GroupKey; @@ -574,26 +574,23 @@ private Map readTxCachesData(File snpDir) throws Ignit } /** */ - public IdleVerifyResultV2 reduceIncrementalResults( - Map results, + public IdleVerifyResult reduceIncrementalResults( + Map> results, Map operationErrors ) { if (!operationErrors.isEmpty()) - return new IdleVerifyResultV2(operationErrors); + return IdleVerifyResult.builder().exceptions(operationErrors).build(); + + IdleVerifyResult.Builder bldr = IdleVerifyResult.builder(); Map> nodeTxHashMap = new HashMap<>(); - List> txHashConflicts = new ArrayList<>(); - Map> partHashes = new HashMap<>(); - Map> partiallyCommittedTxs = new HashMap<>(); - Map errors = new HashMap<>(); - results.forEach((node, res) -> { - if (res.exceptions().isEmpty() && errors.isEmpty()) { + results.forEach((node, resLst) -> resLst.forEach(res -> { + if (res.exceptions().isEmpty() && !bldr.hasErrors()) { if (!F.isEmpty(res.partiallyCommittedTxs())) - partiallyCommittedTxs.put(node, res.partiallyCommittedTxs()); + bldr.addPartiallyCommited(node, res.partiallyCommittedTxs()); - for (Map.Entry entry : res.partHashRes().entrySet()) - partHashes.computeIfAbsent(entry.getKey(), v -> new ArrayList<>()).add(entry.getValue()); + bldr.addPartitionHashes(res.partHashRes()); if (log.isDebugEnabled()) log.debug("Handle VerifyIncrementalSnapshotJob result [node=" + node + ", taskRes=" + res + ']'); @@ -612,47 +609,22 @@ public IdleVerifyResultV2 reduceIncrementalResults( TransactionsHashRecord prevHash = prevNodeTxHash.remove(hash.localConsistentId()); if (prevHash == null || prevHash.transactionHash() != hash.transactionHash()) - txHashConflicts.add(F.asList(hash, prevHash)); + bldr.addTxConflicts(F.asList(hash, prevHash)); resIt.remove(); } } } else if (!res.exceptions().isEmpty()) - errors.put(node, F.first(res.exceptions())); - }); + bldr.addException(node, F.first(res.exceptions())); + })); // Add all missed pairs to conflicts. nodeTxHashMap.values().stream() .flatMap(e -> e.values().stream()) - .forEach(e -> txHashConflicts.add(F.asList(e, null))); - - return errors.isEmpty() - ? new IdleVerifyResultV2(partHashes, txHashConflicts, partiallyCommittedTxs) - : new IdleVerifyResultV2(errors); - } - - /** */ - public static IdleVerifyResultV2 reduceHashesResults( - Map> results, - Map ex - ) { - Map> clusterHashes = new HashMap<>(); + .forEach(e -> bldr.addTxConflicts(F.asList(e, null))); - results.forEach((node, nodeHashes) -> { - assert ex.get(node) == null; - - for (Map.Entry e : nodeHashes.entrySet()) { - List records = clusterHashes.computeIfAbsent(e.getKey(), k -> new ArrayList<>()); - - records.add(e.getValue()); - } - }); - - if (results.size() != ex.size()) - return new IdleVerifyResultV2(clusterHashes, ex); - else - return new IdleVerifyResultV2(ex); + return bldr.build(); } /** */ @@ -1023,33 +995,39 @@ public CompletableFuture> checkPartit } /** - * Checks results of the internal and custon snapshot validation handlres. Throws exception if a validation error occurs. - * + * 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(String, String, String, Collection, boolean) */ public void checkCustomHandlersResults( String snpName, - Map>> results + Map>>> results ) throws Exception { Map>> clusterResults = new HashMap<>(); Collection execNodes = new ArrayList<>(results.size()); - for (Map.Entry>> nodeRes : results.entrySet()) { - ClusterNode node = nodeRes.getKey(); + // Checking node -> Map by snapshot part's consistend id. + for (Map.Entry>>> nodeRes : results.entrySet()) { + // Consistent id -> Map by handler name. + for (Map.Entry>> nodeConsIdRes : nodeRes.getValue().entrySet()) { + ClusterNode node = nodeRes.getKey(); - // Depending on the job mapping, we can get several different results from one node. - execNodes.add(node.id()); + // We can get several different results from one node. + execNodes.add(node.id()); - assert nodeRes.getValue() != null : "At least the default snapshot restore handler should have been executed "; + assert nodeRes.getValue() != null : "At least the default snapshot restore handler should have been executed "; - for (Map.Entry> nodeHndRes : nodeRes.getValue().entrySet()) { - String hndName = nodeHndRes.getKey(); - SnapshotHandlerResult hndRes = nodeHndRes.getValue(); + // Handler name -> handler result. + for (Map.Entry> nodeHndRes : nodeConsIdRes.getValue().entrySet()) { + String hndName = nodeHndRes.getKey(); + SnapshotHandlerResult hndRes = nodeHndRes.getValue(); - if (hndRes.error() != null) - throw hndRes.error(); + if (hndRes.error() != null) + throw hndRes.error(); - clusterResults.computeIfAbsent(hndName, v -> new ArrayList<>()).add(hndRes); + clusterResults.computeIfAbsent(hndName, v -> new ArrayList<>()).add(hndRes); + } } } 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 78db07b74fd8b..706bcd34daf62 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 @@ -17,15 +17,11 @@ package org.apache.ignite.internal.processors.cache.persistence.snapshot; -import java.util.ArrayList; import java.util.Collection; -import java.util.HashMap; -import java.util.List; import java.util.Map; import org.apache.ignite.IgniteCheckedException; import org.apache.ignite.IgniteException; -import org.apache.ignite.cluster.ClusterNode; -import org.apache.ignite.internal.management.cache.IdleVerifyResultV2; +import org.apache.ignite.internal.management.cache.IdleVerifyResult; import org.apache.ignite.internal.management.cache.PartitionKeyV2; import org.apache.ignite.internal.processors.cache.GridCacheSharedContext; import org.apache.ignite.internal.processors.cache.verify.PartitionHashRecordV2; @@ -63,23 +59,22 @@ public SnapshotPartitionsVerifyHandler(GridCacheSharedContext cctx) { /** {@inheritDoc} */ @Override public void complete(String name, Collection>> results) throws IgniteCheckedException { - Map> clusterHashes = new HashMap<>(); - Map errs = new HashMap<>(); + IdleVerifyResult.Builder bldr = IdleVerifyResult.builder(); for (SnapshotHandlerResult> res : results) { if (res.error() != null) { - errs.put(res.node(), res.error()); + bldr.addException(res.node(), res.error()); continue; } for (Map.Entry entry : res.data().entrySet()) - clusterHashes.computeIfAbsent(entry.getKey(), v -> new ArrayList<>()).add(entry.getValue()); + bldr.addPartitionHash(entry.getKey(), entry.getValue()); } - IdleVerifyResultV2 verifyResult = new IdleVerifyResultV2(clusterHashes, errs); + IdleVerifyResult verifyResult = bldr.build(); - if (errs.isEmpty() && !verifyResult.hasConflicts()) + if (!bldr.hasErrors() && !verifyResult.hasConflicts()) return; GridStringBuilder buf = new GridStringBuilder(); diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotPartitionsVerifyResult.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotPartitionsVerifyResult.java index 560f6237c1c67..e188758a1831f 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotPartitionsVerifyResult.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotPartitionsVerifyResult.java @@ -28,7 +28,7 @@ import java.util.stream.Collectors; import org.apache.ignite.cluster.ClusterNode; import org.apache.ignite.internal.dto.IgniteDataTransferObject; -import org.apache.ignite.internal.management.cache.IdleVerifyResultV2; +import org.apache.ignite.internal.management.cache.IdleVerifyResult; import org.apache.ignite.internal.util.GridStringBuilder; import org.apache.ignite.internal.util.typedef.F; import org.apache.ignite.internal.util.typedef.internal.U; @@ -36,7 +36,7 @@ /** * The result of execution snapshot partitions verify task which besides calculating partition hashes of - * {@link IdleVerifyResultV2} also contains the snapshot metadata distribution across the cluster. + * {@link IdleVerifyResult} also contains the snapshot metadata distribution across the cluster. */ public class SnapshotPartitionsVerifyResult extends IgniteDataTransferObject { /** Serial version uid. */ @@ -46,7 +46,7 @@ public class SnapshotPartitionsVerifyResult extends IgniteDataTransferObject { private Map> metas; /** Result of cluster nodes partitions comparison. */ - @Nullable private IdleVerifyResultV2 idleRes; + @Nullable private IdleVerifyResult idleRes; /** Default constructor. */ public SnapshotPartitionsVerifyResult() { @@ -59,7 +59,7 @@ public SnapshotPartitionsVerifyResult() { */ SnapshotPartitionsVerifyResult( Map> metas, - @Nullable IdleVerifyResultV2 idleRes + @Nullable IdleVerifyResult idleRes ) { this.metas = metas; this.idleRes = idleRes; @@ -103,7 +103,7 @@ public void print(Consumer printer) { /** * @return Result of cluster nodes partitions comparison. */ - public IdleVerifyResultV2 idleVerifyResult() { + public IdleVerifyResult idleVerifyResult() { return idleRes; } @@ -123,6 +123,6 @@ public Map exceptions() { /** {@inheritDoc} */ @Override protected void readExternalData(byte protoVer, ObjectInput in) throws IOException, ClassNotFoundException { metas = U.readMap(in); - idleRes = (IdleVerifyResultV2)in.readObject(); + idleRes = (IdleVerifyResult)in.readObject(); } } diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheFastNodeLeftForTransactionTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheFastNodeLeftForTransactionTest.java index c394d4f59bd90..55cf4cd1ba8bc 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheFastNodeLeftForTransactionTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheFastNodeLeftForTransactionTest.java @@ -30,7 +30,7 @@ import org.apache.ignite.configuration.ConnectorConfiguration; import org.apache.ignite.configuration.IgniteConfiguration; import org.apache.ignite.internal.IgniteEx; -import org.apache.ignite.internal.management.cache.IdleVerifyResultV2; +import org.apache.ignite.internal.management.cache.IdleVerifyResult; import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxFinishFuture; import org.apache.ignite.internal.processors.cache.transactions.TransactionProxyImpl; import org.apache.ignite.internal.util.typedef.internal.SB; @@ -236,7 +236,7 @@ private void check( checkCacheData(cacheValues, cacheName); - IdleVerifyResultV2 idleVerifyResV2 = idleVerify(stoppedNode, null); + IdleVerifyResult idleVerifyResV2 = idleVerify(stoppedNode, null); SB sb = new SB(); diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/HistoricalRebalanceCheckpointTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/HistoricalRebalanceCheckpointTest.java index 52e2b39cc92b6..16f764d9cfc61 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/HistoricalRebalanceCheckpointTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/HistoricalRebalanceCheckpointTest.java @@ -44,7 +44,7 @@ import org.apache.ignite.failure.StopNodeFailureHandler; import org.apache.ignite.internal.IgniteEx; import org.apache.ignite.internal.TestRecordingCommunicationSpi; -import org.apache.ignite.internal.management.cache.IdleVerifyResultV2; +import org.apache.ignite.internal.management.cache.IdleVerifyResult; import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxFinishRequest; import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxPrepareRequest; import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxPrepareResponse; @@ -191,7 +191,7 @@ public void testDelayed1PhaseCommitResponses() throws Exception { // Storing the highest counters on backup. forceCheckpoint(); - IdleVerifyResultV2 checkRes = idleVerify(prim, DEFAULT_CACHE_NAME); + IdleVerifyResult checkRes = idleVerify(prim, DEFAULT_CACHE_NAME); Map conflicts = F.flatCollections(checkRes.counterConflicts().values()) .stream().collect(Collectors.toMap(PartitionHashRecordV2::isPrimary, Functions.identity())); @@ -394,7 +394,7 @@ private void doTestDelayedToBackupsRequests(int nodes, boolean putAfterGaps) thr rebalanceLsnr.check(); - IdleVerifyResultV2 checkRes = idleVerify(prim, DEFAULT_CACHE_NAME); + IdleVerifyResult checkRes = idleVerify(prim, DEFAULT_CACHE_NAME); assertFalse(checkRes.hasConflicts()); } 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 0e6fbf134c170..7364c21a3a811 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 @@ -61,9 +61,9 @@ 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.IdleVerifyResultV2; +import org.apache.ignite.internal.management.cache.IdleVerifyResult; import org.apache.ignite.internal.management.cache.PartitionKeyV2; -import org.apache.ignite.internal.management.cache.VerifyBackupPartitionsTaskV2; +import org.apache.ignite.internal.management.cache.VerifyBackupPartitionsTask; import org.apache.ignite.internal.managers.communication.GridMessageListener; import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion; import org.apache.ignite.internal.processors.cache.CacheGroupContext; @@ -145,7 +145,7 @@ public void testClusterSnapshotCheck() throws Exception { createAndCheckSnapshot(ignite, SNAPSHOT_NAME); - IdleVerifyResultV2 res = snp(ignite).checkSnapshot(SNAPSHOT_NAME, null).get().idleVerifyResult(); + IdleVerifyResult res = snp(ignite).checkSnapshot(SNAPSHOT_NAME, null).get().idleVerifyResult(); StringBuilder b = new StringBuilder(); res.print(b::append, true); @@ -169,7 +169,7 @@ public void testClusterSnapshotCheckMissedPart() throws Exception { assertTrue(part0.toString(), part0.toFile().exists()); assertTrue(part0.toFile().delete()); - IdleVerifyResultV2 res = snp(ignite).checkSnapshot(SNAPSHOT_NAME, null).get().idleVerifyResult(); + IdleVerifyResult res = snp(ignite).checkSnapshot(SNAPSHOT_NAME, null).get().idleVerifyResult(); StringBuilder b = new StringBuilder(); res.print(b::append, true); @@ -193,7 +193,7 @@ public void testClusterSnapshotCheckMissedGroup() throws Exception { assertTrue(dir.toString(), dir.toFile().exists()); assertTrue(U.delete(dir)); - IdleVerifyResultV2 res = snp(ignite).checkSnapshot(SNAPSHOT_NAME, null).get().idleVerifyResult(); + IdleVerifyResult res = snp(ignite).checkSnapshot(SNAPSHOT_NAME, null).get().idleVerifyResult(); StringBuilder b = new StringBuilder(); res.print(b::append, true); @@ -236,7 +236,7 @@ public void testClusterSnapshotCheckWithNodeFilter() throws Exception { createAndCheckSnapshot(ig0, SNAPSHOT_NAME); - IdleVerifyResultV2 res = snp(ig0).checkSnapshot(SNAPSHOT_NAME, null).get().idleVerifyResult(); + IdleVerifyResult res = snp(ig0).checkSnapshot(SNAPSHOT_NAME, null).get().idleVerifyResult(); StringBuilder b = new StringBuilder(); res.print(b::append, true); @@ -312,7 +312,7 @@ public void testClusterSnapshotCheckPartitionCounters() throws Exception { pageStore.finishRecover(); } - IdleVerifyResultV2 res = snp(ignite).checkSnapshot(SNAPSHOT_NAME, null).get().idleVerifyResult(); + IdleVerifyResult res = snp(ignite).checkSnapshot(SNAPSHOT_NAME, null).get().idleVerifyResult(); StringBuilder b = new StringBuilder(); res.print(b::append, true); @@ -362,7 +362,7 @@ public void testClusterSnapshotCheckOtherCluster() throws Exception { ignite.cluster().baselineAutoAdjustEnabled(false); ignite.cluster().state(ACTIVE); - IdleVerifyResultV2 res = snp(ignite).checkSnapshot(SNAPSHOT_NAME, null).get().idleVerifyResult(); + IdleVerifyResult res = snp(ignite).checkSnapshot(SNAPSHOT_NAME, null).get().idleVerifyResult(); StringBuilder b = new StringBuilder(); res.print(b::append, true); @@ -386,7 +386,7 @@ public void testClusterSnapshotCheckCRCFail() throws Exception { corruptPartitionFile(ignite, SNAPSHOT_NAME, dfltCacheCfg, PART_ID); - IdleVerifyResultV2 res = snp(ignite).checkSnapshot(SNAPSHOT_NAME, null, null, false, -1, false).get().idleVerifyResult(); + IdleVerifyResult res = snp(ignite).checkSnapshot(SNAPSHOT_NAME, null, null, false, -1, false).get().idleVerifyResult(); assertEquals("Check must be disabled", 0, res.exceptions().size()); @@ -494,7 +494,7 @@ public void testClusterSnapshotCheckFailsOnPartitionDataDiffers() throws Excepti assertNotNull(part0); assertTrue(part0.toString(), part0.toFile().exists()); - IdleVerifyResultV2 res = snp(ignite).checkSnapshot(SNAPSHOT_NAME, null).get().idleVerifyResult(); + IdleVerifyResult res = snp(ignite).checkSnapshot(SNAPSHOT_NAME, null).get().idleVerifyResult(); StringBuilder b = new StringBuilder(); res.print(b::append, true); @@ -521,9 +521,9 @@ public void testClusterSnapshotCheckHashesSameAsIdleVerifyHashes() throws Except arg.cacheFilter(CacheFilterEnum.USER); arg.checkCrc(true); - IdleVerifyResultV2 idleVerifyRes = ignite.compute().execute(new TestVisorBackupPartitionsTask(), arg); + IdleVerifyResult idleVerifyRes = ignite.compute().execute(new TestVisorBackupPartitionsTask(), arg); - IdleVerifyResultV2 snpVerifyRes = snp(ignite).checkSnapshot(SNAPSHOT_NAME, null).get().idleVerifyResult(); + IdleVerifyResult snpVerifyRes = snp(ignite).checkSnapshot(SNAPSHOT_NAME, null).get().idleVerifyResult(); Map> idleVerifyHashes = jobResults.get(TestVisorBackupPartitionsTask.class); Map> snpCheckHashes = jobResults.get(TestVisorBackupPartitionsTask.class); @@ -564,6 +564,48 @@ public void testClusterSnapshotCheckWithTwoCachesCheckNotCorrupted() throws Exce 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 { @@ -1405,10 +1447,10 @@ private static void corruptPartitionFile( } /** */ - private class TestVisorBackupPartitionsTask extends VerifyBackupPartitionsTaskV2 { + private class TestVisorBackupPartitionsTask extends VerifyBackupPartitionsTask { /** {@inheritDoc} */ - @Override public @Nullable IdleVerifyResultV2 reduce(List results) throws IgniteException { - IdleVerifyResultV2 res = super.reduce(results); + @Override public @Nullable IdleVerifyResult reduce(List results) throws IgniteException { + IdleVerifyResult res = super.reduce(results); saveHashes(TestVisorBackupPartitionsTask.class, results); 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 335078fc7abf8..ac7c42dcd39c6 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 @@ -51,7 +51,7 @@ import org.apache.ignite.internal.IgniteInternalFuture; import org.apache.ignite.internal.TestRecordingCommunicationSpi; import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException; -import org.apache.ignite.internal.management.cache.IdleVerifyResultV2; +import org.apache.ignite.internal.management.cache.IdleVerifyResult; import org.apache.ignite.internal.processors.cache.DynamicCacheChangeBatch; import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsSingleMessage; import org.apache.ignite.internal.processors.cache.persistence.file.FileIO; @@ -142,7 +142,7 @@ public void testClusterSnapshotRestoreFromCustomDir() throws Exception { createAndCheckSnapshot(ignite, SNAPSHOT_NAME, snpDir.toString(), TIMEOUT); // Check snapshot. - IdleVerifyResultV2 res = snp(ignite).checkSnapshot(SNAPSHOT_NAME, snpDir.getAbsolutePath()).get(TIMEOUT) + IdleVerifyResult res = snp(ignite).checkSnapshot(SNAPSHOT_NAME, snpDir.getAbsolutePath()).get(TIMEOUT) .idleVerifyResult(); StringBuilder sb = new StringBuilder(); diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotStreamerTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotStreamerTest.java index 941cd8b8571a4..fbec40d362dc1 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotStreamerTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotStreamerTest.java @@ -244,7 +244,7 @@ public void testMetaWarningRestoredByOnlyOneNode() throws Exception { stopGrid(0); stopGrid(1); - createAndCheckSnapshot(client, false, false, DataStreamerUpdatesHandler.WRN_MSG, + createAndCheckSnapshot(client, false, DataStreamerUpdatesHandler.WRN_MSG, SnapshotPartitionsQuickVerifyHandler.WRN_MSG); } @@ -351,7 +351,7 @@ private void doTestDataStreamerWhileSnapshot(IgniteEx snpHnd, boolean allowOverw String notExpWrn = allowOverwrite ? null : SnapshotPartitionsQuickVerifyHandler.WRN_MSG; try { - SnapshotPartitionsVerifyResult checkRes = createAndCheckSnapshot(snpHnd, true, true, expectedWrn, + SnapshotPartitionsVerifyResult checkRes = createAndCheckSnapshot(snpHnd, true, expectedWrn, notExpWrn); if (expectedWrn != null) { @@ -417,9 +417,9 @@ private void doTestDataStreamerFailedBeforeSnapshot(IgniteEx snpHnd, boolean all waitForCondition(loadFut::isDone, getTestTimeout()); if (allowOverwrite) - createAndCheckSnapshot(snpHnd, true, true, null, null); + createAndCheckSnapshot(snpHnd, true, null, null); else { - createAndCheckSnapshot(snpHnd, true, true, SnapshotPartitionsQuickVerifyHandler.WRN_MSG, + createAndCheckSnapshot(snpHnd, true, SnapshotPartitionsQuickVerifyHandler.WRN_MSG, DataStreamerUpdatesHandler.WRN_MSG); } } @@ -462,7 +462,7 @@ private IgniteInternalFuture runLoad(Ignite ldr, boolean allowOverwrite, Atom } /** */ - private SnapshotPartitionsVerifyResult createAndCheckSnapshot(IgniteEx snpHnd, boolean create, boolean expectVerifyErr, + private SnapshotPartitionsVerifyResult createAndCheckSnapshot(IgniteEx snpHnd, boolean create, String expWrn, String notExpWrn) throws Exception { assert notExpWrn == null || expWrn != null; @@ -502,7 +502,7 @@ private SnapshotPartitionsVerifyResult createAndCheckSnapshot(IgniteEx snpHnd, b assertTrue(checkRes.exceptions().isEmpty()); - if (!onlyPrimary && expectVerifyErr) + if (!onlyPrimary) assertTrue((expWrn != null) == checkRes.idleVerifyResult().hasConflicts()); if (expWrn != null) { diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotConsistencyTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotConsistencyTest.java index 370b92cdd07fd..a573e4079a2f7 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotConsistencyTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotConsistencyTest.java @@ -37,7 +37,7 @@ import org.apache.ignite.internal.IgniteEx; import org.apache.ignite.internal.IgniteInternalFuture; import org.apache.ignite.internal.TestRecordingCommunicationSpi; -import org.apache.ignite.internal.management.cache.IdleVerifyResultV2; +import org.apache.ignite.internal.management.cache.IdleVerifyResult; import org.apache.ignite.internal.managers.communication.GridMessageListener; import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion; import org.apache.ignite.internal.processors.cache.GridCacheMessage; @@ -210,7 +210,7 @@ public void testConcurrentPutWithStaleTopologyVersion() throws Exception { snpFut.get(getTestTimeout()); putFut.get(getTestTimeout()); - IdleVerifyResultV2 snpVerifyRes = crd.context().cache().context().snapshotMgr() + IdleVerifyResult snpVerifyRes = crd.context().cache().context().snapshotMgr() .checkSnapshot(SNAPSHOT_NAME, null).get().idleVerifyResult(); assertFalse(snpVerifyRes.hasConflicts()); diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotRestoreFromRemoteTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotRestoreFromRemoteTest.java index cbb5767e9c45c..7451e46d054af 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotRestoreFromRemoteTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotRestoreFromRemoteTest.java @@ -45,7 +45,7 @@ import org.apache.ignite.failure.StopNodeOrHaltFailureHandler; import org.apache.ignite.internal.IgniteEx; import org.apache.ignite.internal.TestRecordingCommunicationSpi; -import org.apache.ignite.internal.management.cache.IdleVerifyResultV2; +import org.apache.ignite.internal.management.cache.IdleVerifyResult; import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionDemandMessage; import org.apache.ignite.internal.processors.cache.persistence.partstate.GroupPartitionId; import org.apache.ignite.internal.util.typedef.F; @@ -246,7 +246,7 @@ public void testRestoreFromAnEmptyNode() throws Exception { awaitPartitionMapExchange(); // Ensure that the snapshot check command succeeds. - IdleVerifyResultV2 res = emptyNode.context().cache().context().snapshotMgr() + IdleVerifyResult res = emptyNode.context().cache().context().snapshotMgr() .checkSnapshot(SNAPSHOT_NAME, null).get(TIMEOUT).idleVerifyResult(); StringBuilder buf = new StringBuilder(); diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackOnTimeoutOnePhaseCommitTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackOnTimeoutOnePhaseCommitTest.java index c0c24bd969f69..38a3f5c2af255 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackOnTimeoutOnePhaseCommitTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackOnTimeoutOnePhaseCommitTest.java @@ -27,7 +27,7 @@ import org.apache.ignite.internal.IgniteInternalFuture; import org.apache.ignite.internal.IgniteInterruptedCheckedException; import org.apache.ignite.internal.TestRecordingCommunicationSpi; -import org.apache.ignite.internal.management.cache.IdleVerifyResultV2; +import org.apache.ignite.internal.management.cache.IdleVerifyResult; import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxPrepareResponse; import org.apache.ignite.internal.util.typedef.X; import org.apache.ignite.internal.util.typedef.internal.U; @@ -193,7 +193,7 @@ private void doTestRollbackOnTimeoutPartitionDesync(TransactionConcurrency concu fut.get(); - IdleVerifyResultV2 res = idleVerify(client, DEFAULT_CACHE_NAME); + IdleVerifyResult res = idleVerify(client, DEFAULT_CACHE_NAME); assertPartitionsSame(res); diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxWithSmallTimeoutAndContentionOneKeyTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxWithSmallTimeoutAndContentionOneKeyTest.java index 29a1ef76ff210..ff2e4099b7a56 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxWithSmallTimeoutAndContentionOneKeyTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxWithSmallTimeoutAndContentionOneKeyTest.java @@ -34,7 +34,7 @@ import org.apache.ignite.configuration.IgniteConfiguration; import org.apache.ignite.internal.IgniteEx; import org.apache.ignite.internal.IgniteInternalFuture; -import org.apache.ignite.internal.management.cache.IdleVerifyResultV2; +import org.apache.ignite.internal.management.cache.IdleVerifyResult; import org.apache.ignite.internal.management.cache.PartitionKeyV2; import org.apache.ignite.internal.processors.cache.verify.PartitionHashRecordV2; import org.apache.ignite.internal.util.typedef.internal.SB; @@ -202,7 +202,7 @@ public void test() throws Exception { f.get(); - IdleVerifyResultV2 idleVerifyResult = idleVerify(igClient, DEFAULT_CACHE_NAME); + IdleVerifyResult idleVerifyResult = idleVerify(igClient, DEFAULT_CACHE_NAME); log.info("Current counter value:" + cnt.get()); diff --git a/modules/core/src/test/java/org/apache/ignite/testframework/junits/common/GridCommonAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/testframework/junits/common/GridCommonAbstractTest.java index 3e909de461028..122cec2361dd2 100755 --- a/modules/core/src/test/java/org/apache/ignite/testframework/junits/common/GridCommonAbstractTest.java +++ b/modules/core/src/test/java/org/apache/ignite/testframework/junits/common/GridCommonAbstractTest.java @@ -81,7 +81,7 @@ import org.apache.ignite.internal.IgniteInterruptedCheckedException; import org.apache.ignite.internal.IgniteKernal; import org.apache.ignite.internal.management.cache.CacheIdleVerifyCommandArg; -import org.apache.ignite.internal.management.cache.IdleVerifyResultV2; +import org.apache.ignite.internal.management.cache.IdleVerifyResult; import org.apache.ignite.internal.management.cache.IdleVerifyTaskV2; import org.apache.ignite.internal.pagemem.wal.record.DataRecord; import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion; @@ -2224,7 +2224,7 @@ protected void forceCheckpoint(Collection nodes, String reason) throws I * @return Conflicts result. * @throws IgniteException If none caches or node found. */ - protected IdleVerifyResultV2 idleVerify(Ignite ig, @Nullable String... caches) throws Exception { + protected IdleVerifyResult idleVerify(Ignite ig, @Nullable String... caches) throws Exception { log.info("Starting idleVerify ..."); IgniteEx ig0 = (IgniteEx)ig; @@ -2248,7 +2248,7 @@ protected IdleVerifyResultV2 idleVerify(Ignite ig, @Nullable String... caches) t taskArg.caches(cacheNames.toArray(U.EMPTY_STRS)); - return ((VisorTaskResult)ig.compute().execute( + return ((VisorTaskResult)ig.compute().execute( IdleVerifyTaskV2.class.getName(), new VisorTaskArgument<>(node.id(), taskArg, false) )).result(); @@ -2447,7 +2447,7 @@ public static void awaitCacheOnClient(Ignite client, String cacheName) { /** * @param res Response. */ - protected static void assertPartitionsSame(IdleVerifyResultV2 res) throws AssertionFailedError { + protected static void assertPartitionsSame(IdleVerifyResult res) throws AssertionFailedError { if (res.hasConflicts()) { StringBuilder b = new StringBuilder(); diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotCheckWithIndexesTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotCheckWithIndexesTest.java index 39c6926a230ee..c7f1316077635 100644 --- a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotCheckWithIndexesTest.java +++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotCheckWithIndexesTest.java @@ -24,7 +24,7 @@ import org.apache.ignite.cluster.ClusterNode; import org.apache.ignite.configuration.CacheConfiguration; import org.apache.ignite.internal.IgniteEx; -import org.apache.ignite.internal.management.cache.IdleVerifyResultV2; +import org.apache.ignite.internal.management.cache.IdleVerifyResult; import org.apache.ignite.internal.util.typedef.F; import org.apache.ignite.lang.IgnitePredicate; import org.junit.Test; @@ -44,7 +44,7 @@ public void testClusterSnapshotCheckEmptyCache() throws Exception { snp(ignite).createSnapshot(SNAPSHOT_NAME, null, false, onlyPrimary).get(TIMEOUT); - IdleVerifyResultV2 res = ignite.context().cache().context().snapshotMgr() + IdleVerifyResult res = ignite.context().cache().context().snapshotMgr() .checkSnapshot(SNAPSHOT_NAME, null).get().idleVerifyResult(); StringBuilder b = new StringBuilder(); @@ -62,7 +62,7 @@ public void testClusterSnapshotCheckWithIndexes() throws Exception { createAndCheckSnapshot(ignite, SNAPSHOT_NAME, null, TIMEOUT); - IdleVerifyResultV2 res = ignite.context().cache().context().snapshotMgr() + IdleVerifyResult res = ignite.context().cache().context().snapshotMgr() .checkSnapshot(SNAPSHOT_NAME, null).get().idleVerifyResult(); StringBuilder b = new StringBuilder(); @@ -89,7 +89,7 @@ public void testClusterSnapshotCheckWithNodeFilter() throws Exception { createAndCheckSnapshot(grid(0), SNAPSHOT_NAME, null, TIMEOUT); - IdleVerifyResultV2 res = grid(0).context().cache().context().snapshotMgr() + IdleVerifyResult res = grid(0).context().cache().context().snapshotMgr() .checkSnapshot(SNAPSHOT_NAME, null).get().idleVerifyResult(); StringBuilder b = new StringBuilder();