ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject incubator-ignite git commit: # ignite-394: GridDataLoaderProcessor -> IgniteDataStreamerProcessor.java + test
Date Tue, 03 Mar 2015 18:42:37 GMT
Repository: incubator-ignite
Updated Branches:
  refs/heads/ignite-394 9b33b6510 -> 36b7c8f74


# ignite-394: GridDataLoaderProcessor -> IgniteDataStreamerProcessor.java + test


Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/36b7c8f7
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/36b7c8f7
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/36b7c8f7

Branch: refs/heads/ignite-394
Commit: 36b7c8f74445ccff81c6370685d85beeddfee76b
Parents: 9b33b65
Author: Artem Shutak <ashutak@gridgain.com>
Authored: Tue Mar 3 21:43:02 2015 +0300
Committer: Artem Shutak <ashutak@gridgain.com>
Committed: Tue Mar 3 21:43:02 2015 +0300

----------------------------------------------------------------------
 .../org/apache/ignite/IgniteDataStreamer.java   |   2 +-
 .../ignite/internal/GridKernalContext.java      |   2 +-
 .../ignite/internal/GridKernalContextImpl.java  |  10 +-
 .../apache/ignite/internal/IgniteKernal.java    |   2 +-
 .../dataload/GridDataLoaderProcessor.java       | 316 -------
 .../dataload/IgniteDataStreamerProcessor.java   | 316 +++++++
 .../GridDataLoaderProcessorSelfTest.java        | 924 -------------------
 .../IgniteDataStreamerProcessorSelfTest.java    | 924 +++++++++++++++++++
 .../ignite/testsuites/IgniteCacheTestSuite.java |   2 +-
 9 files changed, 1249 insertions(+), 1249 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/36b7c8f7/modules/core/src/main/java/org/apache/ignite/IgniteDataStreamer.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/IgniteDataStreamer.java b/modules/core/src/main/java/org/apache/ignite/IgniteDataStreamer.java
index c48d61a..519a0a0 100644
--- a/modules/core/src/main/java/org/apache/ignite/IgniteDataStreamer.java
+++ b/modules/core/src/main/java/org/apache/ignite/IgniteDataStreamer.java
@@ -41,7 +41,7 @@ import java.util.*;
  * data can be loaded from underlying data store on demand, whenever it is accessed -
  * for this no explicit data loading step is needed.
  * <p>
- * {@code IgniteDataLoader} supports the following configuration properties:
+ * {@code IgniteDataStreamer} supports the following configuration properties:
  * <ul>
  *  <li>
  *      {@link #perNodeBufferSize(int)} - when entries are added to data loader via

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/36b7c8f7/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java b/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java
index cb9ffa1..c497c06 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java
@@ -247,7 +247,7 @@ public interface GridKernalContext extends Iterable<GridComponent> {
      *
      * @return Data loader processor.
      */
-    public <K, V> GridDataLoaderProcessor<K, V> dataLoad();
+    public <K, V> IgniteDataStreamerProcessor<K, V> dataLoad();
 
     /**
      * Gets file system processor.

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/36b7c8f7/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java
index 756c16a..71a357e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java
@@ -202,7 +202,7 @@ public class GridKernalContextImpl implements GridKernalContext, Externalizable
 
     /** */
     @GridToStringInclude
-    private GridDataLoaderProcessor dataLdrProc;
+    private IgniteDataStreamerProcessor dataLdrProc;
 
     /** */
     @GridToStringInclude
@@ -446,8 +446,8 @@ public class GridKernalContextImpl implements GridKernalContext, Externalizable
             affProc = (GridAffinityProcessor)comp;
         else if (comp instanceof GridRestProcessor)
             restProc = (GridRestProcessor)comp;
-        else if (comp instanceof GridDataLoaderProcessor)
-            dataLdrProc = (GridDataLoaderProcessor)comp;
+        else if (comp instanceof IgniteDataStreamerProcessor)
+            dataLdrProc = (IgniteDataStreamerProcessor)comp;
         else if (comp instanceof IgfsProcessorAdapter)
             igfsProc = (IgfsProcessorAdapter)comp;
         else if (comp instanceof GridOffHeapProcessor)
@@ -660,8 +660,8 @@ public class GridKernalContextImpl implements GridKernalContext, Externalizable
 
     /** {@inheritDoc} */
     @SuppressWarnings("unchecked")
-    @Override public <K, V> GridDataLoaderProcessor<K, V> dataLoad() {
-        return (GridDataLoaderProcessor<K, V>)dataLdrProc;
+    @Override public <K, V> IgniteDataStreamerProcessor<K, V> dataLoad() {
+        return (IgniteDataStreamerProcessor<K, V>)dataLdrProc;
     }
 
     /** {@inheritDoc} */

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/36b7c8f7/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java b/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
index 336f872..2299f3e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
@@ -753,7 +753,7 @@ public class IgniteKernal implements IgniteEx, IgniteMXBean, Externalizable {
             startProcessor(ctx, new GridTaskProcessor(ctx), attrs);
             startProcessor(ctx, (GridProcessor)SCHEDULE.createOptional(ctx), attrs);
             startProcessor(ctx, new GridRestProcessor(ctx), attrs);
-            startProcessor(ctx, new GridDataLoaderProcessor(ctx), attrs);
+            startProcessor(ctx, new IgniteDataStreamerProcessor(ctx), attrs);
             startProcessor(ctx, new GridStreamProcessor(ctx), attrs);
             startProcessor(ctx, (GridProcessor) IGFS.create(ctx, F.isEmpty(cfg.getIgfsConfiguration())), attrs);
             startProcessor(ctx, new GridContinuousProcessor(ctx), attrs);

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/36b7c8f7/modules/core/src/main/java/org/apache/ignite/internal/processors/dataload/GridDataLoaderProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/dataload/GridDataLoaderProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/dataload/GridDataLoaderProcessor.java
deleted file mode 100644
index b29c9ea..0000000
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/dataload/GridDataLoaderProcessor.java
+++ /dev/null
@@ -1,316 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.processors.dataload;
-
-import org.apache.ignite.*;
-import org.apache.ignite.internal.*;
-import org.apache.ignite.internal.managers.communication.*;
-import org.apache.ignite.internal.managers.deployment.*;
-import org.apache.ignite.internal.processors.*;
-import org.apache.ignite.internal.util.*;
-import org.apache.ignite.internal.util.typedef.*;
-import org.apache.ignite.internal.util.typedef.internal.*;
-import org.apache.ignite.internal.util.worker.*;
-import org.apache.ignite.marshaller.*;
-import org.apache.ignite.thread.*;
-import org.jetbrains.annotations.*;
-
-import java.util.*;
-import java.util.concurrent.*;
-
-import static org.apache.ignite.internal.GridTopic.*;
-import static org.apache.ignite.internal.managers.communication.GridIoPolicy.*;
-
-/**
- *
- */
-public class GridDataLoaderProcessor<K, V> extends GridProcessorAdapter {
-    /** Loaders map (access is not supposed to be highly concurrent). */
-    private Collection<IgniteDataStreamerImpl> ldrs = new GridConcurrentHashSet<>();
-
-    /** Busy lock. */
-    private final GridSpinBusyLock busyLock = new GridSpinBusyLock();
-
-    /** Flushing thread. */
-    private Thread flusher;
-
-    /** */
-    private final DelayQueue<IgniteDataStreamerImpl<K, V>> flushQ = new DelayQueue<>();
-
-    /** Marshaller. */
-    private final Marshaller marsh;
-
-    /**
-     * @param ctx Kernal context.
-     */
-    public GridDataLoaderProcessor(GridKernalContext ctx) {
-        super(ctx);
-
-        ctx.io().addMessageListener(TOPIC_DATALOAD, new GridMessageListener() {
-            @Override public void onMessage(UUID nodeId, Object msg) {
-                assert msg instanceof GridDataLoadRequest;
-
-                processDataLoadRequest(nodeId, (GridDataLoadRequest)msg);
-            }
-        });
-
-        marsh = ctx.config().getMarshaller();
-    }
-
-    /** {@inheritDoc} */
-    @Override public void start() throws IgniteCheckedException {
-        if (ctx.config().isDaemon())
-            return;
-
-        flusher = new IgniteThread(new GridWorker(ctx.gridName(), "grid-data-loader-flusher", log) {
-            @Override protected void body() throws InterruptedException {
-                while (!isCancelled()) {
-                    IgniteDataStreamerImpl<K, V> ldr = flushQ.take();
-
-                    if (!busyLock.enterBusy())
-                        return;
-
-                    try {
-                        if (ldr.isClosed())
-                            continue;
-
-                        ldr.tryFlush();
-
-                        flushQ.offer(ldr);
-                    }
-                    finally {
-                        busyLock.leaveBusy();
-                    }
-                }
-            }
-        });
-
-        flusher.start();
-
-        if (log.isDebugEnabled())
-            log.debug("Started data loader processor.");
-    }
-
-    /** {@inheritDoc} */
-    @Override public void onKernalStop(boolean cancel) {
-        if (ctx.config().isDaemon())
-            return;
-
-        ctx.io().removeMessageListener(TOPIC_DATALOAD);
-
-        busyLock.block();
-
-        U.interrupt(flusher);
-        U.join(flusher, log);
-
-        for (IgniteDataStreamerImpl<?, ?> ldr : ldrs) {
-            if (log.isDebugEnabled())
-                log.debug("Closing active data loader on grid stop [ldr=" + ldr + ", cancel=" + cancel + ']');
-
-            try {
-                ldr.closeEx(cancel);
-            }
-            catch (IgniteInterruptedCheckedException e) {
-                U.warn(log, "Interrupted while waiting for completion of the data loader: " + ldr, e);
-            }
-            catch (IgniteCheckedException e) {
-                U.error(log, "Failed to close data loader: " + ldr, e);
-            }
-        }
-
-        if (log.isDebugEnabled())
-            log.debug("Stopped data loader processor.");
-    }
-
-    /**
-     * @param cacheName Cache name ({@code null} for default cache).
-     * @param compact {@code true} if data loader should transfer data in compact format.
-     * @return Data loader.
-     */
-    public IgniteDataStreamerImpl<K, V> dataLoader(@Nullable String cacheName, boolean compact) {
-        if (!busyLock.enterBusy())
-            throw new IllegalStateException("Failed to create data loader (grid is stopping).");
-
-        try {
-            final IgniteDataStreamerImpl<K, V> ldr = new IgniteDataStreamerImpl<>(ctx, cacheName, flushQ, compact);
-
-            ldrs.add(ldr);
-
-            ldr.internalFuture().listenAsync(new CI1<IgniteInternalFuture<?>>() {
-                @Override public void apply(IgniteInternalFuture<?> f) {
-                    boolean b = ldrs.remove(ldr);
-
-                    assert b : "Loader has not been added to set: " + ldr;
-
-                    if (log.isDebugEnabled())
-                        log.debug("Loader has been completed: " + ldr);
-                }
-            });
-
-            return ldr;
-        }
-        finally {
-            busyLock.leaveBusy();
-        }
-    }
-
-    /**
-     * @param cacheName Cache name ({@code null} for default cache).
-     * @return Data loader.
-     */
-    public IgniteDataStreamer<K, V> dataLoader(@Nullable String cacheName) {
-        return dataLoader(cacheName, true);
-    }
-
-    /**
-     * @param nodeId Sender ID.
-     * @param req Request.
-     */
-    private void processDataLoadRequest(UUID nodeId, GridDataLoadRequest req) {
-        if (!busyLock.enterBusy()) {
-            if (log.isDebugEnabled())
-                log.debug("Ignoring data load request (node is stopping): " + req);
-
-            return;
-        }
-
-        try {
-            if (log.isDebugEnabled())
-                log.debug("Processing data load request: " + req);
-
-            Object topic;
-
-            try {
-                topic = marsh.unmarshal(req.responseTopicBytes(), null);
-            }
-            catch (IgniteCheckedException e) {
-                U.error(log, "Failed to unmarshal topic from request: " + req, e);
-
-                return;
-            }
-
-            ClassLoader clsLdr;
-
-            if (req.forceLocalDeployment())
-                clsLdr = U.gridClassLoader();
-            else {
-                GridDeployment dep = ctx.deploy().getGlobalDeployment(
-                    req.deploymentMode(),
-                    req.sampleClassName(),
-                    req.sampleClassName(),
-                    req.userVersion(),
-                    nodeId,
-                    req.classLoaderId(),
-                    req.participants(),
-                    null);
-
-                if (dep == null) {
-                    sendResponse(nodeId,
-                        topic,
-                        req.requestId(),
-                        new IgniteCheckedException("Failed to get deployment for request [sndId=" + nodeId +
-                            ", req=" + req + ']'),
-                        false);
-
-                    return;
-                }
-
-                clsLdr = dep.classLoader();
-            }
-
-            Collection<Map.Entry<K, V>> col;
-            IgniteDataStreamer.Updater<K, V> updater;
-
-            try {
-                col = marsh.unmarshal(req.collectionBytes(), clsLdr);
-                updater = marsh.unmarshal(req.updaterBytes(), clsLdr);
-            }
-            catch (IgniteCheckedException e) {
-                U.error(log, "Failed to unmarshal message [nodeId=" + nodeId + ", req=" + req + ']', e);
-
-                sendResponse(nodeId, topic, req.requestId(), e, false);
-
-                return;
-            }
-
-            GridDataLoadUpdateJob<K, V> job = new GridDataLoadUpdateJob<>(ctx,
-                log,
-                req.cacheName(),
-                col,
-                req.ignoreDeploymentOwnership(),
-                req.skipStore(),
-                updater);
-
-            Exception err = null;
-
-            try {
-                job.call();
-            }
-            catch (Exception e) {
-                U.error(log, "Failed to finish update job.", e);
-
-                err = e;
-            }
-
-            sendResponse(nodeId, topic, req.requestId(), err, req.forceLocalDeployment());
-        }
-        finally {
-            busyLock.leaveBusy();
-        }
-    }
-
-    /**
-     * @param nodeId Node ID.
-     * @param resTopic Response topic.
-     * @param reqId Request ID.
-     * @param err Error.
-     * @param forceLocDep Force local deployment.
-     */
-    private void sendResponse(UUID nodeId, Object resTopic, long reqId, @Nullable Throwable err,
-        boolean forceLocDep) {
-        byte[] errBytes;
-
-        try {
-            errBytes = err != null ? marsh.marshal(err) : null;
-        }
-        catch (IgniteCheckedException e) {
-            U.error(log, "Failed to marshal message.", e);
-
-            return;
-        }
-
-        GridDataLoadResponse res = new GridDataLoadResponse(reqId, errBytes, forceLocDep);
-
-        try {
-            ctx.io().send(nodeId, resTopic, res, PUBLIC_POOL);
-        }
-        catch (IgniteCheckedException e) {
-            if (ctx.discovery().alive(nodeId))
-                U.error(log, "Failed to respond to node [nodeId=" + nodeId + ", res=" + res + ']', e);
-            else if (log.isDebugEnabled())
-                log.debug("Node has left the grid: " + nodeId);
-        }
-    }
-
-    /** {@inheritDoc} */
-    @Override public void printMemoryStats() {
-        X.println(">>>");
-        X.println(">>> Data loader processor memory stats [grid=" + ctx.gridName() + ']');
-        X.println(">>>   ldrsSize: " + ldrs.size());
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/36b7c8f7/modules/core/src/main/java/org/apache/ignite/internal/processors/dataload/IgniteDataStreamerProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/dataload/IgniteDataStreamerProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/dataload/IgniteDataStreamerProcessor.java
new file mode 100644
index 0000000..a2effbc
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/dataload/IgniteDataStreamerProcessor.java
@@ -0,0 +1,316 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.dataload;
+
+import org.apache.ignite.*;
+import org.apache.ignite.internal.*;
+import org.apache.ignite.internal.managers.communication.*;
+import org.apache.ignite.internal.managers.deployment.*;
+import org.apache.ignite.internal.processors.*;
+import org.apache.ignite.internal.util.*;
+import org.apache.ignite.internal.util.typedef.*;
+import org.apache.ignite.internal.util.typedef.internal.*;
+import org.apache.ignite.internal.util.worker.*;
+import org.apache.ignite.marshaller.*;
+import org.apache.ignite.thread.*;
+import org.jetbrains.annotations.*;
+
+import java.util.*;
+import java.util.concurrent.*;
+
+import static org.apache.ignite.internal.GridTopic.*;
+import static org.apache.ignite.internal.managers.communication.GridIoPolicy.*;
+
+/**
+ *
+ */
+public class IgniteDataStreamerProcessor<K, V> extends GridProcessorAdapter {
+    /** Loaders map (access is not supposed to be highly concurrent). */
+    private Collection<IgniteDataStreamerImpl> ldrs = new GridConcurrentHashSet<>();
+
+    /** Busy lock. */
+    private final GridSpinBusyLock busyLock = new GridSpinBusyLock();
+
+    /** Flushing thread. */
+    private Thread flusher;
+
+    /** */
+    private final DelayQueue<IgniteDataStreamerImpl<K, V>> flushQ = new DelayQueue<>();
+
+    /** Marshaller. */
+    private final Marshaller marsh;
+
+    /**
+     * @param ctx Kernal context.
+     */
+    public IgniteDataStreamerProcessor(GridKernalContext ctx) {
+        super(ctx);
+
+        ctx.io().addMessageListener(TOPIC_DATALOAD, new GridMessageListener() {
+            @Override public void onMessage(UUID nodeId, Object msg) {
+                assert msg instanceof GridDataLoadRequest;
+
+                processDataLoadRequest(nodeId, (GridDataLoadRequest)msg);
+            }
+        });
+
+        marsh = ctx.config().getMarshaller();
+    }
+
+    /** {@inheritDoc} */
+    @Override public void start() throws IgniteCheckedException {
+        if (ctx.config().isDaemon())
+            return;
+
+        flusher = new IgniteThread(new GridWorker(ctx.gridName(), "grid-data-loader-flusher", log) {
+            @Override protected void body() throws InterruptedException {
+                while (!isCancelled()) {
+                    IgniteDataStreamerImpl<K, V> ldr = flushQ.take();
+
+                    if (!busyLock.enterBusy())
+                        return;
+
+                    try {
+                        if (ldr.isClosed())
+                            continue;
+
+                        ldr.tryFlush();
+
+                        flushQ.offer(ldr);
+                    }
+                    finally {
+                        busyLock.leaveBusy();
+                    }
+                }
+            }
+        });
+
+        flusher.start();
+
+        if (log.isDebugEnabled())
+            log.debug("Started data loader processor.");
+    }
+
+    /** {@inheritDoc} */
+    @Override public void onKernalStop(boolean cancel) {
+        if (ctx.config().isDaemon())
+            return;
+
+        ctx.io().removeMessageListener(TOPIC_DATALOAD);
+
+        busyLock.block();
+
+        U.interrupt(flusher);
+        U.join(flusher, log);
+
+        for (IgniteDataStreamerImpl<?, ?> ldr : ldrs) {
+            if (log.isDebugEnabled())
+                log.debug("Closing active data loader on grid stop [ldr=" + ldr + ", cancel=" + cancel + ']');
+
+            try {
+                ldr.closeEx(cancel);
+            }
+            catch (IgniteInterruptedCheckedException e) {
+                U.warn(log, "Interrupted while waiting for completion of the data loader: " + ldr, e);
+            }
+            catch (IgniteCheckedException e) {
+                U.error(log, "Failed to close data loader: " + ldr, e);
+            }
+        }
+
+        if (log.isDebugEnabled())
+            log.debug("Stopped data loader processor.");
+    }
+
+    /**
+     * @param cacheName Cache name ({@code null} for default cache).
+     * @param compact {@code true} if data loader should transfer data in compact format.
+     * @return Data loader.
+     */
+    public IgniteDataStreamerImpl<K, V> dataLoader(@Nullable String cacheName, boolean compact) {
+        if (!busyLock.enterBusy())
+            throw new IllegalStateException("Failed to create data loader (grid is stopping).");
+
+        try {
+            final IgniteDataStreamerImpl<K, V> ldr = new IgniteDataStreamerImpl<>(ctx, cacheName, flushQ, compact);
+
+            ldrs.add(ldr);
+
+            ldr.internalFuture().listenAsync(new CI1<IgniteInternalFuture<?>>() {
+                @Override public void apply(IgniteInternalFuture<?> f) {
+                    boolean b = ldrs.remove(ldr);
+
+                    assert b : "Loader has not been added to set: " + ldr;
+
+                    if (log.isDebugEnabled())
+                        log.debug("Loader has been completed: " + ldr);
+                }
+            });
+
+            return ldr;
+        }
+        finally {
+            busyLock.leaveBusy();
+        }
+    }
+
+    /**
+     * @param cacheName Cache name ({@code null} for default cache).
+     * @return Data loader.
+     */
+    public IgniteDataStreamer<K, V> dataLoader(@Nullable String cacheName) {
+        return dataLoader(cacheName, true);
+    }
+
+    /**
+     * @param nodeId Sender ID.
+     * @param req Request.
+     */
+    private void processDataLoadRequest(UUID nodeId, GridDataLoadRequest req) {
+        if (!busyLock.enterBusy()) {
+            if (log.isDebugEnabled())
+                log.debug("Ignoring data load request (node is stopping): " + req);
+
+            return;
+        }
+
+        try {
+            if (log.isDebugEnabled())
+                log.debug("Processing data load request: " + req);
+
+            Object topic;
+
+            try {
+                topic = marsh.unmarshal(req.responseTopicBytes(), null);
+            }
+            catch (IgniteCheckedException e) {
+                U.error(log, "Failed to unmarshal topic from request: " + req, e);
+
+                return;
+            }
+
+            ClassLoader clsLdr;
+
+            if (req.forceLocalDeployment())
+                clsLdr = U.gridClassLoader();
+            else {
+                GridDeployment dep = ctx.deploy().getGlobalDeployment(
+                    req.deploymentMode(),
+                    req.sampleClassName(),
+                    req.sampleClassName(),
+                    req.userVersion(),
+                    nodeId,
+                    req.classLoaderId(),
+                    req.participants(),
+                    null);
+
+                if (dep == null) {
+                    sendResponse(nodeId,
+                        topic,
+                        req.requestId(),
+                        new IgniteCheckedException("Failed to get deployment for request [sndId=" + nodeId +
+                            ", req=" + req + ']'),
+                        false);
+
+                    return;
+                }
+
+                clsLdr = dep.classLoader();
+            }
+
+            Collection<Map.Entry<K, V>> col;
+            IgniteDataStreamer.Updater<K, V> updater;
+
+            try {
+                col = marsh.unmarshal(req.collectionBytes(), clsLdr);
+                updater = marsh.unmarshal(req.updaterBytes(), clsLdr);
+            }
+            catch (IgniteCheckedException e) {
+                U.error(log, "Failed to unmarshal message [nodeId=" + nodeId + ", req=" + req + ']', e);
+
+                sendResponse(nodeId, topic, req.requestId(), e, false);
+
+                return;
+            }
+
+            GridDataLoadUpdateJob<K, V> job = new GridDataLoadUpdateJob<>(ctx,
+                log,
+                req.cacheName(),
+                col,
+                req.ignoreDeploymentOwnership(),
+                req.skipStore(),
+                updater);
+
+            Exception err = null;
+
+            try {
+                job.call();
+            }
+            catch (Exception e) {
+                U.error(log, "Failed to finish update job.", e);
+
+                err = e;
+            }
+
+            sendResponse(nodeId, topic, req.requestId(), err, req.forceLocalDeployment());
+        }
+        finally {
+            busyLock.leaveBusy();
+        }
+    }
+
+    /**
+     * @param nodeId Node ID.
+     * @param resTopic Response topic.
+     * @param reqId Request ID.
+     * @param err Error.
+     * @param forceLocDep Force local deployment.
+     */
+    private void sendResponse(UUID nodeId, Object resTopic, long reqId, @Nullable Throwable err,
+        boolean forceLocDep) {
+        byte[] errBytes;
+
+        try {
+            errBytes = err != null ? marsh.marshal(err) : null;
+        }
+        catch (IgniteCheckedException e) {
+            U.error(log, "Failed to marshal message.", e);
+
+            return;
+        }
+
+        GridDataLoadResponse res = new GridDataLoadResponse(reqId, errBytes, forceLocDep);
+
+        try {
+            ctx.io().send(nodeId, resTopic, res, PUBLIC_POOL);
+        }
+        catch (IgniteCheckedException e) {
+            if (ctx.discovery().alive(nodeId))
+                U.error(log, "Failed to respond to node [nodeId=" + nodeId + ", res=" + res + ']', e);
+            else if (log.isDebugEnabled())
+                log.debug("Node has left the grid: " + nodeId);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override public void printMemoryStats() {
+        X.println(">>>");
+        X.println(">>> Data loader processor memory stats [grid=" + ctx.gridName() + ']');
+        X.println(">>>   ldrsSize: " + ldrs.size());
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/36b7c8f7/modules/core/src/test/java/org/apache/ignite/internal/processors/dataload/GridDataLoaderProcessorSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/dataload/GridDataLoaderProcessorSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/dataload/GridDataLoaderProcessorSelfTest.java
deleted file mode 100644
index 5959957..0000000
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/dataload/GridDataLoaderProcessorSelfTest.java
+++ /dev/null
@@ -1,924 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.processors.dataload;
-
-import org.apache.ignite.*;
-import org.apache.ignite.cache.*;
-import org.apache.ignite.cache.affinity.*;
-import org.apache.ignite.cache.store.*;
-import org.apache.ignite.cluster.*;
-import org.apache.ignite.configuration.*;
-import org.apache.ignite.events.*;
-import org.apache.ignite.internal.*;
-import org.apache.ignite.internal.processors.cache.*;
-import org.apache.ignite.internal.processors.cache.distributed.near.*;
-import org.apache.ignite.internal.util.typedef.internal.*;
-import org.apache.ignite.lang.*;
-import org.apache.ignite.marshaller.optimized.*;
-import org.apache.ignite.spi.discovery.tcp.*;
-import org.apache.ignite.spi.discovery.tcp.ipfinder.*;
-import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.*;
-import org.apache.ignite.testframework.junits.common.*;
-import org.jetbrains.annotations.*;
-
-import javax.cache.*;
-import javax.cache.configuration.*;
-import java.util.*;
-import java.util.concurrent.*;
-import java.util.concurrent.atomic.*;
-
-import static java.util.concurrent.TimeUnit.*;
-import static org.apache.ignite.cache.CacheAtomicityMode.*;
-import static org.apache.ignite.cache.CacheDistributionMode.*;
-import static org.apache.ignite.cache.CacheMode.*;
-import static org.apache.ignite.cache.CacheWriteSynchronizationMode.*;
-import static org.apache.ignite.events.EventType.*;
-
-/**
- *
- */
-public class GridDataLoaderProcessorSelfTest extends GridCommonAbstractTest {
-    /** */
-    private static ConcurrentHashMap<Object, Object> storeMap;
-
-    /** */
-    private TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true);
-
-    /** */
-    private CacheMode mode = PARTITIONED;
-
-    /** */
-    private boolean nearEnabled = true;
-
-    /** */
-    private boolean useCache;
-
-    /** */
-    private TestStore store;
-
-    /** {@inheritDoc} */
-    @Override public void afterTest() throws Exception {
-        super.afterTest();
-
-        useCache = false;
-    }
-
-    /** {@inheritDoc} */
-    @SuppressWarnings({"IfMayBeConditional", "unchecked"})
-    @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
-        IgniteConfiguration cfg = super.getConfiguration(gridName);
-
-        TcpDiscoverySpi spi = new TcpDiscoverySpi();
-
-        spi.setIpFinder(ipFinder);
-
-        cfg.setDiscoverySpi(spi);
-
-        cfg.setIncludeProperties();
-
-        cfg.setMarshaller(new OptimizedMarshaller(false));
-
-        if (useCache) {
-            CacheConfiguration cc = defaultCacheConfiguration();
-
-            cc.setCacheMode(mode);
-            cc.setAtomicityMode(TRANSACTIONAL);
-            cc.setDistributionMode(nearEnabled ? NEAR_PARTITIONED : PARTITIONED_ONLY);
-            cc.setWriteSynchronizationMode(FULL_SYNC);
-
-            cc.setEvictSynchronized(false);
-            cc.setEvictNearSynchronized(false);
-
-            if (store != null) {
-                cc.setCacheStoreFactory(new FactoryBuilder.SingletonFactory(store));
-                cc.setReadThrough(true);
-                cc.setWriteThrough(true);
-            }
-
-            cfg.setCacheConfiguration(cc);
-        }
-        else
-            cfg.setCacheConfiguration();
-
-        return cfg;
-    }
-
-    /**
-     * @throws Exception If failed.
-     */
-    public void testPartitioned() throws Exception {
-        mode = PARTITIONED;
-
-        checkDataLoader();
-    }
-
-    /**
-     * @throws Exception If failed.
-     */
-    public void testColocated() throws Exception {
-        mode = PARTITIONED;
-        nearEnabled = false;
-
-        checkDataLoader();
-    }
-
-    /**
-     * @throws Exception If failed.
-     */
-    public void testReplicated() throws Exception {
-        mode = REPLICATED;
-
-        checkDataLoader();
-    }
-
-    /**
-     * @throws Exception If failed.
-     */
-    public void testLocal() throws Exception {
-        mode = LOCAL;
-
-        try {
-            checkDataLoader();
-
-            assert false;
-        }
-        catch (IgniteCheckedException e) {
-            // Cannot load local cache configured remotely.
-            info("Caught expected exception: " + e);
-        }
-    }
-
-    /**
-     * @throws Exception If failed.
-     */
-    @SuppressWarnings("ErrorNotRethrown")
-    private void checkDataLoader() throws Exception {
-        try {
-            Ignite g1 = startGrid(1);
-
-            useCache = true;
-
-            Ignite g2 = startGrid(2);
-            startGrid(3);
-
-            final IgniteDataStreamer<Integer, Integer> ldr = g1.dataLoader(null);
-
-            ldr.updater(GridDataLoadCacheUpdaters.<Integer, Integer>batchedSorted());
-
-            final AtomicInteger idxGen = new AtomicInteger();
-            final int cnt = 400;
-            final int threads = 10;
-
-            final CountDownLatch l1 = new CountDownLatch(threads);
-
-            IgniteInternalFuture<?> f1 = multithreadedAsync(new Callable<Object>() {
-                @Override public Object call() throws Exception {
-                    Collection<IgniteFuture<?>> futs = new ArrayList<>(cnt);
-
-                    for (int i = 0; i < cnt; i++) {
-                        int idx = idxGen.getAndIncrement();
-
-                        futs.add(ldr.addData(idx, idx));
-                    }
-
-                    l1.countDown();
-
-                    for (IgniteFuture<?> fut : futs)
-                        fut.get();
-
-                    return null;
-                }
-            }, threads);
-
-            l1.await();
-
-            // This will wait until data loader finishes loading.
-            stopGrid(getTestGridName(1), false);
-
-            f1.get();
-
-            int s2 = internalCache(2).primaryKeySet().size();
-            int s3 = internalCache(3).primaryKeySet().size();
-            int total = threads * cnt;
-
-            assertEquals(total, s2 + s3);
-
-            final IgniteDataStreamer<Integer, Integer> rmvLdr = g2.dataLoader(null);
-
-            rmvLdr.updater(GridDataLoadCacheUpdaters.<Integer, Integer>batchedSorted());
-
-            final CountDownLatch l2 = new CountDownLatch(threads);
-
-            IgniteInternalFuture<?> f2 = multithreadedAsync(new Callable<Object>() {
-                @Override public Object call() throws Exception {
-                    Collection<IgniteFuture<?>> futs = new ArrayList<>(cnt);
-
-                    for (int i = 0; i < cnt; i++) {
-                        final int key = idxGen.decrementAndGet();
-
-                        futs.add(rmvLdr.removeData(key));
-                    }
-
-                    l2.countDown();
-
-                    for (IgniteFuture<?> fut : futs)
-                        fut.get();
-
-                    return null;
-                }
-            }, threads);
-
-            l2.await();
-
-            rmvLdr.close(false);
-
-            f2.get();
-
-            s2 = internalCache(2).primaryKeySet().size();
-            s3 = internalCache(3).primaryKeySet().size();
-
-            assert s2 == 0 && s3 == 0 : "Incorrect entries count [s2=" + s2 + ", s3=" + s3 + ']';
-        }
-        finally {
-            stopAllGrids();
-        }
-    }
-
-    /**
-     * @throws Exception If failed.
-     */
-    public void testPartitionedIsolated() throws Exception {
-        mode = PARTITIONED;
-
-        checkIsolatedDataLoader();
-    }
-
-    /**
-     * @throws Exception If failed.
-     */
-    public void testReplicatedIsolated() throws Exception {
-        mode = REPLICATED;
-
-        checkIsolatedDataLoader();
-    }
-
-    /**
-     * @throws Exception If failed.
-     */
-    private void checkIsolatedDataLoader() throws Exception {
-        try {
-            useCache = true;
-
-            Ignite g1 = startGrid(0);
-            startGrid(1);
-            startGrid(2);
-
-            awaitPartitionMapExchange();
-
-            GridCache<Integer, Integer> cache = ((IgniteKernal)grid(0)).cache(null);
-
-            for (int i = 0; i < 100; i++)
-                cache.put(i, -1);
-
-            final int cnt = 40_000;
-            final int threads = 10;
-
-            try (final IgniteDataStreamer<Integer, Integer> ldr = g1.dataLoader(null)) {
-                final AtomicInteger idxGen = new AtomicInteger();
-
-                IgniteInternalFuture<?> f1 = multithreadedAsync(new Callable<Object>() {
-                    @Override public Object call() throws Exception {
-                        for (int i = 0; i < cnt; i++) {
-                            int idx = idxGen.getAndIncrement();
-
-                            ldr.addData(idx, idx);
-                        }
-
-                        return null;
-                    }
-                }, threads);
-
-                f1.get();
-            }
-
-            for (int g = 0; g < 3; g++) {
-                ClusterNode locNode = grid(g).localNode();
-
-                GridCacheAdapter<Integer, Integer> cache0 = ((IgniteKernal)grid(g)).internalCache(null);
-
-                if (cache0.isNear())
-                    cache0 = ((GridNearCacheAdapter<Integer, Integer>)cache0).dht();
-
-                CacheAffinity<Integer> aff = cache0.affinity();
-
-                for (int key = 0; key < cnt * threads; key++) {
-                    if (aff.isPrimary(locNode, key) || aff.isBackup(locNode, key)) {
-                        GridCacheEntryEx<Integer, Integer> entry = cache0.peekEx(key);
-
-                        assertNotNull("Missing entry for key: " + key, entry);
-                        assertEquals((Integer)(key < 100 ? -1 : key), entry.rawGetOrUnmarshal(false));
-                    }
-                }
-            }
-        }
-        finally {
-            stopAllGrids();
-        }
-    }
-
-    /**
-     * Test primitive arrays can be passed into data loader.
-     *
-     * @throws Exception If failed.
-     */
-    public void testPrimitiveArrays() throws Exception {
-        try {
-            useCache = true;
-            mode = PARTITIONED;
-
-            Ignite g1 = startGrid(1);
-            startGrid(2); // Reproduced only for several nodes in topology (if marshalling is used).
-
-            List<Object> arrays = Arrays.<Object>asList(
-                new byte[] {1}, new boolean[] {true, false}, new char[] {2, 3}, new short[] {3, 4},
-                new int[] {4, 5}, new long[] {5, 6}, new float[] {6, 7}, new double[] {7, 8});
-
-            IgniteDataStreamer<Object, Object> dataLdr = g1.dataLoader(null);
-
-            for (int i = 0, size = arrays.size(); i < 1000; i++) {
-                Object arr = arrays.get(i % size);
-
-                dataLdr.addData(i, arr);
-                dataLdr.addData(i, fixedClosure(arr));
-            }
-
-            dataLdr.close(false);
-        }
-        finally {
-            stopAllGrids();
-        }
-    }
-
-    /**
-     * @throws Exception If failed.
-     */
-    public void testReplicatedMultiThreaded() throws Exception {
-        mode = REPLICATED;
-
-        checkLoaderMultithreaded(1, 2);
-    }
-
-    /**
-     * @throws Exception If failed.
-     */
-    public void testPartitionedMultiThreaded() throws Exception {
-        mode = PARTITIONED;
-
-        checkLoaderMultithreaded(1, 3);
-    }
-
-    /**
-     * Tests loader in multithreaded environment with various count of grids started.
-     *
-     * @param nodesCntNoCache How many nodes should be started without cache.
-     * @param nodesCntCache How many nodes should be started with cache.
-     * @throws Exception If failed.
-     */
-    protected void checkLoaderMultithreaded(int nodesCntNoCache, int nodesCntCache)
-        throws Exception {
-        try {
-            // Start all required nodes.
-            int idx = 1;
-
-            for (int i = 0; i < nodesCntNoCache; i++)
-                startGrid(idx++);
-
-            useCache = true;
-
-            for (int i = 0; i < nodesCntCache; i++)
-                startGrid(idx++);
-
-            Ignite g1 = grid(1);
-
-            // Get and configure loader.
-            final IgniteDataStreamer<Integer, Integer> ldr = g1.dataLoader(null);
-
-            ldr.updater(GridDataLoadCacheUpdaters.<Integer, Integer>individual());
-            ldr.perNodeBufferSize(2);
-
-            // Define count of puts.
-            final AtomicInteger idxGen = new AtomicInteger();
-
-            final AtomicBoolean done = new AtomicBoolean();
-
-            try {
-                final int totalPutCnt = 50000;
-
-                IgniteInternalFuture<?> fut1 = multithreadedAsync(new Callable<Object>() {
-                    @Override public Object call() throws Exception {
-                        Collection<IgniteFuture<?>> futs = new ArrayList<>();
-
-                        while (!done.get()) {
-                            int idx = idxGen.getAndIncrement();
-
-                            if (idx >= totalPutCnt) {
-                                info(">>> Stopping producer thread since maximum count of puts reached.");
-
-                                break;
-                            }
-
-                            futs.add(ldr.addData(idx, idx));
-                        }
-
-                        ldr.flush();
-
-                        for (IgniteFuture<?> fut : futs)
-                            fut.get();
-
-                        return null;
-                    }
-                }, 5, "producer");
-
-                IgniteInternalFuture<?> fut2 = multithreadedAsync(new Callable<Object>() {
-                    @Override public Object call() throws Exception {
-                        while (!done.get()) {
-                            ldr.flush();
-
-                            U.sleep(100);
-                        }
-
-                        return null;
-                    }
-                }, 1, "flusher");
-
-                // Define index of node being restarted.
-                final int restartNodeIdx = nodesCntCache + nodesCntNoCache + 1;
-
-                IgniteInternalFuture<?> fut3 = multithreadedAsync(new Callable<Object>() {
-                    @Override public Object call() throws Exception {
-                        try {
-                            for (int i = 0; i < 5; i++) {
-                                Ignite g = startGrid(restartNodeIdx);
-
-                                UUID id = g.cluster().localNode().id();
-
-                                info(">>>>>>> Started node: " + id);
-
-                                U.sleep(1000);
-
-                                stopGrid(getTestGridName(restartNodeIdx), true);
-
-                                info(">>>>>>> Stopped node: " + id);
-                            }
-                        }
-                        finally {
-                            done.set(true);
-
-                            info("Start stop thread finished.");
-                        }
-
-                        return null;
-                    }
-                }, 1, "start-stop-thread");
-
-                fut1.get();
-                fut2.get();
-                fut3.get();
-            }
-            finally {
-                ldr.close(false);
-            }
-        }
-        finally {
-            stopAllGrids();
-        }
-    }
-
-    /**
-     * @throws Exception If failed.
-     */
-    public void testLoaderApi() throws Exception {
-        useCache = true;
-
-        try {
-            Ignite g1 = startGrid(1);
-
-            IgniteDataStreamer<Object, Object> ldr = g1.dataLoader(null);
-
-            ldr.close(false);
-
-            try {
-                ldr.addData(0, 0);
-
-                assert false;
-            }
-            catch (IllegalStateException e) {
-                info("Caught expected exception: " + e);
-            }
-
-            assert ldr.future().isDone();
-
-            ldr.future().get();
-
-            try {
-                // Create another loader.
-                ldr = g1.dataLoader("UNKNOWN_CACHE");
-
-                assert false;
-            }
-            catch (IllegalStateException e) {
-                info("Caught expected exception: " + e);
-            }
-
-            ldr.close(true);
-
-            assert ldr.future().isDone();
-
-            ldr.future().get();
-
-            // Create another loader.
-            ldr = g1.dataLoader(null);
-
-            // Cancel with future.
-            ldr.future().cancel();
-
-            try {
-                ldr.addData(0, 0);
-
-                assert false;
-            }
-            catch (IllegalStateException e) {
-                info("Caught expected exception: " + e);
-            }
-
-            assert ldr.future().isDone();
-
-            try {
-                ldr.future().get();
-
-                assert false;
-            }
-            catch (IgniteFutureCancelledException e) {
-                info("Caught expected exception: " + e);
-            }
-
-            // Create another loader.
-            ldr = g1.dataLoader(null);
-
-            // This will close loader.
-            stopGrid(getTestGridName(1), false);
-
-            try {
-                ldr.addData(0, 0);
-
-                assert false;
-            }
-            catch (IllegalStateException e) {
-                info("Caught expected exception: " + e);
-            }
-
-            assert ldr.future().isDone();
-
-            ldr.future().get();
-        }
-        finally {
-            stopAllGrids();
-        }
-    }
-
-    /**
-     * Wraps integer to closure returning it.
-     *
-     * @param i Value to wrap.
-     * @return Callable.
-     */
-    private static Callable<Integer> callable(@Nullable final Integer i) {
-        return new Callable<Integer>() {
-            @Override public Integer call() throws Exception {
-                return i;
-            }
-        };
-    }
-
-    /**
-     * Wraps integer to closure returning it.
-     *
-     * @param i Value to wrap.
-     * @return Closure.
-     */
-    private static IgniteClosure<Integer, Integer> closure(@Nullable final Integer i) {
-        return new IgniteClosure<Integer, Integer>() {
-            @Override public Integer apply(Integer e) {
-                return e == null ? i : e + i;
-            }
-        };
-    }
-
-    /**
-     * Wraps object to closure returning it.
-     *
-     * @param obj Value to wrap.
-     * @return Closure.
-     */
-    private static <T> IgniteClosure<T, T> fixedClosure(@Nullable final T obj) {
-        return new IgniteClosure<T, T>() {
-            @Override public T apply(T e) {
-                assert e == null || obj == null || e.getClass() == obj.getClass() :
-                    "Expects the same types [e=" + e + ", obj=" + obj + ']';
-
-                return obj;
-            }
-        };
-    }
-
-    /**
-     * Wraps integer to closure expecting it and returning {@code null}.
-     *
-     * @param exp Expected closure value.
-     * @return Remove expected cache value closure.
-     */
-    private static <T> IgniteClosure<T, T> removeClosure(@Nullable final T exp) {
-        return new IgniteClosure<T, T>() {
-            @Override public T apply(T act) {
-                if (exp == null ? act == null : exp.equals(act))
-                    return null;
-
-                throw new AssertionError("Unexpected value [exp=" + exp + ", act=" + act + ']');
-            }
-        };
-    }
-
-    /**
-     * @throws Exception If failed.
-     */
-    public void testFlush() throws Exception {
-        mode = LOCAL;
-
-        useCache = true;
-
-        try {
-            Ignite g = startGrid();
-
-            final IgniteCache<Integer, Integer> c = g.jcache(null);
-
-            final IgniteDataStreamer<Integer, Integer> ldr = g.dataLoader(null);
-
-            ldr.perNodeBufferSize(10);
-
-            for (int i = 0; i < 9; i++)
-                ldr.addData(i, i);
-
-            assertTrue(c.localSize() == 0);
-
-            multithreaded(new Callable<Void>() {
-                @Override
-                public Void call() throws Exception {
-                    ldr.flush();
-
-                    assertEquals(9, c.size());
-
-                    return null;
-                }
-            }, 5, "flush-checker");
-
-            ldr.addData(100, 100);
-
-            ldr.flush();
-
-            assertEquals(10, c.size());
-
-            ldr.addData(200, 200);
-
-            ldr.close(false);
-
-            ldr.future().get();
-
-            assertEquals(11, c.size());
-        }
-        finally {
-            stopAllGrids();
-        }
-    }
-
-    /**
-     * @throws Exception If failed.
-     */
-    public void testTryFlush() throws Exception {
-        mode = LOCAL;
-
-        useCache = true;
-
-        try {
-            Ignite g = startGrid();
-
-            IgniteCache<Integer, Integer> c = g.jcache(null);
-
-            IgniteDataStreamer<Integer, Integer> ldr = g.dataLoader(null);
-
-            ldr.perNodeBufferSize(10);
-
-            for (int i = 0; i < 9; i++)
-                ldr.addData(i, i);
-
-            assertTrue(c.localSize() == 0);
-
-            ldr.tryFlush();
-
-            Thread.sleep(100);
-
-            assertEquals(9, c.size());
-
-            ldr.close(false);
-        }
-        finally {
-            stopAllGrids();
-        }
-    }
-
-    /**
-     * @throws Exception If failed.
-     */
-    public void testFlushTimeout() throws Exception {
-        mode = LOCAL;
-
-        useCache = true;
-
-        try {
-            Ignite g = startGrid();
-
-            final CountDownLatch latch = new CountDownLatch(9);
-
-            g.events().localListen(new IgnitePredicate<Event>() {
-                @Override public boolean apply(Event evt) {
-                    latch.countDown();
-
-                    return true;
-                }
-            }, EVT_CACHE_OBJECT_PUT);
-
-            IgniteCache<Integer, Integer> c = g.jcache(null);
-
-            assertTrue(c.localSize() == 0);
-
-            IgniteDataStreamer<Integer, Integer> ldr = g.dataLoader(null);
-
-            ldr.perNodeBufferSize(10);
-            ldr.autoFlushFrequency(3000);
-            ldr.allowOverwrite(true);
-
-            for (int i = 0; i < 9; i++)
-                ldr.addData(i, i);
-
-            assertTrue(c.localSize() == 0);
-
-            assertFalse(latch.await(1000, MILLISECONDS));
-
-            assertTrue(c.localSize() == 0);
-
-            assertTrue(latch.await(3000, MILLISECONDS));
-
-            assertEquals(9, c.size());
-
-            ldr.close(false);
-        }
-        finally {
-            stopAllGrids();
-        }
-    }
-
-    /**
-     * @throws Exception If failed.
-     */
-    public void testUpdateStore() throws Exception {
-        storeMap = new ConcurrentHashMap<>();
-
-        try {
-            store = new TestStore();
-
-            useCache = true;
-
-            Ignite ignite = startGrid(1);
-
-            startGrid(2);
-            startGrid(3);
-
-            for (int i = 0; i < 1000; i++)
-                storeMap.put(i, i);
-
-            try (IgniteDataStreamer<Object, Object> ldr = ignite.dataLoader(null)) {
-                ldr.allowOverwrite(true);
-
-                assertFalse(ldr.skipStore());
-
-                for (int i = 0; i < 1000; i++)
-                    ldr.removeData(i);
-
-                for (int i = 1000; i < 2000; i++)
-                    ldr.addData(i, i);
-            }
-
-            for (int i = 0; i < 1000; i++)
-                assertNull(storeMap.get(i));
-
-            for (int i = 1000; i < 2000; i++)
-                assertEquals(i, storeMap.get(i));
-
-            try (IgniteDataStreamer<Object, Object> ldr = ignite.dataLoader(null)) {
-                ldr.allowOverwrite(true);
-
-                ldr.skipStore(true);
-
-                for (int i = 0; i < 1000; i++)
-                    ldr.addData(i, i);
-
-                for (int i = 1000; i < 2000; i++)
-                    ldr.removeData(i);
-            }
-
-            IgniteCache<Object, Object> cache = ignite.jcache(null);
-
-            for (int i = 0; i < 1000; i++) {
-                assertNull(storeMap.get(i));
-
-                assertEquals(i, cache.get(i));
-            }
-
-            for (int i = 1000; i < 2000; i++) {
-                assertEquals(i, storeMap.get(i));
-
-                assertNull(cache.localPeek(i, CachePeekMode.ONHEAP));
-            }
-        }
-        finally {
-            storeMap = null;
-        }
-    }
-
-    /**
-     *
-     */
-    private static class TestObject {
-        /** Value. */
-        private final int val;
-
-        /**
-         * @param val Value.
-         */
-        private TestObject(int val) {
-            this.val = val;
-        }
-
-        /** {@inheritDoc} */
-        @Override public boolean equals(Object o) {
-            if (this == o)
-                return true;
-
-            if (o == null || getClass() != o.getClass())
-                return false;
-
-            TestObject obj = (TestObject)o;
-
-            return val == obj.val;
-        }
-
-        /** {@inheritDoc} */
-        @Override public int hashCode() {
-            return val;
-        }
-    }
-
-    /**
-     *
-     */
-    private static class TestStore extends CacheStoreAdapter<Object, Object> {
-        /** {@inheritDoc} */
-        @Nullable @Override public Object load(Object key) {
-            return storeMap.get(key);
-        }
-
-        /** {@inheritDoc} */
-        @Override public void write(Cache.Entry<?, ?> entry) {
-            storeMap.put(entry.getKey(), entry.getValue());
-        }
-
-        /** {@inheritDoc} */
-        @Override public void delete(Object key) {
-            storeMap.remove(key);
-        }
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/36b7c8f7/modules/core/src/test/java/org/apache/ignite/internal/processors/dataload/IgniteDataStreamerProcessorSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/dataload/IgniteDataStreamerProcessorSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/dataload/IgniteDataStreamerProcessorSelfTest.java
new file mode 100644
index 0000000..393642d
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/dataload/IgniteDataStreamerProcessorSelfTest.java
@@ -0,0 +1,924 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.dataload;
+
+import org.apache.ignite.*;
+import org.apache.ignite.cache.*;
+import org.apache.ignite.cache.affinity.*;
+import org.apache.ignite.cache.store.*;
+import org.apache.ignite.cluster.*;
+import org.apache.ignite.configuration.*;
+import org.apache.ignite.events.*;
+import org.apache.ignite.internal.*;
+import org.apache.ignite.internal.processors.cache.*;
+import org.apache.ignite.internal.processors.cache.distributed.near.*;
+import org.apache.ignite.internal.util.typedef.internal.*;
+import org.apache.ignite.lang.*;
+import org.apache.ignite.marshaller.optimized.*;
+import org.apache.ignite.spi.discovery.tcp.*;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.*;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.*;
+import org.apache.ignite.testframework.junits.common.*;
+import org.jetbrains.annotations.*;
+
+import javax.cache.*;
+import javax.cache.configuration.*;
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.*;
+
+import static java.util.concurrent.TimeUnit.*;
+import static org.apache.ignite.cache.CacheAtomicityMode.*;
+import static org.apache.ignite.cache.CacheDistributionMode.*;
+import static org.apache.ignite.cache.CacheMode.*;
+import static org.apache.ignite.cache.CacheWriteSynchronizationMode.*;
+import static org.apache.ignite.events.EventType.*;
+
+/**
+ *
+ */
+public class IgniteDataStreamerProcessorSelfTest extends GridCommonAbstractTest {
+    /** */
+    private static ConcurrentHashMap<Object, Object> storeMap;
+
+    /** */
+    private TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true);
+
+    /** */
+    private CacheMode mode = PARTITIONED;
+
+    /** */
+    private boolean nearEnabled = true;
+
+    /** */
+    private boolean useCache;
+
+    /** */
+    private TestStore store;
+
+    /** {@inheritDoc} */
+    @Override public void afterTest() throws Exception {
+        super.afterTest();
+
+        useCache = false;
+    }
+
+    /** {@inheritDoc} */
+    @SuppressWarnings({"IfMayBeConditional", "unchecked"})
+    @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+        IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+        TcpDiscoverySpi spi = new TcpDiscoverySpi();
+
+        spi.setIpFinder(ipFinder);
+
+        cfg.setDiscoverySpi(spi);
+
+        cfg.setIncludeProperties();
+
+        cfg.setMarshaller(new OptimizedMarshaller(false));
+
+        if (useCache) {
+            CacheConfiguration cc = defaultCacheConfiguration();
+
+            cc.setCacheMode(mode);
+            cc.setAtomicityMode(TRANSACTIONAL);
+            cc.setDistributionMode(nearEnabled ? NEAR_PARTITIONED : PARTITIONED_ONLY);
+            cc.setWriteSynchronizationMode(FULL_SYNC);
+
+            cc.setEvictSynchronized(false);
+            cc.setEvictNearSynchronized(false);
+
+            if (store != null) {
+                cc.setCacheStoreFactory(new FactoryBuilder.SingletonFactory(store));
+                cc.setReadThrough(true);
+                cc.setWriteThrough(true);
+            }
+
+            cfg.setCacheConfiguration(cc);
+        }
+        else
+            cfg.setCacheConfiguration();
+
+        return cfg;
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testPartitioned() throws Exception {
+        mode = PARTITIONED;
+
+        checkDataLoader();
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testColocated() throws Exception {
+        mode = PARTITIONED;
+        nearEnabled = false;
+
+        checkDataLoader();
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testReplicated() throws Exception {
+        mode = REPLICATED;
+
+        checkDataLoader();
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testLocal() throws Exception {
+        mode = LOCAL;
+
+        try {
+            checkDataLoader();
+
+            assert false;
+        }
+        catch (IgniteCheckedException e) {
+            // Cannot load local cache configured remotely.
+            info("Caught expected exception: " + e);
+        }
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    @SuppressWarnings("ErrorNotRethrown")
+    private void checkDataLoader() throws Exception {
+        try {
+            Ignite g1 = startGrid(1);
+
+            useCache = true;
+
+            Ignite g2 = startGrid(2);
+            startGrid(3);
+
+            final IgniteDataStreamer<Integer, Integer> ldr = g1.dataLoader(null);
+
+            ldr.updater(GridDataLoadCacheUpdaters.<Integer, Integer>batchedSorted());
+
+            final AtomicInteger idxGen = new AtomicInteger();
+            final int cnt = 400;
+            final int threads = 10;
+
+            final CountDownLatch l1 = new CountDownLatch(threads);
+
+            IgniteInternalFuture<?> f1 = multithreadedAsync(new Callable<Object>() {
+                @Override public Object call() throws Exception {
+                    Collection<IgniteFuture<?>> futs = new ArrayList<>(cnt);
+
+                    for (int i = 0; i < cnt; i++) {
+                        int idx = idxGen.getAndIncrement();
+
+                        futs.add(ldr.addData(idx, idx));
+                    }
+
+                    l1.countDown();
+
+                    for (IgniteFuture<?> fut : futs)
+                        fut.get();
+
+                    return null;
+                }
+            }, threads);
+
+            l1.await();
+
+            // This will wait until data loader finishes loading.
+            stopGrid(getTestGridName(1), false);
+
+            f1.get();
+
+            int s2 = internalCache(2).primaryKeySet().size();
+            int s3 = internalCache(3).primaryKeySet().size();
+            int total = threads * cnt;
+
+            assertEquals(total, s2 + s3);
+
+            final IgniteDataStreamer<Integer, Integer> rmvLdr = g2.dataLoader(null);
+
+            rmvLdr.updater(GridDataLoadCacheUpdaters.<Integer, Integer>batchedSorted());
+
+            final CountDownLatch l2 = new CountDownLatch(threads);
+
+            IgniteInternalFuture<?> f2 = multithreadedAsync(new Callable<Object>() {
+                @Override public Object call() throws Exception {
+                    Collection<IgniteFuture<?>> futs = new ArrayList<>(cnt);
+
+                    for (int i = 0; i < cnt; i++) {
+                        final int key = idxGen.decrementAndGet();
+
+                        futs.add(rmvLdr.removeData(key));
+                    }
+
+                    l2.countDown();
+
+                    for (IgniteFuture<?> fut : futs)
+                        fut.get();
+
+                    return null;
+                }
+            }, threads);
+
+            l2.await();
+
+            rmvLdr.close(false);
+
+            f2.get();
+
+            s2 = internalCache(2).primaryKeySet().size();
+            s3 = internalCache(3).primaryKeySet().size();
+
+            assert s2 == 0 && s3 == 0 : "Incorrect entries count [s2=" + s2 + ", s3=" + s3 + ']';
+        }
+        finally {
+            stopAllGrids();
+        }
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testPartitionedIsolated() throws Exception {
+        mode = PARTITIONED;
+
+        checkIsolatedDataLoader();
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testReplicatedIsolated() throws Exception {
+        mode = REPLICATED;
+
+        checkIsolatedDataLoader();
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    private void checkIsolatedDataLoader() throws Exception {
+        try {
+            useCache = true;
+
+            Ignite g1 = startGrid(0);
+            startGrid(1);
+            startGrid(2);
+
+            awaitPartitionMapExchange();
+
+            GridCache<Integer, Integer> cache = ((IgniteKernal)grid(0)).cache(null);
+
+            for (int i = 0; i < 100; i++)
+                cache.put(i, -1);
+
+            final int cnt = 40_000;
+            final int threads = 10;
+
+            try (final IgniteDataStreamer<Integer, Integer> ldr = g1.dataLoader(null)) {
+                final AtomicInteger idxGen = new AtomicInteger();
+
+                IgniteInternalFuture<?> f1 = multithreadedAsync(new Callable<Object>() {
+                    @Override public Object call() throws Exception {
+                        for (int i = 0; i < cnt; i++) {
+                            int idx = idxGen.getAndIncrement();
+
+                            ldr.addData(idx, idx);
+                        }
+
+                        return null;
+                    }
+                }, threads);
+
+                f1.get();
+            }
+
+            for (int g = 0; g < 3; g++) {
+                ClusterNode locNode = grid(g).localNode();
+
+                GridCacheAdapter<Integer, Integer> cache0 = ((IgniteKernal)grid(g)).internalCache(null);
+
+                if (cache0.isNear())
+                    cache0 = ((GridNearCacheAdapter<Integer, Integer>)cache0).dht();
+
+                CacheAffinity<Integer> aff = cache0.affinity();
+
+                for (int key = 0; key < cnt * threads; key++) {
+                    if (aff.isPrimary(locNode, key) || aff.isBackup(locNode, key)) {
+                        GridCacheEntryEx<Integer, Integer> entry = cache0.peekEx(key);
+
+                        assertNotNull("Missing entry for key: " + key, entry);
+                        assertEquals((Integer)(key < 100 ? -1 : key), entry.rawGetOrUnmarshal(false));
+                    }
+                }
+            }
+        }
+        finally {
+            stopAllGrids();
+        }
+    }
+
+    /**
+     * Test primitive arrays can be passed into data loader.
+     *
+     * @throws Exception If failed.
+     */
+    public void testPrimitiveArrays() throws Exception {
+        try {
+            useCache = true;
+            mode = PARTITIONED;
+
+            Ignite g1 = startGrid(1);
+            startGrid(2); // Reproduced only for several nodes in topology (if marshalling is used).
+
+            List<Object> arrays = Arrays.<Object>asList(
+                new byte[] {1}, new boolean[] {true, false}, new char[] {2, 3}, new short[] {3, 4},
+                new int[] {4, 5}, new long[] {5, 6}, new float[] {6, 7}, new double[] {7, 8});
+
+            IgniteDataStreamer<Object, Object> dataLdr = g1.dataLoader(null);
+
+            for (int i = 0, size = arrays.size(); i < 1000; i++) {
+                Object arr = arrays.get(i % size);
+
+                dataLdr.addData(i, arr);
+                dataLdr.addData(i, fixedClosure(arr));
+            }
+
+            dataLdr.close(false);
+        }
+        finally {
+            stopAllGrids();
+        }
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testReplicatedMultiThreaded() throws Exception {
+        mode = REPLICATED;
+
+        checkLoaderMultithreaded(1, 2);
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testPartitionedMultiThreaded() throws Exception {
+        mode = PARTITIONED;
+
+        checkLoaderMultithreaded(1, 3);
+    }
+
+    /**
+     * Tests loader in multithreaded environment with various count of grids started.
+     *
+     * @param nodesCntNoCache How many nodes should be started without cache.
+     * @param nodesCntCache How many nodes should be started with cache.
+     * @throws Exception If failed.
+     */
+    protected void checkLoaderMultithreaded(int nodesCntNoCache, int nodesCntCache)
+        throws Exception {
+        try {
+            // Start all required nodes.
+            int idx = 1;
+
+            for (int i = 0; i < nodesCntNoCache; i++)
+                startGrid(idx++);
+
+            useCache = true;
+
+            for (int i = 0; i < nodesCntCache; i++)
+                startGrid(idx++);
+
+            Ignite g1 = grid(1);
+
+            // Get and configure loader.
+            final IgniteDataStreamer<Integer, Integer> ldr = g1.dataLoader(null);
+
+            ldr.updater(GridDataLoadCacheUpdaters.<Integer, Integer>individual());
+            ldr.perNodeBufferSize(2);
+
+            // Define count of puts.
+            final AtomicInteger idxGen = new AtomicInteger();
+
+            final AtomicBoolean done = new AtomicBoolean();
+
+            try {
+                final int totalPutCnt = 50000;
+
+                IgniteInternalFuture<?> fut1 = multithreadedAsync(new Callable<Object>() {
+                    @Override public Object call() throws Exception {
+                        Collection<IgniteFuture<?>> futs = new ArrayList<>();
+
+                        while (!done.get()) {
+                            int idx = idxGen.getAndIncrement();
+
+                            if (idx >= totalPutCnt) {
+                                info(">>> Stopping producer thread since maximum count of puts reached.");
+
+                                break;
+                            }
+
+                            futs.add(ldr.addData(idx, idx));
+                        }
+
+                        ldr.flush();
+
+                        for (IgniteFuture<?> fut : futs)
+                            fut.get();
+
+                        return null;
+                    }
+                }, 5, "producer");
+
+                IgniteInternalFuture<?> fut2 = multithreadedAsync(new Callable<Object>() {
+                    @Override public Object call() throws Exception {
+                        while (!done.get()) {
+                            ldr.flush();
+
+                            U.sleep(100);
+                        }
+
+                        return null;
+                    }
+                }, 1, "flusher");
+
+                // Define index of node being restarted.
+                final int restartNodeIdx = nodesCntCache + nodesCntNoCache + 1;
+
+                IgniteInternalFuture<?> fut3 = multithreadedAsync(new Callable<Object>() {
+                    @Override public Object call() throws Exception {
+                        try {
+                            for (int i = 0; i < 5; i++) {
+                                Ignite g = startGrid(restartNodeIdx);
+
+                                UUID id = g.cluster().localNode().id();
+
+                                info(">>>>>>> Started node: " + id);
+
+                                U.sleep(1000);
+
+                                stopGrid(getTestGridName(restartNodeIdx), true);
+
+                                info(">>>>>>> Stopped node: " + id);
+                            }
+                        }
+                        finally {
+                            done.set(true);
+
+                            info("Start stop thread finished.");
+                        }
+
+                        return null;
+                    }
+                }, 1, "start-stop-thread");
+
+                fut1.get();
+                fut2.get();
+                fut3.get();
+            }
+            finally {
+                ldr.close(false);
+            }
+        }
+        finally {
+            stopAllGrids();
+        }
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testLoaderApi() throws Exception {
+        useCache = true;
+
+        try {
+            Ignite g1 = startGrid(1);
+
+            IgniteDataStreamer<Object, Object> ldr = g1.dataLoader(null);
+
+            ldr.close(false);
+
+            try {
+                ldr.addData(0, 0);
+
+                assert false;
+            }
+            catch (IllegalStateException e) {
+                info("Caught expected exception: " + e);
+            }
+
+            assert ldr.future().isDone();
+
+            ldr.future().get();
+
+            try {
+                // Create another loader.
+                ldr = g1.dataLoader("UNKNOWN_CACHE");
+
+                assert false;
+            }
+            catch (IllegalStateException e) {
+                info("Caught expected exception: " + e);
+            }
+
+            ldr.close(true);
+
+            assert ldr.future().isDone();
+
+            ldr.future().get();
+
+            // Create another loader.
+            ldr = g1.dataLoader(null);
+
+            // Cancel with future.
+            ldr.future().cancel();
+
+            try {
+                ldr.addData(0, 0);
+
+                assert false;
+            }
+            catch (IllegalStateException e) {
+                info("Caught expected exception: " + e);
+            }
+
+            assert ldr.future().isDone();
+
+            try {
+                ldr.future().get();
+
+                assert false;
+            }
+            catch (IgniteFutureCancelledException e) {
+                info("Caught expected exception: " + e);
+            }
+
+            // Create another loader.
+            ldr = g1.dataLoader(null);
+
+            // This will close loader.
+            stopGrid(getTestGridName(1), false);
+
+            try {
+                ldr.addData(0, 0);
+
+                assert false;
+            }
+            catch (IllegalStateException e) {
+                info("Caught expected exception: " + e);
+            }
+
+            assert ldr.future().isDone();
+
+            ldr.future().get();
+        }
+        finally {
+            stopAllGrids();
+        }
+    }
+
+    /**
+     * Wraps integer to closure returning it.
+     *
+     * @param i Value to wrap.
+     * @return Callable.
+     */
+    private static Callable<Integer> callable(@Nullable final Integer i) {
+        return new Callable<Integer>() {
+            @Override public Integer call() throws Exception {
+                return i;
+            }
+        };
+    }
+
+    /**
+     * Wraps integer to closure returning it.
+     *
+     * @param i Value to wrap.
+     * @return Closure.
+     */
+    private static IgniteClosure<Integer, Integer> closure(@Nullable final Integer i) {
+        return new IgniteClosure<Integer, Integer>() {
+            @Override public Integer apply(Integer e) {
+                return e == null ? i : e + i;
+            }
+        };
+    }
+
+    /**
+     * Wraps object to closure returning it.
+     *
+     * @param obj Value to wrap.
+     * @return Closure.
+     */
+    private static <T> IgniteClosure<T, T> fixedClosure(@Nullable final T obj) {
+        return new IgniteClosure<T, T>() {
+            @Override public T apply(T e) {
+                assert e == null || obj == null || e.getClass() == obj.getClass() :
+                    "Expects the same types [e=" + e + ", obj=" + obj + ']';
+
+                return obj;
+            }
+        };
+    }
+
+    /**
+     * Wraps integer to closure expecting it and returning {@code null}.
+     *
+     * @param exp Expected closure value.
+     * @return Remove expected cache value closure.
+     */
+    private static <T> IgniteClosure<T, T> removeClosure(@Nullable final T exp) {
+        return new IgniteClosure<T, T>() {
+            @Override public T apply(T act) {
+                if (exp == null ? act == null : exp.equals(act))
+                    return null;
+
+                throw new AssertionError("Unexpected value [exp=" + exp + ", act=" + act + ']');
+            }
+        };
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testFlush() throws Exception {
+        mode = LOCAL;
+
+        useCache = true;
+
+        try {
+            Ignite g = startGrid();
+
+            final IgniteCache<Integer, Integer> c = g.jcache(null);
+
+            final IgniteDataStreamer<Integer, Integer> ldr = g.dataLoader(null);
+
+            ldr.perNodeBufferSize(10);
+
+            for (int i = 0; i < 9; i++)
+                ldr.addData(i, i);
+
+            assertTrue(c.localSize() == 0);
+
+            multithreaded(new Callable<Void>() {
+                @Override
+                public Void call() throws Exception {
+                    ldr.flush();
+
+                    assertEquals(9, c.size());
+
+                    return null;
+                }
+            }, 5, "flush-checker");
+
+            ldr.addData(100, 100);
+
+            ldr.flush();
+
+            assertEquals(10, c.size());
+
+            ldr.addData(200, 200);
+
+            ldr.close(false);
+
+            ldr.future().get();
+
+            assertEquals(11, c.size());
+        }
+        finally {
+            stopAllGrids();
+        }
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testTryFlush() throws Exception {
+        mode = LOCAL;
+
+        useCache = true;
+
+        try {
+            Ignite g = startGrid();
+
+            IgniteCache<Integer, Integer> c = g.jcache(null);
+
+            IgniteDataStreamer<Integer, Integer> ldr = g.dataLoader(null);
+
+            ldr.perNodeBufferSize(10);
+
+            for (int i = 0; i < 9; i++)
+                ldr.addData(i, i);
+
+            assertTrue(c.localSize() == 0);
+
+            ldr.tryFlush();
+
+            Thread.sleep(100);
+
+            assertEquals(9, c.size());
+
+            ldr.close(false);
+        }
+        finally {
+            stopAllGrids();
+        }
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testFlushTimeout() throws Exception {
+        mode = LOCAL;
+
+        useCache = true;
+
+        try {
+            Ignite g = startGrid();
+
+            final CountDownLatch latch = new CountDownLatch(9);
+
+            g.events().localListen(new IgnitePredicate<Event>() {
+                @Override public boolean apply(Event evt) {
+                    latch.countDown();
+
+                    return true;
+                }
+            }, EVT_CACHE_OBJECT_PUT);
+
+            IgniteCache<Integer, Integer> c = g.jcache(null);
+
+            assertTrue(c.localSize() == 0);
+
+            IgniteDataStreamer<Integer, Integer> ldr = g.dataLoader(null);
+
+            ldr.perNodeBufferSize(10);
+            ldr.autoFlushFrequency(3000);
+            ldr.allowOverwrite(true);
+
+            for (int i = 0; i < 9; i++)
+                ldr.addData(i, i);
+
+            assertTrue(c.localSize() == 0);
+
+            assertFalse(latch.await(1000, MILLISECONDS));
+
+            assertTrue(c.localSize() == 0);
+
+            assertTrue(latch.await(3000, MILLISECONDS));
+
+            assertEquals(9, c.size());
+
+            ldr.close(false);
+        }
+        finally {
+            stopAllGrids();
+        }
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testUpdateStore() throws Exception {
+        storeMap = new ConcurrentHashMap<>();
+
+        try {
+            store = new TestStore();
+
+            useCache = true;
+
+            Ignite ignite = startGrid(1);
+
+            startGrid(2);
+            startGrid(3);
+
+            for (int i = 0; i < 1000; i++)
+                storeMap.put(i, i);
+
+            try (IgniteDataStreamer<Object, Object> ldr = ignite.dataLoader(null)) {
+                ldr.allowOverwrite(true);
+
+                assertFalse(ldr.skipStore());
+
+                for (int i = 0; i < 1000; i++)
+                    ldr.removeData(i);
+
+                for (int i = 1000; i < 2000; i++)
+                    ldr.addData(i, i);
+            }
+
+            for (int i = 0; i < 1000; i++)
+                assertNull(storeMap.get(i));
+
+            for (int i = 1000; i < 2000; i++)
+                assertEquals(i, storeMap.get(i));
+
+            try (IgniteDataStreamer<Object, Object> ldr = ignite.dataLoader(null)) {
+                ldr.allowOverwrite(true);
+
+                ldr.skipStore(true);
+
+                for (int i = 0; i < 1000; i++)
+                    ldr.addData(i, i);
+
+                for (int i = 1000; i < 2000; i++)
+                    ldr.removeData(i);
+            }
+
+            IgniteCache<Object, Object> cache = ignite.jcache(null);
+
+            for (int i = 0; i < 1000; i++) {
+                assertNull(storeMap.get(i));
+
+                assertEquals(i, cache.get(i));
+            }
+
+            for (int i = 1000; i < 2000; i++) {
+                assertEquals(i, storeMap.get(i));
+
+                assertNull(cache.localPeek(i, CachePeekMode.ONHEAP));
+            }
+        }
+        finally {
+            storeMap = null;
+        }
+    }
+
+    /**
+     *
+     */
+    private static class TestObject {
+        /** Value. */
+        private final int val;
+
+        /**
+         * @param val Value.
+         */
+        private TestObject(int val) {
+            this.val = val;
+        }
+
+        /** {@inheritDoc} */
+        @Override public boolean equals(Object o) {
+            if (this == o)
+                return true;
+
+            if (o == null || getClass() != o.getClass())
+                return false;
+
+            TestObject obj = (TestObject)o;
+
+            return val == obj.val;
+        }
+
+        /** {@inheritDoc} */
+        @Override public int hashCode() {
+            return val;
+        }
+    }
+
+    /**
+     *
+     */
+    private static class TestStore extends CacheStoreAdapter<Object, Object> {
+        /** {@inheritDoc} */
+        @Nullable @Override public Object load(Object key) {
+            return storeMap.get(key);
+        }
+
+        /** {@inheritDoc} */
+        @Override public void write(Cache.Entry<?, ?> entry) {
+            storeMap.put(entry.getKey(), entry.getValue());
+        }
+
+        /** {@inheritDoc} */
+        @Override public void delete(Object key) {
+            storeMap.remove(key);
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/36b7c8f7/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite.java
index 24c8e04..a30eea5 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite.java
@@ -109,7 +109,7 @@ public class IgniteCacheTestSuite extends TestSuite {
         suite.addTestSuite(GridCacheBalancingStoreSelfTest.class);
         suite.addTestSuite(GridCacheAffinityApiSelfTest.class);
         suite.addTestSuite(GridCacheStoreValueBytesSelfTest.class);
-        suite.addTestSuite(GridDataLoaderProcessorSelfTest.class);
+        suite.addTestSuite(IgniteDataStreamerProcessorSelfTest.class);
         suite.addTestSuite(GridDataLoaderImplSelfTest.class);
         suite.addTestSuite(GridCacheEntryMemorySizeSelfTest.class);
         suite.addTestSuite(GridCacheClearAllSelfTest.class);


Mime
View raw message