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 804bad2..2d1981a 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
@@ -35,7 +35,6 @@
 import java.nio.file.Paths;
 import java.nio.file.SimpleFileVisitor;
 import java.nio.file.attribute.BasicFileAttributes;
-import java.util.ArrayDeque;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.BitSet;
@@ -2628,7 +2627,7 @@
         private final Queue<RemoteSnapshotFilesRecevier> queue = new ConcurrentLinkedDeque<>();
 
         /** {@code true} if the node is stopping. */
-        private volatile boolean stopping;
+        private boolean stopping;
 
         /**
          * @param next New task for scheduling.
@@ -2639,14 +2638,6 @@
             if (stopping) {
                 next.acceptException(new IgniteException(SNP_NODE_STOPPING_ERR_MSG));
 
-                if (active != null)
-                    active.acceptException(new IgniteException(SNP_NODE_STOPPING_ERR_MSG));
-
-                RemoteSnapshotFilesRecevier r;
-
-                while ((r = queue.poll()) != null)
-                    r.acceptException(new IgniteException(SNP_NODE_STOPPING_ERR_MSG));
-
                 return;
             }
 
@@ -2674,9 +2665,17 @@
         }
 
         /** Stopping handler. */
-        public void stop() {
+        public synchronized void stop() {
             stopping = true;
 
+            if (active != null)
+                active.acceptException(new IgniteException(SNP_NODE_STOPPING_ERR_MSG));
+
+            RemoteSnapshotFilesRecevier r;
+
+            while ((r = queue.poll()) != null)
+                r.acceptException(new IgniteException(SNP_NODE_STOPPING_ERR_MSG));
+
             Set<RemoteSnapshotFilesRecevier> futs = activeTasks();
             GridCompoundFuture<Void, Void> stopFut = new GridCompoundFuture<>();
 
@@ -2709,7 +2708,6 @@
          * @return The set of currently scheduled tasks, some of them may be already completed.
          */
         private Set<RemoteSnapshotFilesRecevier> activeTasks() {
-
             Set<RemoteSnapshotFilesRecevier> futs = new HashSet<>(queue);
 
             RemoteSnapshotFilesRecevier active0 = active;
@@ -2913,12 +2911,10 @@
                             ", grpId=" + grpId + ", partId=" + partId + ']');
                     }
 
-                    busyLock.enterBusy();
+                    if (!busyLock.enterBusy())
+                        throw new IgniteException(SNP_NODE_STOPPING_ERR_MSG);
 
                     try {
-                        if (stopping)
-                            throw new IgniteException(SNP_NODE_STOPPING_ERR_MSG);
-
                         task0.acceptFile(file);
                     }
                     finally {
@@ -2930,70 +2926,6 @@
     }
 
     /**
-     * Such an executor can executes tasks not in a single thread, but executes them
-     * on different threads sequentially. It's important for some {@link SnapshotSender}'s
-     * to process sub-task sequentially due to all these sub-tasks may share a single socket
-     * channel to send data to.
-     */
-    private static class SequentialExecutorWrapper implements Executor {
-        /** Ignite logger. */
-        private final IgniteLogger log;
-
-        /** Queue of task to execute. */
-        private final Queue<Runnable> tasks = new ArrayDeque<>();
-
-        /** Delegate executor. */
-        private final Executor executor;
-
-        /** Currently running task. */
-        private volatile Runnable active;
-
-        /** If wrapped executor is shutting down. */
-        private volatile boolean stopping;
-
-        /**
-         * @param executor Executor to run tasks on.
-         */
-        public SequentialExecutorWrapper(IgniteLogger log, Executor executor) {
-            this.log = log.getLogger(SequentialExecutorWrapper.class);
-            this.executor = executor;
-        }
-
-        /** {@inheritDoc} */
-        @Override public synchronized void execute(final Runnable r) {
-            assert !stopping : "Task must be cancelled prior to the wrapped executor is shutting down.";
-
-            tasks.offer(() -> {
-                try {
-                    r.run();
-                }
-                finally {
-                    scheduleNext();
-                }
-            });
-
-            if (active == null)
-                scheduleNext();
-        }
-
-        /** */
-        private synchronized void scheduleNext() {
-            if ((active = tasks.poll()) != null) {
-                try {
-                    executor.execute(active);
-                }
-                catch (RejectedExecutionException e) {
-                    tasks.clear();
-
-                    stopping = true;
-
-                    log.warning("Task is outdated. Wrapped executor is shutting down.", e);
-                }
-            }
-        }
-    }
-
-    /**
      *
      */
     private static class RemoteSnapshotSender extends SnapshotSender {
@@ -3017,7 +2949,7 @@
             GridIoManager.TransmissionSender sndr,
             String rqId
         ) {
-            super(log, new SequentialExecutorWrapper(log, exec));
+            super(log, exec);
 
             this.sndr = sndr;
             this.rqId = rqId;
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotFilesRequestMessage.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotFilesRequestMessage.java
index 81c9870..a06ec86 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotFilesRequestMessage.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotFilesRequestMessage.java
@@ -26,6 +26,7 @@
 import java.util.Set;
 import java.util.stream.Collectors;
 import org.apache.ignite.internal.GridDirectMap;
+import org.apache.ignite.internal.util.typedef.F;
 import org.apache.ignite.internal.util.typedef.internal.S;
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.plugin.extensions.communication.MessageCollectionItemType;
@@ -80,7 +81,7 @@
         this.snpPath = snpPath;
         this.parts = new HashMap<>();
 
-        for (Map.Entry<Integer, Set<Integer>> e : parts.entrySet())
+        for (Map.Entry<Integer, Set<Integer>> e : F.view(parts.entrySet(), e -> !F.isEmpty(e.getValue())))
             this.parts.put(e.getKey(), U.toIntArray(e.getValue()));
     }
 
@@ -90,11 +91,8 @@
     public Map<Integer, Set<Integer>> parts() {
         Map<Integer, Set<Integer>> res = new HashMap<>();
 
-        for (Map.Entry<Integer, int[]> e : parts.entrySet()) {
-            res.put(e.getKey(), e.getValue().length == 0 ? null : Arrays.stream(e.getValue())
-                .boxed()
-                .collect(Collectors.toSet()));
-        }
+        for (Map.Entry<Integer, int[]> e : parts.entrySet())
+            res.put(e.getKey(), Arrays.stream(e.getValue()).boxed().collect(Collectors.toSet()));
 
         return res;
     }
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotResponseRemoteFutureTask.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotResponseRemoteFutureTask.java
index 6ccf200..d18a6e3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotResponseRemoteFutureTask.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotResponseRemoteFutureTask.java
@@ -18,13 +18,14 @@
 package org.apache.ignite.internal.processors.cache.persistence.snapshot;
 
 import java.io.File;
-import java.util.ArrayList;
-import java.util.Collections;
+import java.util.Collection;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
+import java.util.function.Function;
 import org.apache.ignite.IgniteException;
 import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
 import org.apache.ignite.internal.processors.cache.persistence.file.FileIOFactory;
@@ -32,7 +33,6 @@
 import org.apache.ignite.internal.util.typedef.F;
 import org.jetbrains.annotations.Nullable;
 
-import static java.util.Optional.ofNullable;
 import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.cacheDirectory;
 import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.getPartitionFile;
 import static org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotManager.databaseRelativePath;
@@ -50,6 +50,7 @@
      * @param tmpWorkDir Working directory for intermediate snapshot results.
      * @param ioFactory Factory to working with snapshot files.
      * @param snpSndr Factory which produces snapshot receiver instance.
+     * @param parts Partition to be processed.
      */
     public SnapshotResponseRemoteFutureTask(
         GridCacheSharedContext<?, ?> cctx,
@@ -72,71 +73,67 @@
             return false;
 
         try {
-            List<GroupPartitionId> handled = new ArrayList<>();
+            List<SnapshotMetadata> metas = cctx.snapshotMgr().readSnapshotMetadatas(snpName, snpPath);
 
-            for (Map.Entry<Integer, Set<Integer>> e : parts.entrySet()) {
-                ofNullable(e.getValue()).orElse(Collections.emptySet())
-                    .forEach(p -> handled.add(new GroupPartitionId(e.getKey(), p)));
+            Function<GroupPartitionId, SnapshotMetadata> findMeta = pair -> {
+                for (SnapshotMetadata meta : metas) {
+                    Map<Integer, Set<Integer>> parts0 = meta.partitions();
+
+                    if (F.isEmpty(parts0))
+                        continue;
+
+                    Set<Integer> locParts = parts0.get(pair.getGroupId());
+
+                    if (locParts != null && locParts.contains(pair.getPartitionId()))
+                        return meta;
+                }
+
+                return null;
+            };
+
+            Map<GroupPartitionId, SnapshotMetadata> partsToSend = new HashMap<>();
+
+            parts.forEach((grpId, parts) -> parts.forEach(
+                part -> partsToSend.computeIfAbsent(new GroupPartitionId(grpId, part), findMeta)));
+
+            if (partsToSend.containsValue(null)) {
+                Collection<GroupPartitionId> missed = F.viewReadOnly(partsToSend.entrySet(), Map.Entry::getKey,
+                    e -> e.getValue() == null);
+
+                throw new IgniteException("Snapshot partitions missed on local node " +
+                    "[snpName=" + snpName + ", missed=" + missed + ']');
             }
 
-            snpSndr.init(handled.size());
+            snpSndr.init(partsToSend.size());
 
             File snpDir = cctx.snapshotMgr().snapshotLocalDir(snpName, snpPath);
 
-            List<CompletableFuture<Void>> futs = new ArrayList<>();
-            List<SnapshotMetadata> metas = cctx.snapshotMgr().readSnapshotMetadatas(snpName, snpPath);
+            CompletableFuture.runAsync(() -> partsToSend.forEach((gp, meta) -> {
+                if (err.get() != null)
+                    return;
 
-            for (SnapshotMetadata meta : metas) {
-                Map<Integer, Set<Integer>> parts0 = meta.partitions();
+                File cacheDir = cacheDirectory(new File(snpDir, databaseRelativePath(meta.folderName())),
+                    gp.getGroupId());
 
-                if (F.isEmpty(parts0))
-                    continue;
+                if (cacheDir == null) {
+                    throw new IgniteException("Cache directory not found [snpName=" + snpName + ", meta=" + meta +
+                        ", pair=" + gp + ']');
+                }
 
-                handled.removeIf(gp -> {
-                    if (ofNullable(parts0.get(gp.getGroupId()))
-                        .orElse(Collections.emptySet())
-                        .contains(gp.getPartitionId())
-                    ) {
-                        futs.add(CompletableFuture.runAsync(() -> {
-                            if (err.get() != null)
-                                return;
+                File snpPart = getPartitionFile(cacheDir.getParentFile(), cacheDir.getName(), gp.getPartitionId());
 
-                            File cacheDir = cacheDirectory(new File(snpDir, databaseRelativePath(meta.folderName())),
-                                gp.getGroupId());
+                if (!snpPart.exists()) {
+                    throw new IgniteException("Snapshot partition file not found [cacheDir=" + cacheDir +
+                        ", pair=" + gp + ']');
+                }
 
-                            if (cacheDir == null) {
-                                throw new IgniteException("Cache directory not found [snpName=" + snpName + ", meta=" + meta +
-                                    ", pair=" + gp + ']');
-                            }
-
-                            File snpPart = getPartitionFile(cacheDir.getParentFile(), cacheDir.getName(), gp.getPartitionId());
-
-                            if (!snpPart.exists()) {
-                                throw new IgniteException("Snapshot partition file not found [cacheDir=" + cacheDir +
-                                    ", pair=" + gp + ']');
-                            }
-
-                            snpSndr.sendPart(snpPart, cacheDir.getName(), gp, snpPart.length());
-                        }, snpSndr.executor())
-                            .whenComplete((r, t) -> err.compareAndSet(null, t)));
-
-                        return true;
-                    }
-
-                    return false;
-                });
-            }
-
-            if (!handled.isEmpty()) {
-                err.compareAndSet(null, new IgniteException("Snapshot partitions missed on local node [snpName=" + snpName +
-                    ", missed=" + handled + ']'));
-            }
-
-            int size = futs.size();
-
-            CompletableFuture.allOf(futs.toArray(new CompletableFuture[size]))
+                snpSndr.sendPart(snpPart, cacheDir.getName(), gp, snpPart.length());
+            }), snpSndr.executor())
                 .whenComplete((r, t) -> {
-                    Throwable th = ofNullable(err.get()).orElse(t);
+                    if (t != null)
+                        err.compareAndSet(null, t);
+
+                    Throwable th = err.get();
 
                     if (th == null && log.isInfoEnabled()) {
                         log.info("Snapshot partitions have been sent to the remote node [snpName=" + snpName +
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotRemoteRequestTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotRemoteRequestTest.java
index d61630b..59525a2 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotRemoteRequestTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotRemoteRequestTest.java
@@ -18,12 +18,16 @@
 package org.apache.ignite.internal.processors.cache.persistence.snapshot;
 
 import java.io.File;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.UUID;
+import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -32,6 +36,7 @@
 import java.util.stream.Collectors;
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.IgniteException;
+import org.apache.ignite.configuration.IgniteConfiguration;
 import org.apache.ignite.internal.IgniteEx;
 import org.apache.ignite.internal.IgniteInternalFuture;
 import org.apache.ignite.internal.IgniteInterruptedCheckedException;
@@ -49,6 +54,7 @@
 import org.apache.ignite.testframework.GridTestUtils;
 import org.junit.Test;
 
+import static org.apache.ignite.configuration.IgniteConfiguration.DFLT_SNAPSHOT_THREAD_POOL_SIZE;
 import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.partId;
 import static org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotRestoreProcess.groupIdFromTmpDir;
 import static org.apache.ignite.testframework.GridTestUtils.assertThrowsAnyCause;
@@ -56,6 +62,18 @@
 
 /** */
 public class IgniteSnapshotRemoteRequestTest extends IgniteClusterSnapshotRestoreBaseTest {
+    /** */
+    private int snapshotThreadPoolSize = DFLT_SNAPSHOT_THREAD_POOL_SIZE;
+
+    /** {@inheritDoc} */
+    @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+        IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+        cfg.setSnapshotThreadPoolSize(snapshotThreadPoolSize);
+
+        return cfg;
+    }
+
     /** @throws Exception If fails. */
     @Test
     public void testSnapshotRemoteRequestFromSingleNode() throws Exception {
@@ -300,6 +318,53 @@
             "Future cancelled prior to the all requested partitions processed");
     }
 
+    /** @throws Exception If fails. */
+    @Test
+    public void testSendPartitonsSequentially() throws Exception {
+        snapshotThreadPoolSize = 1;
+
+        IgniteEx sndr = startGridsWithCache(3, CACHE_KEYS_RANGE, valueBuilder(), dfltCacheCfg);
+
+        UUID sndNode = sndr.localNode().id();
+
+        sndr.snapshot().createSnapshot(SNAPSHOT_NAME).get(TIMEOUT);
+
+        IgniteSnapshotManager mgr0 = snp(grid(0));
+
+        List<UUID> nodes = new CopyOnWriteArrayList<>();
+
+        mgr0.remoteSnapshotSenderFactory((rqId, nodeId) -> new DelegateSnapshotSender(log,
+            snp(sndr).snapshotExecutorService(), mgr0.remoteSnapshotSenderFactory(rqId, nodeId)) {
+            @Override public void sendPart0(File part, String cacheDirName, GroupPartitionId pair, Long length) {
+                nodes.add(nodeId);
+
+                // Single thread must send partitions sequentially node by node.
+                checkDuplicates(nodes);
+
+                super.sendPart0(part, cacheDirName, pair, length);
+            }
+        });
+
+        Collection<IgniteEx> rcvrs = F.viewReadOnly(G.allGrids(), srv -> (IgniteEx)srv,
+            srv -> !F.eq(sndr.localNode(), srv.cluster().localNode()));
+
+        GridCompoundFuture<Void, Void> futs = new GridCompoundFuture<>();
+
+        for (IgniteEx rcv : rcvrs) {
+            Map<Integer, Set<Integer>> expParts = owningParts(rcv, CU.cacheId(DEFAULT_CACHE_NAME), sndNode);
+
+            IgniteInternalFuture<Void> fut = snp(rcv)
+                .requestRemoteSnapshotFiles(sndNode, SNAPSHOT_NAME, null, expParts, () -> false,
+                    defaultPartitionConsumer(expParts, null));
+
+            fut.listen(f -> expParts.values().forEach(integers -> assertTrue(integers.isEmpty())));
+
+            futs.add(fut);
+        }
+
+        futs.markInitialized().get(getTestTimeout());
+    }
+
     /**
      * @param parts Expected partitions.
      * @param latch Latch to await partitions processed.
@@ -315,7 +380,8 @@
             assertTrue("Received partition has not been requested",
                 parts.get(grpId).remove(partId(part.getName())));
 
-            latch.countDown();
+            if (latch != null)
+                latch.countDown();
         };
     }
 
@@ -337,4 +403,25 @@
             .map(Map.Entry::getKey)
             .collect(Collectors.toSet()));
     }
+
+    /**
+     * Checks that the list can contain only neighboring duplicates:
+     * 1, 1, 1, 3, 2, 2 -> ok
+     * 1, 1, 3, 1, 2, 2 -> fail
+     *
+     * @param list List to check.
+     */
+    private <T> void checkDuplicates(List<T> list) {
+        LinkedList<T> grouped = new LinkedList<>();
+
+        for (T item : list) {
+            if (!F.eq(grouped.peekLast(), item))
+                grouped.add(item);
+        }
+
+        if (list.stream().distinct().collect(Collectors.toList()).equals(grouped))
+            return;
+
+        fail("List contains non neighboring duplicates: " + list);
+    }
 }