/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.ignite.internal.processors.cache.distributed;

import java.io.Externalizable;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.stream.Collectors;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
import org.apache.ignite.failure.FailureContext;
import org.apache.ignite.failure.FailureType;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.InvalidEnvironmentException;
import org.apache.ignite.internal.NodeStoppingException;
import org.apache.ignite.internal.pagemem.wal.WALPointer;
import org.apache.ignite.internal.pagemem.wal.record.DataEntry;
import org.apache.ignite.internal.pagemem.wal.record.DataRecord;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
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.GridCacheEntryRemovedException;
import org.apache.ignite.internal.processors.cache.GridCacheFilterFailedException;
import org.apache.ignite.internal.processors.cache.GridCacheMvccCandidate;
import org.apache.ignite.internal.processors.cache.GridCacheOperation;
import org.apache.ignite.internal.processors.cache.GridCacheReturn;
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.near.GridNearCacheEntry;
import org.apache.ignite.internal.processors.cache.persistence.StorageException;
import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxAdapter;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxEntry;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxKey;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxRemoteEx;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxRemoteState;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxState;
import org.apache.ignite.internal.processors.cache.transactions.TxCounters;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersionConflictContext;
import org.apache.ignite.internal.transactions.IgniteTxHeuristicCheckedException;
import org.apache.ignite.internal.util.future.GridFinishedFuture;
import org.apache.ignite.internal.util.lang.GridTuple;
import org.apache.ignite.internal.util.tostring.GridToStringBuilder;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.T2;
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.IgniteBiTuple;
import org.apache.ignite.transactions.TransactionConcurrency;
import org.apache.ignite.transactions.TransactionIsolation;
import org.apache.ignite.transactions.TransactionState;
import org.jetbrains.annotations.Nullable;

import static org.apache.ignite.internal.processors.cache.GridCacheOperation.CREATE;
import static org.apache.ignite.internal.processors.cache.GridCacheOperation.DELETE;
import static org.apache.ignite.internal.processors.cache.GridCacheOperation.NOOP;
import static org.apache.ignite.internal.processors.cache.GridCacheOperation.READ;
import static org.apache.ignite.internal.processors.cache.GridCacheOperation.RELOAD;
import static org.apache.ignite.internal.processors.cache.GridCacheOperation.UPDATE;
import static org.apache.ignite.internal.processors.dr.GridDrType.DR_BACKUP;
import static org.apache.ignite.internal.processors.dr.GridDrType.DR_NONE;
import static org.apache.ignite.transactions.TransactionState.COMMITTED;
import static org.apache.ignite.transactions.TransactionState.COMMITTING;
import static org.apache.ignite.transactions.TransactionState.PREPARED;
import static org.apache.ignite.transactions.TransactionState.PREPARING;
import static org.apache.ignite.transactions.TransactionState.ROLLED_BACK;
import static org.apache.ignite.transactions.TransactionState.ROLLING_BACK;
import static org.apache.ignite.transactions.TransactionState.UNKNOWN;

/**
 * Transaction created by system implicitly on remote nodes.
 */
public abstract class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
    implements IgniteTxRemoteEx {
    /** */
    private static final long serialVersionUID = 0L;

    /** Commit allowed field updater. */
    private static final AtomicIntegerFieldUpdater<GridDistributedTxRemoteAdapter> COMMIT_ALLOWED_UPD =
        AtomicIntegerFieldUpdater.newUpdater(GridDistributedTxRemoteAdapter.class, "commitAllowed");

    /** Explicit versions. */
    @GridToStringInclude
    private List<GridCacheVersion> explicitVers;

    /** Started flag. */
    @GridToStringInclude
    private boolean started;

    /** {@code True} only if all write entries are locked by this transaction. */
    @SuppressWarnings("UnusedDeclaration")
    @GridToStringInclude
    private volatile int commitAllowed;

    /** */
    @GridToStringInclude
    protected IgniteTxRemoteState txState;

    /** {@code True} if tx should skip adding itself to completed version map on finish. */
    private boolean skipCompletedVers;

    /**
     * Empty constructor required for {@link Externalizable}.
     */
    public GridDistributedTxRemoteAdapter() {
        // No-op.
    }

    /**
     * @param ctx Cache registry.
     * @param nodeId Node ID.
     * @param xidVer XID version.
     * @param commitVer Commit version.
     * @param sys System flag.
     * @param plc IO policy.
     * @param concurrency Concurrency level (should be pessimistic).
     * @param isolation Transaction isolation.
     * @param invalidate Invalidate flag.
     * @param timeout Timeout.
     * @param txSize Expected transaction size.
     * @param subjId Subject ID.
     * @param taskNameHash Task name hash code.
     */
    public GridDistributedTxRemoteAdapter(
        GridCacheSharedContext<?, ?> ctx,
        UUID nodeId,
        GridCacheVersion xidVer,
        GridCacheVersion commitVer,
        boolean sys,
        byte plc,
        TransactionConcurrency concurrency,
        TransactionIsolation isolation,
        boolean invalidate,
        long timeout,
        int txSize,
        @Nullable UUID subjId,
        int taskNameHash
    ) {
        super(
            ctx,
            nodeId,
            xidVer,
            ctx.versions().last(),
            Thread.currentThread().getId(),
            sys,
            plc,
            concurrency,
            isolation,
            timeout,
            txSize,
            subjId,
            taskNameHash);

        this.invalidate = invalidate;

        commitVersion(commitVer);

        // Must set started flag after concurrency and isolation.
        started = true;
    }

    /** {@inheritDoc} */
    @Override public IgniteTxState txState() {
        return txState;
    }

    /** {@inheritDoc} */
    @Override public UUID eventNodeId() {
        return nodeId;
    }

    /** {@inheritDoc} */
    @Override public UUID originatingNodeId() {
        return nodeId;
    }

    /** {@inheritDoc} */
    @Override public boolean activeCachesDeploymentEnabled() {
        return false;
    }

    /** {@inheritDoc} */
    @Override public void activeCachesDeploymentEnabled(boolean depEnabled) {
        throw new UnsupportedOperationException("Remote tx doesn't support deployment.");
    }

    /** {@inheritDoc} */
    @Override public void addActiveCache(GridCacheContext cacheCtx, boolean recovery) throws IgniteCheckedException {
        txState.addActiveCache(cacheCtx, recovery, this);
    }

    /**
     * @return Checks if transaction has no entries.
     */
    @Override public boolean empty() {
        return txState.empty();
    }

    /** {@inheritDoc} */
    @Override public void invalidate(boolean invalidate) {
        this.invalidate = invalidate;
    }

    /** {@inheritDoc} */
    @Override public Map<IgniteTxKey, IgniteTxEntry> writeMap() {
        return txState.writeMap();
    }

    /** {@inheritDoc} */
    @Override public Map<IgniteTxKey, IgniteTxEntry> readMap() {
        return txState.readMap();
    }

    /** {@inheritDoc} */
    @Override public void seal() {
        // No-op.
    }

    /** {@inheritDoc} */
    @Override public GridTuple<CacheObject> peek(GridCacheContext cacheCtx,
        boolean failFast,
        KeyCacheObject key)
        throws GridCacheFilterFailedException {
        assert false : "Method peek can only be called on user transaction: " + this;

        throw new IllegalStateException("Method peek can only be called on user transaction: " + this);
    }

    /** {@inheritDoc} */
    @Override public IgniteTxEntry entry(IgniteTxKey key) {
        return txState.entry(key);
    }

    /**
     * Clears entry from transaction as it never happened.
     *
     * @param key key to be removed.
     */
    public void clearEntry(IgniteTxKey key) {
        txState.clearEntry(key);
    }

    /**
     * @param baseVer Base version.
     * @param committedVers Committed versions.
     * @param rolledbackVers Rolled back versions.
     */
    @Override public void doneRemote(GridCacheVersion baseVer,
        Collection<GridCacheVersion> committedVers,
        Collection<GridCacheVersion> rolledbackVers,
        Collection<GridCacheVersion> pendingVers) {
        Map<IgniteTxKey, IgniteTxEntry> readMap = txState.readMap();

        if (readMap != null && !readMap.isEmpty()) {
            for (IgniteTxEntry txEntry : readMap.values())
                doneRemote(txEntry, baseVer, committedVers, rolledbackVers, pendingVers);
        }

        Map<IgniteTxKey, IgniteTxEntry> writeMap = txState.writeMap();

        if (writeMap != null && !writeMap.isEmpty()) {
            for (IgniteTxEntry txEntry : writeMap.values())
                doneRemote(txEntry, baseVer, committedVers, rolledbackVers, pendingVers);
        }
    }

    /** {@inheritDoc} */
    @Override public void setPartitionUpdateCounters(long[] cntrs) {
        if (writeMap() != null && !writeMap().isEmpty() && cntrs != null && cntrs.length > 0) {
            int i = 0;

            for (IgniteTxEntry txEntry : writeMap().values()) {
                txEntry.updateCounter(cntrs[i]);

                ++i;
            }
        }
    }

    /**
     * Adds completed versions to an entry.
     *
     * @param txEntry Entry.
     * @param baseVer Base version for completed versions.
     * @param committedVers Completed versions relative to base version.
     * @param rolledbackVers Rolled back versions relative to base version.
     * @param pendingVers Pending versions.
     */
    private void doneRemote(IgniteTxEntry txEntry,
        GridCacheVersion baseVer,
        Collection<GridCacheVersion> committedVers,
        Collection<GridCacheVersion> rolledbackVers,
        Collection<GridCacheVersion> pendingVers) {
        while (true) {
            GridDistributedCacheEntry entry = (GridDistributedCacheEntry)txEntry.cached();

            try {
                // Handle explicit locks.
                GridCacheVersion doneVer = txEntry.explicitVersion() != null ? txEntry.explicitVersion() : xidVer;

                entry.doneRemote(doneVer, baseVer, pendingVers, committedVers, rolledbackVers, isSystemInvalidate());

                break;
            }
            catch (GridCacheEntryRemovedException ignored) {
                assert entry.obsoleteVersion() != null;

                if (log.isDebugEnabled())
                    log.debug("Replacing obsolete entry in remote transaction [entry=" + entry + ", tx=" + this + ']');

                // Replace the entry.
                txEntry.cached(txEntry.context().cache().entryEx(txEntry.key(), topologyVersion()));
            }
        }
    }

    /** {@inheritDoc} */
    @Override public boolean onOwnerChanged(GridCacheEntryEx entry, GridCacheMvccCandidate owner) {
        if (!hasWriteKey(entry.txKey()))
            return false;

        try {
            commitIfLocked();

            return true;
        }
        catch (IgniteCheckedException e) {
            U.error(log, "Failed to commit remote transaction: " + this, e);

            invalidate(true);
            systemInvalidate(true);

            rollbackRemoteTx();

            return false;
        }
    }

    /** {@inheritDoc} */
    @Override public boolean isStarted() {
        return started;
    }

    /** {@inheritDoc} */
    @Override public boolean hasWriteKey(IgniteTxKey key) {
        return txState.hasWriteKey(key);
    }

    /** {@inheritDoc} */
    @Override public Set<IgniteTxKey> readSet() {
        return txState.readSet();
    }

    /** {@inheritDoc} */
    @Override public Set<IgniteTxKey> writeSet() {
        return txState.writeSet();
    }

    /** {@inheritDoc} */
    @Override public Collection<IgniteTxEntry> allEntries() {
        return txState.allEntries();
    }

    /** {@inheritDoc} */
    @Override public Collection<IgniteTxEntry> writeEntries() {
        return txState.writeEntries();
    }

    /** {@inheritDoc} */
    @Override public Collection<IgniteTxEntry> readEntries() {
        return txState.readEntries();
    }

    /**
     * @throws IgniteCheckedException If failed.
     */
    public final void prepareRemoteTx() throws IgniteCheckedException {
        // If another thread is doing prepare or rollback.
        if (!state(PREPARING)) {
            // In optimistic mode prepare may be called multiple times.
            if (state() != PREPARING || !optimistic()) {
                if (log.isDebugEnabled())
                    log.debug("Invalid transaction state for prepare: " + this);

                return;
            }
        }

        try {
            cctx.tm().prepareTx(this, null);

            if (pessimistic() || isSystemInvalidate())
                state(PREPARED);
        }
        catch (IgniteCheckedException e) {
            setRollbackOnly();

            throw e;
        }
    }

    /**
     * @throws IgniteCheckedException If commit failed.
     */
    @SuppressWarnings({"CatchGenericClass"})
    private void commitIfLocked() throws IgniteCheckedException {
        if (state() == COMMITTING) {
            for (IgniteTxEntry txEntry : writeEntries()) {
                assert txEntry != null : "Missing transaction entry for tx: " + this;

                while (true) {
                    GridCacheEntryEx entry = txEntry.cached();

                    assert entry != null : "Missing cached entry for transaction entry: " + txEntry;

                    try {
                        GridCacheVersion ver = txEntry.explicitVersion() != null ? txEntry.explicitVersion() : xidVer;

                        // If locks haven't been acquired yet, keep waiting.
                        if (!entry.lockedBy(ver)) {
                            if (log.isDebugEnabled())
                                log.debug("Transaction does not own lock for entry (will wait) [entry=" + entry +
                                    ", tx=" + this + ']');

                            return;
                        }

                        break; // While.
                    }
                    catch (GridCacheEntryRemovedException ignore) {
                        if (log.isDebugEnabled())
                            log.debug("Got removed entry while committing (will retry): " + txEntry);

                        txEntry.cached(txEntry.context().cache().entryEx(txEntry.key(), topologyVersion()));
                    }
                }
            }

            // Only one thread gets to commit.
            if (COMMIT_ALLOWED_UPD.compareAndSet(this, 0, 1)) {
                IgniteCheckedException err = null;

                Map<IgniteTxKey, IgniteTxEntry> writeMap = txState.writeMap();

                GridCacheReturnCompletableWrapper wrapper = null;

                if (!F.isEmpty(writeMap) || mvccSnapshot != null) {
                    GridCacheReturn ret = null;

                    if (!near() && !local() && onePhaseCommit()) {
                        if (needReturnValue()) {
                            ret = new GridCacheReturn(null, cctx.localNodeId().equals(otherNodeId()), true, null, true);

                            UUID origNodeId = otherNodeId(); // Originating node.

                            cctx.tm().addCommittedTxReturn(this,
                                wrapper = new GridCacheReturnCompletableWrapper(
                                    !cctx.localNodeId().equals(origNodeId) ? origNodeId : null));
                        }
                        else
                            cctx.tm().addCommittedTx(this, this.nearXidVersion(), null);
                    }

                    // Register this transaction as completed prior to write-phase to
                    // ensure proper lock ordering for removed entries.
                    cctx.tm().addCommittedTx(this);

                    AffinityTopologyVersion topVer = topologyVersion();

                    WALPointer ptr = null;

                    cctx.database().checkpointReadLock();

                    try {
                        assert !txState.mvccEnabled() || mvccSnapshot != null : "Mvcc is not initialized: " + this;

                        Collection<IgniteTxEntry> entries = near() || cctx.snapshot().needTxReadLogging() ? allEntries() : writeEntries();

                        // Data entry to write to WAL and associated with it TxEntry.
                        List<T2<DataEntry, IgniteTxEntry>> dataEntries = null;

                        batchStoreCommit(writeMap().values());

                        try {
                            // Node that for near transactions we grab all entries.
                            for (IgniteTxEntry txEntry : entries) {
                                GridCacheContext cacheCtx = txEntry.context();

                                boolean replicate = cacheCtx.isDrEnabled();

                                try {
                                    while (true) {
                                        try {
                                            GridCacheEntryEx cached = txEntry.cached();

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

                                            if (near() && cacheCtx.dr().receiveEnabled()) {
                                                cached.markObsolete(xidVer);

                                                break;
                                            }

                                            GridNearCacheEntry nearCached = null;

                                            if (updateNearCache(cacheCtx, txEntry.key(), topVer))
                                                nearCached = cacheCtx.dht().near().peekExx(txEntry.key());

                                            if (!F.isEmpty(txEntry.entryProcessors()))
                                                txEntry.cached().unswap(false);

                                            IgniteBiTuple<GridCacheOperation, CacheObject> res =
                                                applyTransformClosures(txEntry, false, ret);

                                            GridCacheOperation op = res.get1();
                                            CacheObject val = res.get2();

                                            GridCacheVersion explicitVer = txEntry.conflictVersion();

                                            if (explicitVer == null)
                                                explicitVer = writeVersion();

                                            if (txEntry.ttl() == CU.TTL_ZERO)
                                                op = DELETE;

                                            boolean conflictNeedResolve = cacheCtx.conflictNeedResolve();

                                            GridCacheVersionConflictContext conflictCtx = null;

                                            if (conflictNeedResolve) {
                                                IgniteBiTuple<GridCacheOperation, GridCacheVersionConflictContext>
                                                    drRes = conflictResolve(op, txEntry, val, explicitVer, cached);

                                                assert drRes != null;

                                                conflictCtx = drRes.get2();

                                                if (conflictCtx.isUseOld())
                                                    op = NOOP;
                                                else if (conflictCtx.isUseNew()) {
                                                    txEntry.ttl(conflictCtx.ttl());
                                                    txEntry.conflictExpireTime(conflictCtx.expireTime());
                                                }
                                                else if (conflictCtx.isMerge()) {
                                                    op = drRes.get1();
                                                    val = txEntry.context().toCacheObject(conflictCtx.mergeValue());
                                                    explicitVer = writeVersion();

                                                    txEntry.ttl(conflictCtx.ttl());
                                                    txEntry.conflictExpireTime(conflictCtx.expireTime());
                                                }
                                            }
                                            else
                                                // Nullify explicit version so that innerSet/innerRemove will work as usual.
                                                explicitVer = null;

                                            GridCacheVersion dhtVer = cached.isNear() ? writeVersion() : null;

                                            if (!near() && cacheCtx.group().persistenceEnabled() && cacheCtx.group().walEnabled() &&
                                                op != NOOP && op != RELOAD && (op != READ || cctx.snapshot().needTxReadLogging())) {
                                                if (dataEntries == null)
                                                    dataEntries = new ArrayList<>(entries.size());

                                                dataEntries.add(
                                                        new T2<>(
                                                                new DataEntry(
                                                                        cacheCtx.cacheId(),
                                                                        txEntry.key(),
                                                                        val,
                                                                        op,
                                                                        nearXidVersion(),
                                                                        writeVersion(),
                                                                        0,
                                                                        txEntry.key().partition(),
                                                                        txEntry.updateCounter()
                                                                ),
                                                                txEntry
                                                        )
                                                );
                                            }

                                            if (op == CREATE || op == UPDATE) {
                                                // Invalidate only for near nodes (backups cannot be invalidated).
                                                if (isSystemInvalidate() || (isInvalidate() && cacheCtx.isNear()))
                                                    cached.innerRemove(this,
                                                        eventNodeId(),
                                                        nodeId,
                                                        false,
                                                        true,
                                                        true,
                                                        txEntry.keepBinary(),
                                                        txEntry.hasOldValue(),
                                                        txEntry.oldValue(),
                                                        topVer,
                                                        null,
                                                        replicate ? DR_BACKUP : DR_NONE,
                                                        near() ? null : explicitVer,
                                                        CU.subjectId(this, cctx),
                                                        resolveTaskName(),
                                                        dhtVer,
                                                        txEntry.updateCounter(),
                                                        mvccSnapshot());
                                                else {
                                                    assert val != null : txEntry;

                                                    GridCacheUpdateTxResult updRes = cached.innerSet(this,
                                                        eventNodeId(),
                                                        nodeId,
                                                        val,
                                                        false,
                                                        false,
                                                        txEntry.ttl(),
                                                        true,
                                                        true,
                                                        txEntry.keepBinary(),
                                                        txEntry.hasOldValue(),
                                                        txEntry.oldValue(),
                                                        topVer,
                                                        null,
                                                        replicate ? DR_BACKUP : DR_NONE,
                                                        txEntry.conflictExpireTime(),
                                                        near() ? null : explicitVer,
                                                        CU.subjectId(this, cctx),
                                                        resolveTaskName(),
                                                        dhtVer,
                                                        txEntry.updateCounter(),
                                                        mvccSnapshot());

                                                    txEntry.updateCounter(updRes.updateCounter());

                                                    if (updRes.loggedPointer() != null)
                                                        ptr = updRes.loggedPointer();

                                                    // Keep near entry up to date.
                                                    if (nearCached != null) {
                                                        CacheObject val0 = cached.valueBytes();

                                                        nearCached.updateOrEvict(xidVer,
                                                            val0,
                                                            cached.expireTime(),
                                                            cached.ttl(),
                                                            nodeId,
                                                            topVer);
                                                    }
                                                }
                                            }
                                            else if (op == DELETE) {
                                                GridCacheUpdateTxResult updRes = cached.innerRemove(this,
                                                    eventNodeId(),
                                                    nodeId,
                                                    false,
                                                    true,
                                                    true,
                                                    txEntry.keepBinary(),
                                                    txEntry.hasOldValue(),
                                                    txEntry.oldValue(),
                                                    topVer,
                                                    null,
                                                    replicate ? DR_BACKUP : DR_NONE,
                                                    near() ? null : explicitVer,
                                                    CU.subjectId(this, cctx),
                                                    resolveTaskName(),
                                                    dhtVer,
                                                    txEntry.updateCounter(),
                                                    mvccSnapshot());

                                                txEntry.updateCounter(updRes.updateCounter());

                                                if (updRes.loggedPointer() != null)
                                                    ptr = updRes.loggedPointer();

                                                // Keep near entry up to date.
                                                if (nearCached != null)
                                                    nearCached.updateOrEvict(xidVer, null, 0, 0, nodeId, topVer);
                                            }
                                            else if (op == RELOAD) {
                                                CacheObject reloaded = cached.innerReload();

                                                if (nearCached != null) {
                                                    nearCached.innerReload();

                                                    nearCached.updateOrEvict(cached.version(),
                                                        reloaded,
                                                        cached.expireTime(),
                                                        cached.ttl(),
                                                        nodeId,
                                                        topVer);
                                                }
                                            }
                                            else if (op == READ) {
                                                assert near();

                                                if (log.isDebugEnabled())
                                                    log.debug("Ignoring READ entry when committing: " + txEntry);
                                            }
                                            // No-op.
                                            else {
                                                if (conflictCtx == null || !conflictCtx.isUseOld()) {
                                                    if (txEntry.ttl() != CU.TTL_NOT_CHANGED)
                                                        cached.updateTtl(null, txEntry.ttl());

                                                    if (nearCached != null) {
                                                        CacheObject val0 = cached.valueBytes();

                                                        nearCached.updateOrEvict(xidVer,
                                                            val0,
                                                            cached.expireTime(),
                                                            cached.ttl(),
                                                            nodeId,
                                                            topVer);
                                                    }
                                                }
                                            }

                                            // Assert after setting values as we want to make sure
                                            // that if we replaced removed entries.
                                            assert
                                                txEntry.op() == READ || onePhaseCommit() ||
                                                    // If candidate is not there, then lock was explicit
                                                    // and we simply allow the commit to proceed.
                                                    !cached.hasLockCandidateUnsafe(xidVer) || cached.lockedByUnsafe(xidVer) :
                                                "Transaction does not own lock for commit [entry=" + cached +
                                                    ", tx=" + this + ']';

                                            // Break out of while loop.
                                            break;
                                        }
                                        catch (GridCacheEntryRemovedException ignored) {
                                            if (log.isDebugEnabled())
                                                log.debug("Attempting to commit a removed entry (will retry): " + txEntry);

                                            // Renew cached entry.
                                            txEntry.cached(cacheCtx.cache().entryEx(txEntry.key(), topologyVersion()));
                                        }
                                    }
                                }
                                catch (Throwable ex) {
                                    boolean isNodeStopping = X.hasCause(ex, NodeStoppingException.class);
                                    boolean hasInvalidEnvironmentIssue = X.hasCause(ex, InvalidEnvironmentException.class);

                                    // In case of error, we still make the best effort to commit,
                                    // as there is no way to rollback at this point.
                                    err = new IgniteTxHeuristicCheckedException("Commit produced a runtime exception " +
                                        "(all transaction entries will be invalidated): " + CU.txString(this), ex);

                                    if (isNodeStopping) {
                                        U.warn(log, "Failed to commit transaction, node is stopping [tx=" + this +
                                            ", err=" + ex + ']');
                                    }
                                    else if (hasInvalidEnvironmentIssue) {
                                        U.warn(log, "Failed to commit transaction, node is in invalid state and will be stopped [tx=" + this +
                                            ", err=" + ex + ']');
                                    }
                                    else
                                        U.error(log, "Commit failed.", err);

                                    state(UNKNOWN);

                                    if (hasInvalidEnvironmentIssue)
                                        cctx.kernalContext().failure().process(new FailureContext(FailureType.CRITICAL_ERROR, ex));
                                    else if (!isNodeStopping) { // Skip fair uncommit in case of node stopping or invalidation.
                                        try {
                                            // Courtesy to minimize damage.
                                            uncommit();
                                        }
                                        catch (Throwable ex1) {
                                            U.error(log, "Failed to uncommit transaction: " + this, ex1);

                                            if (ex1 instanceof Error)
                                                throw ex1;
                                        }
                                    }

                                    if (ex instanceof Error)
                                        throw (Error) ex;

                                    throw err;
                                }
                            }

                            // Apply cache size deltas.
                            applyTxSizes();

                            TxCounters txCntrs = txCounters(false);

                            // Apply update counters.
                            if (txCntrs != null)
                                cctx.tm().txHandler().applyPartitionsUpdatesCounters(txCntrs.updateCounters());

                            cctx.mvccCaching().onTxFinished(this, true);

                            if (!near() && !F.isEmpty(dataEntries) && cctx.wal() != null) {
                                // Set new update counters for data entries received from persisted tx entries.
                                List<DataEntry> entriesWithCounters = dataEntries.stream()
                                    .map(tuple -> tuple.get1().partitionCounter(tuple.get2().updateCounter()))
                                    .collect(Collectors.toList());

                                    cctx.wal().log(new DataRecord(entriesWithCounters));
                            }

                            if (ptr != null && !cctx.tm().logTxRecords())
                                cctx.wal().flush(ptr, false);
                        }
                        catch (StorageException e) {
                            err = e;

                            throw new IgniteCheckedException("Failed to log transaction record " +
                                "(transaction will be rolled back): " + this, e);
                        }
                    }
                    finally {
                        cctx.database().checkpointReadUnlock();

                        if (wrapper != null)
                            wrapper.initialize(ret);
                    }
                }

                cctx.tm().commitTx(this);

                cctx.tm().mvccFinish(this, true);

                state(COMMITTED);
            }
        }
    }

    /** {@inheritDoc} */
    @Override public final void commitRemoteTx() throws IgniteCheckedException {
        if (optimistic())
            state(PREPARED);

        if (!state(COMMITTING)) {
            TransactionState state = state();

            // If other thread is doing commit, then no-op.
            if (state == COMMITTING || state == COMMITTED)
                return;

            if (log.isDebugEnabled())
                log.debug("Failed to set COMMITTING transaction state (will rollback): " + this);

            setRollbackOnly();

            if (!isSystemInvalidate())
                throw new IgniteCheckedException("Invalid transaction state for commit [state=" + state + ", tx=" + this + ']');

            rollbackRemoteTx();
        }

        commitIfLocked();
    }

    /**
     * Forces commit for this tx.
     *
     * @throws IgniteCheckedException If commit failed.
     */
    public void forceCommit() throws IgniteCheckedException {
        commitIfLocked();
    }

    /** {@inheritDoc} */
    @Override public IgniteInternalFuture<IgniteInternalTx> commitAsync() {
        try {
            commitRemoteTx();

            return new GridFinishedFuture<IgniteInternalTx>(this);
        }
        catch (IgniteCheckedException e) {
            return new GridFinishedFuture<>(e);
        }
    }

    /** {@inheritDoc} */
    @Override public final IgniteInternalFuture<?> salvageTx() {
        try {
            systemInvalidate(true);

            prepareRemoteTx();

            if (state() == PREPARING) {
                if (log.isDebugEnabled())
                    log.debug("Ignoring transaction in PREPARING state as it is currently handled " +
                        "by another thread: " + this);

                return null;
            }

            doneRemote(xidVersion(),
                Collections.<GridCacheVersion>emptyList(),
                Collections.<GridCacheVersion>emptyList(),
                Collections.<GridCacheVersion>emptyList());

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

        return null;
    }

    /** {@inheritDoc} */
    @Override public final void rollbackRemoteTx() {
        try {
            // Note that we don't evict near entries here -
            // they will be deleted by their corresponding transactions.
            if (state(ROLLING_BACK) || state() == UNKNOWN) {
                cctx.tm().rollbackTx(this, false, skipCompletedVers);

                TxCounters counters = txCounters(false);

                if (counters != null)
                    cctx.tm().txHandler().applyPartitionsUpdatesCounters(counters.updateCounters());

                state(ROLLED_BACK);

                cctx.mvccCaching().onTxFinished(this, false);

                cctx.tm().mvccFinish(this, false);
            }
        }
        catch (IgniteCheckedException | RuntimeException | Error e) {
            state(UNKNOWN);

            if (e instanceof IgniteCheckedException)
                throw new IgniteException(e);
            else if (e instanceof RuntimeException)
                throw (RuntimeException) e;
            else
                throw (Error) e;
        }
    }

    /** {@inheritDoc} */
    @Override public IgniteInternalFuture<IgniteInternalTx> rollbackAsync() {
        rollbackRemoteTx();

        return new GridFinishedFuture<IgniteInternalTx>(this);
    }

    /** {@inheritDoc} */
    @Override public Collection<GridCacheVersion> alternateVersions() {
        return explicitVers == null ? Collections.<GridCacheVersion>emptyList() : explicitVers;
    }

    /** {@inheritDoc} */
    @Override public void commitError(Throwable e) {
        // No-op.
    }

    /**
     * @return {@code True} if tx should skip adding itself to completed version map on finish.
     */
    public boolean skipCompletedVersions() {
        return skipCompletedVers;
    }

    /**
     * @param skipCompletedVers {@code True} if tx should skip adding itself to completed version map on finish.
     */
    public void skipCompletedVersions(boolean skipCompletedVers) {
        this.skipCompletedVers = skipCompletedVers;
    }

    /**
     * Adds explicit version if there is one.
     *
     * @param e Transaction entry.
     */
    protected void addExplicit(IgniteTxEntry e) {
        if (e.explicitVersion() != null) {
            if (explicitVers == null)
                explicitVers = new LinkedList<>();

            if (!explicitVers.contains(e.explicitVersion())) {
                explicitVers.add(e.explicitVersion());

                if (log.isDebugEnabled())
                    log.debug("Added explicit version to transaction [explicitVer=" + e.explicitVersion() +
                        ", tx=" + this + ']');

                // Register alternate version with TM.
                cctx.tm().addAlternateVersion(e.explicitVersion(), this);
            }
        }
    }

    /** {@inheritDoc} */
    @Override public String toString() {
        return GridToStringBuilder.toString(GridDistributedTxRemoteAdapter.class, this, "super", super.toString());
    }
}