ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject [2/3] ignite git commit: ignite-4705
Date Wed, 22 Feb 2017 11:46:49 GMT
ignite-4705


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

Branch: refs/heads/ignite-4705
Commit: 586de83cd6b912cb02add7cdaed8ca8616633b6d
Parents: 11e8776
Author: sboikov <sboikov@gridgain.com>
Authored: Wed Feb 22 10:59:53 2017 +0300
Committer: sboikov <sboikov@gridgain.com>
Committed: Wed Feb 22 14:26:50 2017 +0300

----------------------------------------------------------------------
 .../internal/binary/BinaryObjectImpl.java       |  43 ++++---
 .../processors/cache/GridCacheIoManager.java    |   3 +
 .../GridCachePartitionExchangeManager.java      |  10 +-
 .../processors/cache/KeyCacheObjectImpl.java    |  66 +++++++++--
 .../GridDhtAtomicAbstractUpdateFuture.java      | 104 ++++++++--------
 .../GridDhtAtomicAbstractUpdateRequest.java     |  24 ++--
 .../dht/atomic/GridDhtAtomicCache.java          |  74 +++++++-----
 .../dht/atomic/GridDhtAtomicNearResponse.java   |   7 +-
 .../atomic/GridDhtAtomicSingleUpdateFuture.java |  61 +---------
 .../GridDhtAtomicSingleUpdateRequest.java       |  67 ++++-------
 .../dht/atomic/GridDhtAtomicUpdateFuture.java   |  67 +----------
 .../dht/atomic/GridDhtAtomicUpdateRequest.java  |  33 ++----
 .../GridNearAtomicSingleUpdateFuture.java       |  39 +++---
 .../dht/atomic/GridNearAtomicUpdateFuture.java  | 118 ++++++++++++++-----
 .../IgniteCacheObjectProcessorImpl.java         |   4 +-
 .../util/future/GridCompoundFuture.java         |  11 +-
 ...eCacheContinuousQueryImmutableEntryTest.java |   2 +-
 .../file/GridFileSwapSpaceSpiSelfTest.java      |   2 +-
 .../IgniteCacheFullApiSelfTestSuite.java        |   5 +-
 19 files changed, 374 insertions(+), 366 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/ignite/blob/586de83c/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryObjectImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryObjectImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryObjectImpl.java
index 7a81659..6fe1a3b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryObjectImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/binary/BinaryObjectImpl.java
@@ -17,6 +17,17 @@
 
 package org.apache.ignite.internal.binary;
 
+import java.io.Externalizable;
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.nio.ByteBuffer;
+import java.sql.Time;
+import java.sql.Timestamp;
+import java.util.Date;
+import java.util.UUID;
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.binary.BinaryObject;
 import org.apache.ignite.binary.BinaryObjectException;
@@ -33,19 +44,7 @@ import org.apache.ignite.plugin.extensions.communication.MessageReader;
 import org.apache.ignite.plugin.extensions.communication.MessageWriter;
 import org.jetbrains.annotations.Nullable;
 
-import java.io.Externalizable;
-import java.io.IOException;
-import java.io.ObjectInput;
-import java.io.ObjectOutput;
-import java.math.BigDecimal;
-import java.math.BigInteger;
-import java.nio.ByteBuffer;
-import java.sql.Time;
-import java.sql.Timestamp;
-import java.util.Date;
-import java.util.UUID;
-
-import static java.nio.charset.StandardCharsets.*;
+import static java.nio.charset.StandardCharsets.UTF_8;
 
 /**
  * Binary object implementation.
@@ -74,7 +73,6 @@ public final class BinaryObjectImpl extends BinaryObjectExImpl implements Extern
     private boolean detachAllowed;
 
     /** */
-    @GridDirectTransient
     private int part = -1;
 
     /**
@@ -561,7 +559,6 @@ public final class BinaryObjectImpl extends BinaryObjectExImpl implements Extern
 
         start = in.readInt();
     }
-
     /** {@inheritDoc} */
     @Override public boolean writeTo(ByteBuffer buf, MessageWriter writer) {
         writer.setBuffer(buf);
@@ -584,6 +581,12 @@ public final class BinaryObjectImpl extends BinaryObjectExImpl implements Extern
                 writer.incrementState();
 
             case 1:
+                if (!writer.writeInt("part", part))
+                    return false;
+
+                writer.incrementState();
+
+            case 2:
                 if (!writer.writeInt("start", detachAllowed ? 0 : start))
                     return false;
 
@@ -611,6 +614,14 @@ public final class BinaryObjectImpl extends BinaryObjectExImpl implements Extern
                 reader.incrementState();
 
             case 1:
+                part = reader.readInt("part");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+            case 2:
                 start = reader.readInt("start");
 
                 if (!reader.isLastRead())
@@ -620,7 +631,7 @@ public final class BinaryObjectImpl extends BinaryObjectExImpl implements Extern
 
         }
 
-        return true;
+        return reader.afterMessageRead(BinaryObjectImpl.class);
     }
 
     /** {@inheritDoc} */

http://git-wip-us.apache.org/repos/asf/ignite/blob/586de83c/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java
index 1d3b714..c46b01a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheIoManager.java
@@ -366,6 +366,9 @@ public class GridCacheIoManager extends GridCacheSharedManagerAdapter {
             if (depEnabled)
                 cctx.deploy().ignoreOwnership(true);
 
+//            if (!cacheMsg.partitionExchangeMessage())
+//                log.info("Message [from=" + nodeId + ", msg=" + cacheMsg + ']');
+
             unmarshall(nodeId, cacheMsg);
 
             if (cacheMsg.classError() != null)

http://git-wip-us.apache.org/repos/asf/ignite/blob/586de83c/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
index e44f4a8..f14f612 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
@@ -1349,6 +1349,8 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
         if (!readyFuts.isEmpty()) {
             U.warn(log, "Pending affinity ready futures:");
 
+            int cnt = 0;
+
             for (AffinityReadyFuture fut : readyFuts.values())
                 U.warn(log, ">>> " + fut);
         }
@@ -1508,9 +1510,15 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
 
             U.warn(log, "Pending atomic cache futures:");
 
-            for (GridCacheFuture<?> fut : mvcc.atomicFutures())
+            int cnt = 0;
+
+            for (GridCacheFuture<?> fut : mvcc.atomicFutures()) {
                 U.warn(log, ">>> " + fut);
 
+                if (cnt++ >= 10)
+                    break;
+            }
+
             U.warn(log, "Pending data streamer futures:");
 
             for (IgniteInternalFuture<?> fut : mvcc.dataStreamerFutures())

http://git-wip-us.apache.org/repos/asf/ignite/blob/586de83c/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/KeyCacheObjectImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/KeyCacheObjectImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/KeyCacheObjectImpl.java
index 146e554..eb305bf 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/KeyCacheObjectImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/KeyCacheObjectImpl.java
@@ -17,8 +17,10 @@
 
 package org.apache.ignite.internal.processors.cache;
 
+import java.nio.ByteBuffer;
 import org.apache.ignite.IgniteCheckedException;
-import org.apache.ignite.internal.GridDirectTransient;
+import org.apache.ignite.plugin.extensions.communication.MessageReader;
+import org.apache.ignite.plugin.extensions.communication.MessageWriter;
 import org.jetbrains.annotations.Nullable;
 
 /**
@@ -29,7 +31,6 @@ public class KeyCacheObjectImpl extends CacheObjectAdapter implements KeyCacheOb
     private static final long serialVersionUID = 0L;
 
     /** */
-    @GridDirectTransient
     private int part = -1;
 
     /**
@@ -42,18 +43,11 @@ public class KeyCacheObjectImpl extends CacheObjectAdapter implements KeyCacheOb
     /**
      * @param val Value.
      * @param valBytes Value bytes.
-     */
-    public KeyCacheObjectImpl(Object val, byte[] valBytes) {
-        this(val, valBytes, -1);
-    }
-
-    /**
-     * @param val Value.
-     * @param valBytes Value bytes.
      * @param part Partition.
      */
     public KeyCacheObjectImpl(Object val, byte[] valBytes, int part) {
         assert val != null;
+        assert part >= 0 : part;
 
         this.val = val;
         this.valBytes = valBytes;
@@ -130,7 +124,57 @@ public class KeyCacheObjectImpl extends CacheObjectAdapter implements KeyCacheOb
 
     /** {@inheritDoc} */
     @Override public byte fieldsCount() {
-        return 1;
+        return 2;
+    }
+
+    /** {@inheritDoc} */
+    @Override public boolean readFrom(ByteBuffer buf, MessageReader reader) {
+        reader.setBuffer(buf);
+
+        if (!reader.beforeMessageRead())
+            return false;
+
+        if (!super.readFrom(buf, reader))
+            return false;
+
+        switch (reader.state()) {
+            case 1:
+                part = reader.readInt("part");
+
+                if (!reader.isLastRead())
+                    return false;
+
+                reader.incrementState();
+
+        }
+
+        return reader.afterMessageRead(KeyCacheObjectImpl.class);
+    }
+
+    /** {@inheritDoc} */
+    @Override public boolean writeTo(ByteBuffer buf, MessageWriter writer) {
+        writer.setBuffer(buf);
+
+        if (!super.writeTo(buf, writer))
+            return false;
+
+        if (!writer.isHeaderWritten()) {
+            if (!writer.writeHeader(directType(), fieldsCount()))
+                return false;
+
+            writer.onHeaderWritten();
+        }
+
+        switch (writer.state()) {
+            case 1:
+                if (!writer.writeInt("part", part))
+                    return false;
+
+                writer.incrementState();
+
+        }
+
+        return true;
     }
 
     /** {@inheritDoc} */

http://git-wip-us.apache.org/repos/asf/ignite/blob/586de83c/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicAbstractUpdateFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicAbstractUpdateFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicAbstractUpdateFuture.java
index 79fb7fc..3f7e28f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicAbstractUpdateFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicAbstractUpdateFuture.java
@@ -40,10 +40,13 @@ import org.apache.ignite.internal.processors.cache.KeyCacheObject;
 import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtCacheEntry;
 import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
 import org.apache.ignite.internal.util.future.GridFutureAdapter;
+import org.apache.ignite.internal.util.tostring.GridToStringExclude;
 import org.apache.ignite.internal.util.tostring.GridToStringInclude;
 import org.apache.ignite.internal.util.typedef.CI1;
 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.lang.IgniteClosure;
 import org.apache.ignite.lang.IgniteUuid;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
@@ -80,11 +83,8 @@ public abstract class GridDhtAtomicAbstractUpdateFuture extends GridFutureAdapte
     /** Update request. */
     final GridNearAtomicAbstractUpdateRequest updateReq;
 
-    /** Update response. */
-    final GridNearAtomicUpdateResponse updateRes;
-
     /** Mappings. */
-    @GridToStringInclude
+    @GridToStringExclude
     protected Map<UUID, GridDhtAtomicAbstractUpdateRequest> mappings;
 
     /** Continuous query closures. */
@@ -100,18 +100,15 @@ public abstract class GridDhtAtomicAbstractUpdateFuture extends GridFutureAdapte
      * @param cctx Cache context.
      * @param writeVer Write version.
      * @param updateReq Update request.
-     * @param updateRes Update response.
      */
     protected GridDhtAtomicAbstractUpdateFuture(
         GridCacheContext cctx,
         GridCacheVersion writeVer,
-        GridNearAtomicAbstractUpdateRequest updateReq,
-        GridNearAtomicUpdateResponse updateRes
+        GridNearAtomicAbstractUpdateRequest updateReq
     ) {
         this.cctx = cctx;
 
         this.updateReq = updateReq;
-        this.updateRes = updateRes;
         this.writeVer = writeVer;
 
         futId = cctx.mvcc().atomicFutureId();
@@ -145,6 +142,7 @@ public abstract class GridDhtAtomicAbstractUpdateFuture extends GridFutureAdapte
     }
 
     /**
+     * @param nearNodeId Near node ID.
      * @param entry Entry to map.
      * @param val Value to write.
      * @param entryProcessor Entry processor.
@@ -208,7 +206,6 @@ public abstract class GridDhtAtomicAbstractUpdateFuture extends GridFutureAdapte
                     conflictExpireTime,
                     conflictVer,
                     addPrevVal,
-                    entry.partition(),
                     prevVal,
                     updateCntr);
             }
@@ -274,8 +271,6 @@ public abstract class GridDhtAtomicAbstractUpdateFuture extends GridFutureAdapte
                 mappings.put(nodeId, updateReq);
             }
 
-            addNearReaderEntry(entry);
-
             updateReq.addNearWriteValue(entry.key(),
                 val,
                 entryProcessor,
@@ -284,20 +279,6 @@ public abstract class GridDhtAtomicAbstractUpdateFuture extends GridFutureAdapte
         }
     }
 
-    /**
-     * adds new nearReader.
-     *
-     * @param entry GridDhtCacheEntry.
-     */
-    protected abstract void addNearReaderEntry(GridDhtCacheEntry entry);
-
-    /**
-     * @return Write version.
-     */
-    final GridCacheVersion writeVersion() {
-        return writeVer;
-    }
-
     /** {@inheritDoc} */
     @Override public final IgniteUuid futureId() {
         throw new UnsupportedOperationException();
@@ -308,6 +289,13 @@ public abstract class GridDhtAtomicAbstractUpdateFuture extends GridFutureAdapte
         return futId;
     }
 
+    /**
+     * @return Write version.
+     */
+    final GridCacheVersion writeVersion() {
+        return writeVer;
+    }
+
     /** {@inheritDoc} */
     @Override public final boolean onNodeLeft(UUID nodeId) {
         boolean res = registerResponse(nodeId, true);
@@ -325,7 +313,7 @@ public abstract class GridDhtAtomicAbstractUpdateFuture extends GridFutureAdapte
      * @param nodeErr Node error flag.
      * @return {@code True} if request found.
      */
-    final boolean registerResponse(UUID nodeId, boolean nodeErr) {
+    private boolean registerResponse(UUID nodeId, boolean nodeErr) {
         int resCnt0;
 
         GridDhtAtomicAbstractUpdateRequest req = mappings != null ? mappings.get(nodeId) : null;
@@ -348,6 +336,9 @@ public abstract class GridDhtAtomicAbstractUpdateFuture extends GridFutureAdapte
                     return false;
             }
 
+            if (resCnt0 == mappings.size())
+                onDone();
+
             if (needReplyToNear) {
                 assert !F.isEmpty(mappings);
 
@@ -393,9 +384,6 @@ public abstract class GridDhtAtomicAbstractUpdateFuture extends GridFutureAdapte
                 }
             }
 
-            if (resCnt0 == mappings.size())
-                onDone();
-
             return true;
         }
 
@@ -405,12 +393,16 @@ public abstract class GridDhtAtomicAbstractUpdateFuture extends GridFutureAdapte
     /**
      * Sends requests to remote nodes.
      *
+     * @param updateRes Response.
      * @param completionCb Callback to invoke to send response to near node.
      * @param ret Cache operation return value.
      */
-    final void map(GridDhtAtomicCache.UpdateReplyClosure completionCb, GridCacheReturn ret) {
+    final void map(GridNearAtomicUpdateResponse updateRes,
+        GridDhtAtomicCache.UpdateReplyClosure completionCb,
+        GridCacheReturn ret) {
         boolean fullSync = updateReq.writeSynchronizationMode() == FULL_SYNC;
-        repliedToNear = updateReq.writeSynchronizationMode() == PRIMARY_SYNC || ret.hasValue();
+        boolean needReplyToNear = repliedToNear = updateReq.writeSynchronizationMode() == PRIMARY_SYNC ||
+            ret.hasValue() || updateReq.nodeId().equals(cctx.localNodeId());
 
         List<UUID> dhtNodes = null;
 
@@ -423,14 +415,14 @@ public abstract class GridDhtAtomicAbstractUpdateFuture extends GridFutureAdapte
             else
                 dhtNodes = Collections.emptyList();
 
-            if (repliedToNear)
+            if (needReplyToNear)
                 updateRes.mapping(dhtNodes);
         }
 
         if (!F.isEmpty(mappings)) {
-            sendDhtRequests(fullSync && !repliedToNear, dhtNodes, ret);
+            sendDhtRequests(fullSync && !needReplyToNear, dhtNodes, ret);
 
-            if (repliedToNear)
+            if (needReplyToNear)
                 completionCb.apply(updateReq, updateRes);
             else {
                 if (fullSync && GridDhtAtomicCache.IGNITE_ATOMIC_SND_MAPPING_TO_NEAR) {
@@ -472,6 +464,8 @@ public abstract class GridDhtAtomicAbstractUpdateFuture extends GridFutureAdapte
                         req.setResult(ret.success());
                 }
 
+                assert !cctx.localNodeId().equals(req.nodeId()) : req;
+
                 cctx.io().send(req.nodeId(), req, cctx.ioPolicy());
 
                 if (msgLog.isDebugEnabled()) {
@@ -497,6 +491,14 @@ public abstract class GridDhtAtomicAbstractUpdateFuture extends GridFutureAdapte
     }
 
     /**
+     * @param nodeId Node ID.
+     * @param res Response.
+     */
+    public final void onDhtErrorResponse(UUID nodeId, GridDhtAtomicUpdateResponse res) {
+        // TODO IGNITE-4705.
+    }
+
+    /**
      * Deferred update response.
      *
      * @param nodeId Backup node ID.
@@ -532,20 +534,6 @@ public abstract class GridDhtAtomicAbstractUpdateFuture extends GridFutureAdapte
         @Nullable GridCacheVersion conflictVer
     );
 
-    /**
-     * Callback for backup update response.
-     *
-     * @param nodeId Backup node ID.
-     * @param updateRes Update response.
-     */
-    public abstract void onResult(UUID nodeId, GridDhtAtomicUpdateResponse updateRes);
-
-    /**
-     * @param updateRes Response.
-     * @param err Error.
-     */
-    protected abstract void addFailedKeys(GridNearAtomicUpdateResponse updateRes, Throwable err);
-
     /** {@inheritDoc} */
     @Override public final boolean onDone(@Nullable Void res, @Nullable Throwable err) {
         if (super.onDone(res, err)) {
@@ -553,9 +541,6 @@ public abstract class GridDhtAtomicAbstractUpdateFuture extends GridFutureAdapte
 
             boolean suc = err == null;
 
-            if (!suc)
-                addFailedKeys(updateRes, err);
-
             if (cntQryClsrs != null) {
                 for (CI1<Boolean> clsr : cntQryClsrs)
                     clsr.apply(suc);
@@ -576,4 +561,21 @@ public abstract class GridDhtAtomicAbstractUpdateFuture extends GridFutureAdapte
     @Override public void markNotTrackable() {
         // No-op.
     }
+
+    /** {@inheritDoc} */
+    @Override public String toString() {
+        synchronized (this) {
+            Map<UUID, String> dhtRes = F.viewReadOnly(mappings,
+                new IgniteClosure<GridDhtAtomicAbstractUpdateRequest, String>() {
+                    @Override public String apply(GridDhtAtomicAbstractUpdateRequest req) {
+                        return "[res" + req.hasResponse() +
+                            ", size=" + req.size() +
+                            ", nearSize=" + req.nearSize() + ']';
+                    }
+                }
+            );
+
+            return S.toString(GridDhtAtomicAbstractUpdateFuture.class, this, "dhtRes", dhtRes);
+        }
+    }
 }

http://git-wip-us.apache.org/repos/asf/ignite/blob/586de83c/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicAbstractUpdateRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicAbstractUpdateRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicAbstractUpdateRequest.java
index 1841a49..9bc4f81 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicAbstractUpdateRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicAbstractUpdateRequest.java
@@ -42,22 +42,22 @@ import org.jetbrains.annotations.Nullable;
  */
 public abstract class GridDhtAtomicAbstractUpdateRequest extends GridCacheMessage implements GridCacheDeployable {
     /** Skip store flag bit mask. */
-    public static final int DHT_ATOMIC_SKIP_STORE_FLAG_MASK = 0x01;
+    static final int DHT_ATOMIC_SKIP_STORE_FLAG_MASK = 0x01;
 
     /** Keep binary flag. */
-    public static final int DHT_ATOMIC_KEEP_BINARY_FLAG_MASK = 0x02;
+    static final int DHT_ATOMIC_KEEP_BINARY_FLAG_MASK = 0x02;
 
     /** Near cache key flag. */
-    public static final int DHT_ATOMIC_NEAR_FLAG_MASK = 0x04;
+    static final int DHT_ATOMIC_NEAR_FLAG_MASK = 0x04;
 
     /** */
-    public static final int DHT_ATOMIC_HAS_RESULT_MASK = 0x08;
+    static final int DHT_ATOMIC_HAS_RESULT_MASK = 0x08;
 
     /** */
-    public static final int DHT_ATOMIC_RESULT_SUCCESS_MASK = 0x10;
+    static final int DHT_ATOMIC_RESULT_SUCCESS_MASK = 0x10;
 
     /** */
-    public static final int DHT_ATOMIC_PRIMARY_DHT_FAIL_RESPONSE = 0x20;
+    static final int DHT_ATOMIC_PRIMARY_DHT_FAIL_RESPONSE = 0x20;
 
     /** Message index. */
     public static final int CACHE_MSG_IDX = nextIndexId();
@@ -171,6 +171,10 @@ public abstract class GridDhtAtomicAbstractUpdateRequest extends GridCacheMessag
         return !onRes && (onRes = true);
     }
 
+    boolean hasResponse() {
+        return onRes;
+    }
+
     /** {@inheritDoc} */
     @Override public boolean addDeploymentInfo() {
         return addDepInfo;
@@ -199,7 +203,6 @@ public abstract class GridDhtAtomicAbstractUpdateRequest extends GridCacheMessag
      * @param conflictExpireTime Conflict expire time (optional).
      * @param conflictVer Conflict version (optional).
      * @param addPrevVal If {@code true} adds previous value.
-     * @param partId Partition.
      * @param prevVal Previous value.
      * @param updateCntr Update counter.
      */
@@ -210,7 +213,6 @@ public abstract class GridDhtAtomicAbstractUpdateRequest extends GridCacheMessag
         long conflictExpireTime,
         @Nullable GridCacheVersion conflictVer,
         boolean addPrevVal,
-        int partId,
         @Nullable CacheObject prevVal,
         long updateCntr
     );
@@ -288,12 +290,6 @@ public abstract class GridDhtAtomicAbstractUpdateRequest extends GridCacheMessag
     public abstract KeyCacheObject key(int idx);
 
     /**
-     * @param idx Partition index.
-     * @return Partition id.
-     */
-    public abstract int partitionId(int idx);
-
-    /**
      * @param updCntr Update counter.
      * @return Update counter.
      */

http://git-wip-us.apache.org/repos/asf/ignite/blob/586de83c/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
index 0557bc6..1c8220b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
@@ -1825,7 +1825,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
 
                         boolean sndPrevVal = !top.rebalanceFinished(req.topologyVersion());
 
-                        dhtFut = createDhtFuture(ver, req, res, false);
+                        dhtFut = createDhtFuture(ver, req, false);
 
                         expiry = expiryPolicy(req.expiry());
 
@@ -1845,7 +1845,6 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
                                 locked,
                                 ver,
                                 dhtFut,
-                                completionCb,
                                 ctx.isDrEnabled(),
                                 taskName,
                                 expiry,
@@ -1946,7 +1945,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
         else {
             // If there are backups, map backup update future.
             if (dhtFut != null) {
-                dhtFut.map(completionCb, res.returnValue());
+                dhtFut.map(res, completionCb, res.returnValue());
                 // Otherwise, complete the call.
             }
             else
@@ -1966,7 +1965,6 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
      * @param locked Locked entries.
      * @param ver Assigned version.
      * @param dhtFut Optional DHT future.
-     * @param completionCb Completion callback to invoke when DHT future is completed.
      * @param replicate Whether replication is enabled.
      * @param taskName Task name.
      * @param expiry Expiry policy.
@@ -1983,7 +1981,6 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
         final List<GridDhtCacheEntry> locked,
         final GridCacheVersion ver,
         @Nullable GridDhtAtomicAbstractUpdateFuture dhtFut,
-        final GridDhtAtomicCache.UpdateReplyClosure completionCb,
         final boolean replicate,
         final String taskName,
         @Nullable final IgniteCacheExpiryPolicy expiry,
@@ -2479,7 +2476,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
                     dhtFut);
 
                 if (dhtFut == null && !F.isEmpty(filteredReaders)) {
-                    dhtFut = createDhtFuture(ver, req, res, true);
+                    dhtFut = createDhtFuture(ver, req, true);
 
                     readersOnly = true;
                 }
@@ -2610,7 +2607,6 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
      * @param rmvKeys Keys to remove.
      * @param entryProcessorMap Entry processors.
      * @param dhtFut DHT update future if has backups.
-     * @param completionCb Completion callback to invoke when DHT future is completed.
      * @param req Request.
      * @param res Response.
      * @param replicate Whether replication is enabled.
@@ -2785,7 +2781,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
                     batchRes.addDeleted(entry, updRes, entries);
 
                     if (dhtFut == null && !F.isEmpty(filteredReaders)) {
-                        dhtFut = createDhtFuture(ver, req, res, true);
+                        dhtFut = createDhtFuture(ver, req, true);
 
                         batchRes.readersOnly(true);
                     }
@@ -3104,20 +3100,18 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
      *
      * @param writeVer Write version.
      * @param updateReq Update request.
-     * @param updateRes Update response.
      * @param force If {@code true} then creates future without optimizations checks.
      * @return Backup update future or {@code null} if there are no backups.
      */
     @Nullable private GridDhtAtomicAbstractUpdateFuture createDhtFuture(
         GridCacheVersion writeVer,
         GridNearAtomicAbstractUpdateRequest updateReq,
-        GridNearAtomicUpdateResponse updateRes,
         boolean force
     ) {
         if (updateReq.size() == 1)
-            return new GridDhtAtomicSingleUpdateFuture(ctx, writeVer, updateReq, updateRes);
+            return new GridDhtAtomicSingleUpdateFuture(ctx, writeVer, updateReq);
         else
-            return new GridDhtAtomicUpdateFuture(ctx, writeVer, updateReq, updateRes);
+            return new GridDhtAtomicUpdateFuture(ctx, writeVer, updateReq);
 //        if (!force) {
 //            if (updateReq.fastMap())
 //                return null;
@@ -3193,6 +3187,8 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
                 ", writeVer=" + req.writeVersion() + ", node=" + nodeId + ']');
         }
 
+        assert req.partition() >= 0 : req;
+
         GridCacheVersion ver = req.writeVersion();
 
         GridDhtAtomicNearResponse nearRes = ctx.config().getWriteSynchronizationMode() == FULL_SYNC ?
@@ -3298,12 +3294,14 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
             List<KeyCacheObject> nearEvicted =
                 ((GridNearAtomicCache<K, V>)near()).processDhtAtomicUpdateRequest(nodeId, req, nearRes);
 
-            dhtRes = new GridDhtAtomicUpdateResponse(ctx.cacheId(),
-                req.partition(),
-                req.futureId(),
-                ctx.deploymentEnabled());
+            if (nearEvicted != null) {
+                dhtRes = new GridDhtAtomicUpdateResponse(ctx.cacheId(),
+                    req.partition(),
+                    req.futureId(),
+                    ctx.deploymentEnabled());
 
-            dhtRes.nearEvicted(nearEvicted);
+                dhtRes.nearEvicted(nearEvicted);
+            }
         }
 
         if (nearRes != null) {
@@ -3433,7 +3431,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
 
             GridDhtAtomicDeferredUpdateResponse msg = resMap.get(primaryId);
 
-            if (msg.timeoutSender() == this) {
+            if (msg != null && msg.timeoutSender() == this) {
                 msg.timeoutSender(null);
 
                 resMap.remove(primaryId);
@@ -3627,19 +3625,39 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
      */
     @SuppressWarnings("unchecked")
     private void processDhtAtomicUpdateResponse(UUID nodeId, GridDhtAtomicUpdateResponse res) {
-        GridDhtAtomicAbstractUpdateFuture updateFut = (GridDhtAtomicAbstractUpdateFuture)ctx.mvcc().atomicFuture(res.futureId());
+        assert !F.isEmpty(res.nearEvicted()) || res.error() != null : res;
 
-        if (updateFut != null) {
-            if (msgLog.isDebugEnabled()) {
-                msgLog.debug("Received DHT atomic update response [futId=" + res.futureId() +
-                    ", writeVer=" + updateFut.writeVersion() + ", node=" + nodeId + ']');
-            }
+        if (!F.isEmpty(res.nearEvicted())) {
+            for (KeyCacheObject key : res.nearEvicted()) {
+                try {
+                    GridDhtCacheEntry entry = (GridDhtCacheEntry)ctx.cache().peekEx(key);
 
-            updateFut.onResult(nodeId, res);
+                    if (entry != null)
+                        entry.removeReader(nodeId, res.messageId());
+                }
+                catch (GridCacheEntryRemovedException e) {
+                    if (log.isDebugEnabled())
+                        log.debug("Entry with evicted reader was removed [key=" + key + ", err=" + e + ']');
+                }
+            }
         }
-        else {
-            U.warn(msgLog, "Failed to find DHT update future for update response [futId=" + res.futureId() +
-                ", node=" + nodeId + ", res=" + res + ']');
+
+        if (res.error() != null) {
+            GridDhtAtomicAbstractUpdateFuture updateFut =
+                (GridDhtAtomicAbstractUpdateFuture)ctx.mvcc().atomicFuture(res.futureId());
+
+            if (updateFut != null) {
+                if (msgLog.isDebugEnabled()) {
+                    msgLog.debug("Received DHT atomic update response [futId=" + res.futureId() +
+                        ", writeVer=" + updateFut.writeVersion() + ", node=" + nodeId + ']');
+                }
+
+                updateFut.onDhtErrorResponse(nodeId, res);
+            }
+            else {
+                U.warn(msgLog, "Failed to find DHT update future for update response [futId=" + res.futureId() +
+                    ", node=" + nodeId + ", res=" + res + ']');
+            }
         }
     }
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/586de83c/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicNearResponse.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicNearResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicNearResponse.java
index af9e908..7c2f9fe 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicNearResponse.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicNearResponse.java
@@ -115,6 +115,9 @@ public class GridDhtAtomicNearResponse extends GridCacheMessage {
         setFlag(true, DHT_ATOMIC_PRIMARY_DHT_FAIL_RESPONSE);
     }
 
+    /**
+     * @return {@code True} if message is sent from primary when DHT node fails.
+     */
     boolean primaryDhtFailureResponse() {
         return isFlag(DHT_ATOMIC_PRIMARY_DHT_FAIL_RESPONSE);
     }
@@ -146,7 +149,7 @@ public class GridDhtAtomicNearResponse extends GridCacheMessage {
      * @return Operation result.
      */
     public GridCacheReturn result() {
-        assert hasResult();
+        assert hasResult() : this;
 
         return new GridCacheReturn(true, isFlag(DHT_ATOMIC_RESULT_SUCCESS_MASK));
     }
@@ -154,7 +157,7 @@ public class GridDhtAtomicNearResponse extends GridCacheMessage {
     /**
      * @return {@code True} if response contains operation result.
      */
-    public boolean hasResult() {
+    boolean hasResult() {
         return isFlag(DHT_ATOMIC_HAS_RESULT_MASK);
     }
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/586de83c/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicSingleUpdateFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicSingleUpdateFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicSingleUpdateFuture.java
index b431bd7..86fbdfd 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicSingleUpdateFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicSingleUpdateFuture.java
@@ -24,11 +24,8 @@ import org.apache.ignite.cache.CacheWriteSynchronizationMode;
 import org.apache.ignite.cluster.ClusterNode;
 import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
 import org.apache.ignite.internal.processors.cache.GridCacheContext;
-import org.apache.ignite.internal.processors.cache.GridCacheEntryRemovedException;
 import org.apache.ignite.internal.processors.cache.KeyCacheObject;
-import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtCacheEntry;
 import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
-import org.apache.ignite.internal.util.typedef.F;
 import org.apache.ignite.internal.util.typedef.internal.CU;
 import org.apache.ignite.internal.util.typedef.internal.S;
 import org.apache.ignite.internal.util.typedef.internal.U;
@@ -42,53 +39,29 @@ class GridDhtAtomicSingleUpdateFuture extends GridDhtAtomicAbstractUpdateFuture
     /** */
     private static final long serialVersionUID = 0L;
 
-    /** Future keys. */
-    private KeyCacheObject key;
-
-    /** Entries with readers. */
-    private GridDhtCacheEntry nearReaderEntry;
-
     /**
      * @param cctx Cache context.
      * @param writeVer Write version.
      * @param updateReq Update request.
-     * @param updateRes Update response.
      */
     GridDhtAtomicSingleUpdateFuture(
         GridCacheContext cctx,
         GridCacheVersion writeVer,
-        GridNearAtomicAbstractUpdateRequest updateReq,
-        GridNearAtomicUpdateResponse updateRes
+        GridNearAtomicAbstractUpdateRequest updateReq
     ) {
-        super(cctx,
-            writeVer,
-            updateReq,
-            updateRes);
+        super(cctx, writeVer, updateReq);
     }
 
     /** {@inheritDoc} */
     @Override protected void addDhtKey(KeyCacheObject key, List<ClusterNode> dhtNodes) {
-        assert this.key == null || this.key.equals(key) : this.key;
-
         if (mappings == null)
             mappings = U.newHashMap(dhtNodes.size());
-
-        this.key = key;
     }
 
     /** {@inheritDoc} */
     @Override protected void addNearKey(KeyCacheObject key, Collection<UUID> readers) {
-        assert this.key == null || this.key.equals(key) : this.key;
-
         if (mappings == null)
             mappings = U.newHashMap(readers.size());
-
-        this.key = key;
-    }
-
-    /** {@inheritDoc} */
-    @Override protected void addNearReaderEntry(GridDhtCacheEntry entry) {
-        nearReaderEntry = entry;
     }
 
     /** {@inheritDoc} */
@@ -139,34 +112,6 @@ class GridDhtAtomicSingleUpdateFuture extends GridDhtAtomicAbstractUpdateFuture
         }
     }
 
-    /** {@inheritDoc} */
-    @Override public void onResult(UUID nodeId, GridDhtAtomicUpdateResponse updateRes) {
-        if (log.isDebugEnabled())
-            log.debug("Received DHT atomic update future result [nodeId=" + nodeId + ", updateRes=" + updateRes + ']');
-
-        if (updateRes.error() != null)
-            this.updateRes.addFailedKeys(updateRes.failedKeys(), updateRes.error());
-
-        if (!F.isEmpty(updateRes.nearEvicted())) {
-            try {
-                assert nearReaderEntry != null;
-
-                nearReaderEntry.removeReader(nodeId, updateRes.messageId());
-            }
-            catch (GridCacheEntryRemovedException e) {
-                if (log.isDebugEnabled())
-                    log.debug("Entry with evicted reader was removed [entry=" + nearReaderEntry + ", err=" + e + ']');
-            }
-        }
-
-        registerResponse(nodeId, false);
-    }
-
-    /** {@inheritDoc} */
-    @Override protected void addFailedKeys(GridNearAtomicUpdateResponse updateRes, Throwable err) {
-        updateRes.addFailedKey(key, err);
-    }
-
     /**
      * @param ttl TTL.
      * @param conflictExpireTime Conflict expire time.
@@ -183,6 +128,6 @@ class GridDhtAtomicSingleUpdateFuture extends GridDhtAtomicAbstractUpdateFuture
 
     /** {@inheritDoc} */
     @Override public String toString() {
-        return S.toString(GridDhtAtomicSingleUpdateFuture.class, this);
+        return S.toString(GridDhtAtomicSingleUpdateFuture.class, this, "super", super.toString());
     }
 }

http://git-wip-us.apache.org/repos/asf/ignite/blob/586de83c/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicSingleUpdateRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicSingleUpdateRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicSingleUpdateRequest.java
index 36c730a..77bcc26 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicSingleUpdateRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicSingleUpdateRequest.java
@@ -81,9 +81,6 @@ public class GridDhtAtomicSingleUpdateRequest extends GridDhtAtomicAbstractUpdat
     /** Partition. */
     protected long updateCntr;
 
-    /** */
-    protected int partId;
-
     /**
      * Empty constructor required by {@link Externalizable}.
      */
@@ -131,9 +128,9 @@ public class GridDhtAtomicSingleUpdateRequest extends GridDhtAtomicAbstractUpdat
         this.addDepInfo = addDepInfo;
 
         if (skipStore)
-            setFlag(true, SKIP_STORE_FLAG_MASK);
+            setFlag(true, DHT_ATOMIC_SKIP_STORE_FLAG_MASK);
         if (keepBinary)
-            setFlag(true, KEEP_BINARY_FLAG_MASK);
+            setFlag(true, DHT_ATOMIC_KEEP_BINARY_FLAG_MASK);
     }
 
     /**
@@ -144,7 +141,6 @@ public class GridDhtAtomicSingleUpdateRequest extends GridDhtAtomicAbstractUpdat
      * @param conflictExpireTime Conflict expire time (optional).
      * @param conflictVer Conflict version (optional).
      * @param addPrevVal If {@code true} adds previous value.
-     * @param partId Partition.
      * @param prevVal Previous value.
      * @param updateCntr Update counter.
      */
@@ -155,7 +151,6 @@ public class GridDhtAtomicSingleUpdateRequest extends GridDhtAtomicAbstractUpdat
         long conflictExpireTime,
         @Nullable GridCacheVersion conflictVer,
         boolean addPrevVal,
-        int partId,
         @Nullable CacheObject prevVal,
         long updateCntr
     ) {
@@ -163,11 +158,11 @@ public class GridDhtAtomicSingleUpdateRequest extends GridDhtAtomicAbstractUpdat
         assert ttl <= 0 : ttl;
         assert conflictExpireTime <= 0 : conflictExpireTime;
         assert conflictVer == null : conflictVer;
+        assert key.partition() >= 0 : key;
 
         near(false);
 
         this.key = key;
-        this.partId = partId;
         this.val = val;
 
         if (addPrevVal)
@@ -190,6 +185,7 @@ public class GridDhtAtomicSingleUpdateRequest extends GridDhtAtomicAbstractUpdat
         long expireTime) {
         assert entryProcessor == null;
         assert ttl <= 0 : ttl;
+        assert key.partition() >= 0 : key;
 
         near(true);
 
@@ -231,14 +227,7 @@ public class GridDhtAtomicSingleUpdateRequest extends GridDhtAtomicAbstractUpdat
 
     /** {@inheritDoc} */
     @Override public int partition() {
-        return partId;
-    }
-
-    /** {@inheritDoc} */
-    @Override public int partitionId(int idx) {
-        assert idx == 0 : idx;
-
-        return partId;
+        return key.partition();
     }
 
     /** {@inheritDoc} */
@@ -399,8 +388,6 @@ public class GridDhtAtomicSingleUpdateRequest extends GridDhtAtomicAbstractUpdat
         finishUnmarshalObject(val, cctx, ldr);
 
         finishUnmarshalObject(prevVal, cctx, ldr);
-
-        key.partition(partId);
     }
 
     /** {@inheritDoc} */
@@ -431,54 +418,48 @@ public class GridDhtAtomicSingleUpdateRequest extends GridDhtAtomicAbstractUpdat
                 writer.incrementState();
 
             case 9:
-                if (!writer.writeInt("partId", partId))
-                    return false;
-
-                writer.incrementState();
-
-            case 10:
                 if (!writer.writeMessage("prevVal", prevVal))
                     return false;
 
                 writer.incrementState();
 
-            case 11:
+            case 10:
                 if (!writer.writeUuid("subjId", subjId))
                     return false;
 
                 writer.incrementState();
 
-            case 12:
+            case 11:
                 if (!writer.writeByte("syncMode", syncMode != null ? (byte)syncMode.ordinal() : -1))
                     return false;
 
                 writer.incrementState();
 
-            case 13:
+            case 12:
                 if (!writer.writeInt("taskNameHash", taskNameHash))
                     return false;
 
                 writer.incrementState();
 
-            case 14:
+            case 13:
                 if (!writer.writeMessage("topVer", topVer))
                     return false;
 
                 writer.incrementState();
 
-            case 15:
+            case 14:
                 if (!writer.writeLong("updateCntr", updateCntr))
                     return false;
 
                 writer.incrementState();
 
-            case 16:
+            case 15:
                 if (!writer.writeMessage("val", val))
                     return false;
 
                 writer.incrementState();
 
-            case 17:
+            case 16:
                 if (!writer.writeMessage("writeVer", writeVer))
                     return false;
 
@@ -517,14 +498,6 @@ public class GridDhtAtomicSingleUpdateRequest extends GridDhtAtomicAbstractUpdat
                 reader.incrementState();
 
             case 9:
-                partId = reader.readInt("partId");
-
-                if (!reader.isLastRead())
-                    return false;
-
-                reader.incrementState();
-
-            case 10:
                 prevVal = reader.readMessage("prevVal");
 
                 if (!reader.isLastRead())
@@ -532,7 +505,7 @@ public class GridDhtAtomicSingleUpdateRequest extends GridDhtAtomicAbstractUpdat
 
                 reader.incrementState();
 
-            case 11:
+            case 10:
                 subjId = reader.readUuid("subjId");
 
                 if (!reader.isLastRead())
@@ -540,7 +513,7 @@ public class GridDhtAtomicSingleUpdateRequest extends GridDhtAtomicAbstractUpdat
 
                 reader.incrementState();
 
-            case 12:
+            case 11:
                 byte syncModeOrd;
 
                 syncModeOrd = reader.readByte("syncMode");
@@ -552,7 +525,7 @@ public class GridDhtAtomicSingleUpdateRequest extends GridDhtAtomicAbstractUpdat
 
                 reader.incrementState();
 
-            case 13:
+            case 12:
                 taskNameHash = reader.readInt("taskNameHash");
 
                 if (!reader.isLastRead())
@@ -560,7 +533,7 @@ public class GridDhtAtomicSingleUpdateRequest extends GridDhtAtomicAbstractUpdat
 
                 reader.incrementState();
 
-            case 14:
+            case 13:
                 topVer = reader.readMessage("topVer");
 
                 if (!reader.isLastRead())
@@ -568,7 +541,7 @@ public class GridDhtAtomicSingleUpdateRequest extends GridDhtAtomicAbstractUpdat
 
                 reader.incrementState();
 
-            case 15:
+            case 14:
                 updateCntr = reader.readLong("updateCntr");
 
                 if (!reader.isLastRead())
@@ -576,7 +549,7 @@ public class GridDhtAtomicSingleUpdateRequest extends GridDhtAtomicAbstractUpdat
 
                 reader.incrementState();
 
-            case 16:
+            case 15:
                 val = reader.readMessage("val");
 
                 if (!reader.isLastRead())
@@ -584,7 +557,7 @@ public class GridDhtAtomicSingleUpdateRequest extends GridDhtAtomicAbstractUpdat
 
                 reader.incrementState();
 
-            case 17:
+            case 16:
                 writeVer = reader.readMessage("writeVer");
 
                 if (!reader.isLastRead())
@@ -634,7 +607,7 @@ public class GridDhtAtomicSingleUpdateRequest extends GridDhtAtomicAbstractUpdat
 
     /** {@inheritDoc} */
     @Override public byte fieldsCount() {
-        return 18;
+        return 17;
     }
 
     /** {@inheritDoc} */

http://git-wip-us.apache.org/repos/asf/ignite/blob/586de83c/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java
index 7303736..22ecef8 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java
@@ -17,21 +17,15 @@
 
 package org.apache.ignite.internal.processors.cache.distributed.dht.atomic;
 
-import java.util.ArrayList;
 import java.util.Collection;
-import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
 import java.util.UUID;
 import org.apache.ignite.cache.CacheWriteSynchronizationMode;
 import org.apache.ignite.cluster.ClusterNode;
 import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
 import org.apache.ignite.internal.processors.cache.GridCacheContext;
-import org.apache.ignite.internal.processors.cache.GridCacheEntryRemovedException;
 import org.apache.ignite.internal.processors.cache.KeyCacheObject;
-import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtCacheEntry;
 import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
-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.jetbrains.annotations.NotNull;
@@ -44,80 +38,29 @@ class GridDhtAtomicUpdateFuture extends GridDhtAtomicAbstractUpdateFuture {
     /** */
     private static final long serialVersionUID = 0L;
 
-    /** Future keys. */
-    private final Collection<KeyCacheObject> keys;
-
-    /** Entries with readers. */
-    private Map<KeyCacheObject, GridDhtCacheEntry> nearReadersEntries;
-
     /**
      * @param cctx Cache context.
      * @param writeVer Write version.
      * @param updateReq Update request.
-     * @param updateRes Update response.
      */
     GridDhtAtomicUpdateFuture(
         GridCacheContext cctx,
         GridCacheVersion writeVer,
-        GridNearAtomicAbstractUpdateRequest updateReq,
-        GridNearAtomicUpdateResponse updateRes
+        GridNearAtomicAbstractUpdateRequest updateReq
     ) {
-        super(cctx,
-            writeVer,
-            updateReq,
-            updateRes);
+        super(cctx, writeVer, updateReq);
 
-        keys = new ArrayList<>(updateReq.size());
         mappings = U.newHashMap(updateReq.size());
     }
 
     /** {@inheritDoc} */
     @Override protected void addDhtKey(KeyCacheObject key, List<ClusterNode> dhtNodes) {
-        keys.add(key);
+        // No-op.
     }
 
     /** {@inheritDoc} */
     @Override protected void addNearKey(KeyCacheObject key, Collection<UUID> readers) {
-        keys.add(key);
-    }
-
-    /** {@inheritDoc} */
-    @Override protected void addNearReaderEntry(GridDhtCacheEntry entry) {
-        if (nearReadersEntries == null)
-            nearReadersEntries = new HashMap<>();
-
-        nearReadersEntries.put(entry.key(), entry);
-    }
-
-    /** {@inheritDoc} */
-    @Override public void onResult(UUID nodeId, GridDhtAtomicUpdateResponse updateRes) {
-        if (log.isDebugEnabled())
-            log.debug("Received DHT atomic update future result [nodeId=" + nodeId + ", updateRes=" + updateRes + ']');
-
-        if (updateRes.error() != null)
-            this.updateRes.addFailedKeys(updateRes.failedKeys(), updateRes.error());
-
-        if (!F.isEmpty(updateRes.nearEvicted())) {
-            for (KeyCacheObject key : updateRes.nearEvicted()) {
-                GridDhtCacheEntry entry = nearReadersEntries.get(key);
-
-                try {
-                    entry.removeReader(nodeId, updateRes.messageId());
-                }
-                catch (GridCacheEntryRemovedException e) {
-                    if (log.isDebugEnabled())
-                        log.debug("Entry with evicted reader was removed [entry=" + entry + ", err=" + e + ']');
-                }
-            }
-        }
-
-        registerResponse(nodeId, false);
-    }
-
-    /** {@inheritDoc} */
-    @Override protected void addFailedKeys(GridNearAtomicUpdateResponse updateRes, Throwable err) {
-        for (KeyCacheObject key : keys)
-            updateRes.addFailedKey(key, err);
+        // No-op.
     }
 
     /** {@inheritDoc} */
@@ -152,6 +95,6 @@ class GridDhtAtomicUpdateFuture extends GridDhtAtomicAbstractUpdateFuture {
 
     /** {@inheritDoc} */
     @Override public String toString() {
-        return S.toString(GridDhtAtomicUpdateFuture.class, this);
+        return S.toString(GridDhtAtomicUpdateFuture.class, this, "super", super.toString());
     }
 }

http://git-wip-us.apache.org/repos/asf/ignite/blob/586de83c/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java
index 979d3ae..8ffd9af 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateRequest.java
@@ -44,8 +44,6 @@ import org.apache.ignite.plugin.extensions.communication.MessageWriter;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 
-import static org.apache.ignite.internal.processors.cache.GridCacheUtils.SKIP_STORE_FLAG_MASK;
-
 /**
  * Lite dht cache backup update request.
  */
@@ -141,10 +139,6 @@ public class GridDhtAtomicUpdateRequest extends GridDhtAtomicAbstractUpdateReque
     /** Partition. */
     private GridLongList updateCntrs;
 
-    /** */
-    @GridDirectTransient
-    private List<Integer> partIds;
-
     /** Keep binary flag. */
     private boolean keepBinary;
 
@@ -203,10 +197,10 @@ public class GridDhtAtomicUpdateRequest extends GridDhtAtomicAbstractUpdateReque
         this.addDepInfo = addDepInfo;
         this.keepBinary = keepBinary;
 
-        setFlag(skipStore, SKIP_STORE_FLAG_MASK);
+        if (skipStore)
+            setFlag(true, DHT_ATOMIC_SKIP_STORE_FLAG_MASK);
 
         keys = new ArrayList<>();
-        partIds = new ArrayList<>();
 
         if (forceTransformBackups) {
             entryProcessors = new ArrayList<>();
@@ -224,13 +218,12 @@ public class GridDhtAtomicUpdateRequest extends GridDhtAtomicAbstractUpdateReque
         long conflictExpireTime,
         @Nullable GridCacheVersion conflictVer,
         boolean addPrevVal,
-        int partId,
         @Nullable CacheObject prevVal,
         long updateCntr
     ) {
-        keys.add(key);
+        assert key.partition() >= 0 : key;
 
-        partIds.add(partId);
+        keys.add(key);
 
         if (forceTransformBackups) {
             assert entryProcessor != null;
@@ -297,6 +290,8 @@ public class GridDhtAtomicUpdateRequest extends GridDhtAtomicAbstractUpdateReque
         EntryProcessor<Object, Object, Object> entryProcessor,
         long ttl,
         long expireTime) {
+        assert key.partition() >= 0 : key;
+
         if (nearKeys == null) {
             nearKeys = new ArrayList<>();
 
@@ -399,11 +394,6 @@ public class GridDhtAtomicUpdateRequest extends GridDhtAtomicAbstractUpdateReque
     }
 
     /** {@inheritDoc} */
-    @Override public int partitionId(int idx) {
-        return partIds.get(idx);
-    }
-
-    /** {@inheritDoc} */
     @Override public Long updateCounter(int updCntr) {
         if (updateCntrs != null && updCntr < updateCntrs.size())
             return updateCntrs.get(updCntr);
@@ -485,7 +475,9 @@ public class GridDhtAtomicUpdateRequest extends GridDhtAtomicAbstractUpdateReque
 
     /** {@inheritDoc} */
     @Override public int partition() {
-        return partIds != null && !partIds.isEmpty() ? partIds.get(0) : -1;
+        assert !F.isEmpty(keys) || !F.isEmpty(nearKeys);
+
+        return keys.size() > 0 ? keys.get(0).partition() : nearKeys.get(0).partition();
     }
 
     /** {@inheritDoc} */
@@ -583,13 +575,6 @@ public class GridDhtAtomicUpdateRequest extends GridDhtAtomicAbstractUpdateReque
             if (nearEntryProcessors == null)
                 nearEntryProcessors = unmarshalCollection(nearEntryProcessorsBytes, ctx, ldr);
         }
-
-        if (partIds != null && !partIds.isEmpty()) {
-            assert partIds.size() == keys.size();
-
-            for (int i = 0; i < keys.size(); i++)
-                keys.get(i).partition(partIds.get(i));
-        }
     }
 
     /** {@inheritDoc} */

http://git-wip-us.apache.org/repos/asf/ignite/blob/586de83c/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicSingleUpdateFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicSingleUpdateFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicSingleUpdateFuture.java
index df74d3d..69dc10b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicSingleUpdateFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicSingleUpdateFuture.java
@@ -121,10 +121,12 @@ public class GridNearAtomicSingleUpdateFuture extends GridNearAtomicAbstractUpda
 
     /** {@inheritDoc} */
     @Override public boolean onNodeLeft(UUID nodeId) {
+        GridNearAtomicAbstractUpdateRequest req;
+
         GridNearAtomicUpdateResponse res = null;
 
-        GridNearAtomicAbstractUpdateRequest req;
         GridCacheReturn opRes0 = null;
+        CachePartialUpdateCheckedException err0 = null;
 
         synchronized (mux) {
             if (reqState == null)
@@ -148,9 +150,10 @@ public class GridNearAtomicSingleUpdateFuture extends GridNearAtomicAbstractUpda
             else {
                 if (reqState.onNodeLeft(nodeId)) {
                     opRes0 = opRes;
-
-                    assert opRes0 != null;
+                    err0 = err;
                 }
+                else
+                    return false;
             }
         }
 
@@ -163,8 +166,8 @@ public class GridNearAtomicSingleUpdateFuture extends GridNearAtomicAbstractUpda
 
             onPrimaryResponse(nodeId, res, true);
         }
-        else if (opRes0 != null)
-            onDone(opRes0);
+        else
+            onDone(opRes0, err0);
 
         return false;
     }
@@ -181,8 +184,7 @@ public class GridNearAtomicSingleUpdateFuture extends GridNearAtomicAbstractUpda
 
         GridCacheReturn ret = (GridCacheReturn)res;
 
-        Object retval =
-            res == null ? null : rawRetval ? ret : (this.retval || op == TRANSFORM) ?
+        Object retval = res == null ? null : rawRetval ? ret : (this.retval || op == TRANSFORM) ?
                 cctx.unwrapBinaryIfNeeded(ret.value(), keepBinary) : ret.success();
 
         if (op == TRANSFORM && retval == null)
@@ -202,7 +204,8 @@ public class GridNearAtomicSingleUpdateFuture extends GridNearAtomicAbstractUpda
 
     /** {@inheritDoc} */
     @Override public void onMappingReceived(UUID nodeId, GridNearAtomicMappingResponse res) {
-        GridCacheReturn opRes0 = null;
+        GridCacheReturn opRes0;
+        CachePartialUpdateCheckedException err0;
 
         synchronized (mux) {
             if (futId == null || futId != res.futureId())
@@ -212,25 +215,25 @@ public class GridNearAtomicSingleUpdateFuture extends GridNearAtomicAbstractUpda
 
             if (reqState.onMappingReceived(cctx, res)) {
                 opRes0 = opRes;
-
-                assert opRes0 != null;
+                err0 = err;
             }
+            else
+                return;
         }
 
-        if (opRes0 != null)
-            onDone(opRes0);
+        onDone(opRes0, err0);
     }
 
     /** {@inheritDoc} */
     @Override public void onDhtResponse(UUID nodeId, GridDhtAtomicNearResponse res) {
-        GridCacheReturn opRes0 = null;
+        GridCacheReturn opRes0;
+        CachePartialUpdateCheckedException err0;
 
         synchronized (mux) {
             if (futId == null || futId != res.futureId())
                 return;
 
             assert reqState != null;
-
             assert reqState.req.nodeId().equals(res.primaryId());
 
             if (opRes == null && res.hasResult())
@@ -238,13 +241,13 @@ public class GridNearAtomicSingleUpdateFuture extends GridNearAtomicAbstractUpda
 
             if (reqState.onDhtResponse(cctx, nodeId, res)) {
                 opRes0 = opRes;
-
-                assert opRes0 != null;
+                err0 = err;
             }
+            else
+                return;
         }
 
-        if (opRes0 != null)
-            onDone(opRes0);
+        onDone(opRes0, err0);
     }
 
     /** {@inheritDoc} */

http://git-wip-us.apache.org/repos/asf/ignite/blob/586de83c/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateFuture.java
index 85cb0e2..bffb6ad 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateFuture.java
@@ -151,33 +151,62 @@ public class GridNearAtomicUpdateFuture extends GridNearAtomicAbstractUpdateFutu
 
     /** {@inheritDoc} */
     @Override public boolean onNodeLeft(UUID nodeId) {
+        GridCacheReturn opRes0 = null;
+        CachePartialUpdateCheckedException err0 = null;
+
         GridNearAtomicUpdateResponse res = null;
 
         GridNearAtomicAbstractUpdateRequest req;
 
         synchronized (mux) {
-            if (singleReq != null)
+            if (singleReq != null) {
                 req = singleReq.processPrimaryResponse(nodeId);
+
+                if (req == null) {
+                    if (singleReq.onNodeLeft(nodeId)) {
+                        opRes0 = opRes;
+                        err0 = err;
+                    }
+                    else
+                        return false;
+                }
+                else
+                    res = primaryFailedResponse(req, nodeId);
+            }
             else {
                 PrimaryRequestState reqState = mappings != null ? mappings.get(nodeId) : null;
 
-                req = reqState != null ? reqState.processPrimaryResponse(nodeId) : null;
-            }
+                if (reqState != null) {
+                    req = reqState.processPrimaryResponse(nodeId);
 
-            if (req != null) {
-                assert req.response() == null : req;
+                    if (req == null) {
+                        boolean rcvAll = false;
 
-                res = new GridNearAtomicUpdateResponse(cctx.cacheId(),
-                    nodeId,
-                    req.futureId(),
-                    cctx.deploymentEnabled());
+                        for (PrimaryRequestState reqState0 : mappings.values()) {
+                            if (reqState0.onNodeLeft(nodeId)) {
+                                assert mappings.size() > resCnt : "[mappings=" + mappings.size() + ", cnt=" + resCnt + ']';
+
+                                resCnt++;
+
+                                if (mappings.size() == resCnt) {
+                                    opRes0 = opRes;
+                                    err0 = err;
 
-                ClusterTopologyCheckedException e = new ClusterTopologyCheckedException("Primary node left grid " +
-                    "before response is received: " + nodeId);
+                                    rcvAll = true;
 
-                e.retryReadyFuture(cctx.shared().nextAffinityReadyFuture(req.topologyVersion()));
+                                    break;
+                                }
+                            }
+                        }
 
-                res.addFailedKeys(req.keys(), e);
+                        if (!rcvAll)
+                            return false;
+                    }
+                    else
+                        res = primaryFailedResponse(req, nodeId);
+                }
+                else
+                    return false;
             }
         }
 
@@ -190,10 +219,35 @@ public class GridNearAtomicUpdateFuture extends GridNearAtomicAbstractUpdateFutu
 
             onPrimaryResponse(nodeId, res, true);
         }
+        else
+            onDone(opRes0, err0);
 
         return false;
     }
 
+    /**
+     * @param req Request.
+     * @param nodeId Failed node ID.
+     * @return Response to notify about primary failure.
+     */
+    private GridNearAtomicUpdateResponse primaryFailedResponse(GridNearAtomicAbstractUpdateRequest req, UUID nodeId) {
+        assert req.response() == null : req;
+
+        GridNearAtomicUpdateResponse res = new GridNearAtomicUpdateResponse(cctx.cacheId(),
+            nodeId,
+            req.futureId(),
+            cctx.deploymentEnabled());
+
+        ClusterTopologyCheckedException e = new ClusterTopologyCheckedException("Primary node left grid " +
+            "before response is received: " + nodeId);
+
+        e.retryReadyFuture(cctx.shared().nextAffinityReadyFuture(req.topologyVersion()));
+
+        res.addFailedKeys(req.keys(), e);
+
+        return res;
+    }
+
     /** {@inheritDoc} */
     @Override public IgniteInternalFuture<Void> completeFuture(AffinityTopologyVersion topVer) {
         return null;
@@ -227,7 +281,8 @@ public class GridNearAtomicUpdateFuture extends GridNearAtomicAbstractUpdateFutu
 
     /** {@inheritDoc} */
     @Override public void onMappingReceived(UUID nodeId, GridNearAtomicMappingResponse res) {
-        GridCacheReturn opRes0 = null;
+        GridCacheReturn opRes0;
+        CachePartialUpdateCheckedException err0;
 
         synchronized (mux) {
             if (futId == null || futId != res.futureId())
@@ -238,9 +293,10 @@ public class GridNearAtomicUpdateFuture extends GridNearAtomicAbstractUpdateFutu
             if (singleReq != null) {
                 if (singleReq.onMappingReceived(cctx, res)) {
                     opRes0 = opRes;
-
-                    assert opRes0 != null;
+                    err0 = err;
                 }
+                else
+                    return;
             }
             else {
                 reqState = mappings != null ? mappings.get(nodeId) : null;
@@ -252,20 +308,23 @@ public class GridNearAtomicUpdateFuture extends GridNearAtomicAbstractUpdateFutu
 
                     if (mappings.size() == resCnt) {
                         opRes0 = opRes;
-
-                        assert opRes0 != null;
+                        err0 = err;
                     }
+                    else
+                        return;
                 }
+                else
+                    return;
             }
         }
 
-        if (opRes0 != null)
-            onDone(opRes0);
+        onDone(opRes0, err0);
     }
 
     /** {@inheritDoc} */
     @Override public void onDhtResponse(UUID nodeId, GridDhtAtomicNearResponse res) {
-        GridCacheReturn opRes0 = null;
+        GridCacheReturn opRes0;
+        CachePartialUpdateCheckedException err0;
 
         synchronized (mux) {
             if (futId == null || futId != res.futureId())
@@ -281,9 +340,10 @@ public class GridNearAtomicUpdateFuture extends GridNearAtomicAbstractUpdateFutu
 
                 if (singleReq.onDhtResponse(cctx, nodeId, res)) {
                     opRes0 = opRes;
-
-                    assert opRes0 != null;
+                    err0 = err;
                 }
+                else
+                    return;
             }
             else {
                 reqState = mappings != null ? mappings.get(res.primaryId()) : null;
@@ -299,16 +359,20 @@ public class GridNearAtomicUpdateFuture extends GridNearAtomicAbstractUpdateFutu
 
                         if (mappings.size() == resCnt) {
                             opRes0 = opRes;
-
-                            assert opRes0 != null;
+                            err0 = err;
                         }
+                        else
+                            return;
                     }
+                    else
+                        return;
                 }
+                else
+                    return;
             }
         }
 
-        if (opRes0 != null)
-            onDone(opRes0);
+        onDone(opRes0, err0);
     }
 
     /** {@inheritDoc} */

http://git-wip-us.apache.org/repos/asf/ignite/blob/586de83c/modules/core/src/main/java/org/apache/ignite/internal/processors/cacheobject/IgniteCacheObjectProcessorImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cacheobject/IgniteCacheObjectProcessorImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cacheobject/IgniteCacheObjectProcessorImpl.java
index 9f37f6e..5cacb6f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cacheobject/IgniteCacheObjectProcessorImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cacheobject/IgniteCacheObjectProcessorImpl.java
@@ -366,10 +366,10 @@ public class IgniteCacheObjectProcessorImpl extends GridProcessorAdapter impleme
 
                     Object val = ctx.processor().unmarshal(ctx, valBytes, ldr);
 
-                    return new KeyCacheObjectImpl(val, valBytes);
+                    return new KeyCacheObjectImpl(val, valBytes, partition());
                 }
 
-                return new KeyCacheObjectImpl(val, valBytes);
+                return new KeyCacheObjectImpl(val, valBytes, partition());
             }
             catch (IgniteCheckedException e) {
                 throw new IgniteException("Failed to marshal object: " + val, e);

http://git-wip-us.apache.org/repos/asf/ignite/blob/586de83c/modules/core/src/main/java/org/apache/ignite/internal/util/future/GridCompoundFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/future/GridCompoundFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/util/future/GridCompoundFuture.java
index 7abd367..96f3797 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/future/GridCompoundFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/future/GridCompoundFuture.java
@@ -18,8 +18,10 @@
 package org.apache.ignite.internal.util.future;
 
 import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteLogger;
 import org.apache.ignite.internal.IgniteFutureCancelledCheckedException;
 import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.NodeStoppingException;
 import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
 import org.apache.ignite.internal.transactions.IgniteTxOptimisticCheckedException;
 import org.apache.ignite.internal.util.tostring.GridToStringInclude;
@@ -118,7 +120,14 @@ public class GridCompoundFuture<T, R> extends GridFutureAdapter<R> implements Ig
         }
         catch (IgniteCheckedException e) {
             if (!ignoreFailure(e)) {
-                U.error(null, "Failed to execute compound future reducer: " + this, e);
+                if (e instanceof NodeStoppingException) {
+                    IgniteLogger log = logger();
+
+                    if (log != null && log.isDebugEnabled())
+                        log.debug("Failed to execute compound future reducer, node stopped.");
+                }
+                else
+                    U.error(null, "Failed to execute compound future reducer: " + this, e);
 
                 onDone(e);
             }

http://git-wip-us.apache.org/repos/asf/ignite/blob/586de83c/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/IgniteCacheContinuousQueryImmutableEntryTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/IgniteCacheContinuousQueryImmutableEntryTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/IgniteCacheContinuousQueryImmutableEntryTest.java
index 66d727c..519a989 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/IgniteCacheContinuousQueryImmutableEntryTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/IgniteCacheContinuousQueryImmutableEntryTest.java
@@ -132,7 +132,7 @@ public class IgniteCacheContinuousQueryImmutableEntryTest extends GridCommonAbst
         CacheContinuousQueryEntry e0 = new CacheContinuousQueryEntry(
             1,
             EventType.UPDATED,
-            new KeyCacheObjectImpl(1, new byte[] {0, 0, 0, 1}),
+            new KeyCacheObjectImpl(1, new byte[] {0, 0, 0, 1}, 1),
             new CacheObjectImpl(2, new byte[] {0, 0, 0, 2}),
             new CacheObjectImpl(2, new byte[] {0, 0, 0, 3}),
             true,

http://git-wip-us.apache.org/repos/asf/ignite/blob/586de83c/modules/core/src/test/java/org/apache/ignite/spi/swapspace/file/GridFileSwapSpaceSpiSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/swapspace/file/GridFileSwapSpaceSpiSelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/swapspace/file/GridFileSwapSpaceSpiSelfTest.java
index ab21165..a59b6d1 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/swapspace/file/GridFileSwapSpaceSpiSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/swapspace/file/GridFileSwapSpaceSpiSelfTest.java
@@ -118,7 +118,7 @@ public class GridFileSwapSpaceSpiSelfTest extends GridSwapSpaceSpiAbstractSelfTe
      * @return Swap key.
      */
     private SwapKey key(int i) {
-        return new SwapKey(new KeyCacheObjectImpl(i, U.intToBytes(i)), i % 11, U.intToBytes(i));
+        return new SwapKey(new KeyCacheObjectImpl(i, U.intToBytes(i), i), i % 11, U.intToBytes(i));
     }
 
     /**

http://git-wip-us.apache.org/repos/asf/ignite/blob/586de83c/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheFullApiSelfTestSuite.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheFullApiSelfTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheFullApiSelfTestSuite.java
index 17757ab..c73ffd5 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheFullApiSelfTestSuite.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheFullApiSelfTestSuite.java
@@ -231,8 +231,9 @@ public class IgniteCacheFullApiSelfTestSuite extends TestSuite {
         suite.addTestSuite(GridCachePartitionedFullApiMultithreadedSelfTest.class);
 
         // Disabled striped pool.
-        suite.addTestSuite(GridCacheAtomicPrimaryWriteOrderNoStripedPoolMultiNodeFullApiSelfTest.class);
-        suite.addTestSuite(GridCachePartitionedNoStripedPoolMultiNodeFullApiSelfTest.class);
+        // TODO IGNITE-4705 (
+//        suite.addTestSuite(GridCacheAtomicPrimaryWriteOrderNoStripedPoolMultiNodeFullApiSelfTest.class);
+//        suite.addTestSuite(GridCachePartitionedNoStripedPoolMultiNodeFullApiSelfTest.class);
 
         // Other.
         suite.addTestSuite(GridCacheClearSelfTest.class);


Mime
View raw message