/*
 * 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.transactions;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.UUID;
import javax.cache.processor.EntryProcessor;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.failure.FailureContext;
import org.apache.ignite.failure.FailureType;
import org.apache.ignite.internal.IgniteFutureTimeoutCheckedException;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.pagemem.wal.record.RollbackRecord;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cache.CacheEntryInfoCollection;
import org.apache.ignite.internal.processors.cache.CacheGroupContext;
import org.apache.ignite.internal.processors.cache.CacheObject;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.GridCacheEntryEx;
import org.apache.ignite.internal.processors.cache.GridCacheEntryInfo;
import org.apache.ignite.internal.processors.cache.GridCacheEntryRemovedException;
import org.apache.ignite.internal.processors.cache.GridCacheMessage;
import org.apache.ignite.internal.processors.cache.GridCacheReturnCompletableWrapper;
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import org.apache.ignite.internal.processors.cache.GridCacheUpdateTxResult;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
import org.apache.ignite.internal.processors.cache.distributed.GridCacheTxRecoveryFuture;
import org.apache.ignite.internal.processors.cache.distributed.GridCacheTxRecoveryRequest;
import org.apache.ignite.internal.processors.cache.distributed.GridCacheTxRecoveryResponse;
import org.apache.ignite.internal.processors.cache.distributed.GridDistributedTxRemoteAdapter;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtCacheAdapter;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtCacheEntry;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTopologyFuture;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxFinishFuture;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxFinishRequest;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxFinishResponse;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxLocal;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxOnePhaseCommitAckRequest;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxPrepareFuture;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxPrepareRequest;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxPrepareResponse;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxRemote;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridInvokeValue;
import org.apache.ignite.internal.processors.cache.distributed.dht.PartitionUpdateCountersMessage;
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtInvalidPartitionException;
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtLocalPartition;
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionTopology;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearCacheAdapter;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxFinishFuture;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxFinishRequest;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxFinishResponse;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxLocal;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxPrepareFutureAdapter;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxPrepareRequest;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxPrepareResponse;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxRemote;
import org.apache.ignite.internal.processors.cache.mvcc.MvccSnapshot;
import org.apache.ignite.internal.processors.cache.persistence.wal.WALPointer;
import org.apache.ignite.internal.processors.cache.query.IgniteQueryErrorCode;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.processors.query.EnlistOperation;
import org.apache.ignite.internal.processors.query.IgniteSQLException;
import org.apache.ignite.internal.processors.tracing.MTC;
import org.apache.ignite.internal.transactions.IgniteTxHeuristicCheckedException;
import org.apache.ignite.internal.transactions.IgniteTxOptimisticCheckedException;
import org.apache.ignite.internal.transactions.IgniteTxRollbackCheckedException;
import org.apache.ignite.internal.transactions.IgniteTxTimeoutCheckedException;
import org.apache.ignite.internal.util.future.GridCompoundFuture;
import org.apache.ignite.internal.util.future.GridFinishedFuture;
import org.apache.ignite.internal.util.typedef.C1;
import org.apache.ignite.internal.util.typedef.CI1;
import org.apache.ignite.internal.util.typedef.CI2;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.X;
import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteFutureCancelledException;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.thread.IgniteThread;
import org.apache.ignite.transactions.TransactionState;
import org.jetbrains.annotations.Nullable;

import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
import static org.apache.ignite.internal.managers.communication.GridIoPolicy.SYSTEM_POOL;
import static org.apache.ignite.internal.managers.communication.GridIoPolicy.UTILITY_CACHE_POOL;
import static org.apache.ignite.internal.processors.cache.GridCacheOperation.NOOP;
import static org.apache.ignite.internal.processors.cache.GridCacheOperation.TRANSFORM;
import static org.apache.ignite.internal.processors.cache.GridCacheUtils.isNearEnabled;
import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState.RENTING;
import static org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx.FinalizationStatus.USER_FINISH;
import static org.apache.ignite.internal.processors.security.SecurityUtils.securitySubjectId;
import static org.apache.ignite.internal.processors.tracing.MTC.TraceSurroundings;
import static org.apache.ignite.internal.processors.tracing.SpanType.TX_NEAR_FINISH_REQ;
import static org.apache.ignite.internal.processors.tracing.SpanType.TX_NEAR_FINISH_RESP;
import static org.apache.ignite.internal.processors.tracing.SpanType.TX_NEAR_PREPARE_REQ;
import static org.apache.ignite.internal.processors.tracing.SpanType.TX_NEAR_PREPARE_RESP;
import static org.apache.ignite.internal.processors.tracing.SpanType.TX_PROCESS_DHT_FINISH_REQ;
import static org.apache.ignite.internal.processors.tracing.SpanType.TX_PROCESS_DHT_FINISH_RESP;
import static org.apache.ignite.internal.processors.tracing.SpanType.TX_PROCESS_DHT_ONE_PHASE_COMMIT_ACK_REQ;
import static org.apache.ignite.internal.processors.tracing.SpanType.TX_PROCESS_DHT_PREPARE_REQ;
import static org.apache.ignite.internal.processors.tracing.SpanType.TX_PROCESS_DHT_PREPARE_RESP;
import static org.apache.ignite.transactions.TransactionConcurrency.OPTIMISTIC;
import static org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC;
import static org.apache.ignite.transactions.TransactionState.PREPARED;
import static org.apache.ignite.transactions.TransactionState.ROLLED_BACK;
import static org.apache.ignite.transactions.TransactionState.ROLLING_BACK;

/**
 * Isolated logic to process cache messages.
 */
public class IgniteTxHandler {
    /** Logger. */
    private IgniteLogger log;

    /** */
    private final IgniteLogger txPrepareMsgLog;

    /** */
    private final IgniteLogger txFinishMsgLog;

    /** */
    private final IgniteLogger txRecoveryMsgLog;

    /** Shared cache context. */
    private GridCacheSharedContext<?, ?> ctx;

    /**
     * @param nearNodeId Sender node ID.
     * @param req Request.
     */
    private void processNearTxPrepareRequest(UUID nearNodeId, GridNearTxPrepareRequest req) {
        try (TraceSurroundings ignored =
                 MTC.support(ctx.kernalContext().tracing().create(TX_NEAR_PREPARE_REQ, MTC.span()))) {
            if (txPrepareMsgLog.isDebugEnabled()) {
                txPrepareMsgLog.debug("Received near prepare request [txId=" + req.version() +
                    ", node=" + nearNodeId + ']');
            }

            ClusterNode nearNode = ctx.node(nearNodeId);

            if (nearNode == null) {
                if (txPrepareMsgLog.isDebugEnabled()) {
                    txPrepareMsgLog.debug("Received near prepare from node that left grid (will ignore) [" +
                        "txId=" + req.version() +
                        ", node=" + nearNodeId + ']');
                }

                return;
            }

            processNearTxPrepareRequest0(nearNode, req);
        }
    }

    /**
     * @param nearNode Sender node.
     * @param req Request.
     */
    private IgniteInternalFuture<GridNearTxPrepareResponse> processNearTxPrepareRequest0(
        ClusterNode nearNode,
        GridNearTxPrepareRequest req
    ) {
        IgniteInternalFuture<GridNearTxPrepareResponse> fut;

        if (req.firstClientRequest() && req.allowWaitTopologyFuture()) {
            for (;;) {
                if (waitForExchangeFuture(nearNode, req))
                    return new GridFinishedFuture<>();

                fut = prepareNearTx(nearNode, req);

                if (fut != null)
                    break;
            }
        }
        else
            fut = prepareNearTx(nearNode, req);

        assert req.txState() != null || fut == null || fut.error() != null ||
            (ctx.tm().tx(req.version()) == null && ctx.tm().nearTx(req.version()) == null);

        return fut;
    }

    /**
     * @param ctx Shared cache context.
     */
    public IgniteTxHandler(GridCacheSharedContext ctx) {
        this.ctx = ctx;

        log = ctx.logger(IgniteTxHandler.class);

        txRecoveryMsgLog = ctx.logger(CU.TX_MSG_RECOVERY_LOG_CATEGORY);
        txPrepareMsgLog = ctx.logger(CU.TX_MSG_PREPARE_LOG_CATEGORY);
        txFinishMsgLog = ctx.logger(CU.TX_MSG_FINISH_LOG_CATEGORY);

        ctx.io().addCacheHandler(0, GridNearTxPrepareRequest.class, new CI2<UUID, GridCacheMessage>() {
            @Override public void apply(UUID nodeId, GridCacheMessage msg) {
                processNearTxPrepareRequest(nodeId, (GridNearTxPrepareRequest)msg);
            }
        });

        ctx.io().addCacheHandler(0, GridNearTxPrepareResponse.class, new CI2<UUID, GridCacheMessage>() {
            @Override public void apply(UUID nodeId, GridCacheMessage msg) {
                processNearTxPrepareResponse(nodeId, (GridNearTxPrepareResponse)msg);
            }
        });

        ctx.io().addCacheHandler(0, GridNearTxFinishRequest.class, new CI2<UUID, GridCacheMessage>() {
            @Override public void apply(UUID nodeId, GridCacheMessage msg) {
                processNearTxFinishRequest(nodeId, (GridNearTxFinishRequest)msg);
            }
        });

        ctx.io().addCacheHandler(0, GridNearTxFinishResponse.class, new CI2<UUID, GridCacheMessage>() {
            @Override public void apply(UUID nodeId, GridCacheMessage msg) {
                processNearTxFinishResponse(nodeId, (GridNearTxFinishResponse)msg);
            }
        });

        ctx.io().addCacheHandler(0, GridDhtTxPrepareRequest.class, new CI2<UUID, GridCacheMessage>() {
            @Override public void apply(UUID nodeId, GridCacheMessage msg) {
                processDhtTxPrepareRequest(nodeId, (GridDhtTxPrepareRequest)msg);
            }
        });

        ctx.io().addCacheHandler(0, GridDhtTxPrepareResponse.class, new CI2<UUID, GridCacheMessage>() {
            @Override public void apply(UUID nodeId, GridCacheMessage msg) {
                processDhtTxPrepareResponse(nodeId, (GridDhtTxPrepareResponse)msg);
            }
        });

        ctx.io().addCacheHandler(0, GridDhtTxFinishRequest.class, new CI2<UUID, GridCacheMessage>() {
            @Override public void apply(UUID nodeId, GridCacheMessage msg) {
                processDhtTxFinishRequest(nodeId, (GridDhtTxFinishRequest)msg);
            }
        });

        ctx.io().addCacheHandler(0, GridDhtTxOnePhaseCommitAckRequest.class, new CI2<UUID, GridCacheMessage>() {
            @Override public void apply(UUID nodeId, GridCacheMessage msg) {
                processDhtTxOnePhaseCommitAckRequest(nodeId, (GridDhtTxOnePhaseCommitAckRequest)msg);
            }
        });

        ctx.io().addCacheHandler(0, GridDhtTxFinishResponse.class, new CI2<UUID, GridCacheMessage>() {
            @Override public void apply(UUID nodeId, GridCacheMessage msg) {
                processDhtTxFinishResponse(nodeId, (GridDhtTxFinishResponse)msg);
            }
        });

        ctx.io().addCacheHandler(0, GridCacheTxRecoveryRequest.class,
            new CI2<UUID, GridCacheTxRecoveryRequest>() {
                @Override public void apply(UUID nodeId, GridCacheTxRecoveryRequest req) {
                    processCheckPreparedTxRequest(nodeId, req);
                }
            });

        ctx.io().addCacheHandler(0, GridCacheTxRecoveryResponse.class,
            new CI2<UUID, GridCacheTxRecoveryResponse>() {
                @Override public void apply(UUID nodeId, GridCacheTxRecoveryResponse res) {
                    processCheckPreparedTxResponse(nodeId, res);
                }
            });
    }

    /**
     * Prepares local colocated tx.
     *
     * @param locTx Local transaction.
     * @param req Near prepare request.
     * @return Prepare future.
     */
    public IgniteInternalFuture<GridNearTxPrepareResponse> prepareColocatedTx(
        final GridNearTxLocal locTx,
        final GridNearTxPrepareRequest req
    ) {
        req.txState(locTx.txState());

        IgniteInternalFuture<GridNearTxPrepareResponse> fut = locTx.prepareAsyncLocal(req);

        return fut.chain(new C1<IgniteInternalFuture<GridNearTxPrepareResponse>, GridNearTxPrepareResponse>() {
            @Override public GridNearTxPrepareResponse apply(IgniteInternalFuture<GridNearTxPrepareResponse> f) {
                try {
                    return f.get();
                }
                catch (Exception e) {
                    locTx.setRollbackOnly(); // Just in case.

                    if (!X.hasCause(e, IgniteTxOptimisticCheckedException.class) &&
                        !X.hasCause(e, IgniteFutureCancelledException.class))
                        U.error(log, "Failed to prepare DHT transaction: " + locTx, e);

                    return new GridNearTxPrepareResponse(
                        req.partition(),
                        req.version(),
                        req.futureId(),
                        req.miniId(),
                        req.version(),
                        req.version(),
                        null,
                        e,
                        null,
                        req.onePhaseCommit(),
                        req.deployInfo() != null);
                }
            }
        });
    }

    /**
     * @param entries Entries.
     * @return First entry.
     * @throws IgniteCheckedException If failed.
     */
    private IgniteTxEntry unmarshal(@Nullable Collection<IgniteTxEntry> entries) throws IgniteCheckedException {
        if (entries == null)
            return null;

        IgniteTxEntry firstEntry = null;

        for (IgniteTxEntry e : entries) {
            e.unmarshal(ctx, false, ctx.deploy().globalLoader());

            if (firstEntry == null)
                firstEntry = e;
        }

        return firstEntry;
    }

    /**
     * @param originTx Transaction for copy.
     * @param req Request.
     * @return Prepare future.
     */
    public IgniteInternalFuture<GridNearTxPrepareResponse> prepareNearTxLocal(
        final GridNearTxLocal originTx,
        final GridNearTxPrepareRequest req) {
        // Make sure not to provide Near entries to DHT cache.
        req.cloneEntries();

        return prepareNearTx(originTx, ctx.localNode(), req);
    }

    /**
     * @param nearNode Node that initiated transaction.
     * @param req Near prepare request.
     * @return Prepare future or {@code null} if need retry operation.
     */
    @Nullable private IgniteInternalFuture<GridNearTxPrepareResponse> prepareNearTx(
        final ClusterNode nearNode,
        final GridNearTxPrepareRequest req
    ) {
        return prepareNearTx(null, nearNode, req);
    }

    /**
     * @param originTx Transaction for copy.
     * @param nearNode Node that initiated transaction.
     * @param req Near prepare request.
     * @return Prepare future or {@code null} if need retry operation.
     */
    @Nullable private IgniteInternalFuture<GridNearTxPrepareResponse> prepareNearTx(
        final GridNearTxLocal originTx,
        final ClusterNode nearNode,
        final GridNearTxPrepareRequest req
    ) {
        IgniteTxEntry firstEntry;

        try {
            IgniteTxEntry firstWrite = unmarshal(req.writes());
            IgniteTxEntry firstRead = unmarshal(req.reads());

            firstEntry = firstWrite != null ? firstWrite : firstRead;
        }
        catch (IgniteCheckedException e) {
            return new GridFinishedFuture<>(e);
        }

        GridDhtTxLocal tx = null;

        GridCacheVersion mappedVer = ctx.tm().mappedVersion(req.version());

        if (mappedVer != null) {
            tx = ctx.tm().tx(mappedVer);

            if (tx == null)
                U.warn(log, "Missing local transaction for mapped near version [nearVer=" + req.version()
                    + ", mappedVer=" + mappedVer + ']');
            else {
                if (req.concurrency() == PESSIMISTIC)
                    tx.nearFutureId(req.futureId());
            }
        }
        else {
            GridDhtPartitionTopology top = null;

            if (req.firstClientRequest()) {
                assert firstEntry != null : req;

                assert req.concurrency() == OPTIMISTIC : req;
                assert nearNode.isClient() : nearNode;

                top = firstEntry.context().topology();

                top.readLock();

                if (req.allowWaitTopologyFuture()) {
                    GridDhtTopologyFuture topFut = top.topologyVersionFuture();

                    if (!topFut.isDone()) {
                        top.readUnlock();

                        return null;
                    }
                }
            }

            try {
                if (top != null ) {
                    boolean retry = false;

                    GridDhtTopologyFuture topFut = top.topologyVersionFuture();

                    if (!req.allowWaitTopologyFuture() && !topFut.isDone()) {
                        retry = true;

                        if (txPrepareMsgLog.isDebugEnabled()) {
                            txPrepareMsgLog.debug("Topology change is in progress, need remap transaction [" +
                                "txId=" + req.version() +
                                ", node=" + nearNode.id() +
                                ", reqTopVer=" + req.topologyVersion() +
                                ", locTopVer=" + top.readyTopologyVersion() +
                                ", req=" + req + ']');
                        }
                    }

                    if (!retry && needRemap(req.topologyVersion(), top.readyTopologyVersion(), req)) {
                        retry = true;

                        if (txPrepareMsgLog.isDebugEnabled()) {
                            txPrepareMsgLog.debug("Topology version mismatch for near prepare, need remap transaction [" +
                                "txId=" + req.version() +
                                ", node=" + nearNode.id() +
                                ", reqTopVer=" + req.topologyVersion() +
                                ", locTopVer=" + top.readyTopologyVersion() +
                                ", req=" + req + ']');
                        }
                    }

                    if (retry) {
                        GridNearTxPrepareResponse res = new GridNearTxPrepareResponse(
                            req.partition(),
                            req.version(),
                            req.futureId(),
                            req.miniId(),
                            req.version(),
                            req.version(),
                            null,
                            null,
                            top.lastTopologyChangeVersion(),
                            req.onePhaseCommit(),
                            req.deployInfo() != null);

                        try {
                            ctx.io().send(nearNode, res, req.policy());

                            if (txPrepareMsgLog.isDebugEnabled()) {
                                txPrepareMsgLog.debug("Sent remap response for near prepare [txId=" + req.version() +
                                    ", node=" + nearNode.id() + ']');
                            }
                        }
                        catch (ClusterTopologyCheckedException ignored) {
                            if (txPrepareMsgLog.isDebugEnabled()) {
                                txPrepareMsgLog.debug("Failed to send remap response for near prepare, node failed [" +
                                    "txId=" + req.version() +
                                    ", node=" + nearNode.id() + ']');
                            }
                        }
                        catch (IgniteCheckedException e) {
                            U.error(txPrepareMsgLog, "Failed to send remap response for near prepare " +
                                "[txId=" + req.version() +
                                ", node=" + nearNode.id() +
                                ", req=" + req + ']', e);
                        }

                        return new GridFinishedFuture<>(res);
                    }

                    assert topFut.isDone();
                }

                tx = new GridDhtTxLocal(
                    ctx,
                    req.topologyVersion(),
                    nearNode.id(),
                    req.version(),
                    req.futureId(),
                    req.miniId(),
                    req.threadId(),
                    req.implicitSingle(),
                    req.implicitSingle(),
                    req.system(),
                    req.explicitLock(),
                    req.policy(),
                    req.concurrency(),
                    req.isolation(),
                    req.timeout(),
                    req.isInvalidate(),
                    true,
                    req.onePhaseCommit(),
                    req.txSize(),
                    req.transactionNodes(),
                    securitySubjectId(ctx),
                    req.taskNameHash(),
                    req.txLabel(),
                    originTx
                );

                tx = ctx.tm().onCreated(null, tx);

                if (tx != null)
                    tx.topologyVersion(req.topologyVersion());
                else
                    U.warn(log, "Failed to create local transaction (was transaction rolled back?) [xid=" +
                        req.version() + ", req=" + req + ']');
            }
            finally {
                if (tx != null)
                    req.txState(tx.txState());

                if (top != null)
                    top.readUnlock();
            }
        }

        if (tx != null) {
            req.txState(tx.txState());

            if (req.explicitLock())
                tx.explicitLock(true);

            tx.transactionNodes(req.transactionNodes());

            if (req.near())
                tx.nearOnOriginatingNode(true);

            if (req.onePhaseCommit()) {
                assert req.last() : req;

                tx.onePhaseCommit(true);
            }

            if (req.needReturnValue())
                tx.needReturnValue(true);

            IgniteInternalFuture<GridNearTxPrepareResponse> fut = tx.prepareAsync(req);

            if (tx.isRollbackOnly() && !tx.commitOnPrepare()) {
                if (tx.state() != TransactionState.ROLLED_BACK && tx.state() != TransactionState.ROLLING_BACK)
                    tx.rollbackDhtLocalAsync();
            }

            final GridDhtTxLocal tx0 = tx;

            fut.listen(new CI1<IgniteInternalFuture<?>>() {
                @Override public void apply(IgniteInternalFuture<?> txFut) {
                    try {
                        txFut.get();
                    }
                    catch (IgniteCheckedException e) {
                        tx0.setRollbackOnly(); // Just in case.

                        if (!X.hasCause(e, IgniteTxOptimisticCheckedException.class) &&
                            !X.hasCause(e, IgniteFutureCancelledException.class) && !ctx.kernalContext().isStopping())
                            U.error(log, "Failed to prepare DHT transaction: " + tx0, e);
                    }
                }
            });

            return fut;
        }
        else
            return new GridFinishedFuture<>((GridNearTxPrepareResponse)null);
    }

    /**
     * @param node Sender node.
     * @param req Request.
     * @return {@code True} if update will be retried from future listener or topology version future is timed out.
     */
    private boolean waitForExchangeFuture(final ClusterNode node, final GridNearTxPrepareRequest req) {
        assert req.firstClientRequest() : req;

        GridDhtTopologyFuture topFut = ctx.exchange().lastTopologyFuture();

        if (!topFut.isDone()) {
            Thread curThread = Thread.currentThread();

            if (curThread instanceof IgniteThread) {
                final IgniteThread thread = (IgniteThread)curThread;

                if (thread.cachePoolThread()) {
                    ctx.time().waitAsync(topFut, req.timeout(), (e, timedOut) -> {
                            if (e != null || timedOut) {
                                sendResponseOnTimeoutOrError(e, topFut, node, req);

                                return;
                            }
                            ctx.kernalContext().closure().runLocalWithThreadPolicy(thread, () -> {
                                try {
                                    processNearTxPrepareRequest0(node, req);
                                }
                                finally {
                                    ctx.io().onMessageProcessed(req);
                                }
                            });
                        }
                    );

                    return true;
                }
            }

            try {
                if (req.timeout() > 0)
                    topFut.get(req.timeout());
                else
                    topFut.get();
            }
            catch (IgniteFutureTimeoutCheckedException e) {
                sendResponseOnTimeoutOrError(null, topFut, node, req);

                return true;
            }
            catch (IgniteCheckedException e) {
                U.error(log, "Topology future failed: " + e, e);
            }
        }

        return false;
    }

    /**
     * @param e Exception or null if timed out.
     * @param topFut Topology future.
     * @param node Node.
     * @param req Prepare request.
     */
    private void sendResponseOnTimeoutOrError(@Nullable IgniteCheckedException e,
        GridDhtTopologyFuture topFut,
        ClusterNode node,
        GridNearTxPrepareRequest req) {
        if (e == null)
            e = new IgniteTxTimeoutCheckedException("Failed to wait topology version for near prepare " +
                "[txId=" + req.version() +
                ", topVer=" + topFut.initialVersion() +
                ", node=" + node.id() +
                ", req=" + req + ']');

        GridNearTxPrepareResponse res = new GridNearTxPrepareResponse(
            req.partition(),
            req.version(),
            req.futureId(),
            req.miniId(),
            req.version(),
            req.version(),
            null,
            e,
            null,
            req.onePhaseCommit(),
            req.deployInfo() != null);

        try {
            ctx.io().send(node.id(), res, req.policy());
        }
        catch (IgniteCheckedException e0) {
            U.error(txPrepareMsgLog, "Failed to send wait topology version response for near prepare " +
                "[txId=" + req.version() +
                ", topVer=" + topFut.initialVersion() +
                ", node=" + node.id() +
                ", req=" + req + ']', e0);
        }
    }

    /**
     * @param expVer Expected topology version.
     * @param curVer Current topology version.
     * @param req Request.
     * @return {@code True} if cache affinity changed and request should be remapped.
     */
    private boolean needRemap(AffinityTopologyVersion expVer,
        AffinityTopologyVersion curVer,
        GridNearTxPrepareRequest req) {
        if (curVer.equals(expVer))
            return false;

        AffinityTopologyVersion lastAffChangedTopVer = ctx.exchange().lastAffinityChangedTopologyVersion(expVer);

        if (curVer.compareTo(expVer) <= 0 && curVer.compareTo(lastAffChangedTopVer) >= 0)
            return false;

        // TODO IGNITE-6754 check mvcc crd for mvcc enabled txs.

        for (IgniteTxEntry e : F.concat(false, req.reads(), req.writes())) {
            GridCacheContext ctx = e.context();

            Collection<ClusterNode> cacheNodes0 = ctx.discovery().cacheGroupAffinityNodes(ctx.groupId(), expVer);
            Collection<ClusterNode> cacheNodes1 = ctx.discovery().cacheGroupAffinityNodes(ctx.groupId(), curVer);

            if (!cacheNodes0.equals(cacheNodes1) || ctx.affinity().affinityTopologyVersion().compareTo(curVer) < 0)
                return true;

            try {
                List<List<ClusterNode>> aff1 = ctx.affinity().assignments(expVer);
                List<List<ClusterNode>> aff2 = ctx.affinity().assignments(curVer);

                if (!aff1.equals(aff2))
                    return true;
            }
            catch (IllegalStateException ignored) {
                return true;
            }
        }

        return false;
    }

    /**
     * @param nodeId Node ID.
     * @param res Response.
     */
    private void processNearTxPrepareResponse(UUID nodeId, GridNearTxPrepareResponse res) {
        try (TraceSurroundings ignored =
                 MTC.support(ctx.kernalContext().tracing().create(TX_NEAR_PREPARE_RESP, MTC.span()))) {
            if (txPrepareMsgLog.isDebugEnabled())
                txPrepareMsgLog.debug("Received near prepare response [txId=" + res.version() + ", node=" +
                    nodeId + ']');

            GridNearTxPrepareFutureAdapter fut = (GridNearTxPrepareFutureAdapter)ctx.mvcc()
                .<IgniteInternalTx>versionedFuture(res.version(), res.futureId());

            if (fut == null) {
                U.warn(log, "Failed to find future for near prepare response [txId=" + res.version() +
                    ", node=" + nodeId +
                    ", res=" + res + ']');

                return;
            }

            IgniteInternalTx tx = fut.tx();

            assert tx != null;

            res.txState(tx.txState());

            fut.onResult(nodeId, res);
        }
    }

    /**
     * @param nodeId Node ID.
     * @param res Response.
     */
    private void processNearTxFinishResponse(UUID nodeId, GridNearTxFinishResponse res) {
        try (TraceSurroundings ignored =
                 MTC.support(ctx.kernalContext().tracing().create(TX_NEAR_FINISH_RESP, MTC.span()))) {
            if (txFinishMsgLog.isDebugEnabled())
                txFinishMsgLog.debug("Received near finish response [txId=" + res.xid() + ", node=" + nodeId + ']');

            GridNearTxFinishFuture fut = (GridNearTxFinishFuture)ctx.mvcc().<IgniteInternalTx>future(res.futureId());

            if (fut == null) {
                if (txFinishMsgLog.isDebugEnabled()) {
                    txFinishMsgLog.debug("Failed to find future for near finish response [txId=" + res.xid() +
                        ", node=" + nodeId +
                        ", res=" + res + ']');
                }

                return;
            }

            fut.onResult(nodeId, res);
        }
    }

    /**
     * @param nodeId Node ID.
     * @param res Response.
     */
    private void processDhtTxPrepareResponse(UUID nodeId, GridDhtTxPrepareResponse res) {
        try (TraceSurroundings ignored =
                 MTC.support(ctx.kernalContext().tracing().create(TX_PROCESS_DHT_PREPARE_RESP, MTC.span()))) {
            GridDhtTxPrepareFuture fut =
                (GridDhtTxPrepareFuture)ctx.mvcc().versionedFuture(res.version(), res.futureId());

            if (fut == null) {
                if (txPrepareMsgLog.isDebugEnabled()) {
                    txPrepareMsgLog.debug("Failed to find future for dht prepare response [txId=null" +
                        ", dhtTxId=" + res.version() +
                        ", node=" + nodeId +
                        ", res=" + res + ']');
                }

                return;
            }
            else if (txPrepareMsgLog.isDebugEnabled())
                txPrepareMsgLog.debug("Received dht prepare response [txId=" + fut.tx().nearXidVersion() +
                    ", node=" + nodeId + ']');

            IgniteInternalTx tx = fut.tx();

            assert tx != null;

            res.txState(tx.txState());

            fut.onResult(nodeId, res);
        }
    }

    /**
     * @param nodeId Node ID.
     * @param res Response.
     */
    private void processDhtTxFinishResponse(UUID nodeId, GridDhtTxFinishResponse res) {
        try (TraceSurroundings ignored =
                 MTC.support(ctx.kernalContext().tracing().create(TX_PROCESS_DHT_FINISH_RESP, MTC.span()))) {
            assert nodeId != null;
            assert res != null;

            if (res.checkCommitted()) {
                GridNearTxFinishFuture fut = (GridNearTxFinishFuture)ctx.mvcc().<IgniteInternalTx>future(res.futureId());

                if (fut == null) {
                    if (txFinishMsgLog.isDebugEnabled()) {
                        txFinishMsgLog.debug("Failed to find future for dht finish check committed response [txId=null" +
                            ", dhtTxId=" + res.xid() +
                            ", node=" + nodeId +
                            ", res=" + res + ']');
                    }

                    return;
                }
                else if (txFinishMsgLog.isDebugEnabled()) {
                    txFinishMsgLog.debug("Received dht finish check committed response [txId=" + fut.tx().nearXidVersion() +
                        ", dhtTxId=" + res.xid() +
                        ", node=" + nodeId + ']');
                }

                fut.onResult(nodeId, res);
            }
            else {
                GridDhtTxFinishFuture fut = (GridDhtTxFinishFuture)ctx.mvcc().<IgniteInternalTx>future(res.futureId());

                if (fut == null) {
                    if (txFinishMsgLog.isDebugEnabled()) {
                        txFinishMsgLog.debug("Failed to find future for dht finish response [txId=null" +
                            ", dhtTxId=" + res.xid() +
                            ", node=" + nodeId +
                            ", res=" + res);
                    }

                    return;
                }
                else if (txFinishMsgLog.isDebugEnabled()) {
                    txFinishMsgLog.debug("Received dht finish response [txId=" + fut.tx().nearXidVersion() +
                        ", dhtTxId=" + res.xid() +
                        ", node=" + nodeId + ']');
                }

                fut.onResult(nodeId, res);
            }
        }
    }

    /**
     * @param nodeId Node ID.
     * @param req Request.
     * @return Future.
     */
    @Nullable private IgniteInternalFuture<IgniteInternalTx> processNearTxFinishRequest(
        UUID nodeId,
        GridNearTxFinishRequest req
    ) {
        try (TraceSurroundings ignored =
                 MTC.support(ctx.kernalContext().tracing().create(TX_NEAR_FINISH_REQ, MTC.span()))) {
            if (txFinishMsgLog.isDebugEnabled())
                txFinishMsgLog.debug("Received near finish request [txId=" + req.version() + ", node=" + nodeId +
                    ']');

            IgniteInternalFuture<IgniteInternalTx> fut = finish(nodeId, null, req);

            assert req.txState() != null || fut == null || fut.error() != null ||
                (ctx.tm().tx(req.version()) == null && ctx.tm().nearTx(req.version()) == null) :
                "[req=" + req + ", fut=" + fut + "]";

            return fut;
        }
    }

    /**
     * @param nodeId Node ID.
     * @param locTx Local transaction.
     * @param req Request.
     * @return Future.
     */
    @Nullable public IgniteInternalFuture<IgniteInternalTx> finish(UUID nodeId,
        @Nullable GridNearTxLocal locTx,
        GridNearTxFinishRequest req
    ) {
        assert nodeId != null;
        assert req != null;

        if (locTx != null)
            req.txState(locTx.txState());

        // Always add near version to rollback history to prevent races with rollbacks.
        if (!req.commit())
            ctx.tm().addRolledbackTx(null, req.version());

        // Transaction on local cache only.
        if (locTx != null && !locTx.nearLocallyMapped() && !locTx.colocatedLocallyMapped())
            return new GridFinishedFuture<IgniteInternalTx>(locTx);

        if (log.isDebugEnabled())
            log.debug("Processing near tx finish request [nodeId=" + nodeId + ", req=" + req + "]");

        IgniteInternalFuture<IgniteInternalTx> colocatedFinishFut = null;

        if (locTx != null && locTx.colocatedLocallyMapped())
            colocatedFinishFut = finishColocatedLocal(req.commit(), locTx);

        IgniteInternalFuture<IgniteInternalTx> nearFinishFut = null;

        if (locTx == null || locTx.nearLocallyMapped())
            nearFinishFut = finishDhtLocal(nodeId, locTx, req);

        if (colocatedFinishFut != null && nearFinishFut != null) {
            GridCompoundFuture<IgniteInternalTx, IgniteInternalTx> res = new GridCompoundFuture<>();

            res.add(colocatedFinishFut);
            res.add(nearFinishFut);

            res.markInitialized();

            return res;
        }

        if (colocatedFinishFut != null)
            return colocatedFinishFut;

        return nearFinishFut;
    }

    /**
     * @param nodeId Node ID initiated commit.
     * @param locTx Optional local transaction.
     * @param req Finish request.
     * @return Finish future.
     */
    private IgniteInternalFuture<IgniteInternalTx> finishDhtLocal(UUID nodeId,
        @Nullable GridNearTxLocal locTx,
        GridNearTxFinishRequest req
    ) {
        GridCacheVersion dhtVer = ctx.tm().mappedVersion(req.version());

        GridDhtTxLocal tx = null;

        if (dhtVer == null) {
            if (log.isDebugEnabled())
                log.debug("Received transaction finish request for unknown near version (was lock explicit?): " + req);
        }
        else
            tx = ctx.tm().tx(dhtVer);

        if (tx != null) {
            tx.mvccSnapshot(req.mvccSnapshot());

            req.txState(tx.txState());
        }

        if (tx == null && locTx != null && !req.commit()) {
            U.warn(log, "DHT local tx not found for near local tx rollback " +
                "[req=" + req + ", dhtVer=" + dhtVer + ", tx=" + locTx + ']');

            return null;
        }

        if (tx == null && !req.explicitLock()) {
            assert locTx == null : "DHT local tx should never be lost for near local tx: " + locTx;

            U.warn(txFinishMsgLog, "Received finish request for completed transaction (the message may be too late) [" +
                "txId=" + req.version() +
                ", dhtTxId=" + dhtVer +
                ", node=" + nodeId +
                ", commit=" + req.commit() + ']');

            // Always send finish response.
            GridCacheMessage res = new GridNearTxFinishResponse(
                req.partition(),
                req.version(),
                req.threadId(),
                req.futureId(),
                req.miniId(),
                new IgniteTxRollbackCheckedException("Transaction has been already completed or not started yet."));

            try {
                ctx.io().send(nodeId, res, req.policy());

                if (txFinishMsgLog.isDebugEnabled()) {
                    txFinishMsgLog.debug("Sent near finish response for completed tx [txId=" + req.version() +
                        ", dhtTxId=" + dhtVer +
                        ", node=" + nodeId + ']');
                }
            }
            catch (Throwable e) {
                // Double-check.
                if (ctx.discovery().node(nodeId) == null) {
                    if (txFinishMsgLog.isDebugEnabled()) {
                        txFinishMsgLog.debug("Failed to send near finish response for completed tx, node failed [" +
                            "txId=" + req.version() +
                            ", dhtTxId=" + dhtVer +
                            ", node=" + nodeId + ']');
                    }
                }
                else {
                    U.error(txFinishMsgLog, "Failed to send near finish response for completed tx, node failed [" +
                        "txId=" + req.version() +
                        ", dhtTxId=" + dhtVer +
                        ", node=" + nodeId +
                        ", req=" + req +
                        ", res=" + res + ']', e);
                }

                if (e instanceof Error)
                    throw (Error)e;
            }

            return null;
        }

        try {
            assert tx != null : "Transaction is null for near finish request [nodeId=" +
                nodeId + ", req=" + req + "]";
            assert req.syncMode() != null : req;

            tx.syncMode(req.syncMode());
            tx.nearFinishFutureId(req.futureId());
            tx.nearFinishMiniId(req.miniId());
            tx.storeEnabled(req.storeEnabled());

            if (!tx.markFinalizing(USER_FINISH)) {
                if (log.isDebugEnabled())
                    log.debug("Will not finish transaction (it is handled by another thread) [commit=" + req.commit() + ", tx=" + tx + ']');

                return null;
            }

            if (req.commit()) {
                IgniteInternalFuture<IgniteInternalTx> commitFut = tx.commitDhtLocalAsync();

                // Only for error logging.
                commitFut.listen(CU.errorLogger(log));

                return commitFut;
            }
            else {
                IgniteInternalFuture<IgniteInternalTx> rollbackFut = tx.rollbackDhtLocalAsync();

                // Only for error logging.
                rollbackFut.listen(CU.errorLogger(log));

                return rollbackFut;
            }
        }
        catch (Throwable e) {
            try {
                if (tx != null) {
                    tx.commitError(e);

                    tx.systemInvalidate(true);

                    try {
                        IgniteInternalFuture<IgniteInternalTx> res = tx.rollbackDhtLocalAsync();

                        // Only for error logging.
                        res.listen(CU.errorLogger(log));

                        return res;
                    }
                    catch (Throwable e1) {
                        e.addSuppressed(e1);
                    }

                    tx.logTxFinishErrorSafe(log, req.commit(), e);
                }

                if (e instanceof Error)
                    throw (Error)e;

                return new GridFinishedFuture<>(e);
            }
            finally {
                ctx.kernalContext().failure().process(new FailureContext(FailureType.CRITICAL_ERROR, e));
            }
        }
    }

    /**
     * @param commit Commit flag (rollback if {@code false}).
     * @param tx Transaction to commit.
     * @return Future.
     */
    public IgniteInternalFuture<IgniteInternalTx> finishColocatedLocal(boolean commit, GridNearTxLocal tx) {
        try {
            if (commit) {
                if (!tx.markFinalizing(USER_FINISH)) {
                    if (log.isDebugEnabled())
                        log.debug("Will not finish transaction (it is handled by another thread): " + tx);

                    return null;
                }

                return tx.commitAsyncLocal();
            }
            else
                return tx.rollbackAsyncLocal();
        }
        catch (Throwable e) {
            try {
                if (tx != null) {
                    try {
                        return tx.rollbackNearTxLocalAsync();
                    }
                    catch (Throwable e1) {
                        e.addSuppressed(e1);
                    }

                    tx.logTxFinishErrorSafe(log, commit, e);
                }

                if (e instanceof Error)
                    throw e;

                return new GridFinishedFuture<>(e);
            }
            finally {
                ctx.kernalContext().failure().process(new FailureContext(FailureType.CRITICAL_ERROR, e));
            }
        }
    }

    /**
     * @param nodeId Sender node ID.
     * @param req Request.
     */
    private void processDhtTxPrepareRequest(final UUID nodeId, final GridDhtTxPrepareRequest req) {
        try (TraceSurroundings ignored =
                 MTC.support(ctx.kernalContext().tracing().create(TX_PROCESS_DHT_PREPARE_REQ, MTC.span()))) {
            if (txPrepareMsgLog.isDebugEnabled()) {
                txPrepareMsgLog.debug("Received dht prepare request [txId=" + req.nearXidVersion() +
                    ", dhtTxId=" + req.version() +
                    ", node=" + nodeId + ']');
            }

            assert nodeId != null;
            assert req != null;

            assert req.transactionNodes() != null;

            GridDhtTxRemote dhtTx = null;
            GridNearTxRemote nearTx = null;

            GridDhtTxPrepareResponse res;

            try {
                res = new GridDhtTxPrepareResponse(
                    req.partition(),
                    req.version(),
                    req.futureId(),
                    req.miniId(),
                    req.deployInfo() != null);

                // Start near transaction first.
                nearTx = !F.isEmpty(req.nearWrites()) ? startNearRemoteTx(ctx.deploy().globalLoader(), nodeId, req) : null;
                dhtTx = startRemoteTx(nodeId, req, res);

                // Set evicted keys from near transaction.
                if (nearTx != null)
                    res.nearEvicted(nearTx.evicted());

                List<IgniteTxKey> writesCacheMissed = req.nearWritesCacheMissed();

                if (writesCacheMissed != null) {
                    Collection<IgniteTxKey> evicted0 = res.nearEvicted();

                    if (evicted0 != null)
                        writesCacheMissed.addAll(evicted0);

                    res.nearEvicted(writesCacheMissed);
                }

                if (dhtTx != null)
                    req.txState(dhtTx.txState());
                else if (nearTx != null)
                    req.txState(nearTx.txState());

                if (dhtTx != null && !F.isEmpty(dhtTx.invalidPartitions()))
                    res.invalidPartitionsByCacheId(dhtTx.invalidPartitions());

                if (req.onePhaseCommit()) {
                    assert req.last();

                    if (dhtTx != null) {
                        dhtTx.onePhaseCommit(true);
                        dhtTx.needReturnValue(req.needReturnValue());

                        finish(dhtTx, req);
                    }

                    if (nearTx != null) {
                        nearTx.onePhaseCommit(true);

                        finish(nearTx, req);
                    }
                }
            }
            catch (IgniteCheckedException e) {
                if (e instanceof IgniteTxRollbackCheckedException)
                    U.error(log, "Transaction was rolled back before prepare completed: " + req, e);
                else if (e instanceof IgniteTxOptimisticCheckedException) {
                    if (log.isDebugEnabled())
                        log.debug("Optimistic failure for remote transaction (will rollback): " + req);
                }
                else
                    U.error(log, "Failed to process prepare request: " + req, e);

                if (nearTx != null)
                    try {
                        nearTx.rollbackRemoteTx();
                    }
                    catch (Throwable e1) {
                        e.addSuppressed(e1);
                    }

                res = new GridDhtTxPrepareResponse(
                    req.partition(),
                    req.version(),
                    req.futureId(),
                    req.miniId(),
                    e,
                    req.deployInfo() != null);
            }

            if (req.onePhaseCommit()) {
                IgniteInternalFuture completeFut;

                IgniteInternalFuture<IgniteInternalTx> dhtFin = dhtTx == null ?
                    null : dhtTx.done() ? null : dhtTx.finishFuture();

                final IgniteInternalFuture<IgniteInternalTx> nearFin = nearTx == null ?
                    null : nearTx.done() ? null : nearTx.finishFuture();

                if (dhtFin != null && nearFin != null) {
                    GridCompoundFuture fut = new GridCompoundFuture();

                    fut.add(dhtFin);
                    fut.add(nearFin);

                    fut.markInitialized();

                    completeFut = fut;
                }
                else
                    completeFut = dhtFin != null ? dhtFin : nearFin;

                if (completeFut != null) {
                    final GridDhtTxPrepareResponse res0 = res;
                    final GridDhtTxRemote dhtTx0 = dhtTx;
                    final GridNearTxRemote nearTx0 = nearTx;

                    completeFut.listen(new CI1<IgniteInternalFuture<IgniteInternalTx>>() {
                        @Override public void apply(IgniteInternalFuture<IgniteInternalTx> fut) {
                            sendReply(nodeId, req, res0, dhtTx0, nearTx0);
                        }
                    });
                }
                else
                    sendReply(nodeId, req, res, dhtTx, nearTx);
            }
            else
                sendReply(nodeId, req, res, dhtTx, nearTx);

            assert req.txState() != null || res.error() != null || (dhtTx == null && nearTx == null) :
                req + " tx=" + dhtTx + " nearTx=" + nearTx;
        }
    }

    /**
     * @param nodeId Node ID.
     * @param req Request.
     */
    private void processDhtTxOnePhaseCommitAckRequest(final UUID nodeId,
        final GridDhtTxOnePhaseCommitAckRequest req) {
        try (TraceSurroundings ignored =
                 MTC.support(ctx.kernalContext().tracing().create(TX_PROCESS_DHT_ONE_PHASE_COMMIT_ACK_REQ, MTC.span()))) {
            assert nodeId != null;
            assert req != null;

            if (log.isDebugEnabled())
                log.debug("Processing dht tx one phase commit ack request [nodeId=" + nodeId + ", req=" + req + ']');

            for (GridCacheVersion ver : req.versions())
                ctx.tm().removeTxReturn(ver);
        }
    }

    /**
     * @param nodeId Node ID.
     * @param req Request.
     */
    @SuppressWarnings({"unchecked"})
    private void processDhtTxFinishRequest(final UUID nodeId, final GridDhtTxFinishRequest req) {
        try (TraceSurroundings ignored =
                 MTC.support(ctx.kernalContext().tracing().create(TX_PROCESS_DHT_FINISH_REQ, MTC.span()))) {
            assert nodeId != null;
            assert req != null;

            if (req.checkCommitted()) {
                boolean committed = req.waitRemoteTransactions() || !ctx.tm().addRolledbackTx(null, req.version());

                if (!committed || req.syncMode() != FULL_SYNC)
                    sendReply(nodeId, req, committed, null);
                else {
                    IgniteInternalFuture<?> fut = ctx.tm().remoteTxFinishFuture(req.version());

                    fut.listen(new CI1<IgniteInternalFuture<?>>() {
                        @Override public void apply(IgniteInternalFuture<?> fut) {
                            sendReply(nodeId, req, true, null);
                        }
                    });
                }

                return;
            }

            // Always add version to rollback history to prevent races with rollbacks.
            if (!req.commit())
                ctx.tm().addRolledbackTx(null, req.version());

            GridDhtTxRemote dhtTx = ctx.tm().tx(req.version());
            GridNearTxRemote nearTx = ctx.tm().nearTx(req.version());

            IgniteInternalTx anyTx = U.<IgniteInternalTx>firstNotNull(dhtTx, nearTx);

            final GridCacheVersion nearTxId = anyTx != null ? anyTx.nearXidVersion() : null;

            if (txFinishMsgLog.isDebugEnabled())
                txFinishMsgLog.debug("Received dht finish request [txId=" + nearTxId + ", dhtTxId=" + req.version() +
                    ", node=" + nodeId + ']');

            if (anyTx == null && req.commit())
                ctx.tm().addCommittedTx(null, req.version(), null);

            if (dhtTx != null)
                finish(nodeId, dhtTx, req);
            else {
                try {
                    applyPartitionsUpdatesCounters(req.updateCounters(), !req.commit(), false);
                }
                catch (IgniteCheckedException e) {
                    throw new IgniteException(e);
                }
            }

            if (nearTx != null)
                finish(nodeId, nearTx, req);

            if (req.replyRequired()) {
                IgniteInternalFuture completeFut;

                IgniteInternalFuture<IgniteInternalTx> dhtFin = dhtTx == null ?
                    null : dhtTx.done() ? null : dhtTx.finishFuture();

                final IgniteInternalFuture<IgniteInternalTx> nearFin = nearTx == null ?
                    null : nearTx.done() ? null : nearTx.finishFuture();

                if (dhtFin != null && nearFin != null) {
                    GridCompoundFuture fut = new GridCompoundFuture();

                    fut.add(dhtFin);
                    fut.add(nearFin);

                    fut.markInitialized();

                    completeFut = fut;
                }
                else
                    completeFut = dhtFin != null ? dhtFin : nearFin;

                if (completeFut != null) {
                    completeFut.listen(new CI1<IgniteInternalFuture<IgniteInternalTx>>() {
                        @Override public void apply(IgniteInternalFuture<IgniteInternalTx> fut) {
                            sendReply(nodeId, req, true, nearTxId);
                        }
                    });
                }
                else
                    sendReply(nodeId, req, true, nearTxId);
            }
            else
                sendReply(nodeId, req, true, null);

            assert req.txState() != null || (dhtTx == null && nearTx == null) : req + " tx=" + dhtTx + " nearTx=" + nearTx;
        }
    }

    /**
     * @param nodeId Node ID.
     * @param tx Transaction.
     * @param req Request.
     */
    protected void finish(
        UUID nodeId,
        IgniteTxRemoteEx tx,
        GridDhtTxFinishRequest req
    ) {
        assert tx != null;

        req.txState(tx.txState());

        try {
            if (req.commit() || req.isSystemInvalidate()) {
                tx.commitVersion(req.commitVersion());
                if (req.isInvalidate())
                    tx.invalidate(true);
                if (req.isSystemInvalidate())
                    tx.systemInvalidate(true);
                tx.mvccSnapshot(req.mvccSnapshot());

                // Complete remote candidates.
                tx.doneRemote(req.baseVersion(), null, null, null);

                tx.commitRemoteTx();
            }
            else {
                if (tx.dht() && req.updateCounters() != null)
                    tx.txCounters(true).updateCounters(req.updateCounters());

                tx.doneRemote(req.baseVersion(), null, null, null);
                tx.mvccSnapshot(req.mvccSnapshot());
                tx.rollbackRemoteTx();
            }
        }
        catch (IgniteTxHeuristicCheckedException ignore) {
            // Already uncommitted.
        }
        catch (Throwable e) {
            // Mark transaction for invalidate.
            tx.invalidate(true);
            tx.systemInvalidate(true);

            try {
                tx.commitRemoteTx();
            }
            catch (IgniteCheckedException ex) {
                U.error(log, "Failed to invalidate transaction: " + tx, ex);
            }

            if (e instanceof Error)
                throw (Error)e;
        }
    }

    /**
     * Finish for one-phase distributed tx.
     *
     * @param tx Transaction.
     * @param req Request.
     */
    protected void finish(
        GridDistributedTxRemoteAdapter tx,
        GridDhtTxPrepareRequest req) throws IgniteTxHeuristicCheckedException {
        assert tx != null : "No transaction for one-phase commit prepare request: " + req;

        try {
            tx.commitVersion(req.writeVersion());
            tx.invalidate(req.isInvalidate());
            tx.mvccSnapshot(req.mvccSnapshot());

            // Complete remote candidates.
            tx.doneRemote(req.version(), null, null, null);

            tx.commitRemoteTx();
        }
        catch (IgniteTxHeuristicCheckedException e) {
            // Just rethrow this exception. Transaction was already uncommitted.
            throw e;
        }
        catch (Throwable e) {
            try {
                // Mark transaction for invalidate.
                tx.invalidate(true);
                tx.systemInvalidate(true);

                try {
                    tx.rollbackRemoteTx();
                }
                catch (Throwable e1) {
                    e.addSuppressed(e1);
                }

                tx.logTxFinishErrorSafe(log, true, e);

                if (e instanceof Error)
                    throw (Error)e;
            }
            finally {
                ctx.kernalContext().failure().process(new FailureContext(FailureType.CRITICAL_ERROR, e));
            }
        }
    }

    /**
     * @param nodeId Node id.
     * @param req Request.
     * @param res Response.
     * @param dhtTx Dht tx.
     * @param nearTx Near tx.
     */
    private void sendReply(UUID nodeId,
        GridDhtTxPrepareRequest req,
        GridDhtTxPrepareResponse res,
        GridDhtTxRemote dhtTx,
        GridNearTxRemote nearTx) {
        try {
            // Reply back to sender.
            ctx.io().send(nodeId, res, req.policy());

            if (txPrepareMsgLog.isDebugEnabled()) {
                txPrepareMsgLog.debug("Sent dht prepare response [txId=" + req.nearXidVersion() +
                    ", dhtTxId=" + req.version() +
                    ", node=" + nodeId + ']');
            }
        }
        catch (IgniteCheckedException e) {
            if (e instanceof ClusterTopologyCheckedException) {
                if (txPrepareMsgLog.isDebugEnabled()) {
                    txPrepareMsgLog.debug("Failed to send dht prepare response, node left [txId=" + req.nearXidVersion() +
                        ", dhtTxId=" + req.version() +
                        ", node=" + nodeId + ']');
                }
            }
            else {
                U.warn(log, "Failed to send tx response to remote node (will rollback transaction) [" +
                    "txId=" + req.nearXidVersion() +
                    ", dhtTxId=" + req.version() +
                    ", node=" + nodeId +
                    ", err=" + e.getMessage() + ']');
            }

            if (nearTx != null)
                try {
                    nearTx.rollbackRemoteTx();
                }
                catch (Throwable e1) {
                    e.addSuppressed(e1);
                }

            if (dhtTx != null)
                try {
                    dhtTx.rollbackRemoteTx();
                }
                catch (Throwable e1) {
                    e.addSuppressed(e1);
                }
        }
    }

    /**
     * Sends tx finish response to remote node, if response is requested.
     *
     * @param nodeId Node id that originated finish request.
     * @param req Request.
     * @param committed {@code True} if transaction committed on this node.
     * @param nearTxId Near tx version.
     */
    private void sendReply(UUID nodeId, GridDhtTxFinishRequest req, boolean committed, GridCacheVersion nearTxId) {
        if (req.replyRequired() || req.checkCommitted()) {
            GridDhtTxFinishResponse res = new GridDhtTxFinishResponse(
                req.partition(),
                req.version(),
                req.futureId(),
                req.miniId());

            if (req.checkCommitted()) {
                res.checkCommitted(true);

                if (committed) {
                    if (req.needReturnValue()) {
                        try {
                            GridCacheReturnCompletableWrapper wrapper = ctx.tm().getCommittedTxReturn(req.version());

                            if (wrapper != null)
                                res.returnValue(wrapper.fut().get());
                            else
                                assert !ctx.discovery().alive(nodeId) : nodeId;
                        }
                        catch (IgniteCheckedException ignored) {
                            if (txFinishMsgLog.isDebugEnabled()) {
                                txFinishMsgLog.debug("Failed to gain entry processor return value. [txId=" + nearTxId +
                                    ", dhtTxId=" + req.version() +
                                    ", node=" + nodeId + ']');
                            }
                        }
                    }
                }
                else {
                    ClusterTopologyCheckedException cause =
                        new ClusterTopologyCheckedException("Primary node left grid.");

                    res.checkCommittedError(new IgniteTxRollbackCheckedException("Failed to commit transaction " +
                        "(transaction has been rolled back on backup node): " + req.version(), cause));
                }
            }

            try {
                ctx.io().send(nodeId, res, req.policy());

                if (txFinishMsgLog.isDebugEnabled()) {
                    txFinishMsgLog.debug("Sent dht tx finish response [txId=" + nearTxId +
                        ", dhtTxId=" + req.version() +
                        ", node=" + nodeId +
                        ", checkCommitted=" + req.checkCommitted() + ']');
                }
            }
            catch (Throwable e) {
                // Double-check.
                if (ctx.discovery().node(nodeId) == null) {
                    if (txFinishMsgLog.isDebugEnabled()) {
                        txFinishMsgLog.debug("Node left while send dht tx finish response [txId=" + nearTxId +
                            ", dhtTxId=" + req.version() +
                            ", node=" + nodeId + ']');
                    }
                }
                else {
                    U.error(log, "Failed to send finish response to node [txId=" + nearTxId +
                        ", dhtTxId=" + req.version() +
                        ", nodeId=" + nodeId +
                        ", res=" + res + ']', e);
                }

                if (e instanceof Error)
                    throw (Error)e;
            }
        }
        else {
            if (txFinishMsgLog.isDebugEnabled()) {
                txFinishMsgLog.debug("Skip send dht tx finish response [txId=" + nearTxId +
                    ", dhtTxId=" + req.version() +
                    ", node=" + nodeId + ']');
            }
        }
    }

    /**
     * @param nodeId Node ID.
     * @param req Request.
     * @param res Response.
     * @return Remote transaction.
     * @throws IgniteCheckedException If failed.
     */
    @Nullable GridDhtTxRemote startRemoteTx(
        UUID nodeId,
        GridDhtTxPrepareRequest req,
        GridDhtTxPrepareResponse res
    ) throws IgniteCheckedException {
        if (req.queryUpdate() || !F.isEmpty(req.writes())) {
            GridDhtTxRemote tx = ctx.tm().tx(req.version());

            if (tx == null) {
                boolean single = req.last() && req.writes().size() == 1;

                tx = new GridDhtTxRemote(
                    ctx,
                    req.nearNodeId(),
                    req.futureId(),
                    nodeId,
                    req.topologyVersion(),
                    req.version(),
                    null,
                    req.system(),
                    req.policy(),
                    req.concurrency(),
                    req.isolation(),
                    req.isInvalidate(),
                    req.timeout(),
                    req.writes() != null ? Math.max(req.writes().size(), req.txSize()) : req.txSize(),
                    req.nearXidVersion(),
                    req.transactionNodes(),
                    securitySubjectId(ctx),
                    req.taskNameHash(),
                    single,
                    req.storeWriteThrough(),
                    req.txLabel());

                tx.onePhaseCommit(req.onePhaseCommit());
                tx.writeVersion(req.writeVersion());

                tx = ctx.tm().onCreated(null, tx);

                if (tx == null || !ctx.tm().onStarted(tx)) {
                    if (log.isDebugEnabled())
                        log.debug("Attempt to start a completed transaction (will ignore): " + tx);

                    applyPartitionsUpdatesCounters(req.updateCounters(), true, false);

                    return null;
                }

                if (ctx.discovery().node(nodeId) == null) {
                    tx.state(ROLLING_BACK);

                    tx.state(ROLLED_BACK);

                    ctx.tm().uncommitTx(tx);

                    applyPartitionsUpdatesCounters(req.updateCounters(), true, false);

                    return null;
                }

                ctx.versions().onReceived(nodeId, req.writeVersion());
            }
            else {
                tx.writeVersion(req.writeVersion());
                tx.transactionNodes(req.transactionNodes());
            }

            TxCounters txCounters = null;

            if (req.updateCounters() != null) {
                txCounters = tx.txCounters(true);

                txCounters.updateCounters(req.updateCounters());
            }

            Set<GridDhtLocalPartition> reservedParts = new HashSet<>();

            try {
                if (!tx.isSystemInvalidate()) {
                    int idx = 0;

                    for (IgniteTxEntry entry : req.writes()) {
                        GridCacheContext cacheCtx = entry.context();

                        int part = cacheCtx.affinity().partition(entry.key());

                        try {
                            GridDhtLocalPartition locPart = cacheCtx.topology().localPartition(part,
                                req.topologyVersion(),
                                false);

                            // Avoid enlisting to invalid partition.
                            boolean reserved = locPart != null && reservedParts.contains(locPart);

                            if (!reserved) {
                                if ((reserved = locPart != null && locPart.reserve()))
                                    reservedParts.add(locPart);
                            }

                            if (reserved) {
                                tx.addWrite(entry, ctx.deploy().globalLoader());

                                if (txCounters != null && entry.op() != NOOP && !(entry.op() == TRANSFORM && entry.noop())) {
                                    Long cntr = txCounters.generateNextCounter(entry.cacheId(), part);

                                    if (cntr != null) // Counter is null if entry is no-op.
                                        entry.updateCounter(cntr);
                                }

                                if (isNearEnabled(cacheCtx) && req.invalidateNearEntry(idx))
                                    invalidateNearEntry(cacheCtx, entry.key(), req.version());

                                if (req.needPreloadKey(idx)) {
                                    GridCacheEntryEx cached = entry.cached();

                                    if (cached == null)
                                        cached = cacheCtx.cache().entryEx(entry.key(), req.topologyVersion());

                                    GridCacheEntryInfo info = cached.info();

                                    if (info != null && !info.isNew() && !info.isDeleted())
                                        res.addPreloadEntry(info);
                                }

                                if (cacheCtx.readThroughConfigured() &&
                                    !entry.skipStore() &&
                                    entry.op() == TRANSFORM &&
                                    entry.oldValueOnPrimary() &&
                                    !entry.hasValue()) {
                                    while (true) {
                                        try {
                                            GridCacheEntryEx cached = entry.cached();

                                            if (cached == null) {
                                                cached = cacheCtx.cache().entryEx(entry.key(), req.topologyVersion());

                                                entry.cached(cached);
                                            }

                                            CacheObject val = cached.innerGet(
                                                /*ver*/null,
                                                tx,
                                                /*readThrough*/false,
                                                /*updateMetrics*/false,
                                                /*evt*/false,
                                                /*transformClo*/null,
                                                tx.resolveTaskName(),
                                                /*expiryPlc*/null,
                                                /*keepBinary*/true);

                                            if (val == null)
                                                val = cacheCtx.toCacheObject(cacheCtx.store().load(null, entry.key()));

                                            if (val != null)
                                                entry.readValue(val);

                                            break;
                                        }
                                        catch (GridCacheEntryRemovedException ignored) {
                                            if (log.isDebugEnabled())
                                                log.debug("Got entry removed exception, will retry: " + entry.txKey());

                                            entry.cached(cacheCtx.cache().entryEx(entry.key(), req.topologyVersion()));
                                        }
                                    }
                                }
                            }
                            else
                                tx.addInvalidPartition(cacheCtx.cacheId(), part);
                        }
                        catch (GridDhtInvalidPartitionException e) {
                            tx.addInvalidPartition(cacheCtx.cacheId(), part);
                        }

                        idx++;
                    }
                }

                // Prepare prior to reordering, so the pending locks added
                // in prepare phase will get properly ordered as well.
                tx.prepareRemoteTx();
            }
            finally {
                reservedParts.forEach(GridDhtLocalPartition::release);
            }

            if (req.last()) {
                assert !F.isEmpty(req.transactionNodes()) :
                    "Received last prepare request with empty transaction nodes: " + req;

                tx.state(PREPARED);
            }

            res.invalidPartitionsByCacheId(tx.invalidPartitions());

            if (!req.queryUpdate() && tx.empty() && req.last()) {
                tx.skipCompletedVersions(req.skipCompletedVersion());

                tx.rollbackRemoteTx();

                return null;
            }

            return tx;
        }

        return null;
    }

    /**
     * Writes updated values on the backup node.
     *
     * @param tx Transaction.
     * @param ctx Cache context.
     * @param op Operation.
     * @param keys Keys.
     * @param vals Values sent from the primary node.
     * @param snapshot Mvcc snapshot.
     * @param batchNum Batch number.
     * @param futId Future id.
     * @throws IgniteCheckedException If failed.
     */
    public void mvccEnlistBatch(GridDhtTxRemote tx, GridCacheContext ctx, EnlistOperation op, List<KeyCacheObject> keys,
        List<Message> vals, MvccSnapshot snapshot, IgniteUuid futId, int batchNum) throws IgniteCheckedException {
        assert keys != null && (vals == null || vals.size() == keys.size());
        assert tx != null;

        GridDhtCacheAdapter dht = ctx.dht();

        tx.addActiveCache(ctx, false);

        for (int i = 0; i < keys.size(); i++) {
            KeyCacheObject key = keys.get(i);

            assert key != null;

            int part = ctx.affinity().partition(key);

            try {
                GridDhtLocalPartition locPart = ctx.topology().localPartition(part, tx.topologyVersion(), false);

                if (locPart != null && locPart.reserve()) {
                    try {
                        // Skip renting partitions.
                        if (locPart.state() == RENTING) {
                            tx.addInvalidPartition(ctx.cacheId(), part);

                            continue;
                        }

                        CacheObject val = null;
                        EntryProcessor entryProc = null;
                        Object[] invokeArgs = null;

                        boolean needOldVal = tx.txState().useMvccCaching(ctx.cacheId());

                        Message val0 = vals != null ? vals.get(i) : null;

                        CacheEntryInfoCollection entries =
                            val0 instanceof CacheEntryInfoCollection ? (CacheEntryInfoCollection)val0 : null;

                        if (entries == null && !op.isDeleteOrLock() && !op.isInvoke())
                            val = (val0 instanceof CacheObject) ? (CacheObject)val0 : null;

                        if (entries == null && op.isInvoke()) {
                            assert val0 instanceof GridInvokeValue;

                            GridInvokeValue invokeVal = (GridInvokeValue)val0;

                            entryProc = invokeVal.entryProcessor();
                            invokeArgs = invokeVal.invokeArgs();
                        }

                        assert entries != null || entryProc != null || !op.isInvoke() : "entryProc=" + entryProc + ", op=" + op;

                        GridDhtCacheEntry entry = dht.entryExx(key, tx.topologyVersion());

                        GridCacheUpdateTxResult updRes;

                        while (true) {
                            ctx.shared().database().checkpointReadLock();

                            try {
                                if (entries == null) {
                                    switch (op) {
                                        case DELETE:
                                            updRes = entry.mvccRemove(
                                                tx,
                                                ctx.localNodeId(),
                                                tx.topologyVersion(),
                                                snapshot,
                                                false,
                                                needOldVal,
                                                null,
                                                false);

                                            break;

                                        case INSERT:
                                        case TRANSFORM:
                                        case UPSERT:
                                        case UPDATE:
                                            updRes = entry.mvccSet(
                                                tx,
                                                ctx.localNodeId(),
                                                val,
                                                entryProc,
                                                invokeArgs,
                                                0,
                                                tx.topologyVersion(),
                                                snapshot,
                                                op.cacheOperation(),
                                                false,
                                                false,
                                                needOldVal,
                                                null,
                                                false,
                                                false);

                                            break;

                                        default:
                                            throw new IgniteSQLException("Cannot acquire lock for operation [op= "
                                                + op + "]" + "Operation is unsupported at the moment ",
                                                IgniteQueryErrorCode.UNSUPPORTED_OPERATION);
                                    }
                                }
                                else {
                                    updRes = entry.mvccUpdateRowsWithPreloadInfo(tx,
                                        ctx.localNodeId(),
                                        tx.topologyVersion(),
                                        entries.infos(),
                                        op.cacheOperation(),
                                        snapshot,
                                        futId,
                                        batchNum);
                                }

                                break;
                            }
                            catch (GridCacheEntryRemovedException ignore) {
                                entry = dht.entryExx(key);
                            }
                            finally {
                                ctx.shared().database().checkpointReadUnlock();
                            }
                        }

                        if (!updRes.filtered())
                            ctx.shared().mvccCaching().addEnlisted(key, updRes.newValue(), 0, 0, tx.xidVersion(),
                                updRes.oldValue(), tx.local(), tx.topologyVersion(), snapshot, ctx.cacheId(), tx, futId, batchNum);

                        assert updRes.updateFuture() == null : "Entry should not be locked on the backup";
                    }

                    finally {
                        locPart.release();
                    }
                }
                else
                    tx.addInvalidPartition(ctx.cacheId(), part);
            }
            catch (GridDhtInvalidPartitionException e) {
                tx.addInvalidPartition(ctx.cacheId(), e.partition());
            }
        }
    }

    /**
     * @param cacheCtx Context.
     * @param key Key
     * @param ver Version.
     * @throws IgniteCheckedException If invalidate failed.
     */
    private void invalidateNearEntry(GridCacheContext cacheCtx, KeyCacheObject key, GridCacheVersion ver)
        throws IgniteCheckedException {
        GridNearCacheAdapter near = cacheCtx.isNear() ? cacheCtx.near() : cacheCtx.dht().near();

        GridCacheEntryEx nearEntry = near.peekEx(key);

        if (nearEntry != null)
            nearEntry.invalidate(ver);
    }

    /**
     * Called while processing dht tx prepare request.
     *
     * @param ldr Loader.
     * @param nodeId Sender node ID.
     * @param req Request.
     * @return Remote transaction.
     * @throws IgniteCheckedException If failed.
     */
    @Nullable private GridNearTxRemote startNearRemoteTx(ClassLoader ldr, UUID nodeId,
        GridDhtTxPrepareRequest req) throws IgniteCheckedException {

        if (!F.isEmpty(req.nearWrites())) {
            GridNearTxRemote tx = ctx.tm().nearTx(req.version());

            if (tx == null) {
                tx = new GridNearTxRemote(
                    ctx,
                    req.topologyVersion(),
                    ldr,
                    nodeId,
                    req.nearNodeId(),
                    req.version(),
                    null,
                    req.system(),
                    req.policy(),
                    req.concurrency(),
                    req.isolation(),
                    req.isInvalidate(),
                    req.timeout(),
                    req.nearWrites(),
                    req.txSize(),
                    securitySubjectId(ctx),
                    req.taskNameHash(),
                    req.txLabel()
                );

                tx.writeVersion(req.writeVersion());

                if (!tx.empty()) {
                    tx = ctx.tm().onCreated(null, tx);

                    if (tx == null || !ctx.tm().onStarted(tx))
                        throw new IgniteTxRollbackCheckedException("Attempt to start a completed transaction: " + tx);
                }
            }
            else
                tx.addEntries(ldr, req.nearWrites());

            tx.ownedVersions(req.owned());

            // Prepare prior to reordering, so the pending locks added
            // in prepare phase will get properly ordered as well.
            tx.prepareRemoteTx();

            if (req.last())
                tx.state(PREPARED);

            return tx;
        }

        return null;
    }

    /**
     * @param nodeId Node ID.
     * @param req Request.
     */
    private void processCheckPreparedTxRequest(final UUID nodeId,
        final GridCacheTxRecoveryRequest req) {
        if (txRecoveryMsgLog.isDebugEnabled()) {
            txRecoveryMsgLog.debug("Received tx recovery request [txId=" + req.nearXidVersion() +
                ", node=" + nodeId + ']');
        }

        IgniteInternalFuture<Boolean> fut = req.nearTxCheck() ? ctx.tm().txCommitted(req.nearXidVersion()) :
            ctx.tm().txsPreparedOrCommitted(req.nearXidVersion(), req.transactions());

        if (fut == null || fut.isDone()) {
            boolean prepared;

            try {
                prepared = fut == null ? true : fut.get();
            }
            catch (IgniteCheckedException e) {
                U.error(log, "Check prepared transaction future failed [req=" + req + ']', e);

                prepared = false;
            }

            sendCheckPreparedResponse(nodeId, req, prepared);
        }
        else {
            fut.listen(new CI1<IgniteInternalFuture<Boolean>>() {
                @Override public void apply(IgniteInternalFuture<Boolean> fut) {
                    boolean prepared;

                    try {
                        prepared = fut.get();
                    }
                    catch (IgniteCheckedException e) {
                        U.error(log, "Check prepared transaction future failed [req=" + req + ']', e);

                        prepared = false;
                    }

                    sendCheckPreparedResponse(nodeId, req, prepared);
                }
            });
        }
    }

    /**
     * @param nodeId Node ID.
     * @param req Request.
     * @param prepared {@code True} if all transaction prepared or committed.
     */
    private void sendCheckPreparedResponse(UUID nodeId,
        GridCacheTxRecoveryRequest req,
        boolean prepared) {
        GridCacheTxRecoveryResponse res = new GridCacheTxRecoveryResponse(req.version(),
            req.futureId(),
            req.miniId(),
            prepared,
            req.deployInfo() != null);

        try {
            ctx.io().send(nodeId, res, req.system() ? UTILITY_CACHE_POOL : SYSTEM_POOL);

            if (txRecoveryMsgLog.isDebugEnabled()) {
                txRecoveryMsgLog.debug("Sent tx recovery response [txId=" + req.nearXidVersion() +
                    ", node=" + nodeId + ", res=" + res + ']');
            }
        }
        catch (ClusterTopologyCheckedException ignored) {
            if (txRecoveryMsgLog.isDebugEnabled())
                txRecoveryMsgLog.debug("Failed to send tx recovery response, node failed [" +
                    ", txId=" + req.nearXidVersion() +
                    ", node=" + nodeId +
                    ", res=" + res + ']');
        }
        catch (IgniteCheckedException e) {
            U.error(txRecoveryMsgLog, "Failed to send tx recovery response [txId=" + req.nearXidVersion() +
                ", node=" + nodeId +
                ", req=" + req +
                ", res=" + res + ']', e);
        }
    }

    /**
     * @param nodeId Node ID.
     * @param res Response.
     */
    protected void processCheckPreparedTxResponse(UUID nodeId, GridCacheTxRecoveryResponse res) {
        if (txRecoveryMsgLog.isInfoEnabled()) {
            txRecoveryMsgLog.info("Received tx recovery response [txId=" + res.version() +
                ", node=" + nodeId +
                ", res=" + res + ']');
        }

        GridCacheTxRecoveryFuture fut = (GridCacheTxRecoveryFuture)ctx.mvcc().future(res.futureId());

        if (fut == null) {
            if (txRecoveryMsgLog.isInfoEnabled()) {
                txRecoveryMsgLog.info("Failed to find future for tx recovery response [txId=" + res.version() +
                    ", node=" + nodeId + ", res=" + res + ']');
            }

            return;
        }
        else
            res.txState(fut.tx().txState());

        fut.onResult(nodeId, res);
    }

    /**
     * Applies partition counter updates for transactions.
     * <p>
     * Called after entries are written to WAL on commit or during rollback to close gaps in update counter sequence.
     *
     * @param counters Counters.
     */
    public void applyPartitionsUpdatesCounters(Iterable<PartitionUpdateCountersMessage> counters)
        throws IgniteCheckedException {
        applyPartitionsUpdatesCounters(counters, false, false);
    }

    /**
     * Applies partition counter updates for transactions.
     * <p>
     * Called after entries are written to WAL on commit or during rollback to close gaps in update counter sequence.
     * <p>
     * On rollback counters should be applied on the primary only after backup nodes, otherwise if the primary fail
     * before sending rollback requests to backups remote transactions can be committed by recovery protocol and
     * partition consistency will not be restored when primary returns to the grid because RollbackRecord was written
     * (actual for persistent mode only).
     *
     * @param counters Counter values to be updated.
     * @param rollback {@code True} if applied during rollbacks.
     * @param rollbackOnPrimary {@code True} if rollback happens on primary node. Passed to CQ engine.
     */
    public void applyPartitionsUpdatesCounters(
        Iterable<PartitionUpdateCountersMessage> counters,
        boolean rollback,
        boolean rollbackOnPrimary
    ) throws IgniteCheckedException {
        if (counters == null)
            return;

        WALPointer ptr = null;

        try {
            for (PartitionUpdateCountersMessage counter : counters) {
                GridCacheContext ctx0 = ctx.cacheContext(counter.cacheId());

                GridDhtPartitionTopology top = ctx0.topology();

                AffinityTopologyVersion topVer = top.readyTopologyVersion();

                assert top != null;

                for (int i = 0; i < counter.size(); i++) {
                    boolean invalid = false;

                    try {
                        GridDhtLocalPartition part = top.localPartition(counter.partition(i));

                        if (part != null && part.reserve()) {
                            try {
                                if (part.state() != RENTING) { // Check is actual only for backup node.
                                    long start = counter.initialCounter(i);
                                    long delta = counter.updatesCount(i);

                                    boolean updated = part.updateCounter(start, delta);

                                    // Need to log rolled back range for logical recovery.
                                    if (updated && rollback) {
                                        CacheGroupContext grpCtx = part.group();

                                        if (grpCtx.persistenceEnabled() && grpCtx.walEnabled() && !grpCtx.mvccEnabled()) {
                                            RollbackRecord rec =
                                                new RollbackRecord(grpCtx.groupId(), part.id(), start, delta);

                                            ptr = ctx.wal().log(rec);
                                        }

                                        for (int cntr = 1; cntr <= delta; cntr++) {
                                            ctx0.continuousQueries().skipUpdateCounter(null, part.id(), start + cntr,
                                                topVer, rollbackOnPrimary);
                                        }
                                    }
                                }
                                else
                                    invalid = true;
                            }
                            finally {
                                part.release();
                            }
                        }
                        else
                            invalid = true;
                    }
                    catch (GridDhtInvalidPartitionException e) {
                        invalid = true;
                    }

                    if (log.isDebugEnabled() && invalid) {
                        log.debug("Received partition update counters message for invalid partition, ignoring: " +
                            "[cacheId=" + counter.cacheId() + ", part=" + counter.partition(i) + ']');
                    }
                }
            }
        }
        finally {
            if (ptr != null)
                ctx.wal().flush(ptr, false);
        }
    }

    /**
     * @param tx Transaction.
     * @param node Backup node.
     * @return Partition counters for the given backup node.
     */
    @Nullable public List<PartitionUpdateCountersMessage> filterUpdateCountersForBackupNode(
        IgniteInternalTx tx, ClusterNode node) {
        TxCounters txCntrs = tx.txCounters(false);
        Collection<PartitionUpdateCountersMessage> updCntrs;

        if (txCntrs == null || F.isEmpty(updCntrs = txCntrs.updateCounters()))
            return null;

        List<PartitionUpdateCountersMessage> res = new ArrayList<>(updCntrs.size());

        AffinityTopologyVersion top = tx.topologyVersionSnapshot();

        for (PartitionUpdateCountersMessage partCntrs : updCntrs) {
            GridDhtPartitionTopology topology = ctx.cacheContext(partCntrs.cacheId()).topology();

            PartitionUpdateCountersMessage resCntrs = new PartitionUpdateCountersMessage(partCntrs.cacheId(), partCntrs.size());

            for (int i = 0; i < partCntrs.size(); i++) {
                int part = partCntrs.partition(i);

                if (topology.nodes(part, top).indexOf(node) > 0)
                    resCntrs.add(part, partCntrs.initialCounter(i), partCntrs.updatesCount(i));
            }

            if (resCntrs.size() > 0)
                res.add(resCntrs);
        }

        return res;
    }
}