/*
 * 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.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.IdentityHashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.ServiceConfigurationError;
import java.util.Set;
import java.util.concurrent.Executor;
import java.util.concurrent.locks.ReentrantLock;

import jakarta.transaction.Status;
import org.apache.logging.log4j.Logger;

import org.apache.geode.CancelCriterion;
import org.apache.geode.CancelException;
import org.apache.geode.InternalGemFireError;
import org.apache.geode.SystemFailure;
import org.apache.geode.annotations.Immutable;
import org.apache.geode.cache.CommitConflictException;
import org.apache.geode.cache.DiskAccessException;
import org.apache.geode.cache.EntryNotFoundException;
import org.apache.geode.cache.FailedSynchronizationException;
import org.apache.geode.cache.Operation;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.Region.Entry;
import org.apache.geode.cache.RegionDestroyedException;
import org.apache.geode.cache.SynchronizationCommitConflictException;
import org.apache.geode.cache.TransactionDataRebalancedException;
import org.apache.geode.cache.TransactionException;
import org.apache.geode.cache.TransactionId;
import org.apache.geode.cache.TransactionWriter;
import org.apache.geode.cache.TransactionWriterException;
import org.apache.geode.cache.UnsupportedOperationInTransactionException;
import org.apache.geode.cache.client.internal.ServerRegionDataAccess;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.TXManagerCancelledException;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.cache.control.MemoryThresholds;
import org.apache.geode.internal.cache.entries.AbstractRegionEntry;
import org.apache.geode.internal.cache.partitioned.PutAllPRMessage;
import org.apache.geode.internal.cache.partitioned.RemoveAllPRMessage;
import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID;
import org.apache.geode.internal.cache.tier.sockets.VersionedObjectList;
import org.apache.geode.internal.cache.tx.TransactionalOperation.ServerRegionOperation;
import org.apache.geode.internal.offheap.annotations.Released;
import org.apache.geode.internal.offheap.annotations.Retained;
import org.apache.geode.internal.statistics.StatisticsClock;
import org.apache.geode.logging.internal.log4j.api.LogService;

/**
 * TXState is the entity that tracks the transaction state on a per thread basis, noting changes to
 * Region entries on a per operation basis. It lives on the node where transaction data exists.
 *
 * @since GemFire 4.0
 * @see TXManagerImpl
 */
public class TXState implements TXStateInterface {
  protected static final Logger logger = LogService.getLogger();

  // The nano-timestamp of when the transaction began
  private final long beginTime;
  // A map of transaction state by Region
  final IdentityHashMap<InternalRegion, TXRegionState> regions;

  /** whether completion has been started */
  protected boolean completionStarted;

  /** whether the transaction has been completed and cleaned up */
  protected boolean closed = false;

  /** guards the completionStarted boolean and the closed boolean */
  protected final Object completionGuard = new Object();

  protected TXLockRequest locks = null;
  // Used for jta commit lifetime
  private long jtaLifeTime;
  /**
   * Used to hand out modification serial numbers used to preserve the order of operation done by
   * this transaction.
   */
  private int modSerialNum;
  private final List<EntryEventImpl> pendingCallbacks = new ArrayList<>();
  // Access this variable should be in synchronized block.
  private boolean beforeCompletionCalled;

  /**
   * for client/server JTA transactions we need to have a single thread handle both beforeCompletion
   * and afterCompletion so that beforeCompletion can obtain locks for the afterCompletion step.
   * This is that thread
   */
  private final SingleThreadJTAExecutor singleThreadJTAExecutor;

  // Internal testing hooks
  private Runnable internalAfterReservation;
  protected Runnable internalAfterConflictCheck;
  protected Runnable internalDuringApplyChanges;
  protected Runnable internalAfterApplyChanges;
  protected Runnable internalAfterReleaseLocalLocks;
  Runnable internalDuringIndividualSend; // package scope allows TXCommitMessage use
  Runnable internalAfterIndividualSend; // package scope allows TXCommitMessage use
  Runnable internalDuringIndividualCommitProcess; // package scope allows TXCommitMessage use
  Runnable internalAfterIndividualCommitProcess; // package scope allows TXCommitMessage use
  protected Runnable internalAfterSend;
  protected Runnable internalBeforeSend;

  /**
   * Used to generate eventIDs
   */
  private byte[] baseMembershipId;
  /**
   * Used to generate eventIDs
   */
  private long baseThreadId;
  /**
   * Used to generate eventIDs
   */
  private long baseSequenceId;
  protected final TXStateProxy proxy;
  protected boolean firedWriter = false;
  protected final boolean onBehalfOfRemoteStub;
  protected boolean gotBucketLocks = false;
  protected TXCommitMessage commitMessage = null;
  ClientProxyMembershipID bridgeContext = null;
  /** keeps track of events, so as not to re-apply events */
  protected Set<EventID> seenEvents = new HashSet<>();
  /** keeps track of results of txPutEntry */
  private final Map<EventID, Boolean> seenResults = new HashMap<>();
  /** keeps track of TransactionDataRebalancedException during txPutEntry */
  private final Map<EventID, TransactionDataRebalancedException> eventExceptions = new HashMap<>();

  @Immutable
  static final TXEntryState ENTRY_EXISTS = new TXEntryState();

  private volatile DistributedMember proxyServer;

  private final StatisticsClock statisticsClock;

  public TXState(TXStateProxy proxy, boolean onBehalfOfRemoteStub,
      StatisticsClock statisticsClock) {
    this(proxy, onBehalfOfRemoteStub, new SingleThreadJTAExecutor(), statisticsClock);
  }

  public TXState(TXStateProxy proxy, boolean onBehalfOfRemoteStub,
      SingleThreadJTAExecutor singleThreadJTAExecutor, StatisticsClock statisticsClock) {
    beginTime = statisticsClock.getTime();
    regions = new IdentityHashMap<>();

    internalAfterConflictCheck = null;
    internalAfterApplyChanges = null;
    internalAfterReleaseLocalLocks = null;
    internalDuringIndividualSend = null;
    internalAfterIndividualSend = null;
    internalBeforeSend = null;
    internalAfterSend = null;
    this.proxy = proxy;
    this.onBehalfOfRemoteStub = onBehalfOfRemoteStub;
    this.singleThreadJTAExecutor = singleThreadJTAExecutor;
    this.statisticsClock = statisticsClock;
  }

  private boolean hasSeenEvent(EntryEventImpl event) {
    assert event != null;
    if (event.getEventId() == null) {
      return false;
    }
    return seenEvents.contains(event.getEventId());
  }

  private void recordEvent(EntryEventImpl event) {
    assert event != null;
    if (event.getEventId() != null) {
      seenEvents.add(event.getEventId());
    }
  }

  void recordEventAndResult(EntryEventImpl event, boolean result) {
    recordEvent(event);
    if (event.getEventId() != null) {
      seenResults.put(event.getEventId(), result);
    }
  }

  boolean getRecordedResult(EntryEventImpl event) {
    assert event != null;
    assert seenResults.containsKey(event.getEventId());
    return seenResults.get(event.getEventId());
  }

  void recordEventException(EntryEventImpl event,
      TransactionDataRebalancedException exception) {
    if (event.getEventId() != null) {
      eventExceptions.put(event.getEventId(), exception);
    }
  }

  boolean getRecordedResultOrException(EntryEventImpl event) {
    boolean result = getRecordedResult(event);
    if (!result && eventExceptions.containsKey(event.getEventId())) {
      throw eventExceptions.get(event.getEventId());
    }
    return result;
  }

  @Override
  public String toString() {
    return getClass() + "@" + System.identityHashCode(this)
        + " onBehalfOfRemoteStub:" + onBehalfOfRemoteStub;
  }

  /*
   * (non-Javadoc)
   *
   * @see org.apache.geode.internal.cache.TXStateInterface#getTransactionId()
   */
  @Override
  public TransactionId getTransactionId() {
    return proxy.getTxId();
  }

  public void firePendingCallbacks() {
    boolean isConfigError = false;
    EntryEventImpl lastTransactionEvent = null;
    try {
      lastTransactionEvent = TXLastEventInTransactionUtils
          .getLastTransactionEventInGroupedTxForWANSender(getPendingCallbacks(), getCache());
    } catch (ServiceConfigurationError ex) {
      logger.error(ex.getMessage());
      isConfigError = true;
    }

    for (EntryEventImpl ee : getPendingCallbacks()) {
      boolean isLastTransactionEvent = TXLastEventInTransactionUtils
          .isLastTransactionEvent(isConfigError, lastTransactionEvent, ee);
      if (ee.getOperation().isDestroy()) {
        ee.getRegion().invokeTXCallbacks(EnumListenerEvent.AFTER_DESTROY, ee, true,
            isLastTransactionEvent);
      } else if (ee.getOperation().isInvalidate()) {
        ee.getRegion().invokeTXCallbacks(EnumListenerEvent.AFTER_INVALIDATE, ee, true,
            isLastTransactionEvent);
      } else if (ee.getOperation().isCreate()) {
        ee.getRegion().invokeTXCallbacks(EnumListenerEvent.AFTER_CREATE, ee, true,
            isLastTransactionEvent);
      } else {
        ee.getRegion().invokeTXCallbacks(EnumListenerEvent.AFTER_UPDATE, ee, true,
            isLastTransactionEvent);
      }
    }
  }

  public void freePendingCallbacks() {
    for (EntryEventImpl ee : getPendingCallbacks()) {
      ee.release();
    }
  }

  public List<EntryEventImpl> getPendingCallbacks() {
    return pendingCallbacks;
  }


  /*
   * (non-Javadoc)
   *
   * @see
   * org.apache.geode.internal.cache.TXStateInterface#readRegion(org.apache.geode.internal.cache.
   * LocalRegion)
   */
  @Override
  public TXRegionState readRegion(InternalRegion r) {
    return regions.get(r);
  }

  @Override
  public void rmRegion(LocalRegion r) {
    TXRegionState txr = regions.remove(r);
    if (txr != null) {
      txr.cleanup(r);
    }
  }

  /*
   * (non-Javadoc)
   *
   * @see
   * org.apache.geode.internal.cache.TXStateInterface#writeRegion(org.apache.geode.internal.cache.
   * LocalRegion)
   */
  @Override
  public TXRegionState writeRegion(InternalRegion r) {
    TXRegionState result = readRegion(r);
    if (result == null) {
      if (r instanceof BucketRegion) {
        result = new TXBucketRegionState((BucketRegion) r, this);
      } else {
        result = new TXRegionState(r, this);
      }
      regions.put(r, result);
    }
    if (logger.isDebugEnabled()) {
      logger.debug("TXState writeRegion flag {} region-state {} ", false,
          result/* , new Throwable() */);
    }
    return result;
  }

  /*
   * (non-Javadoc)
   *
   * @see org.apache.geode.internal.cache.TXStateInterface#getBeginTime()
   */
  @Override
  public long getBeginTime() {
    return beginTime;
  }

  /*
   * (non-Javadoc)
   *
   * @see org.apache.geode.internal.cache.TXStateInterface#getChanges()
   */
  @Override
  public int getChanges() {
    int changes = 0;
    for (final TXRegionState txrs : regions.values()) {
      changes += txrs.getChanges();
    }
    return changes;
  }

  /*
   * (non-Javadoc)
   *
   * @see org.apache.geode.internal.cache.TXStateInterface#isInProgress()
   */
  @Override
  public boolean isInProgress() {
    return !closed;
  }

  /*
   * (non-Javadoc)
   *
   * @see org.apache.geode.internal.cache.TXStateInterface#nextModSerialNum()
   */
  @Override
  public int nextModSerialNum() {
    modSerialNum += 1;
    return modSerialNum;
  }

  /*
   * (non-Javadoc)
   *
   * @see org.apache.geode.internal.cache.TXStateInterface#needsLargeModCount()
   */
  @Override
  public boolean needsLargeModCount() {
    return modSerialNum > Byte.MAX_VALUE;
  }

  protected void reserveAndCheck() throws CommitConflictException {
    if (closed) {
      return;
    }

    final long conflictStart = statisticsClock.getTime();
    locks = createLockRequest();
    locks.obtain(getCache().getInternalDistributedSystem());
    // for now check account the dlock service time
    // later this stat end should be moved to a finally block
    if (statisticsClock.isEnabled()) {
      proxy.getTxMgr().getCachePerfStats()
          .incTxConflictCheckTime(statisticsClock.getTime() - conflictStart);
    }
    if (internalAfterReservation != null) {
      internalAfterReservation.run();
    }
    checkForConflicts();
  }

  byte[] getBaseMembershipId() {
    return baseMembershipId;
  }

  long getBaseThreadId() {
    return baseThreadId;
  }

  long getBaseSequenceId() {
    return baseSequenceId;
  }

  @Override
  public void precommit()
      throws CommitConflictException, UnsupportedOperationInTransactionException {
    throw new UnsupportedOperationInTransactionException(
        String.format("precommit() operation %s meant for Dist Tx is not supported",
            "precommit"));
  }

  /*
   * (non-Javadoc)
   *
   * @see org.apache.geode.internal.cache.TXStateInterface#commit()
   */
  @Override
  public void commit() throws CommitConflictException {
    if (closed) {
      return;
    }

    if (logger.isDebugEnabled()) {
      logger.debug("committing transaction {}", getTransactionId());
    }
    synchronized (completionGuard) {
      completionStarted = true;
    }

    if (onBehalfOfRemoteStub && !proxy.isCommitOnBehalfOfRemoteStub()) {
      throw new UnsupportedOperationInTransactionException(
          "Cannot commit a transaction being run on behalf of a remote thread");
    }
    cleanupNonDirtyRegions();
    try {
      /*
       * Lock buckets so they can't be rebalanced then perform the conflict check to fix #43489
       */
      try {
        lockBucketRegions();
      } catch (PrimaryBucketException pbe) {
        // not sure what to do here yet
        throw new TransactionDataRebalancedException(
            "Transactional data moved, due to rebalancing.", pbe);
      }

      if (locks == null) {
        reserveAndCheck();
      }

      // For internal testing
      if (internalAfterConflictCheck != null) {
        internalAfterConflictCheck.run();
      }

      /*
       * If there is a TransactionWriter plugged in, we need to to give it an opportunity to cleanup
       * the transaction.
       */
      TransactionWriter writer = proxy.getTxMgr().getWriter();
      if (!firedWriter && writer != null) {
        try {
          firedWriter = true;
          TXEvent event = getEvent();
          if (!event.hasOnlyInternalEvents()) {
            writer.beforeCommit(event);
          }
        } catch (TransactionWriterException twe) {
          cleanup();
          throw new CommitConflictException(twe);
        } catch (VirtualMachineError err) {
          // cleanup(); this allocates objects so I don't think we can do it - that leaves the TX
          // open, but we are poison pilling so we should be ok??

          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) {
          cleanup(); // rollback the transaction!
          // 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();
          throw new CommitConflictException(t);
        }
      }

      List/* <TXEntryStateWithRegionAndKey> */ entries = generateEventOffsets();
      TXCommitMessage msg = null;
      try {
        /*
         * In order to preserve data consistency, we need to: 1. Modify the cache first
         * (applyChanges) 2. Ask for advice on who to send to (buildMessage) 3. Send out to other
         * members.
         *
         * If this is done out of order, we will have problems with GII, split brain, and HA.
         */

        lockTXRegions(regions);

        try {
          // apply changes to the cache
          applyChanges(entries);
          // For internal testing
          if (internalAfterApplyChanges != null) {
            internalAfterApplyChanges.run();
          }

          attachFilterProfileInformation(entries);

          // build and send the message
          msg = buildMessage();
          commitMessage = msg;
          if (internalBeforeSend != null) {
            internalBeforeSend.run();
          }

          msg.send(locks.getDistributedLockId());
          // For internal testing
          if (internalAfterSend != null) {
            internalAfterSend.run();
          }

          firePendingCallbacks();
          /*
           * This is to prepare the commit message for the caller, make sure all events are in
           * there.
           */
          commitMessage = buildCompleteMessage();
        } finally {
          unlockTXRegions(regions);
        }
      } finally {
        if (msg != null) {
          msg.releaseViewVersions();
        }
        locks.releaseLocal();
        // For internal testing
        if (internalAfterReleaseLocalLocks != null) {
          internalAfterReleaseLocalLocks.run();
        }
      }
    } finally {
      cleanup();
    }
  }

  private void lockTXRegions(IdentityHashMap<InternalRegion, TXRegionState> regions) {
    for (final Map.Entry<InternalRegion, TXRegionState> me : regions.entrySet()) {
      InternalRegion r = me.getKey();
      r.getRegionMap().lockRegionForAtomicTX(r);
    }
  }

  private void unlockTXRegions(IdentityHashMap<InternalRegion, TXRegionState> regions) {
    for (final Map.Entry<InternalRegion, TXRegionState> me : regions.entrySet()) {
      InternalRegion r = me.getKey();
      r.getRegionMap().unlockRegionForAtomicTX(r);
    }
  }

  protected void attachFilterProfileInformation(List entries) {
    {
      /* <TXEntryStateWithRegionAndKey> */
      for (final Object entry : entries) {
        TXEntryStateWithRegionAndKey o = (TXEntryStateWithRegionAndKey) entry;
        try {
          if (o.r.isUsedForPartitionedRegionBucket()) {
            BucketRegion bucket = (BucketRegion) o.r;
            /*
             * The event must contain the bucket region
             */
            @Released
            EntryEventImpl ev =
                (EntryEventImpl) o.es.getEvent(o.r, o.key, o.es.getTXRegionState().getTXState());
            if (ev.getOperation() == null) {
              // A read op with detect read conflicts does not need filter routing.
              continue;
            }
            try {
              /*
               * The routing information is derived from the PR advisor, not the bucket advisor.
               */
              FilterRoutingInfo fri = bucket.getPartitionedRegion().getRegionAdvisor()
                  .adviseFilterRouting(ev, Collections.EMPTY_SET);
              o.es.setFilterRoutingInfo(fri);
              Set set = bucket.getAdjunctReceivers(ev, Collections.EMPTY_SET, new HashSet(), fri);
              o.es.setAdjunctRecipients(set);

              if (o.es.getPendingCallback() != null) {
                if (fri != null) {
                  // For tx host, local filter info was also calculated.
                  // Set this local filter info in corresponding pending callback so that
                  // notifyBridgeClient has correct routing info.
                  FilterRoutingInfo.FilterInfo localRouting = fri.getLocalFilterInfo();
                  o.es.getPendingCallback().setLocalFilterInfo(localRouting);
                }
                // Do not hold pending callback reference in TXEntryState as it is no longer used.
                o.es.setPendingCallback(null);
              }
            } finally {
              ev.release();
            }
          }
        } catch (RegionDestroyedException ex) {
          // region was destroyed out from under us; after conflict checking
          // passed. So act as if the region destroy happened right after the
          // commit. We act this way by doing nothing; including distribution
          // of this region's commit data.
        } catch (CancelException ex) {
          // cache was closed out from under us; after conflict checking
          // passed. So do nothing.
        }
      }
    }

  }

  /*
   * (non-Javadoc)
   *
   * @see org.apache.geode.internal.cache.TXStateInterface#rollback()
   */
  @Override
  public void rollback() {
    if (closed) {
      return;
    }
    synchronized (completionGuard) {
      completionStarted = true;
    }
    cleanup();
  }

  /**
   * This is a fix for bug #42228 where a client fails over from one server to another but gets a
   * conflict on completion because completion had already been initiated and had not yet completed
   *
   * @return true if a previous completion was in progress
   */
  public boolean waitForPreviousCompletion() {
    synchronized (completionGuard) {// should have already been done, but just to be sure
      if (!completionStarted) {
        return false;
      }
      while (commitMessage == null && !closed) {
        if (logger.isDebugEnabled()) {
          logger.debug("Waiting for previous completion for transaction {}", getTransactionId());
        }
        try {
          completionGuard.wait();
        } catch (InterruptedException e) {
          proxy.getCache().getCancelCriterion().checkCancelInProgress(e);
          Thread.currentThread().interrupt();
          return true;
        }
      } // while
    }
    return true;
  }

  /**
   * Generate an event id for each operation that will be done by this tx during the application
   * phase of its commit.
   *
   * @return a sorted list of TXEntryStateWithRegionAndKey that will be used to apply the ops on the
   *         nearside in the correct order.
   */
  protected List/* <TXEntryStateWithRegionAndKey> */ generateEventOffsets() {
    baseMembershipId = EventID.getMembershipId(proxy.getTxMgr().getDM().getSystem());
    baseThreadId = EventID.getThreadId();
    baseSequenceId = EventID.getSequenceId();

    List/* <TXEntryStateWithRegionAndKey> */ entries = getSortedEntries();
    if (logger.isDebugEnabled()) {
      logger
          .debug("generateEventOffsets() entries " + entries + " RegionState Map=" + regions);
    }
    for (final Object entry : entries) {
      TXEntryStateWithRegionAndKey o = (TXEntryStateWithRegionAndKey) entry;
      o.es.generateEventOffsets(this);
    }
    return entries;
  }

  private TXLockRequest createLockRequest() {
    TXLockRequest result = new TXLockRequest();
    for (final Map.Entry<InternalRegion, TXRegionState> me : regions.entrySet()) {
      InternalRegion r = me.getKey();
      TXRegionState txrs = me.getValue();
      txrs.createLockRequest(r, result);
    }
    return result;
  }

  private void checkForConflicts() throws CommitConflictException, PrimaryBucketException {
    for (final Map.Entry<InternalRegion, TXRegionState> me : regions.entrySet()) {
      InternalRegion r = me.getKey();
      TXRegionState txrs = me.getValue();
      try {
        txrs.checkForConflicts(r);
      } catch (DiskAccessException dae) {
        r.handleDiskAccessException(dae);
        throw dae;
      }
    }
  }


  protected void lockBucketRegions() throws PrimaryBucketException {

    boolean lockingSucceeded;
    do {
      lockingSucceeded = true;
      Iterator<Map.Entry<InternalRegion, TXRegionState>> it = regions.entrySet().iterator();
      Set<BucketRegion> obtained = new HashSet<>();
      while (it.hasNext()) {
        Map.Entry<InternalRegion, TXRegionState> me = it.next();
        InternalRegion r = me.getKey();
        if (r instanceof BucketRegion) {
          if (isDistTx() && !((BucketRegion) r).getBucketAdvisor().isPrimary()) {
            // For distTx we skip for taking locks on secondary.
            continue;
          }
          BucketRegion b = (BucketRegion) r;
          /*
           * Lock the primary bucket so it doesnt get rebalanced until we cleanup!
           */
          boolean lockObtained = false;
          try {
            // use tryLocks to avoid hanging (bug #41708)
            boolean locked = b.doLockForPrimary(true);
            if (locked) {
              obtained.add(b);
              lockObtained = true;
            } else {
              // if we can't get locks then someone has a write-lock. To prevent
              // deadlock (see bug #41708) we release locks and re-acquire them
              r.getCancelCriterion().checkCancelInProgress(null);
              if (logger.isDebugEnabled()) {
                logger.debug("tryLock failed for commit on {}. Releasing locks and retrying",
                    r.getFullPath());
              }
              // release locks and start over
              break;
            }
          } catch (RegionDestroyedException rde) {
            if (logger.isDebugEnabled()) {
              logger.debug("RegionDestroyedException while locking bucket region {}",
                  r.getFullPath(), rde);
            }
            throw new TransactionDataRebalancedException(
                "Bucket rebalanced during commit: " + r.getFullPath());
          } finally {
            if (!lockObtained) {
              // fix for bug #41708 - unlock operation-locks already obtained
              if (logger.isDebugEnabled()) {
                logger.debug("Unexpected exception while locking bucket {}", r.getFullPath());
              }
              for (BucketRegion br : obtained) {
                br.doUnlockForPrimary();
              }
              try {
                Thread.sleep(50);
              } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return;
              }
              lockingSucceeded = false;
            }
          }
        }
      }
    } while (!lockingSucceeded);
    gotBucketLocks = true;
  }


  protected void cleanupNonDirtyRegions() {
    for (final Map.Entry<InternalRegion, TXRegionState> me : regions.entrySet()) {
      InternalRegion r = me.getKey();
      TXRegionState txrs = me.getValue();
      txrs.cleanupNonDirtyEntries(r);
    }
  }

  /**
   * this builds a new TXCommitMessage and returns it
   *
   * @return the new message
   */
  protected TXCommitMessage buildMessage() {
    TXCommitMessage msg =
        new TXCommitMessage(proxy.getTxId(), proxy.getTxMgr().getDM(), this);
    for (final Map.Entry<InternalRegion, TXRegionState> me : regions.entrySet()) {
      InternalRegion r = me.getKey();
      TXRegionState txrs = me.getValue();
      txrs.buildMessage(r, msg);
    }
    return msg;
  }


  /**
   * this builds a new TXCommitMessage and returns it
   *
   * @return the new message
   */
  protected TXCommitMessage buildCompleteMessage() {
    TXCommitMessage msg =
        new TXCommitMessage(proxy.getTxId(), proxy.getTxMgr().getDM(), this);
    for (final Map.Entry<InternalRegion, TXRegionState> me : regions.entrySet()) {
      InternalRegion r = me.getKey();
      TXRegionState txrs = me.getValue();
      txrs.buildCompleteMessage(r, msg);
      // rcl.add(r);
    }
    return msg;
  }

  /**
   * applies this transaction to the cache.
   */
  protected void applyChanges(List/* <TXEntryStateWithRegionAndKey> */ entries) {
    // applyChangesStart for each region
    for (Map.Entry<InternalRegion, TXRegionState> me : regions.entrySet()) {
      InternalRegion r = me.getKey();
      TXRegionState txrs = me.getValue();
      txrs.applyChangesStart(r, this);
    }

    // serializePendingValue for each entry
    for (Object entry : entries) {
      TXEntryStateWithRegionAndKey o = (TXEntryStateWithRegionAndKey) entry;
      o.es.serializePendingValue();
    }

    // applyChanges for each entry
    int size = pendingCallbacks.size();
    for (Object entry : entries) {
      TXEntryStateWithRegionAndKey o = (TXEntryStateWithRegionAndKey) entry;
      if (internalDuringApplyChanges != null) {
        internalDuringApplyChanges.run();
      }
      try {
        o.es.applyChanges(o.r, o.key, this);
        if (pendingCallbacks.size() > size) {
          o.es.setPendingCallback(pendingCallbacks.get(size));
          size = pendingCallbacks.size();
        }
      } catch (RegionDestroyedException ex) {
        // region was destroyed out from under us; after conflict checking
        // passed. So act as if the region destroy happened right after the
        // commit. We act this way by doing nothing; including distribution
        // of this region's commit data.
      } catch (CancelException ex) {
        // cache was closed out from under us; after conflict checking
        // passed. So do nothing.
      }
    }

    // applyChangesEnd for each region
    for (Map.Entry<InternalRegion, TXRegionState> me : regions.entrySet()) {
      InternalRegion r = me.getKey();
      TXRegionState txrs = me.getValue();
      txrs.applyChangesEnd(r, this);
    }
  }

  @Override
  public TXEvent getEvent() {
    return new TXEvent(this, getCache());
  }

  /**
   * Note that cleanup does more than is needed in this method. This method only needs to do stuff
   * that is required when a Cache close is done and we have txs that are still in progress.
   * Currently the only thing that is needed is to decrement off-heap refcounts since off-heap
   * memory lives after a cache close.
   */
  @Override
  public void close() {
    if (!closed) {
      if (locks != null) {
        cleanup();
        return;
      }
      closed = true;
      for (TXRegionState r : regions.values()) {
        r.close();
      }
    }
  }

  protected void cleanup() {
    if (singleThreadJTAExecutor.shouldDoCleanup()) {
      singleThreadJTAExecutor.cleanup();
    } else {
      doCleanup();
    }
  }

  void doCleanup() {
    RuntimeException exception = null;
    try {
      closed = true;
      seenEvents.clear();
      seenResults.clear();
      freePendingCallbacks();
      if (locks != null) {
        final long conflictStart = statisticsClock.getTime();
        try {
          locks.cleanup(getCache().getInternalDistributedSystem());
        } catch (IllegalArgumentException | IllegalMonitorStateException e) {
          exception = e;
        }
        if (statisticsClock.isEnabled()) {
          proxy.getTxMgr().getCachePerfStats()
              .incTxConflictCheckTime(statisticsClock.getTime() - conflictStart);
        }
      }
      for (final Map.Entry<InternalRegion, TXRegionState> me : regions.entrySet()) {
        InternalRegion r = me.getKey();
        TXRegionState txrs = me.getValue();
        /*
         * Need to unlock the primary lock for rebalancing so that rebalancing can resume.
         */
        if (gotBucketLocks) {
          if (r instanceof BucketRegion && (((BucketRegion) r).getBucketAdvisor().isPrimary())) {
            try {
              ((BucketRegion) r).doUnlockForPrimary();
            } catch (RegionDestroyedException rde) {
              // ignore
              if (logger.isDebugEnabled()) {
                logger.debug("RegionDestroyedException while unlocking bucket region {}",
                    r.getFullPath(), rde);
              }
            } catch (Exception rde) {
              // ignore
              if (logger.isDebugEnabled()) {
                logger.debug(
                    "Exception while unlocking bucket region {} this is probably because the bucket was destroyed and never locked initially.",
