/*
 * 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.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.TreeMap;

import org.apache.geode.InternalGemFireException;
import org.apache.geode.InvalidDeltaException;
import org.apache.geode.SystemFailure;
import org.apache.geode.annotations.internal.MutableForTesting;
import org.apache.geode.cache.CacheWriterException;
import org.apache.geode.cache.CommitConflictException;
import org.apache.geode.cache.DiskAccessException;
import org.apache.geode.cache.EntryNotFoundException;
import org.apache.geode.cache.TransactionDataRebalancedException;
import org.apache.geode.cache.TransactionWriter;
import org.apache.geode.cache.TransactionWriterException;
import org.apache.geode.cache.UnsupportedOperationInTransactionException;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.cache.TXEntryState.DistTxThinEntryState;
import org.apache.geode.internal.cache.partitioned.PutAllPRMessage;
import org.apache.geode.internal.cache.partitioned.RemoveAllPRMessage;
import org.apache.geode.internal.cache.tier.sockets.VersionedObjectList;
import org.apache.geode.internal.cache.tx.DistTxEntryEvent;
import org.apache.geode.internal.cache.tx.DistTxKeyInfo;
import org.apache.geode.internal.cache.versions.RegionVersionVector;
import org.apache.geode.internal.offheap.annotations.Released;
import org.apache.geode.internal.statistics.StatisticsClock;

/**
 * TxState on a data node VM
 *
 *
 */
public class DistTXState extends TXState {

  @MutableForTesting
  public static Runnable internalBeforeApplyChanges; // TODO: cleanup this test hook
  @MutableForTesting
  public static Runnable internalBeforeNonTXBasicPut; // TODO: cleanup this test hook

  private boolean updatingTxStateDuringPreCommit = false;

  public DistTXState(TXStateProxy proxy, boolean onBehalfOfRemoteStub,
      StatisticsClock statisticsClock) {
    super(proxy, onBehalfOfRemoteStub, statisticsClock);
  }

  @Override
  protected void cleanup() {
    super.cleanup();
    // Do nothing for now
  }

  /*
   * If this is a primary member, for each entry in TXState, generate next region version and store
   * in the entry.
   */
  public void updateRegionVersions() {

    for (final Entry<InternalRegion, TXRegionState> me : regions.entrySet()) {
      InternalRegion r = me.getKey();
      TXRegionState txrs = me.getValue();

      // Generate next region version only on the primary
      if (!txrs.isCreatedDuringCommit()) {
        try {
          Set entries = txrs.getEntryKeys();
          if (!entries.isEmpty()) {
            for (final Object key : entries) {
              TXEntryState txes = txrs.getTXEntryState(key);
              RegionVersionVector rvv = r.getVersionVector();
              if (rvv != null) {
                long v = rvv.getNextVersion();
                // txes.setNextRegionVersion(v);
                txes.getDistTxEntryStates().setRegionVersion(v);
                if (logger.isDebugEnabled()) {
                  logger.debug("Set next region version to " + v + " for region=" + r.getName()
                      + "in TXEntryState for key" + key);
                }
              }
            }
          }
        } catch (DiskAccessException dae) {
          r.handleDiskAccessException(dae);
          throw dae;
        }
      }
    }
  }

  /*
   * Iterate through all changes and for those changes for which this member hosts a primary bucket,
   * generate a tail key and store in the TXEntryState. From there it is expected to be carried over
   * to the secondaries in phase-2 commit. In phase-2 commit, the both the primary and secondaries
   * should use this tail key to enqueue into parallel queues.
   */
  public void generateTailKeysForParallelDispatcherEvents() {

    for (final Entry<InternalRegion, TXRegionState> me : regions.entrySet()) {
      InternalRegion r = me.getKey();
      TXRegionState txrs = me.getValue();

      InternalRegion region = txrs.getRegion();
      // Check if it is a bucket region
      if (region.isUsedForPartitionedRegionBucket()) {
        // Check if it is a primary bucket
        BucketRegion bRegion = (BucketRegion) region;
        if (!(bRegion instanceof AbstractBucketRegionQueue)) {
          if (bRegion.getBucketAdvisor().isPrimary()) {

            // Generate a tail key for each entry
            Set entries = txrs.getEntryKeys();
            if (!entries.isEmpty()) {
              for (final Object key : entries) {
                TXEntryState txes = txrs.getTXEntryState(key);

                long tailKey = ((BucketRegion) region).generateTailKey();
                txes.getDistTxEntryStates().setTailKey(tailKey);
              }
            }
          } // end if primary
        }
      }
    }
  }


  /*
   * (non-Javadoc)
   *
   * @see org.apache.geode.internal.cache.TXStateInterface#commit()
   *
   * Take Locks Does conflict check on primary ([DISTTX] TODO on primary only) Invoke TxWriter
   */
  @Override
  public void precommit()
      throws CommitConflictException, UnsupportedOperationInTransactionException {
    if (logger.isDebugEnabled()) {
      logger.debug("DistTXState.precommit transaction {} is closed {} ", getTransactionId(),
          closed/* , new Throwable() */);
    }

    if (closed) {
      return;
    }

    synchronized (completionGuard) {
      completionStarted = true;
    }

    if (onBehalfOfRemoteStub && !proxy.isCommitOnBehalfOfRemoteStub()) {
      throw new UnsupportedOperationInTransactionException(
          "Cannot commit a transaction being run on behalf of a remote thread");
    }

    cleanupNonDirtyRegions();

    /*
     * 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
      RuntimeException re = new TransactionDataRebalancedException(
          "Transactional data moved, due to rebalancing.", pbe);
      throw re;
    }

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

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

    updateRegionVersions();

    generateTailKeysForParallelDispatcherEvents();

    /*
     * If there is a TransactionWriter plugged in, we need to to give it an opportunity to abort the
     * transaction.
     */
    TransactionWriter writer = proxy.getTxMgr().getWriter();
    if (!firedWriter && writer != null) {
      try {
        firedWriter = true;
        writer.beforeCommit(getEvent());
      } 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);
      }
    }
  }

  /*
   * (non-Javadoc)
   *
   * @see org.apache.geode.internal.cache.TXStateInterface#commit()
   *
   * Apply changes release locks
   */
  @Override
  public void commit() throws CommitConflictException {
    if (logger.isDebugEnabled()) {
      logger.debug("DistTXState.commit transaction {} is closed {} ", getTransactionId(),
          closed/* , new Throwable() */);
    }

    if (closed) {
      return;
    }

    try {
      List/* <TXEntryStateWithRegionAndKey> */ entries = generateEventOffsets();
      if (logger.isDebugEnabled()) {
        logger.debug("commit entries " + entries);
      }
      TXCommitMessage msg = null;
      try {
        attachFilterProfileInformation(entries);

        if (internalBeforeApplyChanges != null) {
          internalBeforeApplyChanges.run();
        }

        // apply changes to the cache
        applyChanges(entries);

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

        // [DISTTX]TODO:
        // Build a message specifically for those nodes who
        // hold gateway senders and listeners but not a copy of the buckets
        // on which changes in this tx are done.
        // This is applicable only for partitioned regions and
        // serial gateway senders.
        // This works only if the coordinator and sender are not the same node.
        // For same sender as coordinator, this results in a hang, which needs to be addressed.
        // If an another method of notifying adjunct receivers is implemented,
        // the following two lines should be commented out.
        msg = buildMessageForAdjunctReceivers();
        msg.send(locks.getDistributedLockId());

        // Fire callbacks collected in the local txApply* executions
        firePendingCallbacks();

        commitMessage = buildCompleteMessage();

      } finally {
        if (msg != null) {
          msg.releaseViewVersions();
        }
        locks.releaseLocal();
        // For internal testing
        if (internalAfterReleaseLocalLocks != null) {
          internalAfterReleaseLocalLocks.run();
        }
      }
    } finally {
      cleanup();
    }
  }

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

      // only on the primary
      if (r.isUsedForPartitionedRegionBucket() && !txrs.isCreatedDuringCommit()) {
        txrs.buildMessageForAdjunctReceivers(r, msg);
      }
    }
    return msg;
  }


  @Override
  public void rollback() {
    super.rollback();
    // Cleanup is called next
  }

  protected boolean applyOpsOnRedundantCopy(DistributedMember sender,
      ArrayList<DistTxEntryEvent> secondaryTransactionalOperations) {
    boolean returnValue = true;
    try {
      boolean result = true;

      // Start TxState Update During PreCommit phase
      setUpdatingTxStateDuringPreCommit(true);

      if (logger.isDebugEnabled()) {
        logger.debug("DistTXState.applyOpOnRedundantCopy: size of "
            + "secondaryTransactionalOperations = {}", secondaryTransactionalOperations.size());
      }
      /*
       * Handle Put Operations meant for secondary.
       *
       * @see org.apache.geode.internal.cache.partitioned.PutMessage.
       * operateOnPartitionedRegion(DistributionManager, PartitionedRegion, long)
       *
       * [DISTTX] TODO need to handle other operations
       */
      for (DistTxEntryEvent dtop : secondaryTransactionalOperations) {
        if (logger.isDebugEnabled()) {
          logger.debug("DistTXState.applyOpOnRedundantCopy: processing dist " + "tx operation {}",
              dtop);
        }
        dtop.setDistributedMember(sender);
        dtop.setOriginRemote(false);

        /*
         * [DISTTX} TODO handle call back argument version tag and other settings in PutMessage
         */
        String failureReason = null;
        try {
          if (dtop.getRegion() == null) {
            // Tx event from the peer.
            if (dtop.getRegionName() == null) {
              throw new InternalGemFireException("Region is unavailable on DistTxEntryEvent.");
            }
            dtop.setRegion((LocalRegion) getCache().getRegion(dtop.getRegionName()));
          }

          if (dtop.getKeyInfo().isDistKeyInfo()) {
            dtop.getKeyInfo().setCheckPrimary(false);
          } else {
            dtop.setKeyInfo(new DistTxKeyInfo(dtop.getKeyInfo()));
            dtop.getKeyInfo().setCheckPrimary(false);
          }

          // apply the op
          result = applyIndividualOp(dtop);

          if (!result) { // make sure the region hasn't gone away
            dtop.getRegion().checkReadiness();
          }
        } catch (CacheWriterException cwe) {
          result = false;
          failureReason = "CacheWriterException";
        } catch (PrimaryBucketException pbe) {
          result = false;
          failureReason = "PrimaryBucketException";
        } catch (InvalidDeltaException ide) {
          result = false;
          failureReason = "InvalidDeltaException";
        } catch (DataLocationException e) {
          result = false;
          failureReason = "DataLocationException";
        }
        if (logger.isDebugEnabled()) {
          logger.debug("DistTXState.applyOpOnRedundantCopy {} ##op {},  " + "##region {}, ##key {}",
              (result ? " sucessfully applied op " : " failed to apply op due to " + failureReason),
              dtop.getOperation(), dtop.getRegion().getName(), dtop.getKey());
        }
        if (!result) {
          returnValue = false;
          break;
        }
      }
    } finally {
      // End TxState Update During PreCommit phase
      setUpdatingTxStateDuringPreCommit(false);
    }
    return returnValue;
  }

  /**
   * Apply the individual tx op on secondary
   *
   * Calls local function such as putEntry instead of putEntryOnRemote as for this
   * {@link DistTXStateOnCoordinator} as events will always be local. In parent {@link DistTXState}
   * class will call remote version of functions
   *
   */
  protected boolean applyIndividualOp(DistTxEntryEvent dtop) throws DataLocationException {
    boolean result = true;
    if (dtop.op.isUpdate() || dtop.op.isCreate()) {
      if (dtop.op.isPutAll()) {
        assert (dtop.getPutAllOperation() != null);
        // [DISTTX] TODO what do with versions next?
        final VersionedObjectList versions =
            new VersionedObjectList(dtop.getPutAllOperation().putAllDataSize, true,
                dtop.getRegion().getConcurrencyChecksEnabled());
        postPutAll(dtop.getPutAllOperation(), versions, dtop.getRegion());
      } else {
        result = putEntryOnRemote(dtop, false/* ifNew */, false/* ifOld */,
            null/* expectedOldValue */, false/* requireOldValue */, 0L/* lastModified */,
            true/*
                 * overwriteDestroyed *not* used
                 */);
      }
    } else if (dtop.op.isDestroy()) {
      if (dtop.op.isRemoveAll()) {
        assert (dtop.getRemoveAllOperation() != null);
        // [DISTTX] TODO what do with versions next?
        final VersionedObjectList versions =
            new VersionedObjectList(dtop.getRemoveAllOperation().removeAllDataSize, true,
                dtop.getRegion().getConcurrencyChecksEnabled());
        postRemoveAll(dtop.getRemoveAllOperation(), versions, dtop.getRegion());
      } else {
        destroyOnRemote(dtop, false/* TODO [DISTTX] */, null/*
                                                             * TODO [DISTTX]
                                                             */);
      }
    } else if (dtop.op.isInvalidate()) {
      invalidateOnRemote(dtop, true/* TODO [DISTTX] */, false/*
                                                              * TODO [DISTTX]
                                                              */);
    } else {
      logger.debug("DistTXCommitPhaseOneMessage: unsupported TX operation {}", dtop);
      assert (false);
    }
    return result;
  }


  public boolean isUpdatingTxStateDuringPreCommit() {
    return updatingTxStateDuringPreCommit;
  }

  /**
   * For Dist Tx
   *
   * @param updatingTxState if updating TxState during Commit Phase
   */
  private void setUpdatingTxStateDuringPreCommit(boolean updatingTxState)
      throws UnsupportedOperationInTransactionException {
    updatingTxStateDuringPreCommit = updatingTxState;
    if (logger.isDebugEnabled()) {
      logger.debug("DistTXState setUpdatingTxStateDuringPreCommit incoming {} final {} ",
          updatingTxState, updatingTxStateDuringPreCommit);
    }
  }

  @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);
      }
      result.setCreatedDuringCommit(updatingTxStateDuringPreCommit);
      regions.put(r, result);
      if (logger.isDebugEnabled()) {
        logger.debug("DistTXState writeRegion flag {} new region-state {} ",
            updatingTxStateDuringPreCommit, result);
      }
    } else {
      if (logger.isDebugEnabled()) {
        logger.debug("DistTXState writeRegion flag {} region-state {} ",
            updatingTxStateDuringPreCommit, result);
      }
    }

    return result;
  }


  /*
   * [DISTTX] Note: This has been overridden here to associate DistKeyInfo with event to disable
   * primary check(see DistKeyInfo.setCheckPrimary(false)) when this gets called on secondary of a
   * PR
   *
   * For TX this needs to be a PR passed in as region
   *
   *
   * @see org.apache.geode.internal.cache.InternalDataView#postPutAll(org.apache
   * .gemfire.internal.cache.DistributedPutAllOperation, java.util.Map,
   * org.apache.geode.internal.cache.LocalRegion)
   */
  @Override
  public void postPutAll(final DistributedPutAllOperation putallOp,
      final VersionedObjectList successfulPuts, InternalRegion reg) {

    final InternalRegion theRegion;
    if (reg instanceof BucketRegion) {
      theRegion = ((BucketRegion) reg).getPartitionedRegion();
    } else {
      theRegion = reg;
    }
    /*
     * Don't fire events here.
     */
    /*
     * We are on the data store, we don't need to do anything here. Commit will push them out.
     */
    /*
     * We need to put this into the tx state.
     */
    theRegion.syncBulkOp(() -> {
      // final boolean requiresRegionContext =
      // theRegion.keyRequiresRegionContext();
      InternalDistributedMember myId =
          theRegion.getDistributionManager().getDistributionManagerId();
      for (int i = 0; i < putallOp.putAllDataSize; ++i) {
        @Released
        EntryEventImpl ev = PutAllPRMessage.getEventFromEntry(theRegion, myId, myId, i,
            putallOp.putAllData, false, putallOp.getBaseEvent().getContext(), false,
            !putallOp.getBaseEvent().isGenerateCallbacks());
        try {
          // ev.setPutAllOperation(putallOp);

          // below if condition returns true on secondary when TXState is
          // updated in preCommit only on secondary
          // In this case disable the primary check by calling
          // distKeyInfo.setCheckPrimary(false);
          if (isUpdatingTxStateDuringPreCommit()) {
            KeyInfo keyInfo = ev.getKeyInfo();
            DistTxKeyInfo distKeyInfo = new DistTxKeyInfo(keyInfo);
            distKeyInfo.setCheckPrimary(false);
            ev.setKeyInfo(distKeyInfo);
          }
          /*
           * Whenever commit is called, especially when its a DistTxStateOnCoordinator the txState
           * is set to null in @see TXManagerImpl.commit() and thus when @see LocalRegion.basicPut
           * will be called as in this function, they will not found a TxState with call for
           * getDataView()
           */
          if (!(theRegion.getDataView() instanceof TXStateInterface)) {
            if (putEntry(ev, false, false, null, false, 0L, false)) {
              successfulPuts.addKeyAndVersion(putallOp.putAllData[i].key, null);
            }
          } else if (theRegion.basicPut(ev, false, false, null, false)) {
            successfulPuts.addKeyAndVersion(putallOp.putAllData[i].key, null);
          }
        } finally {
          ev.release();
        }
      }
    }, putallOp.getBaseEvent().getEventId());

  }

  @Override
  public void postRemoveAll(final DistributedRemoveAllOperation op,
      final VersionedObjectList successfulOps, InternalRegion reg) {
    final InternalRegion theRegion;
    if (reg instanceof BucketRegion) {
      theRegion = ((BucketRegion) reg).getPartitionedRegion();
    } else {
      theRegion = reg;
    }
    /*
     * Don't fire events here. We are on the data store, we don't need to do anything here. Commit
     * will push them out. We need to put this into the tx state.
     */
    theRegion.syncBulkOp(() -> {
      InternalDistributedMember myId =
          theRegion.getDistributionManager().getDistributionManagerId();
      for (int i = 0; i < op.removeAllDataSize; ++i) {
        @Released
        EntryEventImpl ev = RemoveAllPRMessage.getEventFromEntry(theRegion, myId, myId, i,
            op.removeAllData, false, op.getBaseEvent().getContext(), false,
            !op.getBaseEvent().isGenerateCallbacks());
        try {
          ev.setRemoveAllOperation(op);
          // below if condition returns true on secondary when TXState is
          // updated in preCommit only on secondary
          // In this case disable the primary check by calling
          // distKeyInfo.setCheckPrimary(false);
          if (isUpdatingTxStateDuringPreCommit()) {
            KeyInfo keyInfo = ev.getKeyInfo();
            DistTxKeyInfo distKeyInfo = new DistTxKeyInfo(keyInfo);
            distKeyInfo.setCheckPrimary(false);
            ev.setKeyInfo(distKeyInfo);
          }
          /*
           * Whenever commit is called, especially when its a DistTxStateOnCoordinator the txState
           * is set to null in @see TXManagerImpl.commit() and thus when basicDestroy will be
           * called will be called as in i.e. @see LocalRegion.basicDestroy, they will not found a
           * TxState with call for getDataView()
           *
           * [DISTTX] TODO verify if this is correct to call destroyExistingEntry directly?
           */
          try {
            if (!(theRegion.getDataView() instanceof TXStateInterface)) {
              destroyExistingEntry(ev, true/* should we invoke cacheWriter? */, null);
            } else {
              theRegion.basicDestroy(ev, true/* should we invoke cacheWriter? */, null);
            }
          } catch (EntryNotFoundException ignore) {
          }
          successfulOps.addKeyAndVersion(op.removeAllData[i].key, null);
        } finally {
          ev.release();
        }
      }
    }, op.getBaseEvent().getEventId());

  }

  @Override
  public boolean isDistTx() {
    return true;
  }

  /*
   * Populate list of entry states for each region while replying precommit
   */
  public boolean populateDistTxEntryStateList(
      TreeMap<String, ArrayList<DistTxThinEntryState>> entryStateSortedMap) {
    for (Map.Entry<InternalRegion, TXRegionState> me : regions.entrySet()) {
      InternalRegion r = me.getKey();
      TXRegionState txrs = me.getValue();
      String regionFullPath = r.getFullPath();
      if (!txrs.isCreatedDuringCommit()) {
        ArrayList<DistTxThinEntryState> entryStateList = new ArrayList<>();
        boolean returnValue = txrs.populateDistTxEntryStateList(entryStateList);
        if (returnValue) {
          if (logger.isDebugEnabled()) {
            logger.debug("DistTxState.populateDistTxEntryStateList Adding entries " + " with count="
                + entryStateList.size() + " for region " + regionFullPath + " . Added list="
                + entryStateList);
          }
          entryStateSortedMap.put(regionFullPath, entryStateList);
        } else {
          if (logger.isDebugEnabled()) {
            logger.debug("DistTxState.populateDistTxEntryStateList Got exception for region "
                + regionFullPath);
          }
          return false;
        }
      } else {
        if (logger.isDebugEnabled()) {
          logger.debug("DistTxState.populateDistTxEntryStateList Not adding entries for region "
              + regionFullPath);
        }
      }
    }
    return true;
  }

  /*
   * Set list of entry states for each region while applying commit
   */
  public void setDistTxEntryStates(ArrayList<ArrayList<DistTxThinEntryState>> entryEventList) {
    TreeMap<String, TXRegionState> regionSortedMap = new TreeMap<>();
    for (TXRegionState txrs : regions.values()) {
      if (txrs.isCreatedDuringCommit()) {
        regionSortedMap.put(txrs.getRegion().getFullPath(), txrs);
      }
    }

    int index = 0;
    for (Entry<String, TXRegionState> me : regionSortedMap.entrySet()) {
      String regionFullPath = me.getKey();
      TXRegionState txrs = me.getValue();
      ArrayList<DistTxThinEntryState> entryEvents = entryEventList.get(index++);
      if (logger.isDebugEnabled()) {
        logger.debug("DistTxState.setDistTxEntryStates For region=" + regionFullPath + " ,index="
            + index + " ,entryEvents=(" + entryEvents.size() + ")=" + entryEvents
            + " ,regionSortedMap=" + regionSortedMap.keySet());
      }
      txrs.setDistTxEntryStates(entryEvents);
    }
  }
}