WIP

Author: NSAmelchev

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IncrementalSnapshotVerify.java

@@ -234,31 +234,25 @@ else if (txRec.state() == TransactionState.ROLLED_BACK) {
             for (Map.Entry<GridCacheVersion, Set<Short>> tx: txPrimParticipatingNodes.entrySet())
                 calcTxHash.accept(tx.getKey(), tx.getValue());
 
-            Collection<TransactionsHashRecord> txHashRes = nodesTxHash.entrySet().stream().map(e -> {
-                Short nodeId = e.getKey();
-                HashHolder hashHolder = e.getValue();
-
-                return new TransactionsHashRecord(
+            Collection<TransactionsHashRecord> txHashRes = nodesTxHash.entrySet().stream()
+                .map(e -> new TransactionsHashRecord(
                     sft.consistentId(),
-                    blt.compactIdMapping().get(nodeId),
-                    hashHolder.hash);
-            }).collect(Collectors.toCollection(ArrayList::new));
+                    blt.compactIdMapping().get(e.getKey()),
+                    e.getValue().hash
+                ))
+                .collect(Collectors.toCollection(ArrayList::new));
 
-            Map<PartitionKey, PartitionHashRecord> partHashRes = partMap.entrySet().stream()
-                .collect(Collectors.toMap(
-                    Map.Entry::getKey,
-                    e -> new PartitionHashRecord(
+            Collection<PartitionHashRecord> partHashRes = partMap.entrySet().stream()
+                .map(e -> new PartitionHashRecord(
                         e.getKey(),
                         false,
                         sft.consistentId(),
                         null,
                         0,
                         null,
                         new VerifyPartitionContext(e.getValue())
-                    ),
-                    (a, b) -> a,
-                    HashMap::new
-                ));
+                    )
+                ).collect(Collectors.toCollection(ArrayList::new));
 
             if (log.isInfoEnabled()) {
                 log.info("Verify incremental snapshot procedure finished " +

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IncrementalSnapshotVerifyResult.java

@@ -18,10 +18,8 @@
 package org.apache.ignite.internal.processors.cache.persistence.snapshot;
 
 import java.util.Collection;
-import java.util.Map;
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.internal.Order;
-import org.apache.ignite.internal.management.cache.PartitionKey;
 import org.apache.ignite.internal.managers.communication.ErrorMessage;
 import org.apache.ignite.internal.pagemem.wal.record.DataEntry;
 import org.apache.ignite.internal.processors.cache.verify.PartitionHashRecord;
@@ -46,7 +44,7 @@ public class IncrementalSnapshotVerifyResult implements MarshallableMessage {
      * Partition hashes collection. Value is a hash of data entries {@link DataEntry} from WAL segments included
      * into the incremental snapshot.
      */
-    private Map<PartitionKey, PartitionHashRecord> partHashRes;
+    private Collection<PartitionHashRecord> partHashRes;
 
     /** */
     @Order(1)
@@ -68,7 +66,7 @@ public IncrementalSnapshotVerifyResult() {
     /** */
     IncrementalSnapshotVerifyResult(
         Collection<TransactionsHashRecord> txHashRes,
-        Map<PartitionKey, PartitionHashRecord> partHashRes,
+        Collection<PartitionHashRecord> partHashRes,
         Collection<GridCacheVersion> partiallyCommittedTxs,
         Collection<Exception> exceptions
     ) {
@@ -79,7 +77,7 @@ public IncrementalSnapshotVerifyResult() {
     }
 
     /** */
-    public Map<PartitionKey, PartitionHashRecord> partHashRes() {
+    public Collection<PartitionHashRecord> partHashRes() {
         return partHashRes;
     }
 

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotChecker.java

@@ -33,7 +33,9 @@
 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.apache.ignite.internal.processors.cache.verify.TransactionsHashRecord;
 import org.apache.ignite.internal.util.typedef.F;
 import org.apache.ignite.plugin.extensions.communication.Message;
@@ -102,13 +104,16 @@ public IdleVerifyResult reduceIncrementalResults(
             if (!F.isEmpty(res.partiallyCommittedTxs()))
                 bldr.addPartiallyCommited(nodeRes.getKey(), res.partiallyCommittedTxs());
 
-            bldr.addPartitionHashes(res.partHashRes());
+            Map<PartitionKey, PartitionHashRecord> partHashRes = res.partHashRes().stream()
+                .collect(Collectors.toMap(PartitionHashRecord::partitionKey, Function.identity()));
+
+            bldr.addPartitionHashes(partHashRes);
 
             if (log.isDebugEnabled())
                 log.debug("Handle VerifyIncrementalSnapshotJob result [node=" + nodeRes.getKey() + ", taskRes=" + res + ']');
 
             Map<Object, TransactionsHashRecord> txHashRes = res.txHashRes().stream().collect(
-                Collectors.toMap(TransactionsHashRecord::remoteConsistentId, Function.identity(), (a, b) -> a, HashMap::new));
+                Collectors.toMap(TransactionsHashRecord::remoteConsistentId, Function.identity()));
 
             bldr.addIncrementalHashRecords(nodeRes.getKey(), txHashRes);
         }