/*
 * 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 static org.apache.geode.internal.cache.LocalRegion.InitializationLevel.BEFORE_INITIAL_IMAGE;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.ArrayList;
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 org.apache.logging.log4j.Logger;

import org.apache.geode.CancelException;
import org.apache.geode.DataSerializer;
import org.apache.geode.SystemFailure;
import org.apache.geode.annotations.Immutable;
import org.apache.geode.annotations.internal.MakeNotStatic;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheClosedException;
import org.apache.geode.cache.CacheRuntimeException;
import org.apache.geode.cache.CommitDistributionException;
import org.apache.geode.cache.CommitIncompleteException;
import org.apache.geode.cache.Operation;
import org.apache.geode.cache.RegionDestroyedException;
import org.apache.geode.cache.RegionDistributionException;
import org.apache.geode.cache.TransactionId;
import org.apache.geode.cache.TransactionListener;
import org.apache.geode.distributed.internal.ClusterDistributionManager;
import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.DistributionMessage;
import org.apache.geode.distributed.internal.MembershipListener;
import org.apache.geode.distributed.internal.MessageWithReply;
import org.apache.geode.distributed.internal.PooledDistributionMessage;
import org.apache.geode.distributed.internal.ReliableReplyProcessor21;
import org.apache.geode.distributed.internal.ReplyException;
import org.apache.geode.distributed.internal.ReplyMessage;
import org.apache.geode.distributed.internal.ReplyProcessor21;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.HeapDataOutputStream;
import org.apache.geode.internal.InternalDataSerializer;
import org.apache.geode.internal.cache.LocalRegion.InitializationLevel;
import org.apache.geode.internal.cache.locks.TXLockId;
import org.apache.geode.internal.cache.locks.TXLockIdImpl;
import org.apache.geode.internal.cache.locks.TXLockService;
import org.apache.geode.internal.cache.partitioned.Bucket;
import org.apache.geode.internal.cache.persistence.PersistentMemberID;
import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID;
import org.apache.geode.internal.cache.versions.VersionSource;
import org.apache.geode.internal.cache.versions.VersionTag;
import org.apache.geode.internal.offheap.annotations.Released;
import org.apache.geode.internal.serialization.DeserializationContext;
import org.apache.geode.internal.serialization.KnownVersion;
import org.apache.geode.internal.serialization.SerializationContext;
import org.apache.geode.internal.serialization.StaticSerialization;
import org.apache.geode.logging.internal.executors.LoggingThread;
import org.apache.geode.logging.internal.log4j.api.LogService;

/**
 * TXCommitMessage is the message that contains all the information that needs to be distributed, on
 * commit, to other cache members.
 *
 * @since GemFire 4.0
 */
public class TXCommitMessage extends PooledDistributionMessage
    implements MembershipListener, MessageWithReply {

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

  // Keep a 60 second history @ an estimated 1092 transactions/second ~= 16^4
  @MakeNotStatic
  protected static final TXFarSideCMTracker txTracker = new TXFarSideCMTracker((60 * 1092));

  private ArrayList<RegionCommit> regions; // list of RegionCommit instances
  protected TXId txIdent;
  protected int processorId; // 0 unless needsAck is true
  protected TXLockIdImpl lockId;
  protected HashSet<InternalDistributedMember> farSiders;
  protected transient DistributionManager dm; // Used on the sending side of this message
  private transient int sequenceNum = 0;

  // Maps receiver Serializables to RegionCommitList instances
  private transient HashMap<InternalDistributedMember, RegionCommitList> msgMap = null;

  private transient RegionCommit currentRegion;
  protected transient TXState txState = null;
  private transient boolean wasProcessed;
  private transient boolean isProcessing;
  private transient boolean dontProcess;
  private transient boolean departureNoticed = false;
  private transient boolean lockNeedsUpdate = false;
  private transient boolean ackRequired = true;
  /**
   * List of operations to do when processing this tx. Valid on farside only.
   */
  protected transient ArrayList<RegionCommit.FarSideEntryOp> farSideEntryOps;
  private byte[] farsideBaseMembershipId; // only available on farside
  private long farsideBaseThreadId; // only available on farside
  private long farsideBaseSequenceId; // only available on farside

  /**
   * (Nearside) true of any regions in this TX have required roles
   */
  private transient boolean hasReliableRegions = false;

  /**
   * Set of all caching exceptions produced while processing this tx
   */
  private transient Set<Exception> processingExceptions = Collections.emptySet();

  private ClientProxyMembershipID bridgeContext = null;

  /**
   * Version of the client that this TXCommitMessage is being sent to. Used for backwards
   * compatibility
   */
  private transient KnownVersion clientVersion;

  /**
   * A token to be put in TXManagerImpl#failoverMap to represent a CommitConflictException while
   * committing a transaction
   */
  @Immutable
  public static final TXCommitMessage CMT_CONFLICT_MSG = new TXCommitMessage();
  /**
   * A token to be put in TXManagerImpl#failoverMap to represent a
   * TransactionDataNodeHasDepartedException
   * while committing a transaction
   */
  @Immutable
  public static final TXCommitMessage REBALANCE_MSG = new TXCommitMessage();
  /**
   * A token to be put in TXManagerImpl#failoverMap to represent an exception while committing a
   * transaction
   */
  @Immutable
  public static final TXCommitMessage EXCEPTION_MSG = new TXCommitMessage();
  /**
   * A token to be put in TXManagerImpl#failoverMap to represent a rolled back transaction
   */
  @Immutable
  public static final TXCommitMessage ROLLBACK_MSG = new TXCommitMessage();

  public TXCommitMessage(TXId txIdent, DistributionManager dm, TXState txState) {
    this.dm = dm;
    this.txIdent = txIdent;
    lockId = null;
    regions = null;
    this.txState = txState;
    wasProcessed = false;
    isProcessing = false;
    dontProcess = false;
    farSiders = null;
    bridgeContext = txState.bridgeContext;
  }

  public TXCommitMessage() {
    // zero arg constructor for DataSerializer
  }

  public static TXFarSideCMTracker getTracker() {
    return TXCommitMessage.txTracker;
  }

  /**
   * Create and return an eventId given its offset.
   *
   * @since GemFire 5.7
   */
  protected EventID getEventId(int eventOffset) {
    return new EventID(farsideBaseMembershipId, farsideBaseThreadId,
        farsideBaseSequenceId + eventOffset);
  }

  /**
   * Return the TXCommitMessage we have already received that is associated with id. Note because of
   * bug 37657 we may need to wait for it to show up.
   */
  public static TXCommitMessage waitForMessage(Object id, DistributionManager dm) {
    TXFarSideCMTracker map = getTracker();
    return map.waitForMessage(id, dm);
  }

  void startRegion(InternalRegion r, int maxSize) {
    currentRegion = new RegionCommit(this, r, maxSize);
    if (r.requiresReliabilityCheck()) {
      hasReliableRegions = true;
    }
  }

  void finishRegion(Set<InternalDistributedMember> s) {
    // make sure we have some changes and someone to send them to
    if (!currentRegion.isEmpty() && s != null && !s.isEmpty()) {
      // Get the persistent ids for the current region and save them
      currentRegion.persistentIds = getPersistentIds(currentRegion.internalRegion);

      if (msgMap == null) {
        msgMap = new HashMap<>();
      }
      {
        RegionCommitList newRCL = null;
        for (final InternalDistributedMember recipient : s) {
          if (!dm.getDistributionManagerIds().contains(recipient)) {
            if (logger.isDebugEnabled()) {
              logger.debug("Skipping member {} due to dist list absence", recipient);
            }
            // skip this member since the dm no longer knows about it
            continue;
          }
          RegionCommitList rcl = msgMap.get(recipient);
          if (rcl == null) {
            if (newRCL == null) {
              rcl = new RegionCommitList();
              rcl.add(currentRegion);
              newRCL = rcl;
            } else {
              rcl = newRCL;
            }
            msgMap.put(recipient, rcl);
          } else if (rcl.get(rcl.size() - 1) != currentRegion) {
            rcl.add(currentRegion);
          }
        }
      }

      // Now deal with each existing recipient that does not care
      // about this region
      for (final Map.Entry<InternalDistributedMember, RegionCommitList> me : msgMap.entrySet()) {
        if (!s.contains(me.getKey())) {
          RegionCommitList rcl = me.getValue();
          RegionCommitList trimmedRcl = rcl.trim(currentRegion);
          if (trimmedRcl != rcl) {
            me.setValue(trimmedRcl);
          }
        }
      }
    }
    currentRegion = null;
  }

  private Map<InternalDistributedMember, PersistentMemberID> getPersistentIds(InternalRegion r) {
    if (r instanceof DistributedRegion) {
      return ((CacheDistributionAdvisee) r).getCacheDistributionAdvisor().advisePersistentMembers();
    } else {
      return Collections.emptyMap();
    }
  }

  void finishRegionComplete() {
    // make sure we have some changes and someone to send them to
    if (!currentRegion.isEmpty()) {
      {
        if (regions == null) {
          regions = new RegionCommitList();
        }
        regions.add(currentRegion);
      }
    }
    currentRegion = null;
  }

  Map<DistributedRegion, Long> viewVersions = new HashMap<>();

  private Boolean needsLargeModCount;

  private transient boolean disableListeners = false;

  /**
   * record CacheDistributionAdvisor.startOperation versions for later cleanup
   */
  protected void addViewVersion(DistributedRegion dr, long version) {
    viewVersions.put(dr, version);
  }

  protected void releaseViewVersions() {
    RuntimeException rte = null;
    for (final Map.Entry<DistributedRegion, Long> e : viewVersions.entrySet()) {
      final DistributedRegion dr = e.getKey();
      final Long viewVersion = e.getValue();
      // need to continue the iteration if one of the regions is destroyed
      // since others may still be okay
      try {
        dr.getDistributionAdvisor().endOperation(viewVersion);
      } catch (RuntimeException ex) {
        rte = ex;
      }
    }
    if (rte != null) {
      throw rte;
    }
  }

  private boolean isEmpty() {
    return msgMap == null || msgMap.isEmpty();
  }

  void addOp(InternalRegion r, Object key, TXEntryState entry) {
    currentRegion.addOp(key, entry);
  }

  void send(TXLockId lockId) {
    if (isEmpty()) {
      if (logger.isDebugEnabled()) {
        logger.debug("empty transaction - nothing to distribute");
      }
      return;
    }
    Assert.assertTrue(txState != null, "Send must have transaction state.");
    this.lockId = (TXLockIdImpl) lockId;
    updateLockMembers();

    // Map of RegionCommitList keys to Sets of receivers
    IdentityHashMap<RegionCommitList, Set<InternalDistributedMember>> distMap =
        new IdentityHashMap<>();
    HashSet<InternalDistributedMember> ackReceivers = null;
    for (final Map.Entry<InternalDistributedMember, RegionCommitList> entry : msgMap.entrySet()) {
      final RegionCommitList rcl = entry.getValue();
      if (rcl.getNeedsAck()) {
        if (ackReceivers == null) {
          ackReceivers = new HashSet<>();
        }
        ackReceivers.add(entry.getKey());
      }
      final Set<InternalDistributedMember> receivers =
          distMap.computeIfAbsent(rcl, k -> new HashSet<>());
      receivers.add(entry.getKey());
    }

    CommitReplyProcessor processor = null;
    {
      if (ackReceivers != null) {
        processor = new CommitReplyProcessor(dm, ackReceivers, msgMap);
        if (ackReceivers.size() > 1) {
          farSiders = ackReceivers;
        }
        processor.enableSevereAlertProcessing();
      }
      {
        for (final Map.Entry<RegionCommitList, Set<InternalDistributedMember>> me : distMap
            .entrySet()) {
          RegionCommitList rcl = me.getKey();
          Set<InternalDistributedMember> recipients = me.getValue();
          // now remove from the recipients any recipients that the dm no
          // longer knows about
          recipients.retainAll(dm.getDistributionManagerIds());
          if (!recipients.isEmpty()) {
            if (txState.internalDuringIndividualSend != null) {
              // Run in test mode, splitting out individual recipients,
              // so we can control who gets what
              for (final InternalDistributedMember recipient : recipients) {
                txState.internalDuringIndividualSend.run();
                setRecipientsSendData(Collections.singleton(recipient), processor, rcl);
              }
            } else {
              // Run in normal mode sending to multiple recipients in one shot
              setRecipientsSendData(recipients, processor, rcl);
            }
          }
        }
      }
    }
    if (txState.internalAfterIndividualSend != null) {
      txState.internalAfterIndividualSend.run();
    }

    if (processor != null) {
      // Send the CommitProcessMessage
      final CommitProcessMessage cpMsg;
      if (this.lockId != null) {
        cpMsg = new CommitProcessForLockIdMessage(this.lockId);
      } else {
        cpMsg = new CommitProcessForTXIdMessage(txIdent);
      }
      if (txState.internalDuringIndividualCommitProcess != null) {
        // Run in test mode
        for (final InternalDistributedMember ackReceiver : ackReceivers) {
          txState.internalDuringIndividualCommitProcess.run();
          cpMsg.setRecipients(Collections.singleton(ackReceiver));
          dm.putOutgoing(cpMsg);
          cpMsg.resetRecipients();
        }
      } else {
        // Run in normal mode
        cpMsg.setRecipients(ackReceivers);
        dm.putOutgoing(cpMsg);
      }

      if (txState.internalAfterIndividualCommitProcess != null) {
        // Testing callback
        txState.internalAfterIndividualCommitProcess.run();
      }

      // for() loop removed for bug 36983 - you can't loop on waitForReplies()
      dm.getCancelCriterion().checkCancelInProgress(null);
      processor.waitForCommitCompletion();
      dm.getStats().incCommitWaits();
    }
    if (hasReliableRegions) {
      checkDistributionReliability(distMap, processor);
    }
  }

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

  /**
   * Checks reliable regions and throws CommitDistributionException if any required roles may not
   * have received the commit message.
   *
   * @param distMap map of RegionCommitList keys to Sets of receivers
   * @param processor the reply processor
   * @throws CommitDistributionException if any required roles may not have received the commit
   *         message
   */
  private void checkDistributionReliability(
      IdentityHashMap<RegionCommitList, Set<InternalDistributedMember>> distMap,
      CommitReplyProcessor processor) {
    // key=RegionCommit, value=Set of recipients
    Map<RegionCommit, Set<InternalDistributedMember>> regionToRecipients = new IdentityHashMap<>();

    // build up the keys in regionToRecipients and add all receivers
    for (final Map.Entry<RegionCommitList, Set<InternalDistributedMember>> me : distMap
        .entrySet()) {
      RegionCommitList rcl = me.getKey();
      Set<InternalDistributedMember> recipients = me.getValue();

      for (RegionCommit rc : rcl) {
        // skip region if no required roles
        if (!rc.internalRegion.requiresReliabilityCheck()) {
          continue;
        }

        final Set<InternalDistributedMember> recipientsForRegion =
            regionToRecipients.computeIfAbsent(rc, k -> new HashSet<>());
        // get the receiver Set for rcl and perform addAll
        if (recipients != null) {
          recipientsForRegion.addAll(recipients);
        }
      }
    }

    Set<InternalDistributedMember> cacheClosedMembers =
        (processor == null) ? Collections.emptySet() : processor.getCacheClosedMembers();
    Set<InternalDistributedMember> departedMembers =
        (processor == null) ? Collections.emptySet() : processor.getDepartedMembers();

    // check reliability on each region
    @SuppressWarnings("deprecation")
    Set<RegionDistributionException> regionDistributionExceptions = Collections.emptySet();
    Set<String> failedRegionNames = Collections.emptySet();
    for (final Map.Entry<RegionCommit, Set<InternalDistributedMember>> me : regionToRecipients
        .entrySet()) {
      final RegionCommit rc = me.getKey();
      final Set<InternalDistributedMember> successfulRecipients = new HashSet<>(msgMap.keySet());
      successfulRecipients.removeAll(departedMembers);

      // remove members who destroyed that region or closed their cache
      Set<InternalDistributedMember> regionDestroyedMembers =
          (processor == null) ? Collections.emptySet()
              : processor.getRegionDestroyedMembers(rc.internalRegion.getFullPath());

      successfulRecipients.removeAll(cacheClosedMembers);
      successfulRecipients.removeAll(regionDestroyedMembers);

      try {
        rc.internalRegion.handleReliableDistribution(successfulRecipients);
      } catch (@SuppressWarnings("deprecation") RegionDistributionException e) {
        if (regionDistributionExceptions.isEmpty()) {
          regionDistributionExceptions = new HashSet<>();
          failedRegionNames = new HashSet<>();
        }
        regionDistributionExceptions.add(e);
        failedRegionNames.add(rc.internalRegion.getFullPath());
      }
    }

    if (!regionDistributionExceptions.isEmpty()) {
      throw new CommitDistributionException(
          String.format(
              "These regions experienced reliability failure during distribution of the operation: %s",
              failedRegionNames),
          regionDistributionExceptions);
    }
  }

  /**
   * Helper method for send
   */
  private void setRecipientsSendData(Set<InternalDistributedMember> recipients,
      ReplyProcessor21 processor,
      RegionCommitList rcl) {
    setRecipients(recipients);
    regions = rcl;
    if (rcl.getNeedsAck()) {
      processorId = processor.getProcessorId();
    } else {
      processorId = 0;
    }
    dm.getStats().incSentCommitMessages(1L);
    sequenceNum++;
    dm.putOutgoing(this);
    resetRecipients();
  }

  @Override
  protected void process(ClusterDistributionManager dm) {
    this.dm = dm;
    // Remove this node from the set of recipients
    if (farSiders != null) {
      farSiders.remove(dm.getId());
    }

    if (processorId != 0) {
      TXLockService.createDTLS(this.dm.getSystem()); // fix bug 38843; no-op if already created
      synchronized (this) {
        // Handle potential origin departure
        this.dm.addMembershipListener(this);
        // Assume ACK mode, defer processing until we receive a
        // CommitProcess message
        if (logger.isDebugEnabled()) {
          final Object key = getTrackerKey();
          logger.debug("Adding key:{} class{} to tracker list", key, key.getClass().getName());
        }
        txTracker.add(this);
      }
      if (!this.dm.getDistributionManagerIds().contains(getSender())) {
        memberDeparted(this.dm, getSender(), false /* don't care */);
      }

    } else {
      basicProcess();
    }
  }

  /**
   * Adds an entry op for this tx to do on the far side
   */
  void addFarSideEntryOp(RegionCommit.FarSideEntryOp entryOp) {
    farSideEntryOps.add(entryOp);
  }

  protected void addProcessingException(Exception e) {
    // clear all previous exceptions if e is a CacheClosedException
    if (processingExceptions.isEmpty() || e instanceof CancelException) {
      processingExceptions = new HashSet<>();
    }
    processingExceptions.add(e);
  }

  public void setDM(DistributionManager dm) {
    this.dm = dm;
  }

  public void basicProcess() {
    final DistributionManager dm = this.dm;

    synchronized (this) {
      if (isProcessing()) {
        if (logger.isDebugEnabled()) {
          logger.debug("TXCommitMessage {} is already in process, returning", this);
        }
        return;
      } else {
        setIsProcessing(true);
      }
    }

    if (logger.isDebugEnabled()) {
      logger.debug("begin processing TXCommitMessage for {}", txIdent);
    }
    final InitializationLevel oldLevel =
        LocalRegion.setThreadInitLevelRequirement(BEFORE_INITIAL_IMAGE);
    boolean forceListener = false; // this gets flipped if we need to fire tx listener
    // it needs to default to false because we don't want to fire listeners on pr replicates
    try {
      TXRmtEvent txEvent = null;
      final Cache cache = dm.getExistingCache();
      if (cache == null) {
        addProcessingException(new CacheClosedException());
        // return ... this cache is closed so we can't do anything.
        return;
      }
      final TransactionListener[] tls = cache.getCacheTransactionManager().getListeners();
      if (tls.length > 0) {
        txEvent = new TXRmtEvent(txIdent, cache);
      }
      try {
        // Pre-process each Region in the tx
        try {
          Iterator<RegionCommit> it = regions.iterator();
          while (it.hasNext()) {
            boolean failedBeginProcess = true;
            RegionCommit rc = it.next();
            try {
              failedBeginProcess = !rc.beginProcess(dm, txIdent, txEvent);
            } catch (CacheRuntimeException problem) {
              processCacheRuntimeException(problem);
            } finally {
              if (failedBeginProcess) {
                rc.internalRegion = null; // Cause related FarSideEntryOps to skip processing
                it.remove(); // Skip endProcessing as well
              }
            }
          }
          basicProcessOps();
        } finally {
          // post-process each Region in the tx
          Iterator<RegionCommit> it = regions.iterator();
          while (it.hasNext()) {
            try {
              RegionCommit rc = it.next();
              rc.endProcess();
              if (rc.isForceFireEvent(dm)) {
                forceListener = true;
              }
            } catch (CacheRuntimeException problem) {
              processCacheRuntimeException(problem);
            }
          }
        }

        /*
         * We need to make sure that we should fire a TX afterCommit event.
         */
        boolean internalEvent = (txEvent != null && txEvent.hasOnlyInternalEvents());
        if (!disableListeners && !internalEvent
            && (forceListener || (txEvent != null && !txEvent.isEmpty()))) {
          for (final TransactionListener tl : tls) {
            try {
              tl.afterCommit(txEvent);
            } 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);
            }
          }
        }
      } catch (CancelException e) {
        processCacheRuntimeException(e);
      } finally {
        if (txEvent != null) {
          txEvent.freeOffHeapResources();
        }
      }
    } finally {
      LocalRegion.setThreadInitLevelRequirement(oldLevel);
      if (isAckRequired()) {
        ack();
      }
      if (!dm.getExistingCache().isClient() && bridgeContext != null) {
        getTracker().saveTXForClientFailover(txIdent, this);
      }
      if (logger.isDebugEnabled()) {
        logger.debug("completed processing TXCommitMessage for {}", txIdent);
      }
    }
  }

  public void basicProcessOps() {
    List<EntryEventImpl> pendingCallbacks = new ArrayList<>(farSideEntryOps.size());
    farSideEntryOps.sort(null);
    Iterator<RegionCommit.FarSideEntryOp> it = farSideEntryOps.iterator();
    while (it.hasNext()) {
      try {
        RegionCommit.FarSideEntryOp entryOp = it.next();
        entryOp.process(pendingCallbacks);
      } catch (CacheRuntimeException problem) {
        processCacheRuntimeException(problem);
      } catch (Exception e) {
        addProcessingException(e);
      }
    }
    firePendingCallbacks(pendingCallbacks);
  }

  void firePendingCallbacks(List<EntryEventImpl> callbacks) {
    boolean isConfigError = false;
    EntryEventImpl lastTransactionEvent = null;
    try {
      lastTransactionEvent = getLastTransactionEvent(callbacks);
    } catch (ServiceConfigurationError ex) {
      logger.error(ex.getMessage());
      isConfigError = true;
    }

    for (EntryEventImpl ee : callbacks) {
      boolean isLastTransactionEvent = TXLastEventInTransactionUtils
          .isLastTransactionEvent(isConfigError, lastTransactionEvent, ee);
      try {
        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 {
          if (!ee.hasNewValue()) { // GEODE-8964, fixes GII and TX create conflict that
            ee.getRegion(). // produces an Update with null value
                invokeTXCallbacks(EnumListenerEvent.AFTER_CREATE, ee, true, isLastTransactionEvent);
          } else {
            ee.getRegion().invokeTXCallbacks(EnumListenerEvent.AFTER_UPDATE, ee, true,
                isLastTransactionEvent);
          }
        }
      } finally {
        ee.release();
      }
    }
  }

  EntryEventImpl getLastTransactionEvent(List<EntryEventImpl> callbacks) {
    return TXLastEventInTransactionUtils.getLastTransactionEventInGroupedTxForWANSender(callbacks,
        dm.getCache());
  }

  protected void processCacheRuntimeException(CacheRuntimeException problem) {
    if (problem instanceof RegionDestroyedException) { // catch RegionDestroyedException
      addProcessingException(problem);
    } else if (problem instanceof CancelException) { // catch CacheClosedException
      addProcessingException(problem);
      throw problem;
    } else { // catch CacheRuntimeException
      addProcessingException(problem);
      logger.error(
          "Transaction message {} from sender {} failed processing, unknown transaction state: {}",
          new Object[] {this, getSender(), problem});
    }
  }

  private void ack() {
    if (processorId != 0) {
      CommitReplyException replyEx = null;
      if (!processingExceptions.isEmpty()) {
        replyEx = new CommitReplyException(
            String.format("Commit operation generated one or more exceptions from %s",
                getSender()),
            processingExceptions);
      }
      ReplyMessage.send(getSender(), processorId, replyEx, dm);
    }
  }

  @Override
  public int getDSFID() {
    return TX_COMMIT_MESSAGE;
  }

  @Override
  public void fromData(DataInput in, DeserializationContext context)
      throws IOException, ClassNotFoundException {
    int pId = in.readInt();

    if (isAckRequired()) {
      processorId = pId;
      ReplyProcessor21.setMessageRPId(processorId);
    } else {
      processorId = -1;
    }

    txIdent = TXId.createFromData(in);
    if (in.readBoolean()) {
      lockId = TXLockIdImpl.createFromData(in);
    }
    int totalMaxSize = in.readInt();

    farsideBaseMembershipId = DataSerializer.readByteArray(in);
    farsideBaseThreadId = in.readLong();
    farsideBaseSequenceId = in.readLong();

    needsLargeModCount = in.readBoolean();

    final boolean hasShadowKeys = hasFlagsField(in) ? in.readBoolean() : useShadowKey();

    int regionsSize = in.readInt();
    regions = new ArrayList<>(regionsSize);
    farSideEntryOps = new ArrayList<>(totalMaxSize);
    for (int i = 0; i < regionsSize; i++) {
      RegionCommit rc = new RegionCommit(this);
      try {
        rc.fromData(in, hasShadowKeys);
      } catch (CacheClosedException cce) {
        addProcessingException(cce);
        // return to avoid serialization error being sent in reply
        return;
      }
      regions.add(rc);
    }

    bridgeContext = ClientProxyMembershipID.readCanonicalized(in);
    farSiders = DataSerializer.readHashSet(in);
  }

  /**
   * Return true if a distributed ack message is required. On the client side of a transaction, this
   * returns false, while returning true elsewhere.
   *
   * @return requires ack message or not
   */
  private boolean isAckRequired() {
    return ackRequired;
  }


  /**
   * Indicate whether an ack is required. Defaults to true.
   *
   * @param a true if we require an ack. false if not. false on clients.
   */
  public void setAckRequired(boolean a) {
    ackRequired = a;
    if (!a) {
      processorId = -1;
    }
  }

  @Override
  public void toData(DataOutput out, SerializationContext context) throws IOException {
    out.writeInt(processorId);
    InternalDataSerializer.invokeToData(txIdent, out);
    {
      boolean hasLockId = lockId != null;
      out.writeBoolean(hasLockId);
      if (hasLockId) {
        InternalDataSerializer.invokeToData(lockId, out);
      }
    }
    int regionsSize = 0;
    {
      int totalMaxSize = 0;
      if (regions != null) {
        regionsSize = regions.size();
        for (RegionCommit region : regions) {
          totalMaxSize += region.maxSize;
        }
      }
      out.writeInt(totalMaxSize);
    }

    if (txState != null) {
      DataSerializer.writeByteArray(txState.getBaseMembershipId(), out);
      out.writeLong(txState.getBaseThreadId());
      out.writeLong(txState.getBaseSequenceId());
    } else {
      DataSerializer.writeByteArray(farsideBaseMembershipId, out);
      out.writeLong(farsideBaseThreadId);
      out.writeLong(farsideBaseSequenceId);
    }

    if (txState != null) {
      DataSerializer.writeBoolean(txState.needsLargeModCount(), out);
    } else {
      DataSerializer.writeBoolean(needsLargeModCount, out);
    }

    final boolean useShadowKey = useShadowKey();
    if (hasFlagsField(out)) {
      out.writeBoolean(useShadowKey);
    }

    out.writeInt(regionsSize);
    {
      if (regionsSize > 0) {
        for (RegionCommit region : regions) {
          region.toData(out, context, useShadowKey);
        }
      }
    }

    DataSerializer.writeObject(bridgeContext, out);

    DataSerializer.writeHashSet(farSiders, out);
  }

  private boolean hasFlagsField(final DataOutput out) {
    return hasFlagsField(StaticSerialization.getVersionForDataStream(out));
  }

