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

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.Collection;
import java.util.Set;

import org.apache.logging.log4j.Logger;

import org.apache.geode.CancelException;
import org.apache.geode.DataSerializer;
import org.apache.geode.InternalGemFireError;
import org.apache.geode.SystemFailure;
import org.apache.geode.annotations.Immutable;
import org.apache.geode.cache.CacheClosedException;
import org.apache.geode.cache.CacheException;
import org.apache.geode.cache.DiskAccessException;
import org.apache.geode.cache.LowMemoryException;
import org.apache.geode.cache.Operation;
import org.apache.geode.cache.RegionDestroyedException;
import org.apache.geode.cache.TransactionException;
import org.apache.geode.cache.query.QueryException;
import org.apache.geode.cache.query.RegionNotFoundException;
import org.apache.geode.distributed.DistributedSystemDisconnectedException;
import org.apache.geode.distributed.internal.ClusterDistributionManager;
import org.apache.geode.distributed.internal.DirectReplyProcessor;
import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.DistributionMessage;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.MessageWithReply;
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.InternalDataSerializer;
import org.apache.geode.internal.Version;
import org.apache.geode.internal.cache.DataLocationException;
import org.apache.geode.internal.cache.EntryEventImpl;
import org.apache.geode.internal.cache.FilterRoutingInfo;
import org.apache.geode.internal.cache.ForceReattemptException;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.PartitionedRegion;
import org.apache.geode.internal.cache.PartitionedRegionException;
import org.apache.geode.internal.cache.PrimaryBucketException;
import org.apache.geode.internal.cache.TXManagerImpl;
import org.apache.geode.internal.cache.TXStateProxy;
import org.apache.geode.internal.cache.TransactionMessage;
import org.apache.geode.internal.logging.LogService;
import org.apache.geode.internal.logging.log4j.LogMarker;
import org.apache.geode.internal.sequencelog.EntryLogger;

/**
 * The base PartitionedRegion message type upon which other messages should be based.
 *
 * @since GemFire 5.0
 */
public abstract class PartitionMessage extends DistributionMessage
    implements MessageWithReply, TransactionMessage {
  private static final Logger logger = LogService.getLogger();

  /** default exception to ensure a false-positive response is never returned */
  @Immutable
  static final ForceReattemptException UNHANDLED_EXCEPTION =
      (ForceReattemptException) new ForceReattemptException(
          "Unknown exception")
              .fillInStackTrace();

  int regionId;

  int processorId;

  /**
   * whether this message is being sent for listener notification
   */
  boolean notificationOnly;

  protected short flags = 0;

  /*
   * these bit masks are used for encoding the bits of a short on the wire instead of transmitting
   * booleans. Any subclasses interested in saving bits on the wire should add a mask here and then
   * override computeCompressedShort and setBooleans
   *
   */
  /** flag to indicate notification message */
  protected static final short NOTIFICATION_ONLY = DistributionMessage.UNRESERVED_FLAGS_START;
  /** flag to indicate ifNew in PutMessages */
  protected static final short IF_NEW = (NOTIFICATION_ONLY << 1);
  /** flag to indicate ifOld in PutMessages */
  protected static final short IF_OLD = (IF_NEW << 1);
  /** flag to indicate that oldValue is required for PutMessages and others */
  protected static final short REQUIRED_OLD_VAL = (IF_OLD << 1);
  /** flag to indicate filterInfo in message */
  protected static final short HAS_FILTER_INFO = (REQUIRED_OLD_VAL << 1);
  /** flag to indicate delta as value in message */
  protected static final short HAS_DELTA = (HAS_FILTER_INFO << 1);
  /** the unreserved flags start for child classes */
  protected static final short UNRESERVED_FLAGS_START = (HAS_DELTA << 1);

  private InternalDistributedMember txMemberId = null;

  /**
   * The unique transaction Id on the sending member, used to construct a TXId on the receiving side
   */
  private int txUniqId = TXManagerImpl.NOTX;

  protected boolean sendDeltaWithFullValue = true;

  /* TODO [DISTTX] Convert into flag */
  protected boolean isTransactionDistributed = false;

  public PartitionMessage() {}


  public PartitionMessage(InternalDistributedMember recipient, int regionId,
      ReplyProcessor21 processor) {
    Assert.assertTrue(recipient != null, "PartitionMesssage recipient can not be null");
    setRecipient(recipient);
    this.regionId = regionId;
    this.processorId = processor == null ? 0 : processor.getProcessorId();
    if (processor != null && isSevereAlertCompatible()) {
      processor.enableSevereAlertProcessing();
    }
    initTxMemberId();
  }

  public PartitionMessage(Collection<InternalDistributedMember> recipients, int regionId,
      ReplyProcessor21 processor) {
    setRecipients(recipients);
    this.regionId = regionId;
    this.processorId = processor == null ? 0 : processor.getProcessorId();
    if (processor != null && isSevereAlertCompatible()) {
      processor.enableSevereAlertProcessing();
    }
    initTxMemberId();
  }


  public void initTxMemberId() {
    this.txUniqId = TXManagerImpl.getCurrentTXUniqueId();
    TXStateProxy txState = TXManagerImpl.getCurrentTXState();
    if (txState != null) {
      // [DISTTX] Lets not throw this exception for Dist Tx
      if (canStartRemoteTransaction() && txState.isRealDealLocal() && !txState.isDistTx()) {
        // logger.error("sending rmt txId even though tx is local! txState=" + txState, new
        // RuntimeException("STACK");
        throw new IllegalStateException(
            "Sending remote txId even though transaction is local. This should never happen: txState="
                + txState);
      }
      txMemberId = txState.getOriginatingMember();
    }
  }

  /**
   * Copy constructor that initializes the fields declared in this class
   */
  public PartitionMessage(PartitionMessage other) {
    this.regionId = other.regionId;
    this.processorId = other.processorId;
    this.notificationOnly = other.notificationOnly;
    this.txUniqId = other.getTXUniqId();
    this.txMemberId = other.getTXOriginatorClient();
    this.isTransactionDistributed = other.isTransactionDistributed;
  }

  @Override
  public InternalDistributedMember getTXOriginatorClient() {
    return txMemberId;
  }

  @Override
  public InternalDistributedMember getMemberToMasqueradeAs() {
    if (txMemberId == null) {
      return getSender();
    }
    return txMemberId;
  }

  /**
   * Severe alert processing enables suspect processing at the ack-wait-threshold and issuing of a
   * severe alert at the end of the ack-severe-alert-threshold. Some messages should not support
   * this type of processing (e.g., GII, or DLockRequests)
   *
   * @return whether severe-alert processing may be performed on behalf of this message
   */
  @Override
  public boolean isSevereAlertCompatible() {
    return true;
  }

  @Override
  public int getProcessorType() {
    if (this.notificationOnly) {
      return ClusterDistributionManager.SERIAL_EXECUTOR;
    } else {
      return ClusterDistributionManager.PARTITIONED_REGION_EXECUTOR;
    }
  }

  /**
   * @return the compact value that will be sent which represents the PartitionedRegion
   * @see PartitionedRegion#getPRId()
   */
  public int getRegionId() {
    return regionId;
  }

  /**
   * @return the {@link ReplyProcessor21}id associated with the message, null if no acknowlegement
   *         is required.
   */
  @Override
  public int getProcessorId() {
    return this.processorId;
  }

  /**
   * @param processorId1 the {@link org.apache.geode.distributed.internal.ReplyProcessor21} id
   *        associated with the message, null if no acknowlegement is required.
   */
  public void registerProcessor(int processorId1) {
    this.processorId = processorId1;
  }

  /**
   * @return return the message that should be sent to listeners, or null if this message should not
   *         be relayed
   */
  public PartitionMessage getMessageForRelayToListeners(EntryEventImpl event, Set recipients) {
    return null;
  }

  /**
   * check to see if the cache is closing
   */
  public boolean checkCacheClosing(ClusterDistributionManager dm) {
    if (dm == null) {
      return true;
    }
    InternalCache cache = dm.getCache();
    return cache == null || cache.isClosed();
  }

  /**
   * check to see if the distributed system is closing
   *
   * @return true if the distributed system is closing
   */
  public boolean checkDSClosing(ClusterDistributionManager dm) {
    InternalDistributedSystem ds = dm.getSystem();
    return (ds == null || ds.isDisconnecting());
  }

  PartitionedRegion getPartitionedRegion() throws PRLocallyDestroyedException {
    return PartitionedRegion.getPRFromId(this.regionId);
  }

  TXManagerImpl getTXManagerImpl(InternalCache cache) {
    return cache.getTxManager();
  }

  long getStartPartitionMessageProcessingTime(PartitionedRegion pr) {
    return pr.getPrStats().startPartitionMessageProcessing();
  }


  /**
   * Upon receipt of the message, both process the message and send an acknowledgement, not
   * necessarily in that order. Note: Any hang in this message may cause a distributed deadlock for
   * those threads waiting for an acknowledgement.
   *
   * @throws PartitionedRegionException if the region does not exist (typically, if it has been
   *         destroyed)
   */
  @Override
  public void process(final ClusterDistributionManager dm) {
    Throwable thr = null;
    boolean sendReply = true;
    PartitionedRegion pr = null;
    long startTime = 0;
    EntryLogger.setSource(getSender(), "PR");
    try {
      InternalCache cache = dm.getCache();
      if (checkCacheClosing(dm) || checkDSClosing(dm)) {
        if (cache != null) {
          thr = cache
              .getCacheClosedException(String.format("Remote cache is closed: %s",
                  dm.getId()));
        } else {
          thr = new CacheClosedException(String.format("Remote cache is closed: %s",
              dm.getId()));
        }
        return;
      }
      pr = getPartitionedRegion();
      Throwable forcedReattempt = processCheckForPR(pr, dm);
      if (forcedReattempt != null) {
        thr = forcedReattempt;
        return;
      }

      if (pr != null) {
        startTime = getStartPartitionMessageProcessingTime(pr);
      }
      thr = UNHANDLED_EXCEPTION;

      if (cache == null) {
        throw new ForceReattemptException("Remote cache is closed");
      }
      TXManagerImpl txMgr = getTXManagerImpl(cache);
      TXStateProxy tx = txMgr.masqueradeAs(this);
      if (tx == null) {
        sendReply = operateOnPartitionedRegion(dm, pr, startTime);
      } else {
        try {
          if (txMgr.isClosed()) {
            // NO DISTRIBUTED MESSAGING CAN BE DONE HERE!
            sendReply = false;
          } else if (tx.isInProgress()) {
            sendReply = operateOnPartitionedRegion(dm, pr, startTime);
            tx.updateProxyServer(this.getSender());
          } else {
            /*
             * This can occur when processing an in-flight message after the transaction has
             * been failed over and committed.
             */
            throw new TransactionException("transactional operation elided because transaction {"
                + tx.getTxId() + "} is closed");
          }
        } finally {
          txMgr.unmasquerade(tx);
        }
      }
      thr = null;

    } catch (ForceReattemptException | TransactionException e) {
      thr = e;
    } catch (DataLocationException fre) {
      thr = new ForceReattemptException(fre.getMessage(), fre);
    } catch (DistributedSystemDisconnectedException se) {
      // bug 37026: this is too noisy...
      // throw new CacheClosedException("remote system shutting down");
      // thr = se; cache is closed, no point trying to send a reply
      thr = null;
      sendReply = false;
      if (logger.isDebugEnabled()) {
        logger.debug("shutdown caught, abandoning message: {}", se.getMessage(), se);
      }
    } catch (RegionDestroyedException | RegionNotFoundException rde) {
      // [bruce] RDE does not always mean that the sender's region is also
      // destroyed, so we must send back an exception. If the sender's
      // region is also destroyed, who cares if we send it an exception
      // if (pr != null && pr.isClosed) {
      thr = new ForceReattemptException(String.format("Region is destroyed in %s",
          dm.getDistributionManagerId()), rde);
      // }
    } 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();
      // log the exception at fine level if there is no reply to the message
      thr = null;
      if (sendReply) {
        if (!checkDSClosing(dm)) {
          thr = t;
        } else {
          // don't pass arbitrary runtime exceptions and errors back if this
          // cache/vm is closing
          thr = new ForceReattemptException(
              "Distributed system is disconnecting");
        }
      }
      if (logger.isTraceEnabled(LogMarker.DM_VERBOSE) && t instanceof RuntimeException) {
        logger.trace(LogMarker.DM_VERBOSE, "Exception caught while processing message: {}",
            t.getMessage(), t);
      }
    } finally {
      if (sendReply) {
        ReplyException rex = null;

        if (thr != null) {
          // don't transmit the exception if this message was to a listener
          // and this listener is shutting down
          boolean excludeException = this.notificationOnly
              && ((thr instanceof CancelException) || (thr instanceof ForceReattemptException));

          if (!excludeException) {
            rex = new ReplyException(thr);
          }
        }

        // Send the reply if the operateOnPartitionedRegion returned true
        sendReply(getSender(), this.processorId, dm, rex, pr, startTime);
        EntryLogger.clearSource();
      }
    }
  }

  /**
   * If the PR is missing or isn't ready for use we may want to return a
   * ForceReattemptException to have the sender retry after a bit
   */
  protected Throwable processCheckForPR(PartitionedRegion pr,
      DistributionManager distributionManager) {
    if ((pr == null || !pr.getDistributionAdvisor().isInitialized()) && failIfRegionMissing()) {
      // if the distributed system is disconnecting, don't send a reply saying
      // the partitioned region can't be found (bug 36585)
      Throwable thr = new ForceReattemptException(
          String.format("%s : could not find partitioned region with Id %s",
              distributionManager.getDistributionManagerId(), regionId));
      return thr; // reply sent in finally block below
    }
    return null;
  }

  /**
   * Send a generic ReplyMessage. This is in a method so that subclasses can override the reply
   * message type
   *
   * @param pr the Partitioned Region for the message whose statistics are incremented
   * @param startTime the start time of the operation in nanoseconds
   * @see PutMessage#sendReply
   */
  protected void sendReply(InternalDistributedMember member, int procId, DistributionManager dm,
      ReplyException ex, PartitionedRegion pr, long startTime) {
    if (pr != null && startTime > 0) {
      pr.getPrStats().endPartitionMessagesProcessing(startTime);
    }

    ReplyMessage.send(member, procId, ex, getReplySender(dm), pr != null && pr.isInternalRegion());
  }

  /**
   * Allow classes that over-ride to choose whether a RegionDestroyException is thrown if no
   * partitioned region is found (typically occurs if the message will be sent before the
   * PartitionedRegion has been fully constructed.
   *
   * @return true if throwing a {@link RegionDestroyedException} is acceptable
   */
  protected boolean failIfRegionMissing() {
    return true;
  }

  /**
   * relay this message to another set of recipients for event notification
   *
   * @param cacheOpRecipients recipients of associated bucket CacheOperationMessage
   * @param adjunctRecipients recipients who unconditionally get the message
   * @param filterRoutingInfo routing information for all recipients
   * @param event the event causing this message
   * @param r the region being operated on
   * @param processor the reply processor to be notified
   */
  public Set relayToListeners(Set cacheOpRecipients, Set adjunctRecipients,
      FilterRoutingInfo filterRoutingInfo, EntryEventImpl event, PartitionedRegion r,
      DirectReplyProcessor processor) {
    this.processorId = processor == null ? 0 : processor.getProcessorId();
    this.notificationOnly = true;

    this.setFilterInfo(filterRoutingInfo);
    Set failures1 = null;
    if (!adjunctRecipients.isEmpty()) {
      if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) {
        logger.trace(LogMarker.DM_VERBOSE,
            "Relaying partition message to other processes for listener notification");
      }
      resetRecipients();
      setRecipients(adjunctRecipients);
      failures1 = r.getDistributionManager().putOutgoing(this);
    }

    return failures1;
  }

  /**
   * return a new reply processor for this class, for use in relaying a response. This <b>must</b>
   * be an instance method so subclasses can override it properly.
   */
  PartitionResponse createReplyProcessor(PartitionedRegion r, Set recipients) {
    return new PartitionResponse(r.getSystem(), recipients);
  }


  protected boolean operateOnRegion(ClusterDistributionManager dm, PartitionedRegion pr) {
    throw new InternalGemFireError(
        "Sorry, use operateOnPartitionedRegion for PR messages");
  }

  /**
   * An operation upon the messages partitioned region which each subclassing message must implement
   *
   * @param dm the manager that received the message
   * @param pr the partitioned region that should be modified
   * @param startTime the start time of the operation
   * @return true if a reply message should be sent
   * @throws CacheException if an error is generated in the remote cache
   * @throws DataLocationException if the peer is no longer available
   */
  protected abstract boolean operateOnPartitionedRegion(ClusterDistributionManager dm,
      PartitionedRegion pr, long startTime) throws CacheException, QueryException,
      DataLocationException, InterruptedException, IOException;

  /**
   * Fill out this instance of the message using the <code>DataInput</code> Required to be a
   * {@link org.apache.geode.DataSerializable}Note: must be symmetric with
   * {@link #toData(DataOutput)}in what it reads
   */
  @Override
  public void fromData(DataInput in) throws IOException, ClassNotFoundException {
    super.fromData(in);
    this.flags = in.readShort();
    setBooleans(this.flags, in);
    this.regionId = in.readInt();
    // extra field post 9.0
    if (InternalDataSerializer.getVersionForDataStream(in).compareTo(Version.GFE_90) >= 0) {
      this.isTransactionDistributed = in.readBoolean();
    }
  }

  /**
   * Re-construct the booleans using the compressed short. A subclass must override this method if
   * it is using bits in the compressed short.
   */
  protected void setBooleans(short s, DataInput in) throws IOException, ClassNotFoundException {
    if ((s & HAS_PROCESSOR_ID) != 0) {
      this.processorId = in.readInt();
      ReplyProcessor21.setMessageRPId(this.processorId);
    }
    if ((s & NOTIFICATION_ONLY) != 0)
      this.notificationOnly = true;
    if ((s & HAS_TX_ID) != 0)
      this.txUniqId = in.readInt();
    if ((s & HAS_TX_MEMBERID) != 0) {
      this.txMemberId = (InternalDistributedMember) DataSerializer.readObject(in);
    }
  }

  /**
   * Send the contents of this instance to the DataOutput Required to be a
   * {@link org.apache.geode.DataSerializable}Note: must be symmetric with
   * {@link #fromData(DataInput)}in what it writes
   */
  @Override
  public void toData(DataOutput out) throws IOException {
    super.toData(out);
    short compressedShort = 0;
    compressedShort = computeCompressedShort(compressedShort);
    out.writeShort(compressedShort);
    if (this.processorId != 0)
      out.writeInt(this.processorId);
    if (this.txUniqId != TXManagerImpl.NOTX)
      out.writeInt(this.txUniqId);
    if (this.txMemberId != null)
      DataSerializer.writeObject(this.txMemberId, out);
    out.writeInt(this.regionId);
    // extra field post 9.0
    if (InternalDataSerializer.getVersionForDataStream(out).compareTo(Version.GFE_90) >= 0) {
      out.writeBoolean(this.isTransactionDistributed);
    }
  }

  /**
   * Sets the bits of a short by using the bit masks. A subclass must override this method if it is
   * using bits in the compressed short.
   *
   * @return short with appropriate bits set
   */
  protected short computeCompressedShort(short s) {
    if (this.processorId != 0)
      s |= HAS_PROCESSOR_ID;
    if (this.notificationOnly)
      s |= NOTIFICATION_ONLY;
    if (this.getTXUniqId() != TXManagerImpl.NOTX) {
      s |= HAS_TX_ID;
      if (this.txMemberId != null) {
        s |= HAS_TX_MEMBERID;
      }
    }
    return s;
  }

  public static final String PN_TOKEN = ".cache.";

  @Override
  public String toString() {
    StringBuilder buff = new StringBuilder();
    String className = getClass().getName();
    // className.substring(className.lastIndexOf('.', className.lastIndexOf('.') - 1) + 1); //
    // partition.<foo> more generic version
    buff.append(className.substring(className.indexOf(PN_TOKEN) + PN_TOKEN.length())); // partition.<foo>
    buff.append("(prid="); // make sure this is the first one
    buff.append(this.regionId);

    // Append name, if we have it
    String name = null;
    try {
      PartitionedRegion pr = PartitionedRegion.getPRFromId(this.regionId);
      if (pr != null) {
        name = pr.getFullPath();
      }
    } catch (Exception ignore) {
      /* ignored */
      name = null;
    }
    if (name != null) {
      buff.append(" (name = \"").append(name).append("\")");
    }

    appendFields(buff);
    buff.append(" ,distTx=");
    buff.append(this.isTransactionDistributed);
    buff.append(")");
    return buff.toString();
  }

  /**
   * Helper class of {@link #toString()}
   *
   * @param buff buffer in which to append the state of this instance
   */
  protected void appendFields(StringBuilder buff) {
    buff.append(" processorId=").append(this.processorId);
    if (this.notificationOnly) {
      buff.append(" notificationOnly=").append(this.notificationOnly);
    }
    if (this.txUniqId != TXManagerImpl.NOTX) {
      buff.append(" txId=").append(this.txUniqId);
    }
    if (this.txMemberId != null) {
      buff.append(" txMemberId=").append(this.txMemberId);
    }
  }

  public InternalDistributedMember getRecipient() {
    return getRecipients()[0];
  }

  public void setOperation(Operation op) {
    // override in subclasses holding operations
  }

  /**
   * added to support old value to be written on wire.
   *
   * @param value true or false
   * @since GemFire 5.5
   */
  public void setHasOldValue(boolean value) {
    // override in subclasses which need old value to be serialized.
    // overridden by classes like PutMessage, DestroyMessage.
  }

  /**
   * added to support routing of notification-only messages to clients
   */
  public void setFilterInfo(FilterRoutingInfo filterInfo) {
    // subclasses that support routing to clients should reimplement this method
  }

  /**
   * @return the txUniqId
   */
  @Override
  public int getTXUniqId() {
    return txUniqId;
  }

  @Override
  public boolean canStartRemoteTransaction() {
    return false;
  }

  public void setSendDeltaWithFullValue(boolean bool) {
    this.sendDeltaWithFullValue = bool;
  }

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

  protected boolean notifiesSerialGatewaySender(ClusterDistributionManager dm) {
    try {
      PartitionedRegion pr = PartitionedRegion.getPRFromId(this.regionId);
      if (pr == null) {
        return false;
      }
      return pr.notifiesSerialGatewaySender();
    } catch (PRLocallyDestroyedException ignore) {
      return false;
    } catch (RuntimeException ignore) {
      return false;
    }
  }

  /**
   * A processor on which to await a response from the {@link PartitionMessage} recipient, capturing
   * any CacheException thrown by the recipient and handle it as an expected exception.
   *
   * @since GemFire 5.0
   * @see #waitForCacheException()
   */
  public static class PartitionResponse extends DirectReplyProcessor {
    /**
     * The exception thrown when the recipient does not reply
     */
    volatile ForceReattemptException prce;

    /**
     * Whether a response has been received
     */
    volatile boolean responseReceived;

    /**
     * whether a response is required
     */
    boolean responseRequired;

    public PartitionResponse(InternalDistributedSystem dm, Set initMembers) {
      this(dm, initMembers, true);
    }

    public PartitionResponse(InternalDistributedSystem dm, Set initMembers, boolean register) {
      super(dm, initMembers);
      if (register) {
        register();
      }
    }

    public PartitionResponse(InternalDistributedSystem dm, InternalDistributedMember member) {
      this(dm, member, true);
    }

    public PartitionResponse(InternalDistributedSystem dm, InternalDistributedMember member,
        boolean register) {
      super(dm, member);
      if (register) {
        register();
      }
    }

    /**
     * require a response message to be received
     */
    public void requireResponse() {
      this.responseRequired = true;
    }

    @Override
    public void memberDeparted(DistributionManager distributionManager,
        final InternalDistributedMember id, final boolean crashed) {
      if (id != null) {
        if (removeMember(id, true)) {
          this.prce = new ForceReattemptException(
              String.format("memberDeparted event for < %s > crashed, %s",
                  id, crashed));
        }
        checkIfDone();
      } else {
        Exception e = new Exception(
            "memberDeparted got null memberId");
        logger.info(String.format("memberDeparted got null memberId crashed=%s", crashed),
            e);
      }
    }

    /**
     * Waits for the response from the {@link PartitionMessage}'s recipient
     *
     * @throws CacheException if the recipient threw a cache exception during message processing
     * @throws ForceReattemptException if the recipient left the distributed system before the
     *         response was received.
     */
    public void waitForCacheException()
        throws CacheException, ForceReattemptException, PrimaryBucketException {
      try {
        waitForRepliesUninterruptibly();
        if (this.prce != null || (this.responseRequired && !this.responseReceived)) {
          throw new ForceReattemptException(
              "Attempt failed", this.prce);
        }
      } catch (ReplyException e) {
        Throwable t = e.getCause();
        if (t instanceof CacheException) {
          throw (CacheException) t;
        } else if (t instanceof ForceReattemptException) {
          ForceReattemptException ft = (ForceReattemptException) t;
          // See FetchEntriesMessage, which can marshal a ForceReattempt
          // across to the sender
          ForceReattemptException fre = new ForceReattemptException(
              "Peer requests reattempt", t);
          if (ft.hasHash()) {
            fre.setHash(ft.getHash());
          }
          throw fre;
        } else if (t instanceof PrimaryBucketException) {
          // See FetchEntryMessage, GetMessage, InvalidateMessage,
          // PutMessage
          // which can marshal a ForceReattemptacross to the sender
          throw new PrimaryBucketException(
              "Peer failed primary test", t);
        } else if (t instanceof CancelException) {
          logger.debug(
              "PartitionResponse got CacheClosedException from {}, throwing ForceReattemptException",
              e.getSender(), t);
          throw new ForceReattemptException(
              "PartitionResponse got remote CacheClosedException",
              t);
        } else if (t instanceof DiskAccessException) {
          logger.debug(
              "PartitionResponse got DiskAccessException from {}, throwing ForceReattemptException",
              e.getSender(), t);
          throw new ForceReattemptException(
              "PartitionResponse got remote CacheClosedException",
              t);
        } else if (t instanceof LowMemoryException) {
          logger.debug("PartitionResponse re-throwing remote LowMemoryException from {}",
              e.getSender(), t);
          throw (LowMemoryException) t;
        }
        e.handleCause();
      }
    }

    /* overridden from ReplyProcessor21 */
    @Override
    public void process(DistributionMessage msg) {
      this.responseReceived = true;
      super.process(msg);
    }
  }

  @Override
  public boolean isTransactionDistributed() {
    return this.isTransactionDistributed;
  }

  /*
   * For Distributed Tx
   */
  public void setTransactionDistributed(boolean isDistTx) {
    this.isTransactionDistributed = isDistTx;
  }
}