ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From agoncha...@apache.org
Subject [04/49] incubator-ignite git commit: #ingite-9655 - Manual merge of changes.
Date Sun, 15 Feb 2015 08:18:24 GMT
#ingite-9655 - Manual merge of changes.


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

Branch: refs/heads/sprint-1
Commit: 0354a41b94e891273a06587c6cea95aaee15d995
Parents: 31ce585
Author: Alexey Goncharuk <agoncharuk@gridgain.com>
Authored: Wed Jan 28 18:48:49 2015 -0800
Committer: Alexey Goncharuk <agoncharuk@gridgain.com>
Committed: Wed Jan 28 18:48:49 2015 -0800

----------------------------------------------------------------------
 .../processors/cache/GridCacheAdapter.java      |  34 +-
 .../processors/cache/GridCacheUtils.java        |  15 -
 .../distributed/GridCacheCommittedTxInfo.java   |   2 -
 .../GridCachePerThreadTxCommitBuffer.java       | 185 ------
 ...dCachePessimisticCheckCommittedTxFuture.java | 380 ------------
 ...CachePessimisticCheckCommittedTxRequest.java | 292 ---------
 ...achePessimisticCheckCommittedTxResponse.java | 231 -------
 .../distributed/GridCacheTxCommitBuffer.java    |  60 --
 .../distributed/GridDistributedLockRequest.java |  83 ---
 .../GridDistributedTxFinishRequest.java         |  90 +--
 .../GridDistributedTxPrepareRequest.java        |  49 +-
 .../GridDistributedTxRemoteAdapter.java         |  53 +-
 .../distributed/dht/GridDhtLockFuture.java      |  41 +-
 .../distributed/dht/GridDhtLockRequest.java     |   6 +-
 .../dht/GridDhtTransactionalCacheAdapter.java   |  19 +-
 .../distributed/dht/GridDhtTxFinishFuture.java  |  16 +-
 .../distributed/dht/GridDhtTxFinishRequest.java |  70 +--
 .../cache/distributed/dht/GridDhtTxLocal.java   |  21 +-
 .../distributed/dht/GridDhtTxLocalAdapter.java  | 108 ++--
 .../distributed/dht/GridDhtTxPrepareFuture.java | 503 ++++++++++-----
 .../dht/GridDhtTxPrepareRequest.java            |   4 +-
 .../cache/distributed/dht/GridDhtTxRemote.java  |  12 +-
 .../colocated/GridDhtColocatedLockFuture.java   |  41 --
 .../distributed/near/GridNearLockFuture.java    |   8 -
 .../distributed/near/GridNearLockRequest.java   |   6 +-
 .../near/GridNearTransactionalCache.java        |   3 +-
 .../near/GridNearTxFinishFuture.java            |   2 -
 .../near/GridNearTxFinishRequest.java           |   6 +-
 .../cache/distributed/near/GridNearTxLocal.java |  46 +-
 .../near/GridNearTxPrepareFuture.java           | 308 ++++++++--
 .../near/GridNearTxPrepareRequest.java          |  29 +-
 .../near/GridNearTxPrepareResponse.java         |  63 +-
 .../cache/transactions/IgniteTxAdapter.java     |  31 +-
 .../cache/transactions/IgniteTxEntry.java       |  17 -
 .../cache/transactions/IgniteTxEx.java          |   7 -
 .../cache/transactions/IgniteTxHandler.java     | 608 +++----------------
 .../transactions/IgniteTxLocalAdapter.java      | 118 +++-
 .../cache/transactions/IgniteTxLocalEx.java     |   5 +
 .../cache/transactions/IgniteTxManager.java     | 208 ++-----
 .../cache/transactions/IgniteTxProxyImpl.java   |   7 +
 .../util/GridBoundedConcurrentOrderedMap.java   |  20 +-
 .../cache/GridCacheAbstractFullApiSelfTest.java | 182 +++---
 .../cache/GridCacheAbstractMetricsSelfTest.java |   4 +-
 .../cache/IgnitePutAllLargeBatchSelfTest.java   | 301 +++++++++
 ...tAllUpdateNonPreloadedPartitionSelfTest.java | 129 ++++
 .../distributed/GridCacheEventAbstractTest.java |   2 +-
 ...cOriginatingNodeFailureAbstractSelfTest.java |  29 +-
 ...ssimisticOriginatingNodeFailureSelfTest.java |   2 +-
 .../near/GridCacheNearMetricsSelfTest.java      |   8 +-
 .../near/GridCacheNearReadersSelfTest.java      |   2 +-
 ...ssimisticOriginatingNodeFailureSelfTest.java |   2 +-
 ...ssimisticOriginatingNodeFailureSelfTest.java |   5 +-
 .../ignite/testsuites/IgniteCacheTestSuite.java |   3 +
 53 files changed, 1758 insertions(+), 2718 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0354a41b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
index 9271e29..216c13c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
@@ -4190,7 +4190,7 @@ public abstract class GridCacheAdapter<K, V> implements GridCache<K, V>,
                 true,
                 op.single(),
                 ctx.system(),
-                PESSIMISTIC,
+                OPTIMISTIC,
                 READ_COMMITTED,
                 tCfg.getDefaultTxTimeout(),
                 ctx.hasFlag(INVALIDATE),
@@ -4265,7 +4265,7 @@ public abstract class GridCacheAdapter<K, V> implements GridCache<K, V>,
                 true,
                 op.single(),
                 ctx.system(),
-                PESSIMISTIC,
+                OPTIMISTIC,
                 READ_COMMITTED,
                 ctx.kernalContext().config().getTransactionsConfiguration().getDefaultTxTimeout(),
                 ctx.hasFlag(INVALIDATE),
@@ -4301,13 +4301,24 @@ public abstract class GridCacheAdapter<K, V> implements GridCache<K, V>,
         try {
             IgniteFuture fut = holder.future();
 
-            if (fut != null && !fut.isDone()) {
-                final IgniteTxLocalAdapter<K, V> tx0 = tx;
+            final IgniteTxLocalAdapter<K, V> tx0 = tx;
 
+            if (fut != null && !fut.isDone()) {
                 IgniteFuture<T> f = new GridEmbeddedFuture<>(fut,
                     new C2<T, Exception, IgniteFuture<T>>() {
                         @Override public IgniteFuture<T> apply(T t, Exception e) {
-                            return op.op(tx0);
+                            return op.op(tx0).chain(new CX1<IgniteFuture<T>, T>() {
+                                @Override public T applyx(IgniteFuture<T> tFut) throws IgniteCheckedException {
+                                    try {
+                                        return tFut.get();
+                                    }
+                                    catch (IgniteCheckedException e1) {
+                                        tx0.rollbackAsync();
+
+                                        throw e1;
+                                    }
+                                }
+                            });
                         }
                     }, ctx.kernalContext());
 
@@ -4316,7 +4327,18 @@ public abstract class GridCacheAdapter<K, V> implements GridCache<K, V>,
                 return f;
             }
 
-            IgniteFuture<T> f = op.op(tx);
+            IgniteFuture<T> f = op.op(tx).chain(new CX1<IgniteFuture<T>, T>() {
+                @Override public T applyx(IgniteFuture<T> tFut) throws IgniteCheckedException {
+                    try {
+                        return tFut.get();
+                    }
+                    catch (IgniteCheckedException e1) {
+                        tx0.rollbackAsync();
+
+                        throw e1;
+                    }
+                }
+            });
 
             saveFuture(holder, f);
 

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0354a41b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUtils.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUtils.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUtils.java
index 129dc54..1ce00b4 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUtils.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUtils.java
@@ -134,13 +134,6 @@ public class GridCacheUtils {
         }
     };
 
-    /** Transfer required predicate. */
-    private static final IgnitePredicate TRANSFER_REQUIRED_PREDICATE = new P1<IgniteTxEntry>() {
-        @Override public boolean apply(IgniteTxEntry e) {
-            return e.transferRequired();
-        }
-    };
-
     /** Transaction entry to key. */
     private static final IgniteClosure tx2key = new C1<IgniteTxEntry, Object>() {
         @Override public Object apply(IgniteTxEntry e) {
@@ -843,14 +836,6 @@ public class GridCacheUtils {
     }
 
     /**
-     * @return Transfer required predicate.
-     */
-    @SuppressWarnings("unchecked")
-    public static <K, V> IgnitePredicate<IgniteTxEntry<K, V>> transferRequired() {
-        return TRANSFER_REQUIRED_PREDICATE;
-    }
-
-    /**
      * Gets type filter for projections.
      *
      * @param keyType Key type.

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0354a41b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheCommittedTxInfo.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheCommittedTxInfo.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheCommittedTxInfo.java
index 8231465..23ce77b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheCommittedTxInfo.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheCommittedTxInfo.java
@@ -58,8 +58,6 @@ public class GridCacheCommittedTxInfo<K, V> implements Externalizable {
 
         originatingTxId = tx.nearXidVersion();
         originatingNodeId = tx.eventNodeId();
-
-        recoveryWrites = tx.recoveryWrites();
     }
 
     /**

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0354a41b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCachePerThreadTxCommitBuffer.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCachePerThreadTxCommitBuffer.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCachePerThreadTxCommitBuffer.java
deleted file mode 100644
index 0612a5d..0000000
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCachePerThreadTxCommitBuffer.java
+++ /dev/null
@@ -1,185 +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.cache.distributed;
-
-import org.apache.ignite.*;
-import org.apache.ignite.internal.processors.cache.*;
-import org.apache.ignite.internal.util.*;
-import org.apache.ignite.internal.processors.cache.transactions.*;
-import org.apache.ignite.internal.processors.timeout.*;
-import org.apache.ignite.internal.util.typedef.internal.*;
-import org.jdk8.backport.*;
-import org.jetbrains.annotations.*;
-
-import java.util.*;
-
-/**
- * Committed tx buffer which should be used in synchronous commit mode.
- */
-public class GridCachePerThreadTxCommitBuffer<K, V> implements GridCacheTxCommitBuffer<K, V> {
-    /** Logger. */
-    private IgniteLogger log;
-
-    /** Cache context. */
-    private GridCacheSharedContext<K, V> cctx;
-
-    /** Store map. */
-    private Map<StoreKey, GridCacheCommittedTxInfo<K, V>> infoMap;
-
-    /**
-     * @param cctx Cache context.
-     */
-    public GridCachePerThreadTxCommitBuffer(GridCacheSharedContext<K, V> cctx) {
-        this.cctx = cctx;
-
-        log = cctx.logger(GridCachePerThreadTxCommitBuffer.class);
-
-        int logSize = cctx.txConfig().getPessimisticTxLogSize();
-
-        infoMap = logSize > 0 ?
-            new GridBoundedConcurrentLinkedHashMap<StoreKey, GridCacheCommittedTxInfo<K, V>>(logSize) :
-            new ConcurrentHashMap8<StoreKey, GridCacheCommittedTxInfo<K, V>>();
-    }
-
-    /** {@inheritDoc} */
-    @Override public void addCommittedTx(IgniteTxEx<K, V> tx) {
-        long threadId = tx.threadId();
-
-        StoreKey key = new StoreKey(tx.eventNodeId(), threadId);
-
-        if (log.isDebugEnabled())
-            log.debug("Adding committed transaction [locNodeId=" + cctx.localNodeId() + ", key=" + key +
-                ", tx=" + tx + ']');
-
-        infoMap.put(key, new GridCacheCommittedTxInfo<>(tx));
-    }
-
-    /** {@inheritDoc} */
-    @Nullable @Override public GridCacheCommittedTxInfo<K, V> committedTx(GridCacheVersion originatingTxVer,
-        UUID nodeId, long threadId) {
-        assert originatingTxVer != null;
-
-        StoreKey key = new StoreKey(nodeId, threadId);
-
-        GridCacheCommittedTxInfo<K, V> txInfo = infoMap.get(key);
-
-        if (log.isDebugEnabled())
-            log.debug("Got committed transaction info by key [locNodeId=" + cctx.localNodeId() +
-                ", key=" + key + ", originatingTxVer=" + originatingTxVer + ", txInfo=" + txInfo + ']');
-
-        if (txInfo == null || !originatingTxVer.equals(txInfo.originatingTxId()))
-            return null;
-
-        return txInfo;
-    }
-
-    /**
-     * @param nodeId Left node ID.
-     */
-    @Override public void onNodeLeft(UUID nodeId) {
-        // Clear all node's records after clear interval.
-        cctx.kernalContext().timeout().addTimeoutObject(
-            new NodeLeftTimeoutObject(cctx.txConfig().getPessimisticTxLogLinger(), nodeId));
-    }
-
-    /** {@inheritDoc} */
-    @Override public int size() {
-        return infoMap.size();
-    }
-
-    /**
-     * Store key.
-     */
-    private static class StoreKey {
-        /** Node ID which started transaction. */
-        private UUID nodeId;
-
-        /** Thread ID which started transaction. */
-        private long threadId;
-
-        /**
-         * @param nodeId Node ID.
-         * @param threadId Thread ID.
-         */
-        private StoreKey(UUID nodeId, long threadId) {
-            this.nodeId = nodeId;
-            this.threadId = threadId;
-        }
-
-        /**
-         * @return Node ID.
-         */
-        public UUID nodeId() {
-            return nodeId;
-        }
-
-        /** {@inheritDoc} */
-        @Override public boolean equals(Object o) {
-            if (this == o)
-                return true;
-
-            StoreKey storeKey = (StoreKey)o;
-
-            return threadId == storeKey.threadId && nodeId.equals(storeKey.nodeId);
-        }
-
-        /** {@inheritDoc} */
-        @Override public int hashCode() {
-            int res = nodeId.hashCode();
-
-            res = 31 * res + (int)(threadId ^ (threadId >>> 32));
-
-            return res;
-        }
-
-        /** {@inheritDoc} */
-        public String toString() {
-            return S.toString(StoreKey.class, this);
-        }
-    }
-
-    /**
-     * Node left timeout object which will clear all committed records from left node.
-     */
-    private class NodeLeftTimeoutObject extends GridTimeoutObjectAdapter {
-        /** Left node ID. */
-        private UUID leftNodeId;
-
-        /**
-         * @param timeout Timeout.
-         * @param leftNodeId Left node ID.
-         */
-        protected NodeLeftTimeoutObject(long timeout, UUID leftNodeId) {
-            super(timeout);
-
-            this.leftNodeId = leftNodeId;
-        }
-
-        /** {@inheritDoc} */
-        @Override public void onTimeout() {
-            Iterator<StoreKey> it = infoMap.keySet().iterator();
-
-            while (it.hasNext()) {
-                StoreKey key = it.next();
-
-                if (leftNodeId.equals(key.nodeId()))
-                    it.remove();
-            }
-        }
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0354a41b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCachePessimisticCheckCommittedTxFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCachePessimisticCheckCommittedTxFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCachePessimisticCheckCommittedTxFuture.java
deleted file mode 100644
index 12b9177..0000000
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCachePessimisticCheckCommittedTxFuture.java
+++ /dev/null
@@ -1,380 +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.cache.distributed;
-
-import org.apache.ignite.*;
-import org.apache.ignite.cluster.*;
-import org.apache.ignite.internal.processors.cache.*;
-import org.apache.ignite.internal.util.*;
-import org.apache.ignite.lang.*;
-import org.apache.ignite.internal.processors.cache.distributed.dht.*;
-import org.apache.ignite.internal.processors.cache.transactions.*;
-import org.apache.ignite.internal.util.future.*;
-import org.apache.ignite.internal.util.typedef.internal.*;
-import org.jetbrains.annotations.*;
-
-import java.io.*;
-import java.util.*;
-import java.util.concurrent.atomic.*;
-
-/**
- * Future verifying that all remote transactions related to some
- * optimistic transaction were prepared.
- */
-public class GridCachePessimisticCheckCommittedTxFuture<K, V> extends GridCompoundIdentityFuture<GridCacheCommittedTxInfo<K, V>>
-    implements GridCacheFuture<GridCacheCommittedTxInfo<K, V>> {
-    /** */
-    private static final long serialVersionUID = 0L;
-
-    /** Trackable flag. */
-    private boolean trackable = true;
-
-    /** Context. */
-    private final GridCacheSharedContext<K, V> cctx;
-
-    /** Future ID. */
-    private final IgniteUuid futId = IgniteUuid.randomUuid();
-
-    /** Transaction. */
-    private final IgniteTxEx<K, V> tx;
-
-    /** All involved nodes. */
-    private final Map<UUID, ClusterNode> nodes;
-
-    /** ID of failed node started transaction. */
-    private final UUID failedNodeId;
-
-    /** Flag indicating that future checks near node instead of checking all topology in case of primary node crash. */
-    private boolean nearCheck;
-
-    /**
-     * @param cctx Context.
-     * @param tx Transaction.
-     * @param failedNodeId ID of failed node started transaction.
-     */
-    @SuppressWarnings("ConstantConditions")
-    public GridCachePessimisticCheckCommittedTxFuture(GridCacheSharedContext<K, V> cctx, IgniteTxEx<K, V> tx,
-        UUID failedNodeId) {
-        super(cctx.kernalContext(), new SingleReducer<K, V>());
-
-        this.cctx = cctx;
-        this.tx = tx;
-        this.failedNodeId = failedNodeId;
-
-        nodes = new GridLeanMap<>();
-
-        for (ClusterNode node : CU.allNodes(cctx, tx.topologyVersion()))
-            nodes.put(node.id(), node);
-    }
-
-    /**
-     * Initializes future.
-     */
-    public void prepare() {
-        if (log.isDebugEnabled())
-            log.debug("Checking if transaction was committed on remote nodes: " + tx);
-
-        // Check local node first (local node can be a backup node for some part of this transaction).
-        long originatingThreadId = tx.threadId();
-
-        if (tx instanceof IgniteTxRemoteEx)
-            originatingThreadId = ((IgniteTxRemoteEx)tx).remoteThreadId();
-
-        GridCacheCommittedTxInfo<K, V> txInfo = cctx.tm().txCommitted(tx.nearXidVersion(), tx.eventNodeId(),
-            originatingThreadId);
-
-        if (txInfo != null) {
-            onDone(txInfo);
-
-            markInitialized();
-
-            return;
-        }
-
-        Collection<ClusterNode> checkNodes = CU.remoteNodes(cctx, tx.topologyVersion());
-
-        if (tx instanceof GridDhtTxRemote) {
-            // If we got primary node failure and near node has not failed.
-            if (tx.nodeId().equals(failedNodeId) && !tx.eventNodeId().equals(failedNodeId)) {
-                nearCheck = true;
-
-                ClusterNode nearNode = cctx.discovery().node(tx.eventNodeId());
-
-                if (nearNode == null) {
-                    // Near node failed, separate check prepared future will take care of it.
-                    onDone(new ClusterTopologyException("Failed to check near transaction state (near node left grid): " +
-                        tx.eventNodeId()));
-
-                    return;
-                }
-
-                checkNodes = Collections.singletonList(nearNode);
-            }
-        }
-
-        for (ClusterNode rmtNode : checkNodes) {
-            // Skip left nodes and local node.
-            if (rmtNode.id().equals(failedNodeId))
-                continue;
-
-            GridCachePessimisticCheckCommittedTxRequest<K, V> req = new GridCachePessimisticCheckCommittedTxRequest<>(
-                tx,
-                originatingThreadId, futureId(), nearCheck);
-
-            if (rmtNode.isLocal())
-                add(cctx.tm().checkPessimisticTxCommitted(req));
-            else {
-                MiniFuture fut = new MiniFuture(rmtNode.id());
-
-                req.miniId(fut.futureId());
-
-                add(fut);
-
-                try {
-                    cctx.io().send(rmtNode.id(), req);
-                }
-                catch (ClusterTopologyException ignored) {
-                    fut.onNodeLeft();
-                }
-                catch (IgniteCheckedException e) {
-                    fut.onError(e);
-
-                    break;
-                }
-            }
-        }
-
-        markInitialized();
-    }
-
-    /**
-     * @param nodeId Node ID.
-     * @param res Response.
-     */
-    public void onResult(UUID nodeId, GridCachePessimisticCheckCommittedTxResponse<K, V> res) {
-        if (!isDone()) {
-            for (IgniteFuture<GridCacheCommittedTxInfo<K, V>> fut : pending()) {
-                if (isMini(fut)) {
-                    MiniFuture f = (MiniFuture)fut;
-
-                    if (f.futureId().equals(res.miniId())) {
-                        assert f.nodeId().equals(nodeId);
-
-                        f.onResult(res);
-
-                        break;
-                    }
-                }
-            }
-        }
-    }
-
-    /** {@inheritDoc} */
-    @Override public IgniteUuid futureId() {
-        return futId;
-    }
-
-    /** {@inheritDoc} */
-    @Override public GridCacheVersion version() {
-        return tx.xidVersion();
-    }
-
-    /** {@inheritDoc} */
-    @Override public Collection<? extends ClusterNode> nodes() {
-        return nodes.values();
-    }
-
-    /** {@inheritDoc} */
-    @Override public boolean onNodeLeft(UUID nodeId) {
-        for (IgniteFuture<?> fut : futures())
-            if (isMini(fut)) {
-                MiniFuture f = (MiniFuture)fut;
-
-                if (f.nodeId().equals(nodeId)) {
-                    f.onNodeLeft();
-
-                    return true;
-                }
-            }
-
-        return false;
-    }
-
-    /** {@inheritDoc} */
-    @Override public boolean trackable() {
-        return trackable;
-    }
-
-    /** {@inheritDoc} */
-    @Override public void markNotTrackable() {
-        trackable = false;
-    }
-
-    /** {@inheritDoc} */
-    @Override public boolean onDone(@Nullable GridCacheCommittedTxInfo<K, V> res, @Nullable Throwable err) {
-        if (super.onDone(res, err)) {
-            cctx.mvcc().removeFuture(this);
-
-            if (log.isDebugEnabled())
-                log.debug("Completing check committed tx future for transaction [tx=" + tx + ", res=" + res +
-                    ", err=" + err + ']');
-
-            if (err == null)
-                cctx.tm().finishPessimisticTxOnRecovery(tx, res);
-            else {
-                if (log.isDebugEnabled())
-                    log.debug("Failed to check prepared transactions, " +
-                        "invalidating transaction [err=" + err + ", tx=" + tx + ']');
-
-                if (nearCheck)
-                    return true;
-
-                cctx.tm().salvageTx(tx);
-            }
-
-            return true;
-        }
-
-        return false;
-    }
-
-    /**
-     * @param f Future.
-     * @return {@code True} if mini-future.
-     */
-    private boolean isMini(IgniteFuture<?> f) {
-        return f.getClass().equals(MiniFuture.class);
-    }
-
-
-    /** {@inheritDoc} */
-    @Override public String toString() {
-        return S.toString(GridCachePessimisticCheckCommittedTxFuture.class, this, "super", super.toString());
-    }
-
-    /**
-     *
-     */
-    private class MiniFuture extends GridFutureAdapter<GridCacheCommittedTxInfo<K, V>> {
-        /** */
-        private static final long serialVersionUID = 0L;
-
-        /** Mini future ID. */
-        private final IgniteUuid futId = IgniteUuid.randomUuid();
-
-        /** Node ID. */
-        private UUID nodeId;
-
-        /**
-         * Empty constructor required by {@link Externalizable}
-         */
-        public MiniFuture() {
-            // No-op.
-        }
-
-        /**
-         * @param nodeId Node ID.
-         */
-        private MiniFuture(UUID nodeId) {
-            super(cctx.kernalContext());
-
-            this.nodeId = nodeId;
-        }
-
-        /**
-         * @return Node ID.
-         */
-        private UUID nodeId() {
-            return nodeId;
-        }
-
-        /**
-         * @return Future ID.
-         */
-        private IgniteUuid futureId() {
-            return futId;
-        }
-
-        /**
-         * @param e Error.
-         */
-        private void onError(Throwable e) {
-            if (log.isDebugEnabled())
-                log.debug("Failed to get future result [fut=" + this + ", err=" + e + ']');
-
-            onDone(e);
-        }
-
-        /**
-         */
-        private void onNodeLeft() {
-            if (log.isDebugEnabled())
-                log.debug("Transaction node left grid (will ignore) [fut=" + this + ']');
-
-            if (nearCheck) {
-                onDone(new ClusterTopologyException("Failed to check near transaction state (near node left grid): " +
-                    nodeId));
-
-                return;
-            }
-
-            onDone((GridCacheCommittedTxInfo<K, V>)null);
-        }
-
-        /**
-         * @param res Result callback.
-         */
-        private void onResult(GridCachePessimisticCheckCommittedTxResponse<K, V> res) {
-            onDone(res.committedTxInfo());
-        }
-
-        /** {@inheritDoc} */
-        @Override public String toString() {
-            return S.toString(MiniFuture.class, this, "done", isDone(), "err", error());
-        }
-    }
-
-    /**
-     * Single value reducer.
-     */
-    private static class SingleReducer<K, V> implements
-        IgniteReducer<GridCacheCommittedTxInfo<K, V>, GridCacheCommittedTxInfo<K, V>> {
-        /** */
-        private static final long serialVersionUID = 0L;
-
-        /** */
-        private AtomicReference<GridCacheCommittedTxInfo<K, V>> collected = new AtomicReference<>();
-
-        /** {@inheritDoc} */
-        @Override public boolean collect(@Nullable GridCacheCommittedTxInfo<K, V> info) {
-            if (info != null) {
-                collected.compareAndSet(null, info);
-
-                // Stop collecting on first collected info.
-                return false;
-            }
-
-            return true;
-        }
-
-        /** {@inheritDoc} */
-        @Override public GridCacheCommittedTxInfo<K, V> reduce() {
-            return collected.get();
-        }
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0354a41b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCachePessimisticCheckCommittedTxRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCachePessimisticCheckCommittedTxRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCachePessimisticCheckCommittedTxRequest.java
deleted file mode 100644
index 012106d..0000000
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCachePessimisticCheckCommittedTxRequest.java
+++ /dev/null
@@ -1,292 +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.cache.distributed;
-
-import org.apache.ignite.internal.*;
-import org.apache.ignite.internal.processors.cache.*;
-import org.apache.ignite.lang.*;
-import org.apache.ignite.internal.processors.cache.transactions.*;
-import org.apache.ignite.internal.util.direct.*;
-import org.apache.ignite.internal.util.typedef.internal.*;
-
-import java.io.*;
-import java.nio.*;
-import java.util.*;
-
-/**
- * Message sent to check that transactions related to some pessimistic transaction
- * were prepared on remote node.
- */
-public class GridCachePessimisticCheckCommittedTxRequest<K, V> extends GridDistributedBaseMessage<K, V> {
-    /** */
-    private static final long serialVersionUID = 0L;
-
-    /** Future ID. */
-    private IgniteUuid futId;
-
-    /** Mini future ID. */
-    private IgniteUuid miniId;
-
-    /** Near transaction ID. */
-    private GridCacheVersion nearXidVer;
-
-    /** Originating node ID. */
-    private UUID originatingNodeId;
-
-    /** Originating thread ID. */
-    private long originatingThreadId;
-
-    /** Flag indicating that this is near-only check. */
-    @GridDirectVersion(1)
-    private boolean nearOnlyCheck;
-
-    /**
-     * Empty constructor required by {@link Externalizable}
-     */
-    public GridCachePessimisticCheckCommittedTxRequest() {
-        // No-op.
-    }
-
-    /**
-     * @param tx Transaction.
-     * @param originatingThreadId Originating thread ID.
-     * @param futId Future ID.
-     */
-    public GridCachePessimisticCheckCommittedTxRequest(IgniteTxEx<K, V> tx, long originatingThreadId, IgniteUuid futId,
-        boolean nearOnlyCheck) {
-        super(tx.xidVersion(), 0);
-
-        this.futId = futId;
-        this.nearOnlyCheck = nearOnlyCheck;
-
-        nearXidVer = tx.nearXidVersion();
-        originatingNodeId = tx.eventNodeId();
-        this.originatingThreadId = originatingThreadId;
-    }
-
-    /**
-     * @return Near version.
-     */
-    public GridCacheVersion nearXidVersion() {
-        return nearXidVer;
-    }
-
-    /**
-     * @return Tx originating node ID.
-     */
-    public UUID originatingNodeId() {
-        return originatingNodeId;
-    }
-
-    /**
-     * @return Tx originating thread ID.
-     */
-    public long originatingThreadId() {
-        return originatingThreadId;
-    }
-
-    /**
-     * @return Future ID.
-     */
-    public IgniteUuid futureId() {
-        return futId;
-    }
-
-    /**
-     * @return Mini future ID.
-     */
-    public IgniteUuid miniId() {
-        return miniId;
-    }
-
-    /**
-     * @param miniId Mini ID to set.
-     */
-    public void miniId(IgniteUuid miniId) {
-        this.miniId = miniId;
-    }
-
-    /**
-     * @return Flag indicating that this request was sent only to near node. If this flag is set, no finalizing
-     *      will be executed on receiving (near) node since this is a user node.
-     */
-    public boolean nearOnlyCheck() {
-        return nearOnlyCheck;
-    }
-
-    /** {@inheritDoc} */
-    @SuppressWarnings({"CloneDoesntCallSuperClone", "CloneCallsConstructors"})
-    @Override public GridTcpCommunicationMessageAdapter clone() {
-        GridCachePessimisticCheckCommittedTxRequest _clone = new GridCachePessimisticCheckCommittedTxRequest();
-
-        clone0(_clone);
-
-        return _clone;
-    }
-
-    /** {@inheritDoc} */
-    @Override protected void clone0(GridTcpCommunicationMessageAdapter _msg) {
-        super.clone0(_msg);
-
-        GridCachePessimisticCheckCommittedTxRequest _clone = (GridCachePessimisticCheckCommittedTxRequest)_msg;
-
-        _clone.futId = futId;
-        _clone.miniId = miniId;
-        _clone.nearXidVer = nearXidVer;
-        _clone.originatingNodeId = originatingNodeId;
-        _clone.originatingThreadId = originatingThreadId;
-        _clone.nearOnlyCheck = nearOnlyCheck;
-    }
-
-    /** {@inheritDoc} */
-    @SuppressWarnings("all")
-    @Override public boolean writeTo(ByteBuffer buf) {
-        commState.setBuffer(buf);
-
-        if (!super.writeTo(buf))
-            return false;
-
-        if (!commState.typeWritten) {
-            if (!commState.putByte(directType()))
-                return false;
-
-            commState.typeWritten = true;
-        }
-
-        switch (commState.idx) {
-            case 8:
-                if (!commState.putGridUuid(futId))
-                    return false;
-
-                commState.idx++;
-
-            case 9:
-                if (!commState.putGridUuid(miniId))
-                    return false;
-
-                commState.idx++;
-
-            case 10:
-                if (!commState.putCacheVersion(nearXidVer))
-                    return false;
-
-                commState.idx++;
-
-            case 11:
-                if (!commState.putUuid(originatingNodeId))
-                    return false;
-
-                commState.idx++;
-
-            case 12:
-                if (!commState.putLong(originatingThreadId))
-                    return false;
-
-                commState.idx++;
-
-            case 13:
-                if (!commState.putBoolean(nearOnlyCheck))
-                    return false;
-
-                commState.idx++;
-
-        }
-
-        return true;
-    }
-
-    /** {@inheritDoc} */
-    @SuppressWarnings("all")
-    @Override public boolean readFrom(ByteBuffer buf) {
-        commState.setBuffer(buf);
-
-        if (!super.readFrom(buf))
-            return false;
-
-        switch (commState.idx) {
-            case 8:
-                IgniteUuid futId0 = commState.getGridUuid();
-
-                if (futId0 == GRID_UUID_NOT_READ)
-                    return false;
-
-                futId = futId0;
-
-                commState.idx++;
-
-            case 9:
-                IgniteUuid miniId0 = commState.getGridUuid();
-
-                if (miniId0 == GRID_UUID_NOT_READ)
-                    return false;
-
-                miniId = miniId0;
-
-                commState.idx++;
-
-            case 10:
-                GridCacheVersion nearXidVer0 = commState.getCacheVersion();
-
-                if (nearXidVer0 == CACHE_VER_NOT_READ)
-                    return false;
-
-                nearXidVer = nearXidVer0;
-
-                commState.idx++;
-
-            case 11:
-                UUID originatingNodeId0 = commState.getUuid();
-
-                if (originatingNodeId0 == UUID_NOT_READ)
-                    return false;
-
-                originatingNodeId = originatingNodeId0;
-
-                commState.idx++;
-
-            case 12:
-                if (buf.remaining() < 8)
-                    return false;
-
-                originatingThreadId = commState.getLong();
-
-                commState.idx++;
-
-            case 13:
-                if (buf.remaining() < 1)
-                    return false;
-
-                nearOnlyCheck = commState.getBoolean();
-
-                commState.idx++;
-
-        }
-
-        return true;
-    }
-
-    /** {@inheritDoc} */
-    @Override public byte directType() {
-        return 20;
-    }
-
-    /** {@inheritDoc} */
-    @Override public String toString() {
-        return S.toString(GridCachePessimisticCheckCommittedTxRequest.class, this, "super", super.toString());
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0354a41b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCachePessimisticCheckCommittedTxResponse.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCachePessimisticCheckCommittedTxResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCachePessimisticCheckCommittedTxResponse.java
deleted file mode 100644
index 8b50645..0000000
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCachePessimisticCheckCommittedTxResponse.java
+++ /dev/null
@@ -1,231 +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.cache.distributed;
-
-import org.apache.ignite.*;
-import org.apache.ignite.internal.*;
-import org.apache.ignite.internal.processors.cache.*;
-import org.apache.ignite.lang.*;
-import org.apache.ignite.internal.util.direct.*;
-import org.apache.ignite.internal.util.typedef.internal.*;
-import org.jetbrains.annotations.*;
-
-import java.io.*;
-import java.nio.*;
-
-/**
- * Check prepared transactions response.
- */
-public class GridCachePessimisticCheckCommittedTxResponse<K, V> extends GridDistributedBaseMessage<K, V> {
-    /** */
-    private static final long serialVersionUID = 0L;
-
-    /** Future ID. */
-    private IgniteUuid futId;
-
-    /** Mini future ID. */
-    private IgniteUuid miniId;
-
-    /** Committed transaction info. */
-    @GridDirectTransient
-    private GridCacheCommittedTxInfo<K, V> committedTxInfo;
-
-    /** Serialized transaction info. */
-    private byte[] committedTxInfoBytes;
-
-    /**
-     * Empty constructor required by {@link Externalizable}
-     */
-    public GridCachePessimisticCheckCommittedTxResponse() {
-        // No-op.
-    }
-
-    /**
-     * @param txId Transaction ID.
-     * @param futId Future ID.
-     * @param miniId Mini future ID.
-     * @param committedTxInfo Committed transaction info.
-     */
-    public GridCachePessimisticCheckCommittedTxResponse(GridCacheVersion txId, IgniteUuid futId, IgniteUuid miniId,
-        @Nullable GridCacheCommittedTxInfo<K, V> committedTxInfo) {
-        super(txId, 0);
-
-        this.futId = futId;
-        this.miniId = miniId;
-        this.committedTxInfo = committedTxInfo;
-    }
-
-    /**
-     * @return Future ID.
-     */
-    public IgniteUuid futureId() {
-        return futId;
-    }
-
-    /**
-     * @return Mini future ID.
-     */
-    public IgniteUuid miniId() {
-        return miniId;
-    }
-
-    /**
-     * @return {@code True} if all remote transactions were prepared.
-     */
-    public GridCacheCommittedTxInfo<K, V> committedTxInfo() {
-        return committedTxInfo;
-    }
-
-    /** {@inheritDoc}
-     * @param ctx*/
-    @Override public void prepareMarshal(GridCacheSharedContext<K, V> ctx) throws IgniteCheckedException {
-        super.prepareMarshal(ctx);
-
-        if (committedTxInfo != null) {
-            marshalTx(committedTxInfo.recoveryWrites(), ctx);
-
-            committedTxInfoBytes = ctx.marshaller().marshal(committedTxInfo);
-        }
-    }
-
-    /** {@inheritDoc} */
-    @Override public void finishUnmarshal(GridCacheSharedContext<K, V> ctx, ClassLoader ldr) throws IgniteCheckedException {
-        super.finishUnmarshal(ctx, ldr);
-
-        if (committedTxInfoBytes != null) {
-            committedTxInfo = ctx.marshaller().unmarshal(committedTxInfoBytes, ldr);
-
-            unmarshalTx(committedTxInfo.recoveryWrites(), false, ctx, ldr);
-        }
-    }
-
-    /** {@inheritDoc} */
-    @SuppressWarnings({"CloneDoesntCallSuperClone", "CloneCallsConstructors"})
-    @Override public GridTcpCommunicationMessageAdapter clone() {
-        GridCachePessimisticCheckCommittedTxResponse _clone = new GridCachePessimisticCheckCommittedTxResponse();
-
-        clone0(_clone);
-
-        return _clone;
-    }
-
-    /** {@inheritDoc} */
-    @Override protected void clone0(GridTcpCommunicationMessageAdapter _msg) {
-        super.clone0(_msg);
-
-        GridCachePessimisticCheckCommittedTxResponse _clone = (GridCachePessimisticCheckCommittedTxResponse)_msg;
-
-        _clone.futId = futId;
-        _clone.miniId = miniId;
-        _clone.committedTxInfo = committedTxInfo;
-        _clone.committedTxInfoBytes = committedTxInfoBytes;
-    }
-
-    /** {@inheritDoc} */
-    @SuppressWarnings("all")
-    @Override public boolean writeTo(ByteBuffer buf) {
-        commState.setBuffer(buf);
-
-        if (!super.writeTo(buf))
-            return false;
-
-        if (!commState.typeWritten) {
-            if (!commState.putByte(directType()))
-                return false;
-
-            commState.typeWritten = true;
-        }
-
-        switch (commState.idx) {
-            case 8:
-                if (!commState.putByteArray(committedTxInfoBytes))
-                    return false;
-
-                commState.idx++;
-
-            case 9:
-                if (!commState.putGridUuid(futId))
-                    return false;
-
-                commState.idx++;
-
-            case 10:
-                if (!commState.putGridUuid(miniId))
-                    return false;
-
-                commState.idx++;
-
-        }
-
-        return true;
-    }
-
-    /** {@inheritDoc} */
-    @SuppressWarnings("all")
-    @Override public boolean readFrom(ByteBuffer buf) {
-        commState.setBuffer(buf);
-
-        if (!super.readFrom(buf))
-            return false;
-
-        switch (commState.idx) {
-            case 8:
-                byte[] committedTxInfoBytes0 = commState.getByteArray();
-
-                if (committedTxInfoBytes0 == BYTE_ARR_NOT_READ)
-                    return false;
-
-                committedTxInfoBytes = committedTxInfoBytes0;
-
-                commState.idx++;
-
-            case 9:
-                IgniteUuid futId0 = commState.getGridUuid();
-
-                if (futId0 == GRID_UUID_NOT_READ)
-                    return false;
-
-                futId = futId0;
-
-                commState.idx++;
-
-            case 10:
-                IgniteUuid miniId0 = commState.getGridUuid();
-
-                if (miniId0 == GRID_UUID_NOT_READ)
-                    return false;
-
-                miniId = miniId0;
-
-                commState.idx++;
-
-        }
-
-        return true;
-    }
-
-    /** {@inheritDoc} */
-    @Override public byte directType() {
-        return 21;
-    }
-
-    /** {@inheritDoc} */
-    @Override public String toString() {
-        return S.toString(GridCachePessimisticCheckCommittedTxResponse.class, this, "super", super.toString());
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0354a41b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheTxCommitBuffer.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheTxCommitBuffer.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheTxCommitBuffer.java
deleted file mode 100644
index 26a78f5..0000000
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheTxCommitBuffer.java
+++ /dev/null
@@ -1,60 +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.cache.distributed;
-
-import org.apache.ignite.internal.processors.cache.*;
-import org.apache.ignite.internal.processors.cache.transactions.*;
-import org.jetbrains.annotations.*;
-
-import java.util.*;
-
-/**
- * Buffer that stores transaction commit values in order to restore them in case of originating node crash.
- */
-public interface GridCacheTxCommitBuffer<K, V> {
-    /**
-     * Adds committed transaction to commit buffer.
-     *
-     * @param tx Committed transaction.
-     */
-    public void addCommittedTx(IgniteTxEx<K, V> tx);
-
-    /**
-     * Gets transaction from commit buffer.
-     *
-     * @param originatingTxVer Originating tx version.
-     * @param nodeId Originating node ID.
-     * @param threadId Originating thread ID.
-     * @return Committed info, if any.
-     */
-    @Nullable public GridCacheCommittedTxInfo<K, V> committedTx(GridCacheVersion originatingTxVer, UUID nodeId,
-        long threadId);
-
-    /**
-     * Callback called when lode left grid. Used to eventually cleanup the queue from committed tx info from
-     * left node.
-     *
-     * @param nodeId Left node ID.
-     */
-    public void onNodeLeft(UUID nodeId);
-
-    /**
-     * @return Buffer size.
-     */
-    public int size();
-}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0354a41b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedLockRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedLockRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedLockRequest.java
index 4356dc4..da6ca72 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedLockRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedLockRequest.java
@@ -73,14 +73,6 @@ public class GridDistributedLockRequest<K, V> extends GridDistributedBaseMessage
     @GridDirectTransient
     private List<K> keys;
 
-    /** Write entries. */
-    @GridToStringInclude
-    @GridDirectTransient
-    private List<IgniteTxEntry<K, V>> writeEntries;
-
-    /** Serialized write entries. */
-    private byte[] writeEntriesBytes;
-
     /** Array indicating whether value should be returned for a key. */
     @GridToStringInclude
     private boolean[] retVals;
@@ -102,10 +94,6 @@ public class GridDistributedLockRequest<K, V> extends GridDistributedBaseMessage
     /** Partition lock flag. Only if group-lock transaction. */
     private boolean partLock;
 
-    /** DR versions. */
-    @GridToStringInclude
-    private GridCacheVersion[] drVersByIdx;
-
     /**
      * Empty constructor.
      */
@@ -252,13 +240,6 @@ public class GridDistributedLockRequest<K, V> extends GridDistributedBaseMessage
     }
 
     /**
-     * @return Write entries list.
-     */
-    public List<IgniteTxEntry<K, V>> writeEntries() {
-        return writeEntries;
-    }
-
-    /**
      * @return Tx size.
      */
     public int txSize() {
@@ -271,19 +252,15 @@ public class GridDistributedLockRequest<K, V> extends GridDistributedBaseMessage
      * @param key Key.
      * @param retVal Flag indicating whether value should be returned.
      * @param keyBytes Key bytes.
-     * @param writeEntry Write entry.
      * @param cands Candidates.
-     * @param drVer DR version.
      * @param ctx Context.
      * @throws IgniteCheckedException If failed.
      */
     public void addKeyBytes(
         K key,
         @Nullable byte[] keyBytes,
-        @Nullable IgniteTxEntry<K, V> writeEntry,
         boolean retVal,
         @Nullable Collection<GridCacheMvccCandidate<K>> cands,
-        @Nullable GridCacheVersion drVer,
         GridCacheContext<K, V> ctx
     ) throws IgniteCheckedException {
         if (ctx.deploymentEnabled())
@@ -302,21 +279,9 @@ public class GridDistributedLockRequest<K, V> extends GridDistributedBaseMessage
         keys.add(key);
 
         candidatesByIndex(idx, cands);
-        drVersionByIndex(idx, drVer);
 
         retVals[idx] = retVal;
 
-        if (writeEntry != null) {
-            if (writeEntries == null) {
-                assert idx == 0 : "Cannot start adding write entries in the middle of lock message [idx=" + idx +
-                    ", writeEntry=" + writeEntry + ']';
-
-                writeEntries = new ArrayList<>(keysCount());
-            }
-
-            writeEntries.add(writeEntry);
-        }
-
         idx++;
     }
 
@@ -355,39 +320,6 @@ public class GridDistributedLockRequest<K, V> extends GridDistributedBaseMessage
         return timeout;
     }
 
-    /**
-     * @param idx Key index.
-     * @param drVer DR version.
-     */
-    @SuppressWarnings({"unchecked"})
-    public void drVersionByIndex(int idx, GridCacheVersion drVer) {
-        assert idx < keysCount();
-
-        // If nothing to add.
-        if (drVer == null)
-            return;
-
-        if (drVersByIdx == null)
-            drVersByIdx = new GridCacheVersion[keysCount()];
-
-        drVersByIdx[idx] = drVer;
-    }
-
-    /**
-     * @param idx Key index.
-     * @return DR versions for given key.
-     */
-    public GridCacheVersion drVersionByIndex(int idx) {
-        return drVersByIdx == null ? null : drVersByIdx[idx];
-    }
-
-    /**
-     * @return All DR versions.
-     */
-    public GridCacheVersion[] drVersions() {
-        return drVersByIdx;
-    }
-
     /** {@inheritDoc}
      * @param ctx*/
     @Override public void prepareMarshal(GridCacheSharedContext<K, V> ctx) throws IgniteCheckedException {
@@ -399,12 +331,6 @@ public class GridDistributedLockRequest<K, V> extends GridDistributedBaseMessage
 
             grpLockKeyBytes = CU.marshal(ctx, grpLockKey);
         }
-
-        if (writeEntries != null) {
-            marshalTx(writeEntries, ctx);
-
-            writeEntriesBytes = ctx.marshaller().marshal(writeEntries);
-        }
     }
 
     /** {@inheritDoc} */
@@ -416,12 +342,6 @@ public class GridDistributedLockRequest<K, V> extends GridDistributedBaseMessage
 
         if (grpLockKey == null && grpLockKeyBytes != null)
             grpLockKey = ctx.marshaller().unmarshal(grpLockKeyBytes, ldr);
-
-        if (writeEntriesBytes != null) {
-            writeEntries = ctx.marshaller().unmarshal(writeEntriesBytes, ldr);
-
-            unmarshalTx(writeEntries, false, ctx, ldr);
-        }
     }
 
     /** {@inheritDoc} */
@@ -452,15 +372,12 @@ public class GridDistributedLockRequest<K, V> extends GridDistributedBaseMessage
         _clone.isolation = isolation;
         _clone.keyBytes = keyBytes;
         _clone.keys = keys;
-        _clone.writeEntries = writeEntries;
-        _clone.writeEntriesBytes = writeEntriesBytes;
         _clone.retVals = retVals;
         _clone.idx = idx;
         _clone.txSize = txSize;
         _clone.grpLockKey = grpLockKey;
         _clone.grpLockKeyBytes = grpLockKeyBytes;
         _clone.partLock = partLock;
-        _clone.drVersByIdx = drVersByIdx;
     }
 
     /** {@inheritDoc} */

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0354a41b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxFinishRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxFinishRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxFinishRequest.java
index 3792536..389cf30 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxFinishRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxFinishRequest.java
@@ -24,7 +24,6 @@ import org.apache.ignite.lang.*;
 import org.apache.ignite.internal.processors.cache.transactions.*;
 import org.apache.ignite.internal.util.direct.*;
 import org.apache.ignite.internal.util.tostring.*;
-import org.apache.ignite.internal.util.typedef.*;
 import org.apache.ignite.internal.util.typedef.internal.*;
 import org.jetbrains.annotations.*;
 
@@ -63,24 +62,6 @@ public class GridDistributedTxFinishRequest<K, V> extends GridDistributedBaseMes
     /** Min version used as base for completed versions. */
     private GridCacheVersion baseVer;
 
-    /** Transaction write entries. */
-    @GridToStringInclude
-    @GridDirectTransient
-    private Collection<IgniteTxEntry<K, V>> writeEntries;
-
-    /** */
-    @GridDirectCollection(byte[].class)
-    private Collection<byte[]> writeEntriesBytes;
-
-    /** Write entries which have not been transferred to nodes during lock request. */
-    @GridToStringInclude
-    @GridDirectTransient
-    private Collection<IgniteTxEntry<K, V>> recoveryWrites;
-
-    /** */
-    @GridDirectCollection(byte[].class)
-    private Collection<byte[]> recoveryWritesBytes;
-
     /** Expected txSize. */
     private int txSize;
 
@@ -113,9 +94,6 @@ public class GridDistributedTxFinishRequest<K, V> extends GridDistributedBaseMes
      * @param committedVers Committed versions.
      * @param rolledbackVers Rolled back versions.
      * @param txSize Expected transaction size.
-     * @param writeEntries Write entries.
-     * @param recoveryWrites Recover entries. In pessimistic mode entries which were not transferred to remote nodes
-     *      with lock requests. {@code Null} for optimistic mode.
      * @param grpLockKey Group lock key if this is a group-lock transaction.
      */
     public GridDistributedTxFinishRequest(
@@ -132,11 +110,9 @@ public class GridDistributedTxFinishRequest<K, V> extends GridDistributedBaseMes
         Collection<GridCacheVersion> committedVers,
         Collection<GridCacheVersion> rolledbackVers,
         int txSize,
-        Collection<IgniteTxEntry<K, V>> writeEntries,
-        Collection<IgniteTxEntry<K, V>> recoveryWrites,
         @Nullable IgniteTxKey grpLockKey
     ) {
-        super(xidVer, writeEntries == null ? 0 : writeEntries.size());
+        super(xidVer, 0);
         assert xidVer != null;
 
         this.futId = futId;
@@ -149,36 +125,12 @@ public class GridDistributedTxFinishRequest<K, V> extends GridDistributedBaseMes
         this.syncRollback = syncRollback;
         this.baseVer = baseVer;
         this.txSize = txSize;
-        this.writeEntries = writeEntries;
-        this.recoveryWrites = recoveryWrites;
         this.grpLockKey = grpLockKey;
 
         completedVersions(committedVers, rolledbackVers);
     }
 
     /**
-     * Clones write entries so that near entries are not passed to DHT cache.
-     */
-    public void cloneEntries() {
-        if (F.isEmpty(writeEntries))
-            return;
-
-        Collection<IgniteTxEntry<K, V>> cp = new ArrayList<>(writeEntries.size());
-
-        for (IgniteTxEntry<K, V> e : writeEntries) {
-            GridCacheContext<K, V> cacheCtx = e.context();
-
-            // Clone only if it is a near cache.
-            if (cacheCtx.isNear())
-                cp.add(e.cleanCopy(cacheCtx.nearTx().dht().context()));
-            else
-                cp.add(e);
-        }
-
-        writeEntries = cp;
-    }
-
-    /**
      * @return System flag.
      */
     public boolean system() {
@@ -290,24 +242,6 @@ public class GridDistributedTxFinishRequest<K, V> extends GridDistributedBaseMes
     @Override public void prepareMarshal(GridCacheSharedContext<K, V> ctx) throws IgniteCheckedException {
         super.prepareMarshal(ctx);
 
-        if (writeEntries != null) {
-            marshalTx(writeEntries, ctx);
-
-            writeEntriesBytes = new ArrayList<>(writeEntries.size());
-
-            for (IgniteTxEntry<K, V> e : writeEntries)
-                writeEntriesBytes.add(ctx.marshaller().marshal(e));
-        }
-
-        if (recoveryWrites != null) {
-            marshalTx(recoveryWrites, ctx);
-
-            recoveryWritesBytes = new ArrayList<>(recoveryWrites.size());
-
-            for (IgniteTxEntry<K, V> e : recoveryWrites)
-                recoveryWritesBytes.add(ctx.marshaller().marshal(e));
-        }
-
         if (grpLockKey != null && grpLockKeyBytes == null) {
             if (ctx.deploymentEnabled())
                 prepareObject(grpLockKey, ctx);
@@ -320,24 +254,6 @@ public class GridDistributedTxFinishRequest<K, V> extends GridDistributedBaseMes
     @Override public void finishUnmarshal(GridCacheSharedContext<K, V> ctx, ClassLoader ldr) throws IgniteCheckedException {
         super.finishUnmarshal(ctx, ldr);
 
-        if (writeEntriesBytes != null) {
-            writeEntries = new ArrayList<>(writeEntriesBytes.size());
-
-            for (byte[] arr : writeEntriesBytes)
-                writeEntries.add(ctx.marshaller().<IgniteTxEntry<K, V>>unmarshal(arr, ldr));
-
-            unmarshalTx(writeEntries, false, ctx, ldr);
-        }
-
-        if (recoveryWritesBytes != null) {
-            recoveryWrites = new ArrayList<>(recoveryWritesBytes.size());
-
-            for (byte[] arr : recoveryWritesBytes)
-                recoveryWrites.add(ctx.marshaller().<IgniteTxEntry<K, V>>unmarshal(arr, ldr));
-
-            unmarshalTx(recoveryWrites, false, ctx, ldr);
-        }
-
         if (grpLockKeyBytes != null && grpLockKey == null)
             grpLockKey = ctx.marshaller().unmarshal(grpLockKeyBytes, ldr);
     }
@@ -365,10 +281,6 @@ public class GridDistributedTxFinishRequest<K, V> extends GridDistributedBaseMes
         _clone.invalidate = invalidate;
         _clone.commit = commit;
         _clone.baseVer = baseVer;
-        _clone.writeEntries = writeEntries;
-        _clone.writeEntriesBytes = writeEntriesBytes;
-        _clone.recoveryWrites = recoveryWrites;
-        _clone.recoveryWritesBytes = recoveryWritesBytes;
         _clone.txSize = txSize;
         _clone.grpLockKey = grpLockKey;
         _clone.grpLockKeyBytes = grpLockKeyBytes;

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0354a41b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxPrepareRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxPrepareRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxPrepareRequest.java
index f0655b9..5fc3607 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxPrepareRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxPrepareRequest.java
@@ -53,7 +53,7 @@ public class GridDistributedTxPrepareRequest<K, V> extends GridDistributedBaseMe
 
     /** Commit version for EC transactions. */
     @GridToStringInclude
-    private GridCacheVersion commitVer;
+    private GridCacheVersion writeVer;
 
     /** Transaction timeout. */
     @GridToStringInclude
@@ -112,6 +112,9 @@ public class GridDistributedTxPrepareRequest<K, V> extends GridDistributedBaseMe
     /** */
     private byte[] txNodesBytes;
 
+    /** One phase commit flag. */
+    private boolean onePhaseCommit;
+
     /** System flag. */
     private boolean sys;
 
@@ -129,6 +132,7 @@ public class GridDistributedTxPrepareRequest<K, V> extends GridDistributedBaseMe
      * @param grpLockKey Group lock key.
      * @param partLock {@code True} if preparing group-lock transaction with partition lock.
      * @param txNodes Transaction nodes mapping.
+     * @param onePhaseCommit One phase commit flag.
      */
     public GridDistributedTxPrepareRequest(
         IgniteTxEx<K, V> tx,
@@ -136,11 +140,12 @@ public class GridDistributedTxPrepareRequest<K, V> extends GridDistributedBaseMe
         Collection<IgniteTxEntry<K, V>> writes,
         IgniteTxKey grpLockKey,
         boolean partLock,
-        Map<UUID, Collection<UUID>> txNodes
+        Map<UUID, Collection<UUID>> txNodes,
+        boolean onePhaseCommit
     ) {
         super(tx.xidVersion(), 0);
 
-        commitVer = null;
+        writeVer = tx.writeVersion();
         threadId = tx.threadId();
         concurrency = tx.concurrency();
         isolation = tx.isolation();
@@ -154,6 +159,29 @@ public class GridDistributedTxPrepareRequest<K, V> extends GridDistributedBaseMe
         this.grpLockKey = grpLockKey;
         this.partLock = partLock;
         this.txNodes = txNodes;
+        this.onePhaseCommit = onePhaseCommit;
+    }
+
+    /**
+     * Clones write entries so that near entries are not passed to DHT cache.
+     */
+    public void cloneEntries() {
+        if (F.isEmpty(writes))
+            return;
+
+        Collection<IgniteTxEntry<K, V>> cp = new ArrayList<>(writes.size());
+
+        for (IgniteTxEntry<K, V> e : writes) {
+            GridCacheContext<K, V> cacheCtx = e.context();
+
+            // Clone only if it is a near cache.
+            if (cacheCtx.isNear())
+                cp.add(e.cleanCopy(cacheCtx.nearTx().dht().context()));
+            else
+                cp.add(e);
+        }
+
+        writes = cp;
     }
 
     /**
@@ -200,7 +228,7 @@ public class GridDistributedTxPrepareRequest<K, V> extends GridDistributedBaseMe
     /**
      * @return Commit version.
      */
-    public GridCacheVersion commitVersion() { return commitVer; }
+    public GridCacheVersion writeVersion() { return writeVer; }
 
     /**
      * @return Invalidate flag.
@@ -277,6 +305,13 @@ public class GridDistributedTxPrepareRequest<K, V> extends GridDistributedBaseMe
         return txSize;
     }
 
+    /**
+     * @return One phase commit flag.
+     */
+    public boolean onePhaseCommit() {
+        return onePhaseCommit;
+    }
+
     /** {@inheritDoc}
      * @param ctx*/
     @Override public void prepareMarshal(GridCacheSharedContext<K, V> ctx) throws IgniteCheckedException {
@@ -420,7 +455,7 @@ public class GridDistributedTxPrepareRequest<K, V> extends GridDistributedBaseMe
         _clone.threadId = threadId;
         _clone.concurrency = concurrency;
         _clone.isolation = isolation;
-        _clone.commitVer = commitVer;
+        _clone.writeVer = writeVer;
         _clone.timeout = timeout;
         _clone.invalidate = invalidate;
         _clone.reads = reads;
@@ -455,7 +490,7 @@ public class GridDistributedTxPrepareRequest<K, V> extends GridDistributedBaseMe
 
         switch (commState.idx) {
             case 8:
-                if (!commState.putCacheVersion(commitVer))
+                if (!commState.putCacheVersion(writeVer))
                     return false;
 
                 commState.idx++;
@@ -599,7 +634,7 @@ public class GridDistributedTxPrepareRequest<K, V> extends GridDistributedBaseMe
                 if (commitVer0 == CACHE_VER_NOT_READ)
                     return false;
 
-                commitVer = commitVer0;
+                writeVer = commitVer0;
 
                 commState.idx++;
 

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0354a41b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java
index 9d07c52..a3ea0502 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java
@@ -503,34 +503,33 @@ public class GridDistributedTxRemoteAdapter<K, V> extends IgniteTxAdapter<K, V>
 
                                     GridCacheVersion explicitVer = txEntry.drVersion();
 
-                                    if (finalizationStatus() == FinalizationStatus.RECOVERY_FINISH || optimistic()) {
-                                        // Primary node has left the grid so we have to process conflicts on backups.
-                                        if (explicitVer == null)
-                                            explicitVer = writeVersion(); // Force write version to be used.
-
-                                        GridDrResolveResult<V> drRes = cacheCtx.dr().resolveTx(cached,
-                                            txEntry,
-                                            explicitVer,
-                                            op,
-                                            val,
-                                            valBytes,
-                                            txEntry.ttl(),
-                                            txEntry.drExpireTime());
-
-                                        if (drRes != null) {
-                                            op = drRes.operation();
-                                            val = drRes.value();
-                                            valBytes = drRes.valueBytes();
-
-                                            if (drRes.isMerge())
-                                                explicitVer = writeVersion();
-                                            else if (op == NOOP)
-                                                txEntry.ttl(-1L);
-                                        }
-                                        else
-                                            // Nullify explicit version so that innerSet/innerRemove will work as usual.
-                                            explicitVer = null;
+
+                                    // Primary node has left the grid so we have to process conflicts on backups.
+                                    if (explicitVer == null)
+                                        explicitVer = writeVersion(); // Force write version to be used.
+
+                                    GridDrResolveResult<V> drRes = cacheCtx.dr().resolveTx(cached,
+                                        txEntry,
+                                        explicitVer,
+                                        op,
+                                        val,
+                                        valBytes,
+                                        txEntry.ttl(),
+                                        txEntry.drExpireTime());
+
+                                    if (drRes != null) {
+                                        op = drRes.operation();
+                                        val = drRes.value();
+                                        valBytes = drRes.valueBytes();
+
+                                        if (drRes.isMerge())
+                                            explicitVer = writeVersion();
+                                        else if (op == NOOP)
+                                            txEntry.ttl(-1L);
                                     }
+                                    else
+                                        // Nullify explicit version so that innerSet/innerRemove will work as usual.
+                                        explicitVer = null;
 
                                     if (op == CREATE || op == UPDATE) {
                                         // Invalidate only for near nodes (backups cannot be invalidated).

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0354a41b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockFuture.java
index 74d9163..adba497 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockFuture.java
@@ -779,8 +779,8 @@ public final class GridDhtLockFuture<K, V> extends GridCompoundIdentityFuture<Bo
             }
 
             if (tx != null) {
-                tx.addDhtMapping(dhtMap);
-                tx.addNearMapping(nearMap);
+                tx.addDhtNodeEntryMapping(dhtMap);
+                tx.addNearNodeEntryMapping(nearMap);
 
                 tx.needsCompletedVersions(hasRmtNodes);
             }
@@ -851,6 +851,29 @@ public final class GridDhtLockFuture<K, V> extends GridCompoundIdentityFuture<Bo
                             // Must unswap entry so that isNewLocked returns correct value.
                             e.unswap(true, false);
 
+                            boolean needVal = false;
+
+                            try {
+                                needVal = e.isNewLocked();
+
+                                if (needVal) {
+                                    List<ClusterNode> owners = cctx.topology().owners(e.partition(),
+                                        tx != null ? tx.topologyVersion() : cctx.affinity().affinityTopologyVersion());
+
+                                    // Do not preload if local node is partition owner.
+                                    if (owners.contains(cctx.localNode()))
+                                        needVal = false;
+                                }
+                            }
+                            catch (GridCacheEntryRemovedException ex) {
+                                assert false : "Entry cannot become obsolete when DHT local candidate is added " +
+                                    "[e=" + e + ", ex=" + ex + ']';
+                            }
+
+                            // Skip entry if it is not new and is not present in updated mapping.
+                            if (tx != null && !needVal)
+                                continue;
+
                             boolean invalidateRdr = e.readerId(n.id()) != null;
 
                             IgniteTxEntry<K, V> entry = tx != null ? tx.entry(e.txKey()) : null;
@@ -858,20 +881,12 @@ public final class GridDhtLockFuture<K, V> extends GridCompoundIdentityFuture<Bo
                             req.addDhtKey(
                                 e.key(),
                                 e.getOrMarshalKeyBytes(),
-                                tx != null ? tx.writeMap().get(e.txKey()) : null,
-                                entry != null ? entry.drVersion() : null,
                                 invalidateRdr,
                                 cctx);
 
-                            try {
-                                if (e.isNewLocked())
-                                    // Mark last added key as needed to be preloaded.
-                                    req.markLastKeyForPreload();
-                            }
-                            catch (GridCacheEntryRemovedException ex) {
-                                assert false : "Entry cannot become obsolete when DHT local candidate is added " +
-                                    "[e=" + e + ", ex=" + ex + ']';
-                            }
+                            if (needVal)
+                                // Mark last added key as needed to be preloaded.
+                                req.markLastKeyForPreload();
 
                             it.set(addOwned(req, e));
                         }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0354a41b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockRequest.java
index a4efd13..050748a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockRequest.java
@@ -237,8 +237,6 @@ public class GridDhtLockRequest<K, V> extends GridDistributedLockRequest<K, V> {
      *
      * @param key Key.
      * @param keyBytes Key bytes.
-     * @param writeEntry Write entry.
-     * @param drVer DR version.
      * @param invalidateEntry Flag indicating whether node should attempt to invalidate reader.
      * @param ctx Context.
      * @throws IgniteCheckedException If failed.
@@ -246,14 +244,12 @@ public class GridDhtLockRequest<K, V> extends GridDistributedLockRequest<K, V> {
     public void addDhtKey(
         K key,
         byte[] keyBytes,
-        IgniteTxEntry<K, V> writeEntry,
-        @Nullable GridCacheVersion drVer,
         boolean invalidateEntry,
         GridCacheContext<K, V> ctx
     ) throws IgniteCheckedException {
         invalidateEntries.set(idx, invalidateEntry);
 
-        addKeyBytes(key, keyBytes, writeEntry, false, null, drVer, ctx);
+        addKeyBytes(key, keyBytes, false, null, ctx);
     }
 
     /**

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0354a41b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java
index f6e69eb..4e06bc7 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java
@@ -136,8 +136,6 @@ public abstract class GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
         GridDhtLockResponse<K, V> res)
         throws IgniteCheckedException, GridDistributedLockCancelledException {
         List<K> keys = req.keys();
-        List<IgniteTxEntry<K, V>> writes = req.writeEntries();
-
         GridDhtTxRemote<K, V> tx = null;
 
         int size = F.size(keys);
@@ -150,12 +148,8 @@ public abstract class GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
 
             IgniteTxKey<K> txKey = ctx.txKey(key);
 
-            IgniteTxEntry<K, V> writeEntry = writes == null ? null : writes.get(i);
-
             assert F.isEmpty(req.candidatesByIndex(i));
 
-            GridCacheVersion drVer = req.drVersionByIndex(i);
-
             if (log.isDebugEnabled())
                 log.debug("Unmarshalled key: " + key);
 
@@ -218,13 +212,12 @@ public abstract class GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
 
                             tx.addWrite(
                                 ctx,
-                                writeEntry == null ? NOOP : writeEntry.op(),
+                                NOOP,
                                 txKey,
                                 req.keyBytes() != null ? req.keyBytes().get(i) : null,
-                                writeEntry == null ? null : writeEntry.value(),
-                                writeEntry == null ? null : writeEntry.valueBytes(),
-                                writeEntry == null ? null : writeEntry.entryProcessors(),
-                                drVer,
+                                null,
+                                null,
+                                null,
                                 req.accessTtl());
 
                             if (req.groupLock())
@@ -828,14 +821,10 @@ public abstract class GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
                             if (log.isDebugEnabled())
                                 log.debug("Performing DHT lock [tx=" + tx + ", entries=" + entries + ']');
 
-                            assert req.writeEntries() == null || req.writeEntries().size() == entries.size();
-
                             IgniteFuture<GridCacheReturn<V>> txFut = tx.lockAllAsync(
                                 cacheCtx,
                                 entries,
-                                req.writeEntries(),
                                 req.onePhaseCommit(),
-                                req.drVersions(),
                                 req.messageId(),
                                 req.implicitTx(),
                                 req.txRead(),

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0354a41b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishFuture.java
index ea68904..fdf6e87 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishFuture.java
@@ -92,8 +92,6 @@ public final class GridDhtTxFinishFuture<K, V> extends GridCompoundIdentityFutur
     public GridDhtTxFinishFuture(GridCacheSharedContext<K, V> cctx, GridDhtTxLocalAdapter<K, V> tx, boolean commit) {
         super(cctx.kernalContext(), F.<IgniteTx>identityReducer(tx));
 
-        assert cctx != null;
-
         this.cctx = cctx;
         this.tx = tx;
         this.commit = commit;
@@ -283,6 +281,9 @@ public final class GridDhtTxFinishFuture<K, V> extends GridCompoundIdentityFutur
      */
     private boolean finish(Map<UUID, GridDistributedTxMapping<K, V>> dhtMap,
         Map<UUID, GridDistributedTxMapping<K, V>> nearMap) {
+        if (tx.onePhaseCommit())
+            return false;
+
         boolean res = false;
 
         boolean sync = commit ? tx.syncCommit() : tx.syncRollback();
@@ -323,10 +324,6 @@ public final class GridDhtTxFinishFuture<K, V> extends GridCompoundIdentityFutur
                 tx.rolledbackVersions(),
                 tx.pendingVersions(),
                 tx.size(),
-                tx.pessimistic() ? dhtMapping.writes() : null,
-                tx.pessimistic() && nearMapping != null ? nearMapping.writes() : null,
-                tx.recoveryWrites(),
-                tx.onePhaseCommit(),
                 tx.groupLockKey(),
                 tx.subjectId(),
                 tx.taskNameHash());
@@ -345,9 +342,6 @@ public final class GridDhtTxFinishFuture<K, V> extends GridCompoundIdentityFutur
                 }
             }
 
-            if (tx.onePhaseCommit())
-                req.writeVersion(tx.writeVersion());
-
             try {
                 cctx.io().send(n, req, tx.system() ? UTILITY_CACHE_POOL : SYSTEM_POOL);
 
@@ -395,10 +389,6 @@ public final class GridDhtTxFinishFuture<K, V> extends GridCompoundIdentityFutur
                     tx.rolledbackVersions(),
                     tx.pendingVersions(),
                     tx.size(),
-                    null,
-                    tx.pessimistic() ? nearMapping.writes() : null,
-                    tx.recoveryWrites(),
-                    tx.onePhaseCommit(),
                     tx.groupLockKey(),
                     tx.subjectId(),
                     tx.taskNameHash());

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/0354a41b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishRequest.java
index da02065..1d311d4 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishRequest.java
@@ -47,15 +47,6 @@ public class GridDhtTxFinishRequest<K, V> extends GridDistributedTxFinishRequest
     /** Transaction isolation. */
     private IgniteTxIsolation isolation;
 
-    /** Near writes. */
-    @GridToStringInclude
-    @GridDirectTransient
-    private Collection<IgniteTxEntry<K, V>> nearWrites;
-
-    /** Serialized near writes. */
-    @GridDirectCollection(byte[].class)
-    private Collection<byte[]> nearWritesBytes;
-
     /** Mini future ID. */
     private IgniteUuid miniId;
 
@@ -70,9 +61,6 @@ public class GridDhtTxFinishRequest<K, V> extends GridDistributedTxFinishRequest
     @GridDirectCollection(GridCacheVersion.class)
     private Collection<GridCacheVersion> pendingVers;
 
-    /** One phase commit flag for fast-commit path. */
-    private boolean onePhaseCommit;
-
     /** One phase commit write version. */
     private GridCacheVersion writeVer;
 
@@ -117,10 +105,6 @@ public class GridDhtTxFinishRequest<K, V> extends GridDistributedTxFinishRequest
      * @param rolledbackVers Rolled back versions.
      * @param pendingVers Pending versions.
      * @param txSize Expected transaction size.
-     * @param writes Write entries.
-     * @param nearWrites Near cache writes.
-     * @param recoverWrites Recovery write entries.
-     * @param onePhaseCommit One phase commit flag.
      * @param grpLockKey Group lock key.
      * @param subjId Subject ID.
      * @param taskNameHash Task name hash.
@@ -145,16 +129,12 @@ public class GridDhtTxFinishRequest<K, V> extends GridDistributedTxFinishRequest
         Collection<GridCacheVersion> rolledbackVers,
         Collection<GridCacheVersion> pendingVers,
         int txSize,
-        Collection<IgniteTxEntry<K, V>> writes,
-        Collection<IgniteTxEntry<K, V>> nearWrites,
-        Collection<IgniteTxEntry<K, V>> recoverWrites,
-        boolean onePhaseCommit,
         @Nullable IgniteTxKey grpLockKey,
         @Nullable UUID subjId,
         int taskNameHash
     ) {
         super(xidVer, futId, commitVer, threadId, commit, invalidate, sys, syncCommit, syncRollback, baseVer,
-            committedVers, rolledbackVers, txSize, writes, recoverWrites, grpLockKey);
+            committedVers, rolledbackVers, txSize, grpLockKey);
 
         assert miniId != null;
         assert nearNodeId != null;
@@ -164,10 +144,8 @@ public class GridDhtTxFinishRequest<K, V> extends GridDistributedTxFinishRequest
         this.topVer = topVer;
         this.nearNodeId = nearNodeId;
         this.isolation = isolation;
-        this.nearWrites = nearWrites;
         this.miniId = miniId;
         this.sysInvalidate = sysInvalidate;
-        this.onePhaseCommit = onePhaseCommit;
         this.subjId = subjId;
         this.taskNameHash = taskNameHash;
     }
@@ -178,13 +156,6 @@ public class GridDhtTxFinishRequest<K, V> extends GridDistributedTxFinishRequest
     }
 
     /**
-     * @return Near writes.
-     */
-    public Collection<IgniteTxEntry<K, V>> nearWrites() {
-        return nearWrites == null ? Collections.<IgniteTxEntry<K, V>>emptyList() : nearWrites;
-    }
-
-    /**
      * @return Mini ID.
      */
     public IgniteUuid miniId() {
@@ -227,13 +198,6 @@ public class GridDhtTxFinishRequest<K, V> extends GridDistributedTxFinishRequest
     }
 
     /**
-     * @return One phase commit flag.
-     */
-    public boolean onePhaseCommit() {
-        return onePhaseCommit;
-    }
-
-    /**
      * @return Write version for one-phase commit transactions.
      */
     public GridCacheVersion writeVersion() {
@@ -314,35 +278,6 @@ public class GridDhtTxFinishRequest<K, V> extends GridDistributedTxFinishRequest
         return nearTtls;
     }
 
-    /** {@inheritDoc}
-     * @param ctx*/
-    @Override public void prepareMarshal(GridCacheSharedContext<K, V> ctx) throws IgniteCheckedException {
-        super.prepareMarshal(ctx);
-
-        if (nearWrites != null) {
-            marshalTx(nearWrites, ctx);
-
-            nearWritesBytes = new ArrayList<>(nearWrites.size());
-
-            for (IgniteTxEntry<K, V> e : nearWrites)
-                nearWritesBytes.add(ctx.marshaller().marshal(e));
-        }
-    }
-
-    /** {@inheritDoc} */
-    @Override public void finishUnmarshal(GridCacheSharedContext<K, V> ctx, ClassLoader ldr) throws IgniteCheckedException {
-        super.finishUnmarshal(ctx, ldr);
-
-        if (nearWritesBytes != null) {
-            nearWrites = new ArrayList<>(nearWritesBytes.size());
-
-            for (byte[] arr : nearWritesBytes)
-                nearWrites.add(ctx.marshaller().<IgniteTxEntry<K, V>>unmarshal(arr, ldr));
-
-            unmarshalTx(nearWrites, true, ctx, ldr);
-        }
-    }
-
     /** {@inheritDoc} */
     @Override public String toString() {
         return S.toString(GridDhtTxFinishRequest.class, this, super.toString());
@@ -366,13 +301,10 @@ public class GridDhtTxFinishRequest<K, V> extends GridDistributedTxFinishRequest
 
         _clone.nearNodeId = nearNodeId;
         _clone.isolation = isolation;
-        _clone.nearWrites = nearWrites;
-        _clone.nearWritesBytes = nearWritesBytes;
         _clone.miniId = miniId;
         _clone.sysInvalidate = sysInvalidate;
         _clone.topVer = topVer;
         _clone.pendingVers = pendingVers;
-        _clone.onePhaseCommit = onePhaseCommit;
         _clone.writeVer = writeVer;
         _clone.subjId = subjId;
         _clone.taskNameHash = taskNameHash;


Mime
View raw message