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

import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.cluster.ClusterTopologyException;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.cluster.ClusterTopologyServerNotFoundException;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
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.GridCacheMvccCandidate;
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import org.apache.ignite.internal.processors.cache.distributed.GridDistributedTxMapping;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxMapping;
import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxEntry;
import org.apache.ignite.internal.processors.tracing.MTC;
import org.apache.ignite.internal.transactions.IgniteTxOptimisticCheckedException;
import org.apache.ignite.internal.transactions.IgniteTxRollbackCheckedException;
import org.apache.ignite.internal.util.future.GridCompoundFuture;
import org.apache.ignite.internal.util.future.GridFinishedFuture;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
import org.apache.ignite.internal.util.typedef.C1;
import org.apache.ignite.internal.util.typedef.CI1;
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.S;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteBiInClosure;
import org.apache.ignite.lang.IgniteReducer;
import org.jetbrains.annotations.Nullable;
import static org.apache.ignite.internal.processors.cache.GridCacheOperation.TRANSFORM;
import static org.apache.ignite.internal.processors.tracing.MTC.TraceSurroundings;
import static org.apache.ignite.transactions.TransactionState.PREPARED;
import static org.apache.ignite.transactions.TransactionState.PREPARING;

/**
 *
 */
public class GridNearOptimisticSerializableTxPrepareFuture extends GridNearOptimisticTxPrepareFutureAdapter {
    /** */
    @GridToStringExclude
    private ClientRemapFuture remapFut;

    /** */
    private int miniId;

    /**
     * @param cctx Context.
     * @param tx Transaction.
     */
    public GridNearOptimisticSerializableTxPrepareFuture(GridCacheSharedContext cctx,
        GridNearTxLocal tx) {
        super(cctx, tx);

        assert tx.optimistic() && tx.serializable() : tx;
    }

    /** {@inheritDoc} */
    @Override protected boolean ignoreFailure(Throwable err) {
        return IgniteCheckedException.class.isAssignableFrom(err.getClass());
    }

    /** {@inheritDoc} */
    @Override public boolean onOwnerChanged(GridCacheEntryEx entry, GridCacheMvccCandidate owner) {
        if (log.isDebugEnabled())
            log.debug("Transaction future received owner changed callback: " + entry);

        if (entry.context().isNear() && owner != null) {
            IgniteTxEntry txEntry = tx.entry(entry.txKey());

            if (txEntry != null) {
                if (keyLockFut != null)
                    keyLockFut.onKeyLocked(entry.txKey());

                return true;
            }
        }

        return false;
    }

    /** {@inheritDoc} */
    @Override public boolean onNodeLeft(UUID nodeId) {
        boolean found = false;

        for (IgniteInternalFuture<?> fut : futures()) {
            if (isMini(fut)) {
                MiniFuture f = (MiniFuture)fut;

                if (f.primary().id().equals(nodeId)) {
                    ClusterTopologyCheckedException e = new ClusterTopologyCheckedException("Remote node left grid: " +
                        nodeId);

                    e.retryReadyFuture(cctx.nextAffinityReadyFuture(tx.topologyVersion()));

                    f.onNodeLeft(e);

                    found = true;
                }
            }
        }

        return found;
    }

    /**
     * @param m Failed mapping.
     * @param e Error.
     */
    private void onError(@Nullable GridDistributedTxMapping m, Throwable e) {
        try (TraceSurroundings ignored = MTC.support(span)) {
            if (X.hasCause(e, ClusterTopologyCheckedException.class) || X.hasCause(e, ClusterTopologyException.class)) {
                if (tx.onePhaseCommit()) {
                    tx.markForBackupCheck();

                    onComplete();

                    return;
                }
            }

            if (e instanceof IgniteTxOptimisticCheckedException) {
                if (m != null)
                    tx.removeMapping(m.primary().id());
            }

            prepareError(e);
        }
    }

    /**
     * @param e Error.
     */
    private void prepareError(Throwable e) {
        ERR_UPD.compareAndSet(this, null, e);

        if (keyLockFut != null)
            keyLockFut.onDone(e);
    }

    /** {@inheritDoc} */
    @Override public void onResult(UUID nodeId, GridNearTxPrepareResponse res) {
        if (!isDone()) {
            MiniFuture mini = miniFuture(res.miniId());

            if (mini != null)
                mini.onResult(res, true);
        }
    }

    /** {@inheritDoc} */
    @Override public boolean onDone(IgniteInternalTx t, Throwable err) {
        try (TraceSurroundings ignored = MTC.support(span)) {
            if (isDone())
                return false;

            if (err != null) {
                ERR_UPD.compareAndSet(this, null, err);

                if (keyLockFut != null)
                    keyLockFut.onDone(err);
            }

            return onComplete();
        }
    }

    /**
     * Finds pending mini future by the given mini ID.
     *
     * @param miniId Mini ID to find.
     * @return Mini future.
     */
    private MiniFuture miniFuture(int miniId) {
        // We iterate directly over the futs collection here to avoid copy.
        compoundsReadLock();

        try {
            int size = futuresCountNoLock();

            // Avoid iterator creation.
            for (int i = 0; i < size; i++) {
                IgniteInternalFuture fut = future(i);

                if (!isMini(fut))
                    continue;

                MiniFuture mini = (MiniFuture)fut;

                if (mini.futureId() == miniId) {
                    if (!mini.isDone())
                        return mini;
                    else
                        return null;
                }
            }
        }
        finally {
            compoundsReadUnlock();
        }

        return null;
    }

    /**
     * @param f Future.
     * @return {@code True} if mini-future.
     */
    private boolean isMini(IgniteInternalFuture<?> f) {
        return f.getClass().equals(MiniFuture.class);
    }

    /**
     * Completeness callback.
     *
     * @return {@code True} if future was finished by this call.
     */
    private boolean onComplete() {
        Throwable err0 = err;

        if ((!tx.onePhaseCommit() || tx.mappings().get(cctx.localNodeId()) == null) &&
            (err0 == null || tx.needCheckBackup()))
            tx.state(PREPARED);

        if (super.onDone(tx, err0)) {
            if (err0 != null)
                tx.setRollbackOnly();

            // Don't forget to clean up.
            cctx.mvcc().removeVersionedFuture(this);

            return true;
        }

        return false;
    }

    /**
     * Initializes future.
     *
     * @param remap Remap flag.
     */
    @Override protected void prepare0(boolean remap, boolean topLocked) {
        boolean txStateCheck = remap ? tx.state() == PREPARING : tx.state(PREPARING);

        if (!txStateCheck) {
            if (tx.isRollbackOnly() || tx.setRollbackOnly()) {
                if (tx.timedOut())
                    onDone(null, tx.timeoutException());
                else
                    onDone(null, tx.rollbackException());
            }
            else
                onDone(null, new IgniteCheckedException("Invalid transaction state for " +
                    "prepare [state=" + tx.state() + ", tx=" + this + ']'));

            return;
        }

        boolean set = cctx.tm().setTxTopologyHint(tx.topologyVersionSnapshot());

        try {
            prepare(tx.readEntries(), tx.writeEntries(), remap, topLocked);

            markInitialized();
        }
        finally {
            if (set)
                cctx.tm().setTxTopologyHint(null);
        }
    }

    /**
     * @param reads Read entries.
     * @param writes Write entries.
     * @param remap Remap flag.
     * @param topLocked Topology locked flag.
     */
    @SuppressWarnings("unchecked")
    private void prepare(
        Iterable<IgniteTxEntry> reads,
        Iterable<IgniteTxEntry> writes,
        boolean remap,
        boolean topLocked
    ) {
        AffinityTopologyVersion topVer = tx.topologyVersion();

        assert topVer.topologyVersion() > 0;

        GridDhtTxMapping txMapping = new GridDhtTxMapping();

        Map<UUID, GridDistributedTxMapping> mappings = new HashMap<>();

        boolean hasNearCache = false;

        for (IgniteTxEntry write : writes) {
            map(write, topVer, mappings, txMapping, remap, topLocked);

            if (write.context().isNear())
                hasNearCache = true;
        }

        for (IgniteTxEntry read : reads) {
            map(read, topVer, mappings, txMapping, remap, topLocked);

            if (read.context().isNear())
                hasNearCache = true;
        }

        if (keyLockFut != null)
            keyLockFut.onAllKeysAdded();

        if (isDone()) {
            if (log.isDebugEnabled())
                log.debug("Abandoning (re)map because future is done: " + this);

            return;
        }

        tx.addEntryMapping(mappings.values());

        cctx.mvcc().recheckPendingLocks();

        tx.transactionNodes(txMapping.transactionNodes());

        if (!hasNearCache)
            checkOnePhase(txMapping);

        MiniFuture locNearEntriesFut = null;

        // Create futures in advance to have all futures when process {@link GridNearTxPrepareResponse#clientRemapVersion}.
        for (GridDistributedTxMapping m : mappings.values()) {
            assert !m.empty();

            MiniFuture fut = new MiniFuture(this, m, ++miniId);

            add((IgniteInternalFuture)fut);

            if (m.primary().isLocal() && m.hasNearCacheEntries() && m.hasColocatedCacheEntries()) {
                assert locNearEntriesFut == null;

                locNearEntriesFut = fut;

                add((IgniteInternalFuture)new MiniFuture(this, m, ++miniId));
            }
        }

        Collection<IgniteInternalFuture<?>> futs = (Collection)futures();

        Iterator<IgniteInternalFuture<?>> it = futs.iterator();

        while (it.hasNext()) {
            IgniteInternalFuture<?> fut0 = it.next();

            if (skipFuture(remap, fut0))
                continue;

            MiniFuture fut = (MiniFuture)fut0;

            IgniteCheckedException err = prepare(fut, txMapping.transactionNodes(), locNearEntriesFut);

            if (err != null) {
                while (it.hasNext()) {
                    fut0 = it.next();

                    if (skipFuture(remap, fut0))
                        continue;

                    fut = (MiniFuture)fut0;

                    tx.removeMapping(fut.mapping().primary().id());

                    fut.onResult(new IgniteCheckedException("Failed to prepare transaction.", err));
                }

                break;
            }
        }

        markInitialized();
    }

    /**
     * @param remap Remap flag.
     * @param fut Future.
     * @return {@code True} if skip future during remap.
     */
    private boolean skipFuture(boolean remap, IgniteInternalFuture<?> fut) {
        return !(isMini(fut)) || (remap && (((MiniFuture)fut).rcvRes == 1));
    }

    /**
     * @param fut Mini future.
     * @param txNodes Tx nodes.
     * @param locNearEntriesFut Local future for near cache entries prepare.
     * @return Prepare error if any.
     */
    @Nullable private IgniteCheckedException prepare(final MiniFuture fut,
        Map<UUID, Collection<UUID>> txNodes,
        @Nullable MiniFuture locNearEntriesFut) {
        GridDistributedTxMapping m = fut.mapping();

        final ClusterNode primary = m.primary();

        long timeout = tx.remainingTime();

        if (timeout == -1) {
            IgniteCheckedException err = tx.timeoutException();

            fut.onResult(err);

            return err;
        }

        // Must lock near entries separately.
        if (m.hasNearCacheEntries()) {
            try {
                cctx.tm().prepareTx(tx, m.nearCacheEntries());
            }
            catch (IgniteCheckedException e) {
                fut.onResult(e);

                return e;
            }
        }

        if (primary.isLocal()) {
            if (locNearEntriesFut != null) {
                boolean nearEntries = fut == locNearEntriesFut;

                GridNearTxPrepareRequest req = createRequest(txNodes,
                    fut,
                    timeout,
                    nearEntries ? m.nearEntriesReads() : m.colocatedEntriesReads(),
                    nearEntries ? m.nearEntriesWrites() : m.colocatedEntriesWrites());

                prepareLocal(req, fut, nearEntries);
            }
            else {
                GridNearTxPrepareRequest req = createRequest(txNodes,
                    fut,
                    timeout,
                    m.reads(),
                    m.writes());

                prepareLocal(req, fut, m.hasNearCacheEntries());
            }
        }
        else {
            try {
                GridNearTxPrepareRequest req = createRequest(txNodes,
                    fut,
                    timeout,
                    m.reads(),
                    m.writes());

                cctx.io().send(primary, req, tx.ioPolicy());
            }
            catch (ClusterTopologyCheckedException e) {
                e.retryReadyFuture(cctx.nextAffinityReadyFuture(tx.topologyVersion()));

                fut.onNodeLeft(e);

                return e;
            }
            catch (IgniteCheckedException e) {
                fut.onResult(e);

                return e;
            }
        }

        return null;
    }

    /**
     * @param txNodes Tx nodes.
     * @param fut Future.
     * @param timeout Timeout.
     * @param reads Read entries.
     * @param writes Write entries.
     * @return Request.
     */
    private GridNearTxPrepareRequest createRequest(
        Map<UUID, Collection<UUID>> txNodes,
        MiniFuture fut,
        long timeout,
        Collection<IgniteTxEntry> reads,
        Collection<IgniteTxEntry> writes) {
        GridDistributedTxMapping m = fut.mapping();

        GridNearTxPrepareRequest req = new GridNearTxPrepareRequest(
            futId,
            tx.topologyVersion(),
            tx,
            timeout,
            reads,
            writes,
            m.hasNearCacheEntries(),
            txNodes,
            m.last(),
            tx.onePhaseCommit(),
            tx.needReturnValue() && tx.implicit(),
            tx.implicitSingle(),
            m.explicitLock(),
            tx.taskNameHash(),
            m.clientFirst(),
            txNodes.size() == 1,
            tx.activeCachesDeploymentEnabled(),
            tx.txState().recovery());

        for (IgniteTxEntry txEntry : writes) {
            if (txEntry.op() == TRANSFORM)
                req.addDhtVersion(txEntry.txKey(), null);
        }

        req.miniId(fut.futureId());

        return req;
    }

    /**
     * @param req Request.
     * @param fut Future.
     * @param nearEntries {@code True} if prepare near cache entries.
     */
    private void prepareLocal(GridNearTxPrepareRequest req,
        final MiniFuture fut,
        final boolean nearEntries) {
        IgniteInternalFuture<GridNearTxPrepareResponse> prepFut = nearEntries ?
            cctx.tm().txHandler().prepareNearTxLocal(tx, req) :
            cctx.tm().txHandler().prepareColocatedTx(tx, req);

        prepFut.listen(new CI1<IgniteInternalFuture<GridNearTxPrepareResponse>>() {
            @Override public void apply(IgniteInternalFuture<GridNearTxPrepareResponse> prepFut) {
                try {
                    fut.onResult(prepFut.get(), nearEntries);
                }
                catch (IgniteCheckedException e) {
                    fut.onResult(e);
                }
            }
        });
    }

    /**
     * @param entry Transaction entry.
     * @param topVer Topology version.
     * @param curMapping Current mapping.
     * @param txMapping Mapping.
     * @param remap Remap flag.
     * @param topLocked Topology locked flag.
     */
    private void map(
        IgniteTxEntry entry,
        AffinityTopologyVersion topVer,
        Map<UUID, GridDistributedTxMapping> curMapping,
        GridDhtTxMapping txMapping,
        boolean remap,
        boolean topLocked
    ) {
        GridCacheContext cacheCtx = entry.context();

        List<ClusterNode> nodes = cacheCtx.topology().nodes(cacheCtx.affinity().partition(entry.key()), topVer);

        if (F.isEmpty(nodes)) {
            onDone(new ClusterTopologyServerNotFoundException("Failed to map keys to nodes " +
                "(partition is not mapped to any node) [key=" + entry.key() +
                ", partition=" + cacheCtx.affinity().partition(entry.key()) + ", topVer=" + topVer + ']'));

            return;
        }

        txMapping.addMapping(nodes);

        ClusterNode primary = F.first(nodes);

        assert primary != null;

        if (log.isDebugEnabled()) {
            log.debug("Mapped key to primary node [key=" + entry.key() +
                ", part=" + cacheCtx.affinity().partition(entry.key()) +
                ", primary=" + U.toShortString(primary) + ", topVer=" + topVer + ']');
        }

        // Must re-initialize cached entry while holding topology lock.
        if (cacheCtx.isNear())
            entry.cached(cacheCtx.nearTx().entryExx(entry.key(), topVer));
        else
            entry.cached(cacheCtx.colocated().entryExx(entry.key(), topVer, true));

        if (!remap && cacheCtx.isNear()) {
            if (entry.explicitVersion() == null) {
                if (keyLockFut == null) {
                    keyLockFut = new KeyLockFuture();

                    add((IgniteInternalFuture)keyLockFut);
                }

                keyLockFut.addLockKey(entry.txKey());
            }
        }

        GridDistributedTxMapping cur = curMapping.get(primary.id());

        if (cur == null) {
            cur = new GridDistributedTxMapping(primary);

            curMapping.put(primary.id(), cur);

            cur.clientFirst(!topLocked && cctx.kernalContext().clientNode());

            cur.last(true);
        }

        if (primary.isLocal()) {
            if (cacheCtx.isNear())
                tx.nearLocallyMapped(true);
            else if (cacheCtx.isColocated())
                tx.colocatedLocallyMapped(true);
        }

        cur.add(entry);

        if (entry.explicitVersion() != null) {
            tx.markExplicit(primary.id());

            cur.markExplicitLock();
        }

        entry.nodeId(primary.id());

        if (cacheCtx.isNear()) {
            while (true) {
                try {
                    GridNearCacheEntry cached = (GridNearCacheEntry)entry.cached();

                    cached.dhtNodeId(tx.xidVersion(), primary.id());

                    break;
                }
                catch (GridCacheEntryRemovedException ignore) {
                    entry.cached(cacheCtx.near().entryEx(entry.key(), topVer));
                }
            }
        }
    }

    /** {@inheritDoc} */
    @Override public String toString() {
        Collection<String> futs = F.viewReadOnly(futures(),
            new C1<IgniteInternalFuture<?>, String>() {
                @Override public String apply(IgniteInternalFuture<?> f) {
                    if (isMini(f)) {
                        return "[node=" + ((MiniFuture)f).primary().id() +
                            ", loc=" + ((MiniFuture)f).primary().isLocal() +
                            ", done=" + f.isDone() +
                            ", err=" + f.error() + "]";
                    }
                    else
                        return f.toString();
                }
            });

        return S.toString(GridNearOptimisticSerializableTxPrepareFuture.class, this,
            "innerFuts", futs,
            "remap", remapFut != null,
            "tx", tx,
            "super", super.toString());
    }

    /**
     * Client remap future.
     */
    private static class ClientRemapFuture extends GridCompoundFuture<GridNearTxPrepareResponse, Boolean> {
        /** */
        private static final long serialVersionUID = 0L;

        /**
         * Constructor.
         */
        ClientRemapFuture() {
            super(new ClientRemapFutureReducer());
        }
    }

    /**
     * Client remap future reducer.
     */
    private static class ClientRemapFutureReducer implements IgniteReducer<GridNearTxPrepareResponse, Boolean> {
        /** */
        private static final long serialVersionUID = 0L;

        /** Remap flag. */
        private boolean remap = true;

        /** {@inheritDoc} */
        @Override public boolean collect(@Nullable GridNearTxPrepareResponse res) {
            assert res != null;

            if (res.clientRemapVersion() == null)
                remap = false;

            return true;
        }

        /** {@inheritDoc} */
        @Override public Boolean reduce() {
            return remap;
        }
    }

    /**
     *
     */
    private static class MiniFuture extends GridFutureAdapter<GridNearTxPrepareResponse> {
        /** Receive result flag updater. */
        private static final AtomicIntegerFieldUpdater<MiniFuture> RCV_RES_UPD =
            AtomicIntegerFieldUpdater.newUpdater(MiniFuture.class, "rcvRes");

        /** */
        private final int futId;

        /** Parent future. */
        private final GridNearOptimisticSerializableTxPrepareFuture parent;

        /** Keys. */
        @GridToStringInclude
        private GridDistributedTxMapping m;

        /** Flag to signal some result being processed. */
        private volatile int rcvRes;

        /**
         * @param parent Parent future.
         * @param m Mapping.
         * @param futId Mini future ID.
         */
        MiniFuture(GridNearOptimisticSerializableTxPrepareFuture parent, GridDistributedTxMapping m, int futId) {
            this.parent = parent;
            this.m = m;
            this.futId = futId;
        }

        /**
         * @return Future ID.
         */
        int futureId() {
            return futId;
        }

        /**
         * @return Primary node.
         */
        public ClusterNode primary() {
            return m.primary();
        }

        /**
         * @return Keys.
         */
        public GridDistributedTxMapping mapping() {
            return m;
        }

        /**
         * @param e Error.
         */
        void onResult(Throwable e) {
            if (RCV_RES_UPD.compareAndSet(this, 0, 1)) {
                parent.onError(m, e);

                if (log.isDebugEnabled())
                    log.debug("Failed to get future result [fut=" + this + ", err=" + e + ']');

                // Fail.
                onDone(e);
            }
            else
                U.warn(log, "Received error after another result has been processed [fut=" +
                    parent + ", mini=" + this + ']', e);
        }

        /**
         * @param e Node failure.
         */
        void onNodeLeft(ClusterTopologyCheckedException e) {
            if (isDone())
                return;

            if (RCV_RES_UPD.compareAndSet(this, 0, 1)) {
                if (log.isDebugEnabled())
                    log.debug("Remote node left grid while sending or waiting for reply (will not retry): " + this);

                parent.onError(null, e);

                onDone(e);
            }
        }

        /**
         * @param res Result callback.
         * @param updateMapping Update mapping flag.
         */
        @SuppressWarnings({"unchecked"})
        void onResult(final GridNearTxPrepareResponse res, boolean updateMapping) {
            if (isDone())
                return;

            if (RCV_RES_UPD.compareAndSet(this, 0, 1)) {
                if (res.error() != null) {
                    // Fail the whole compound future.
                    parent.onError(m, res.error());

                    onDone(res.error());
                }
                else {
                    if (res.clientRemapVersion() != null) {
                        assert parent.cctx.kernalContext().clientNode();
                        assert m.clientFirst();

                        parent.tx.removeMapping(m.primary().id());

                        ClientRemapFuture remapFut0 = null;

                        synchronized (parent) {
                            if (parent.remapFut == null) {
                                parent.remapFut = new ClientRemapFuture();

                                remapFut0 = parent.remapFut;
                            }
                        }

                        if (remapFut0 != null) {
                            Collection<IgniteInternalFuture<?>> futs = (Collection)parent.futures();

                            for (IgniteInternalFuture<?> fut : futs) {
                                if (parent.isMini(fut) && fut != this)
                                    remapFut0.add((MiniFuture)fut);
                            }

                            remapFut0.markInitialized();

                            remapFut0.listen(new CI1<IgniteInternalFuture<Boolean>>() {
                                @Override public void apply(IgniteInternalFuture<Boolean> remapFut0) {
                                    try {
                                        IgniteInternalFuture<?> affFut =
                                            parent.cctx.exchange().affinityReadyFuture(res.clientRemapVersion());

                                        if (affFut == null)
                                            affFut = new GridFinishedFuture<Object>();

                                        if (parent.remapFut.get()) {
                                            if (log.isDebugEnabled()) {
                                                log.debug("Will remap client tx [" +
                                                    "fut=" + parent +
                                                    ", topVer=" + res.topologyVersion() + ']');
                                            }

                                            synchronized (parent) {
                                                assert remapFut0 == parent.remapFut;

                                                parent.remapFut = null;
                                            }

                                            parent.cctx.time().waitAsync(
                                                affFut,
                                                parent.tx.remainingTime(),
                                                new IgniteBiInClosure<IgniteCheckedException, Boolean>() {
                                                    @Override public void apply(IgniteCheckedException e, Boolean timedOut) {
                                                        if (parent.errorOrTimeoutOnTopologyVersion(e, timedOut))
                                                            return;

                                                        remap(res);
                                                    }
                                                }
                                            );
                                        }
                                        else {
                                            ClusterTopologyCheckedException err0 = new ClusterTopologyCheckedException(
                                                "Cluster topology changed while client transaction is preparing.");

                                            err0.retryReadyFuture(affFut);

                                            parent.prepareError(err0);

                                            onDone(err0);
                                        }
                                    }
                                    catch (IgniteCheckedException e) {
                                        if (log.isDebugEnabled()) {
                                            log.debug("Prepare failed, will not remap tx: " +
                                                parent);
                                        }

                                        parent.prepareError(e);

                                        onDone(e);
                                    }
                                }
                            });
                        }
                        else
                            onDone(res);
                    }
                    else {
                        parent.onPrepareResponse(m, res, updateMapping);

                        // Finish this mini future (need result only on client node).
                        onDone(parent.cctx.kernalContext().clientNode() ? res : null);
                    }
                }
            }
        }

        /**
         * @param res Response.
         */
        private void remap(final GridNearTxPrepareResponse res) {
            if (parent.tx.isRollbackOnly()) {
                onDone(new IgniteTxRollbackCheckedException(
                    "Failed to prepare the transaction, due to the transaction is marked as rolled back " +
                        "[tx=" + CU.txString(parent.tx) + ']'));

                return;
            }

            parent.prepareOnTopology(true, new Runnable() {
                @Override public void run() {
                    onDone(res);
                }
            });
        }

        /** {@inheritDoc} */
        @Override public String toString() {
            return S.toString(MiniFuture.class, this, "done", isDone(), "cancelled", isCancelled(), "err", error());
        }
    }
}