/*
 * 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.geode.internal.cache;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.LockSupport;

import org.apache.logging.log4j.Logger;

import org.apache.geode.GemFireException;
import org.apache.geode.InternalGemFireError;
import org.apache.geode.SystemFailure;
import org.apache.geode.annotations.Immutable;
import org.apache.geode.annotations.VisibleForTesting;
import org.apache.geode.annotations.internal.MakeNotStatic;
import org.apache.geode.annotations.internal.MutableForTesting;
import org.apache.geode.cache.CacheTransactionManager;
import org.apache.geode.cache.CommitConflictException;
import org.apache.geode.cache.TransactionDataRebalancedException;
import org.apache.geode.cache.TransactionId;
import org.apache.geode.cache.TransactionInDoubtException;
import org.apache.geode.cache.TransactionListener;
import org.apache.geode.cache.TransactionWriter;
import org.apache.geode.cache.UnsupportedOperationInTransactionException;
import org.apache.geode.distributed.TXManagerCancelledException;
import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.MembershipListener;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.SystemTimer.SystemTimerTask;
import org.apache.geode.internal.cache.entries.AbstractRegionEntry;
import org.apache.geode.internal.cache.tier.sockets.Message;
import org.apache.geode.internal.statistics.StatisticsClock;
import org.apache.geode.internal.util.concurrent.CustomEntryConcurrentHashMap;
import org.apache.geode.internal.util.concurrent.CustomEntryConcurrentHashMap.HashEntry;
import org.apache.geode.internal.util.concurrent.CustomEntryConcurrentHashMap.MapCallback;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.util.internal.GeodeGlossary;

/**
 * The internal implementation of the {@link CacheTransactionManager} interface returned by
 * {@link InternalCache#getCacheTransactionManager}. Internal operations
 *
 * {@code TransactionListener} invocation, Region synchronization, transaction statistics and
 *
 * transaction logging are handled here
 *
 * @since GemFire 4.0
 *
 * @see CacheTransactionManager
 */
public class TXManagerImpl implements CacheTransactionManager, MembershipListener {

  private static final Logger logger = LogService.getLogger();

  // Thread specific context container
  private final ThreadLocal<TXStateProxy> txContext;

  private final ThreadLocal<Boolean> pauseJTA;

  @MakeNotStatic
  private static final AtomicReference<TXManagerImpl> currentInstance = new AtomicReference<>();

  // The unique transaction ID for this Manager
  private final AtomicInteger uniqId;

  private final DistributionManager dm;
  private final InternalCache cache;

  // The DistributionMemberID used to construct TXId's
  private final InternalDistributedMember distributionMgrId;

  private final CachePerfStats cachePerfStats;

  @Immutable
  private static final TransactionListener[] EMPTY_LISTENERS = new TransactionListener[0];

  /**
   * Default transaction id to indicate no transaction
   */
  public static final int NOTX = -1;

  private final List<TransactionListener> txListeners = new ArrayList<>(8);

  public TransactionWriter writer = null;

  private volatile boolean closed = false;

  private final Map<TXId, TXStateProxy> hostedTXStates;

  // Used for testing only.
  private final Set<TXId> scheduledToBeRemovedTx =
      Boolean.getBoolean(GeodeGlossary.GEMFIRE_PREFIX + "trackScheduledToBeRemovedTx")
          ? ConcurrentHashMap.newKeySet() : null;

  /**
   * the number of client initiated transactions to store for client failover
   */
  public static final int FAILOVER_TX_MAP_SIZE =
      Integer.getInteger(GeodeGlossary.GEMFIRE_PREFIX + "transactionFailoverMapSize", 1000);

  /**
   * used to store TXCommitMessages for client initiated transactions, so that when a client
   * failsover, (after the delegate dies) the commit message can be sent to client. //TODO we really
   * need to keep around only one msg for each thread on a client
   */
  private final Map<TXId, TXCommitMessage> failoverMap =
      Collections.synchronizedMap(new LinkedHashMap<TXId, TXCommitMessage>() {
        // TODO: inner class is serializable but outer class is not
        private static final long serialVersionUID = -4156018226167594134L;

        @Override
        protected boolean removeEldestEntry(Entry eldest) {
          if (logger.isDebugEnabled()) {
            logger.debug("TX: removing client initiated transaction from failover map:{} :{}",
                eldest.getKey(), (size() > FAILOVER_TX_MAP_SIZE));
          }
          return size() > FAILOVER_TX_MAP_SIZE;
        }
      });

  /**
   * A flag to allow persistent transactions. public for testing.
   */
  @MutableForTesting
  public static boolean ALLOW_PERSISTENT_TRANSACTIONS =
      Boolean.getBoolean(GeodeGlossary.GEMFIRE_PREFIX + "ALLOW_PERSISTENT_TRANSACTIONS");

  @MutableForTesting
  static int INITIAL_UNIQUE_ID_VALUE = 0;

  /**
   * this keeps track of all the transactions that were initiated locally.
   */
  private final ConcurrentMap<TXId, TXStateProxy> localTxMap = new ConcurrentHashMap<>();

  /**
   * the time in minutes after which any suspended transaction are rolled back. default is 30
   * minutes
   */
  private volatile long suspendedTXTimeout =
      Long.getLong(GeodeGlossary.GEMFIRE_PREFIX + "suspendedTxTimeout", 30);

  /**
   * Thread-specific flag to indicate whether the transactions managed by this
   * CacheTransactionManager for this thread should be distributed
   */
  private final ThreadLocal<Boolean> isTXDistributed;

  /**
   * The number of seconds to keep transaction states for disconnected clients. This allows the
   * client to fail over to another server and still find the transaction state to complete the
   * transaction.
   */
  private int transactionTimeToLive;

  private final StatisticsClock statisticsClock;

  /**
   * Constructor that implements the {@link CacheTransactionManager} interface. Only only one
   * instance per {@link org.apache.geode.cache.Cache}
   */
  public TXManagerImpl(CachePerfStats cachePerfStats, InternalCache cache,
      StatisticsClock statisticsClock) {
    this.cache = cache;
    dm = ((InternalDistributedSystem) cache.getDistributedSystem()).getDistributionManager();
    distributionMgrId = dm.getDistributionManagerId();
    uniqId = new AtomicInteger(INITIAL_UNIQUE_ID_VALUE);
    this.cachePerfStats = cachePerfStats;
    hostedTXStates = new HashMap<>();
    txContext = new ThreadLocal<>();
    pauseJTA = new ThreadLocal<>();
    isTXDistributed = new ThreadLocal<>();
    transactionTimeToLive = Integer
        .getInteger(GeodeGlossary.GEMFIRE_PREFIX + "cacheServer.transactionTimeToLive", 180);
    currentInstance.set(this);
    this.statisticsClock = statisticsClock;
  }

  public static TXManagerImpl getCurrentInstanceForTest() {
    return currentInstance.get();
  }

  public static void setCurrentInstanceForTest(TXManagerImpl instance) {
    currentInstance.set(instance);
  }

  InternalCache getCache() {
    return cache;
  }

  /**
   * Get the TransactionWriter for the cache
   *
   * @return the current TransactionWriter
   * @see TransactionWriter
   */
  @Override
  public TransactionWriter getWriter() {
    return writer;
  }

  @Override
  public void setWriter(TransactionWriter writer) {
    if (cache.isClient()) {
      throw new IllegalStateException(
          "A TransactionWriter cannot be registered on a client");
    }
    this.writer = writer;
  }

  @Override
  public TransactionListener getListener() {
    synchronized (txListeners) {
      if (txListeners.isEmpty()) {
        return null;
      } else if (txListeners.size() == 1) {
        return txListeners.get(0);
      } else {
        throw new IllegalStateException(
            "More than one transaction listener exists.");
      }
    }
  }

  @Override
  public TransactionListener[] getListeners() {
    synchronized (txListeners) {
      int size = txListeners.size();
      if (size == 0) {
        return EMPTY_LISTENERS;
      } else {
        TransactionListener[] result = new TransactionListener[size];
        txListeners.toArray(result);
        return result;
      }
    }
  }

  @Override
  public TransactionListener setListener(TransactionListener newListener) {
    synchronized (txListeners) {
      TransactionListener result = getListener();
      txListeners.clear();
      if (newListener != null) {
        txListeners.add(newListener);
      }
      if (result != null) {
        closeListener(result);
      }
      return result;
    }
  }

  @Override
  public void addListener(TransactionListener aListener) {
    if (aListener == null) {
      throw new IllegalArgumentException(
          "addListener parameter was null");
    }
    synchronized (txListeners) {
      if (!txListeners.contains(aListener)) {
        txListeners.add(aListener);
      }
    }
  }

  @Override
  public void removeListener(TransactionListener aListener) {
    if (aListener == null) {
      throw new IllegalArgumentException(
          "removeListener parameter was null");
    }
    synchronized (txListeners) {
      if (txListeners.remove(aListener)) {
        closeListener(aListener);
      }
    }
  }

  @Override
  public void initListeners(TransactionListener[] newListeners) {
    synchronized (txListeners) {
      if (!txListeners.isEmpty()) {
        for (final TransactionListener txListener : txListeners) {
          closeListener(txListener);
        }
        txListeners.clear();
      }
      if (newListeners != null && newListeners.length > 0) {
        List<TransactionListener> nl = Arrays.asList(newListeners);
        if (nl.contains(null)) {
          throw new IllegalArgumentException(
              "initListeners parameter had a null element");
        }
        txListeners.addAll(nl);
      }
    }
  }

  CachePerfStats getCachePerfStats() {
    return cachePerfStats;
  }

  /**
   * Get a new {@link TXId}.
   * This method builds a new {@link TXId} with uniqueID incremented by one.
   * Special case: If uniqueID has reached Integer.MAX_VALUE, then it is not incremented,
   * but set to one. This is done to prevent memory overflow of uniqueID to Integer.MIN_VALUE.
   */
  private TXId getNewTXId() {
    return new TXId(distributionMgrId,
        uniqId.updateAndGet(i -> i == Integer.MAX_VALUE ? 1 : i + 1));
  }

  /**
   * Build a new {@link TXId}, use it as part of the transaction state and associate with the
   * current thread using a {@link ThreadLocal}.
   */
  @Override
  public void begin() {
    checkClosed();
    {
      TransactionId tid = getTransactionId();
      if (tid != null) {
        throw new java.lang.IllegalStateException(
            String.format("Transaction %s already in progress",
                tid));
      }
    }
    {
      TXStateProxy curProxy = txContext.get();
      if (curProxy == PAUSED) {
        throw new java.lang.IllegalStateException(
            "Current thread has paused its transaction so it can not start a new transaction");
      }
    }
    TXId id = getNewTXId();
    TXStateProxyImpl proxy = null;
    if (isDistributed()) {
      proxy = new DistTXStateProxyImplOnCoordinator(cache, this, id, null, statisticsClock);
    } else {
      proxy = new TXStateProxyImpl(cache, this, id, null, statisticsClock);
    }
    setTXState(proxy);
    if (logger.isDebugEnabled()) {
      logger.debug("begin tx: {}", proxy);
    }
    localTxMap.put(id, proxy);
  }


  /**
   * Build a new {@link TXId}, use it as part of the transaction state and associate with the
   * current thread using a {@link ThreadLocal}. Flag the transaction to be enlisted with a JTA
   * Transaction. Should only be called in a context where we know there is no existing transaction.
   */
  public TXStateProxy beginJTA() {
    checkClosed();
    TXId id = getNewTXId();
    TXStateProxy newState = null;

    if (isDistributed()) {
      newState = new DistTXStateProxyImplOnCoordinator(cache, this, id, true, statisticsClock);
    } else {
      newState = new TXStateProxyImpl(cache, this, id, true, statisticsClock);
    }
    setTXState(newState);
    return newState;
  }

  /*
   * Only applicable for Distributed transaction.
   */
  public void precommit() throws CommitConflictException {
    checkClosed();

    final TXStateProxy tx = getTXState();
    if (tx == null) {
      throw new IllegalStateException(
          "Thread does not have an active transaction");
    }

    tx.checkJTA(
        "Can not commit this transaction because it is enlisted with a JTA transaction, use the JTA manager to perform the commit.");

    tx.precommit();
  }

  /**
   * Complete the transaction associated with the current thread. When this method completes, the
   * thread is no longer associated with a transaction.
   *
   */
  @Override
  public void commit() throws CommitConflictException {
    checkClosed();

    final TXStateProxy tx = getTXState();
    if (tx == null) {
      throw new IllegalStateException(
          "Thread does not have an active transaction");
    }

    tx.checkJTA(
        "Can not commit this transaction because it is enlisted with a JTA transaction, use the JTA manager to perform the commit.");

    final long opStart = statisticsClock.getTime();
    final long lifeTime = opStart - tx.getBeginTime();
    try {
      setTXState(null);
      tx.commit();
    } catch (CommitConflictException ex) {
      saveTXStateForClientFailover(tx, TXCommitMessage.CMT_CONFLICT_MSG); // fixes #43350
      noteCommitFailure(opStart, lifeTime, tx);
      cleanup(tx.getTransactionId()); // fixes #52086
      throw ex;
    } catch (TransactionDataRebalancedException reb) {
      saveTXStateForClientFailover(tx, TXCommitMessage.REBALANCE_MSG);
      cleanup(tx.getTransactionId()); // fixes #52086
      throw reb;
    } catch (UnsupportedOperationInTransactionException e) {
      // fix for #42490
      setTXState(tx);
      throw e;
    } catch (RuntimeException e) {
      saveTXStateForClientFailover(tx, TXCommitMessage.EXCEPTION_MSG);
      cleanup(tx.getTransactionId()); // fixes #52086
      throw e;
    }
    saveTXStateForClientFailover(tx);
    cleanup(tx.getTransactionId());
    noteCommitSuccess(opStart, lifeTime, tx);
  }

  void noteCommitFailure(long opStart, long lifeTime, TXStateInterface tx) {
    long opEnd = statisticsClock.getTime();
    cachePerfStats.txFailure(opEnd - opStart, lifeTime, tx.getChanges());
    TransactionListener[] listeners = getListeners();
    if (tx.isFireCallbacks() && listeners.length > 0) {
      final TXEvent e = tx.getEvent();
      try {
        for (final TransactionListener listener : listeners) {
          try {
            listener.afterFailedCommit(e);
          } catch (VirtualMachineError err) {
            SystemFailure.initiateFailure(err);
            // If this ever returns, rethrow the error. We're poisoned
            // now, so don't let this thread continue.
            throw err;
          } catch (Throwable t) {
            // Whenever you catch Error or Throwable, you must also
            // catch VirtualMachineError (see above). However, there is
            // _still_ a possibility that you are dealing with a cascading
            // error condition, so you also need to check to see if the JVM
            // is still usable:
            SystemFailure.checkFailure();
            logger.error("Exception occurred in TransactionListener", t);
          }
        }
      } finally {
        e.release();
      }
    }
  }

  void noteCommitSuccess(long opStart, long lifeTime, TXStateInterface tx) {
    long opEnd = statisticsClock.getTime();
    cachePerfStats.txSuccess(opEnd - opStart, lifeTime, tx.getChanges());
    TransactionListener[] listeners = getListeners();
    if (tx.isFireCallbacks() && listeners.length > 0) {
      final TXEvent e = tx.getEvent();
      try {
        for (final TransactionListener listener : listeners) {
          try {
            listener.afterCommit(e);
          } catch (VirtualMachineError err) {
            SystemFailure.initiateFailure(err);
            // If this ever returns, rethrow the error. We're poisoned
            // now, so don't let this thread continue.
            throw err;
          } catch (Throwable t) {
            // Whenever you catch Error or Throwable, you must also
            // catch VirtualMachineError (see above). However, there is
            // _still_ a possibility that you are dealing with a cascading
            // error condition, so you also need to check to see if the JVM
            // is still usable:
            SystemFailure.checkFailure();
            logger.error("Exception occurred in TransactionListener", t);
          }
        }
      } finally {
        e.release();
      }
    }
  }

  /**
   * Roll back the transaction associated with the current thread. When this method completes, the
   * thread is no longer associated with a transaction.
   */
  @Override
  public void rollback() {
    checkClosed();
    TXStateProxy tx = getTXState();
    if (tx == null) {
      throw new IllegalStateException(
          "Thread does not have an active transaction");
    }

    tx.checkJTA(
        "Can not rollback this transaction is enlisted with a JTA transaction, use the JTA manager to perform the rollback.");

    final long opStart = statisticsClock.getTime();
    final long lifeTime = opStart - tx.getBeginTime();
    setTXState(null);
    tx.rollback();
    saveTXStateForClientFailover(tx);
    cleanup(tx.getTransactionId());
    noteRollbackSuccess(opStart, lifeTime, tx);
  }

  void noteRollbackSuccess(long opStart, long lifeTime, TXStateInterface tx) {
    long opEnd = statisticsClock.getTime();
    cachePerfStats.txRollback(opEnd - opStart, lifeTime, tx.getChanges());
    TransactionListener[] listeners = getListeners();
    if (tx.isFireCallbacks() && listeners.length > 0) {
      final TXEvent e = tx.getEvent();
      try {
        for (final TransactionListener listener : listeners) {
          try {
            listener.afterRollback(e);
          } catch (VirtualMachineError err) {
            SystemFailure.initiateFailure(err);
            // If this ever returns, rethrow the error. We're poisoned
            // now, so don't let this thread continue.
            throw err;
          } catch (Throwable t) {
            // Whenever you catch Error or Throwable, you must also
            // catch VirtualMachineError (see above). However, there is
            // _still_ a possibility that you are dealing with a cascading
            // error condition, so you also need to check to see if the JVM
            // is still usable:
            SystemFailure.checkFailure();
            logger.error("Exception occurred in TransactionListener", t);
          }
        }
      } finally {
        e.release();
      }
    }
  }

  /**
   * Called from Commit and Rollback to unblock waiting threads
   */
  private void cleanup(TransactionId txId) {
    TXStateProxy proxy = localTxMap.remove(txId);
    if (proxy != null) {
      proxy.close();
    }
    Queue<Thread> waitingThreads = waitMap.get(txId);
    if (waitingThreads != null && !waitingThreads.isEmpty()) {
      for (Thread waitingThread : waitingThreads) {
        LockSupport.unpark(waitingThread);
      }
      waitMap.remove(txId);
    }
  }

  /**
   * Reports the existence of a Transaction for this thread
   *
   */
  @Override
  public boolean exists() {
    return null != getTXState();
  }

  /**
   * Gets the current transaction identifier or null if no transaction exists
   *
   */
  @Override
  public TransactionId getTransactionId() {
    TXStateProxy t = getTXState();
    TransactionId ret = null;
    if (t != null) {
      ret = t.getTransactionId();
    }
    return ret;
  }

  /**
   * Returns the TXStateProxyInterface of the current thread; null if no transaction.
   */
  public TXStateProxy getTXState() {
    TXStateProxy tsp = txContext.get();
    if (tsp == PAUSED) {
      // treats paused transaction as no transaction.
      return null;
    }
    if (tsp != null && !tsp.isInProgress()) {
      txContext.set(null);
      tsp = null;
    }
    return tsp;
  }

  /**
   * sets {@link TXStateProxy#setInProgress(boolean)} when a txContext is present. This method must
   * only be used in fail-over scenarios.
   *
   * @param progress value of the progress flag to be set
   * @return the previous value of inProgress flag
   * @see TXStateProxy#setInProgress(boolean)
   */
  public boolean setInProgress(boolean progress) {
    boolean retVal = false;
    TXStateProxy tsp = txContext.get();
    assert tsp != PAUSED;
    if (tsp != null) {
      retVal = tsp.isInProgress();
      tsp.setInProgress(progress);
    }
    return retVal;
  }

  public void setTXState(TXStateProxy val) {
    txContext.set(val);
  }


  public void close() {
    if (isClosed()) {
      return;
    }
    TXStateProxy[] proxies = null;
    synchronized (hostedTXStates) {
      // After this, newly added TXStateProxy would not operate on the TXState.
      closed = true;

      proxies = hostedTXStates.values().toArray(new TXStateProxy[0]);
    }

    for (TXStateProxy proxy : proxies) {
      proxy.getLock().lock();
      try {
        proxy.close();
      } finally {
        proxy.getLock().unlock();
      }
    }
    for (TXStateProxy proxy : localTxMap.values()) {
      proxy.close();
    }
    TransactionListener[] listeners = getListeners();
    for (final TransactionListener listener : listeners) {
      closeListener(listener);
    }

    currentInstance.compareAndSet(this, null);
  }

  private void closeListener(TransactionListener tl) {
    try {
      tl.close();
    } catch (VirtualMachineError err) {
      SystemFailure.initiateFailure(err);
      // If this ever returns, rethrow the error. We're poisoned
      // now, so don't let this thread continue.
      throw err;
    } catch (Throwable t) {
      // Whenever you catch Error or Throwable, you must also
      // catch VirtualMachineError (see above). However, there is
      // _still_ a possibility that you are dealing with a cascading
      // error condition, so you also need to check to see if the JVM
      // is still usable:
      SystemFailure.checkFailure();
      logger.error("Exception occurred in TransactionListener", t);
    }
  }

  @Immutable
  private static final TXStateProxy PAUSED = new PausedTXStateProxyImpl();

  /**
   * If the current thread is in a transaction then pause will cause it to no longer be in a
   * transaction. The same thread is expected to unpause/resume the transaction later. The thread
   * should not start a new transaction after it paused a transaction.
   *
   * @return the state of the transaction or null. Pass this value to
   *         {@link TXManagerImpl#unpauseTransaction} to reactivate the puased/suspended
   *         transaction.
   */
  public TXStateProxy pauseTransaction() {
    return internalSuspend(true);
  }

  /**
   * If the current thread is in a transaction then internal suspend will cause it to no longer be
   * in a transaction. The thread can start a new transaction after it internal suspended a
   * transaction.
   *
   * @return the state of the transaction or null. to reactivate the suspended transaction.
   */
  public TXStateProxy internalSuspend() {
    return internalSuspend(false);
  }

  /**
   * If the current thread is in a transaction then suspend will cause it to no longer be in a
   * transaction.
   *
   * @param needToResumeBySameThread whether a suspended transaction needs to be resumed by the same
   *        thread.
   * @return the state of the transaction or null. Pass this value to
   *         {@link TXManagerImpl#internalResume(TXStateProxy, boolean)} to reactivate the suspended
   *         transaction.
   */
  private TXStateProxy internalSuspend(boolean needToResumeBySameThread) {
    TXStateProxy result = getTXState();
    if (result != null) {
      result.suspend();
      if (needToResumeBySameThread) {
        setTXState(PAUSED);
      } else {
        setTXState(null);
      }
    } else {
      if (needToResumeBySameThread) {
        // pausedJTA is set to true when JTA is not yet bootstrapped.
        pauseJTA.set(true);
      }
    }
    return result;
  }

  /**
   * Activates the specified transaction on the calling thread. Only the same thread that pause the
   * transaction can unpause it.
   *
   * @param tx the transaction to be unpaused.
   * @throws IllegalStateException if this thread already has an active transaction or this thread
   *         did not pause the transaction.
   */
  public void unpauseTransaction(TXStateProxy tx) {
    internalResume(tx, true);
  }

  /**
   * Activates the specified transaction on the calling thread. Does not require the same thread to
   * resume it.
   *
   * @param tx the transaction to activate.
   * @throws IllegalStateException if this thread already has an active transaction
   */
  public void internalResume(TXStateProxy tx) {
    internalResume(tx, false);
  }

  /**
   * Activates the specified transaction on the calling thread.
   *
   * @param tx the transaction to activate.
   * @param needToResumeBySameThread whether a suspended transaction needs to be resumed by the same
   *        thread.
   * @throws IllegalStateException if this thread already has an active transaction
   */
  private void internalResume(TXStateProxy tx, boolean needToResumeBySameThread) {
    if (tx != null) {
      TransactionId tid = getTransactionId();
      if (tid != null) {
        throw new java.lang.IllegalStateException(
            String.format("Transaction %s already in progress",
                tid));
      }
      if (needToResumeBySameThread) {
        TXStateProxy result = txContext.get();
        if (result != PAUSED) {
          throw new java.lang.IllegalStateException(
              "try to unpause a transaction not paused by the same thread");
        }
      }
      setTXState(tx);

      tx.resume();
    } else {
      if (needToResumeBySameThread) {
        pauseJTA.set(false);
      }
    }
  }

  public boolean isTransactionPaused() {
    return txContext.get() == PAUSED;
  }

  public boolean isJTAPaused() {
    Boolean jtaPaused = pauseJTA.get();
    if (jtaPaused == null) {
      return false;
    }
    return jtaPaused;
  }

  /**
   * @deprecated use internalResume instead
   */
  @Deprecated
  public void resume(TXStateProxy tx) {
    internalResume(tx);
  }

  public boolean isClosed() {
    return closed;
  }

  private void checkClosed() {
    cache.getCancelCriterion().checkCancelInProgress(null);
    if (closed) {
      throw new TXManagerCancelledException("This transaction manager is closed.");
    }
  }

  DistributionManager getDM() {
    return dm;
  }

  public static int getCurrentTXUniqueId() {
    TXManagerImpl instance = currentInstance.get();
    if (instance == null) {
      return NOTX;
    }
    return instance.getMyTXUniqueId();
  }

  public static TXStateProxy getCurrentTXState() {
    TXManagerImpl instance = currentInstance.get();
    if (instance == null) {
      return null;
    }
    return instance.getTXState();
  }

  public int getMyTXUniqueId() {
    TXStateProxy t = txContext.get();
    if (t != null && t != PAUSED) {
      return t.getTxId().getUniqId();
    } else {
      return NOTX;
    }
  }

  /**
   * Associate the remote txState with the thread processing this message. Also, we acquire a lock
   * on the txState, on which this thread operates. Some messages like SizeMessage should not create
   * a new txState.
   *
   * @return {@link TXStateProxy} the txProxy for the transactional message
   */
  public TXStateProxy masqueradeAs(TransactionMessage msg) throws InterruptedException {
    if (msg.getTXUniqId() == NOTX || !msg.canParticipateInTransaction()) {
      return null;
    }
    TXId key = new TXId(msg.getMemberToMasqueradeAs(), msg.getTXUniqId());
    TXStateProxy val = getOrSetHostedTXState(key, msg);

    if (val != null) {
      boolean success = getLock(val, key);
      while (!success) {
        val = getOrSetHostedTXState(key, msg);
        if (val != null) {
          success = getLock(val, key);
        } else {
          break;
        }
      }
    }
    if (logger.isDebugEnabled()) {
      logger.debug("masqueradeAs tx {} for msg {} ", val, msg);
    }
    setTXState(val);
    return val;
  }

  TXStateProxy getOrSetHostedTXState(TXId key, TransactionMessage msg) {
    TXStateProxy val = hostedTXStates.get(key);
    if (val == null) {
      synchronized (hostedTXStates) {
        val = hostedTXStates.get(key);
        if (val == null && msg.canStartRemoteTransaction()) {
          if (msg.isTransactionDistributed()) {
            val = new DistTXStateProxyImplOnDatanode(cache, this, key, msg.getTXOriginatorClient(),
                statisticsClock);
            val.setLocalTXState(new DistTXState(val, true, statisticsClock));
          } else {
            val = new TXStateProxyImpl(cache, this, key, msg.getTXOriginatorClient(),
                statisticsClock);
            val.setLocalTXState(new TXState(val, true, statisticsClock));
            val.setTarget(cache.getDistributedSystem().getDistributedMember());
          }
          hostedTXStates.put(key, val);
        }
      }
    }
    return val;
  }

  boolean getLock(TXStateProxy val, TXId key) {
    if (!val.getLock().isHeldByCurrentThread()) {
      val.getLock().lock();
      synchronized (hostedTXStates) {
        TXStateProxy curVal = hostedTXStates.get(key);
        // Inflight op could be received later than TXFailover operation.
        if (curVal == null) {
          if (!isHostedTxRecentlyCompleted(key)) {
            // Failover op removed the val
            // It is possible that the same operation can be executed
            // twice by two threads, but data is consistent.
            hostedTXStates.put(key, val);
          } else {
            // Another thread should complete the transaction
            logger.info("{} has already finished.", val.getTxId());
          }
        } else {
          if (val != curVal) {
            // Failover op replaced with a new TXStateProxyImpl
            // Use the new one instead.
            val.getLock().unlock();
            return false;
          }
        }
      }
    }
    return true;
  }

  /**
   * Associate the remote txState with the thread processing this message. Also, we acquire a lock
   * on the txState, on which this thread operates. Some messages like SizeMessage should not create
   * a new txState.
   *
   * @param probeOnly - do not masquerade; just look up the TX state
   * @return {@link TXStateProxy} the txProxy for the transactional message
   */
  public TXStateProxy masqueradeAs(Message msg, InternalDistributedMember memberId,
      boolean probeOnly) throws InterruptedException {
    if (msg.getTransactionId() == NOTX) {
      return null;
    }
    TXId key = new TXId(memberId, msg.getTransactionId());
    TXStateProxy val;
