/*
 * 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.ANY_INIT;
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.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;

import org.apache.logging.log4j.Logger;

import org.apache.geode.CancelException;
import org.apache.geode.DataSerializer;
import org.apache.geode.InvalidDeltaException;
import org.apache.geode.InvalidVersionException;
import org.apache.geode.SystemFailure;
import org.apache.geode.annotations.internal.MutableForTesting;
import org.apache.geode.cache.CacheEvent;
import org.apache.geode.cache.DiskAccessException;
import org.apache.geode.cache.EntryNotFoundException;
import org.apache.geode.cache.EntryOperation;
import org.apache.geode.cache.Operation;
import org.apache.geode.cache.RegionDestroyedException;
import org.apache.geode.cache.query.internal.cq.CqService;
import org.apache.geode.cache.query.internal.cq.ServerCQ;
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.ReplySender;
import org.apache.geode.distributed.internal.SerialDistributionMessage;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.distributed.internal.membership.api.CacheOperationMessageMarker;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.CopyOnWriteHashSet;
import org.apache.geode.internal.InternalDataSerializer;
import org.apache.geode.internal.cache.CacheDistributionAdvisor.CacheProfile;
import org.apache.geode.internal.cache.EntryEventImpl.OldValueImporter;
import org.apache.geode.internal.cache.FilterRoutingInfo.FilterInfo;
import org.apache.geode.internal.cache.LocalRegion.InitializationLevel;
import org.apache.geode.internal.cache.UpdateOperation.UpdateMessage;
import org.apache.geode.internal.cache.partitioned.Bucket;
import org.apache.geode.internal.cache.partitioned.PartitionMessage;
import org.apache.geode.internal.cache.persistence.PersistentMemberID;
import org.apache.geode.internal.cache.tier.MessageType;
import org.apache.geode.internal.cache.tx.RemoteOperationMessage;
import org.apache.geode.internal.cache.versions.DiskVersionTag;
import org.apache.geode.internal.cache.versions.RegionVersionVector;
import org.apache.geode.internal.cache.versions.VersionSource;
import org.apache.geode.internal.cache.versions.VersionTag;
import org.apache.geode.internal.logging.log4j.LogMarker;
import org.apache.geode.internal.offheap.Releasable;
import org.apache.geode.internal.offheap.StoredObject;
import org.apache.geode.internal.offheap.annotations.Released;
import org.apache.geode.internal.offheap.annotations.Unretained;
import org.apache.geode.internal.sequencelog.EntryLogger;
import org.apache.geode.internal.serialization.DeserializationContext;
import org.apache.geode.internal.serialization.SerializationContext;
import org.apache.geode.internal.util.DelayedAction;
import org.apache.geode.logging.internal.log4j.api.LogService;

public abstract class DistributedCacheOperation {

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

  @MutableForTesting
  public static double LOSS_SIMULATION_RATIO = 0; // test hook

  @MutableForTesting
  public static Random LOSS_SIMULATION_GENERATOR;

  @MutableForTesting
  public static long SLOW_DISTRIBUTION_MS = 0; // test hook

  // constants used in subclasses and distribution messages
  // should use enum in source level 1.5+

  /**
   * Deserialization policy: do not deserialize (for byte array, null or cases where the value
   * should stay serialized)
   *
   * @since GemFire 5.7
   */
  public static final byte DESERIALIZATION_POLICY_NONE = (byte) 0;

  /**
   * Deserialization policy: deserialize lazily (for all other objects)
   *
   * @since GemFire 5.7
   */
  public static final byte DESERIALIZATION_POLICY_LAZY = (byte) 2;

  /**
   * @param deserializationPolicy must be one of the following: DESERIALIZATION_POLICY_NONE,
   *        DESERIALIZATION_POLICY_LAZY.
   */
  public static void writeValue(final byte deserializationPolicy, final Object vObj,
      final byte[] vBytes, final DataOutput out) throws IOException {
    if (vObj != null) {
      if (deserializationPolicy == DESERIALIZATION_POLICY_NONE) {
        // We only have NONE with a vObj when vObj is off-heap and not serialized.
        StoredObject so = (StoredObject) vObj;
        assert !so.isSerialized();
        so.sendAsByteArray(out);
      } else { // LAZY
        DataSerializer.writeObjectAsByteArray(vObj, out);
      }
    } else {
      DataSerializer.writeByteArray(vBytes, out);
    }
  }

  /**
   * Given a VALUE_IS_* constant convert and return the corresponding DESERIALIZATION_POLICY_*.
   */
  public static byte valueIsToDeserializationPolicy(boolean oldValueIsSerialized) {
    if (!oldValueIsSerialized) {
      return DESERIALIZATION_POLICY_NONE;
    }
    return DESERIALIZATION_POLICY_LAZY;
  }


  public static final byte DESERIALIZATION_POLICY_NUMBITS =
      DistributionMessage.getNumBits(DESERIALIZATION_POLICY_LAZY);

  public static final short DESERIALIZATION_POLICY_END =
      (short) (1 << DESERIALIZATION_POLICY_NUMBITS);

  public static final short DESERIALIZATION_POLICY_MASK = (short) (DESERIALIZATION_POLICY_END - 1);

  @MutableForTesting
  public static boolean testSendingOldValues;

  protected InternalCacheEvent event;

  protected CacheOperationReplyProcessor processor = null;

  protected Set<InternalDistributedMember> departedMembers;

  protected Set<InternalDistributedMember> originalRecipients;

  @MutableForTesting
  static Runnable internalBeforePutOutgoing;

  public static String deserializationPolicyToString(byte policy) {
    switch (policy) {
      case DESERIALIZATION_POLICY_NONE:
        return "NONE";
      case DESERIALIZATION_POLICY_LAZY:
        return "LAZY";
      default:
        throw new AssertionError("unknown deserialization policy");
    }
  }

  /** Creates a new instance of DistributedCacheOperation */
  public DistributedCacheOperation(CacheEvent<?, ?> event) {
    this.event = (InternalCacheEvent) event;
  }

  /**
   * Return true if this operation needs to check for reliable delivery. Return false if not.
   * Currently the only case it doesn't need to be is a DestroyRegionOperation doing a "local"
   * destroy.
   *
   * @since GemFire 5.0
   */
  boolean isOperationReliable() {
    Operation op = event.getOperation();
    if (!op.isRegionDestroy()) {
      return true;
    }
    // must be a region destroy that is "local" which means
    // Region.localDestroyRegion or Region.close or Cache.clsoe
    // none of these should do reliability checks
    return op.isDistributed();
  }

  public boolean supportsDirectAck() {
    // force use of shared connection if we're already in a secondary
    // thread-owned reader thread. See Connection#processNIOBuffer
    return true;
  }

  /**
   * returns true if multicast can be used for this operation. The default is true.
   */
  public boolean supportsMulticast() {
    return true;
  }

  /** returns true if the receiver can distribute during cache closure */
  public boolean canBeSentDuringShutdown() {
    return getRegion().isUsedForPartitionedRegionAdmin();
  }

  /**
   * returns true if adjunct messaging (piggybacking) is allowed for this operation. Region-oriented
   * operations typically do not allow adjunct messaging, while Entry-oriented operations do. The
   * default implementation returns true.
   */
  protected boolean supportsAdjunctMessaging() {
    return true;
  }

  /**
   * returns true if this operation supports propagation of delta values instead of full changes
   */
  protected boolean supportsDeltaPropagation() {
    return false;
  }

  /** does this operation change region content? */
  public boolean containsRegionContentChange() {
    return true;
  }

  @MutableForTesting
  public static volatile DelayedAction test_InvalidVersionAction;

  /**
   * region's distribution advisor marked that a distribution is about to start, then distribute. It
   * returns a token, which is view version. Return -1 means the method did not succeed. This method
   * must be invoked before toDistribute(). This method should pair with endOperation() in
   * try/finally block.
   */
  public long startOperation() {
    DistributedRegion region = getRegion();
    long viewVersion = -1;
    try {
      if (containsRegionContentChange()) {
        viewVersion = region.getDistributionAdvisor().startOperation();
      }
      if (logger.isTraceEnabled(LogMarker.STATE_FLUSH_OP_VERBOSE)) {
        logger.trace(LogMarker.STATE_FLUSH_OP_VERBOSE, "dispatching operation in view version {}",
            viewVersion);
      }
      try {
        _distribute();
      } catch (InvalidVersionException e) {
        if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) {
          logger.trace(LogMarker.DM_VERBOSE,
              "PutAll failed since versions were missing; retrying", e);
        }

        if (test_InvalidVersionAction != null) {
          test_InvalidVersionAction.run();
        }
        _distribute();
      }
    } catch (RuntimeException | Error e) {
      endOperation(viewVersion);
      throw e;
    }
    return viewVersion;
  }

  /**
   * region's distribution advisor marked that a distribution is ended. This method should pair with
   * startOperation in try/finally block.
   */
  public void endOperation(long viewVersion) {
    DistributedRegion region = getRegion();
    if (viewVersion != -1) {
      region.getDistributionAdvisor().endOperation(viewVersion);
      if (logger.isTraceEnabled()) {
        logger.trace(LogMarker.STATE_FLUSH_OP_VERBOSE,
            "done dispatching operation in view version {}", viewVersion);
      }
    }
  }

  /**
   * Distribute a cache operation to other members of the distributed system. This method determines
   * who the recipients are and handles careful delivery of the operation to those members.
   */
  public void distribute() {
    long token = -1;
    try {
      token = startOperation();
    } finally {
      endOperation(token);
    }
  }

  /**
   * About to distribute a cache operation to other members of the distributed system. This method
   * determines who the recipients are and handles careful delivery of the operation to those
   * members. This method should wrapped by startOperation() and endOperation() in try/finally
   * block.
   */
  protected void _distribute() {
    DistributedRegion region = getRegion();
    DistributionManager mgr = region.getDistributionManager();
    boolean reliableOp = isOperationReliable() && region.requiresReliabilityCheck();

    if (SLOW_DISTRIBUTION_MS > 0) { // test hook
      try {
        Thread.sleep(SLOW_DISTRIBUTION_MS);
      } catch (InterruptedException ignore) {
        Thread.currentThread().interrupt();
      }
      SLOW_DISTRIBUTION_MS = 0;
    }

    boolean isPutAll = (this instanceof DistributedPutAllOperation);
    boolean isRemoveAll = (this instanceof DistributedRemoveAllOperation);

    try {
      // Recipients with CacheOp
      Set<InternalDistributedMember> recipients = new HashSet<>(getRecipients());
      Map<InternalDistributedMember, PersistentMemberID> persistentIds = null;
      if (region.getDataPolicy().withPersistence()) {
        persistentIds = region.getDistributionAdvisor().adviseInitializedPersistentMembers();
      }

      // some members requiring old value are also in the cache op recipients set
      Set<InternalDistributedMember> needsOldValueInCacheOp = Collections.emptySet();

      // set client routing information into the event
      boolean routingComputed = false;
      FilterRoutingInfo filterRouting = null;
      // recipients that will get a cacheop msg and also a PR message
      Set<InternalDistributedMember> twoMessages = Collections.emptySet();
      if (region.isUsedForPartitionedRegionBucket()) {
        twoMessages = ((Bucket) region).getBucketAdvisor().adviseRequiresTwoMessages();
        routingComputed = true;
        filterRouting = getRecipientFilterRouting(recipients);
        if (filterRouting != null) {
          if (logger.isDebugEnabled()) {
            logger.debug("Computed this filter routing: {}", filterRouting);
          }
        }
      }

      // some members need PR notification of the change for client/wan
      // notification
      Set<InternalDistributedMember> adjunctRecipients = Collections.emptySet();

      // Partitioned region listener notification messages piggyback on this
      // operation's replyprocessor and need to be sent at the same time as
      // the operation's message
      if (supportsAdjunctMessaging() && region.isUsedForPartitionedRegionBucket()) {
        BucketRegion br = (BucketRegion) region;
        adjunctRecipients = getAdjunctReceivers(br, recipients, twoMessages, filterRouting);
      }

      EntryEventImpl entryEvent = event.getOperation().isEntry() ? getEvent() : null;

      if (entryEvent != null && entryEvent.hasOldValue()) {
        if (testSendingOldValues) {
          needsOldValueInCacheOp = new HashSet<>(recipients);
        } else {
          needsOldValueInCacheOp =
              region.getCacheDistributionAdvisor().adviseRequiresOldValueInCacheOp();
        }
        recipients.removeAll(needsOldValueInCacheOp);
      }

      Set<InternalDistributedMember> cachelessNodes = Collections.emptySet();
      Set<InternalDistributedMember> adviseCacheServers;
      Set<InternalDistributedMember> cachelessNodesWithNoCacheServer = Collections.emptySet();
      if (region.getDistributionConfig().getDeltaPropagation() && supportsDeltaPropagation()) {
        cachelessNodes = region.getCacheDistributionAdvisor().adviseEmptys();
        if (!cachelessNodes.isEmpty()) {
          List<InternalDistributedMember> list = new ArrayList<>(cachelessNodes);
          for (InternalDistributedMember member : cachelessNodes) {
            if (!recipients.contains(member) || adjunctRecipients.contains(member)) {
              // Don't include those originally excluded.
              list.remove(member);
            }
          }
          cachelessNodes.clear();
          recipients.removeAll(list);
          cachelessNodes.addAll(list);
        }
        if (!cachelessNodes.isEmpty()) {
          cachelessNodesWithNoCacheServer = new HashSet<>(cachelessNodes);
          adviseCacheServers = region.getCacheDistributionAdvisor().adviseCacheServers();
          cachelessNodesWithNoCacheServer.removeAll(adviseCacheServers);
        }
      }

      if (recipients.isEmpty() && adjunctRecipients.isEmpty() && needsOldValueInCacheOp.isEmpty()
          && cachelessNodes.isEmpty()) {
        if (region.isInternalRegion()) {
          if (logger.isDebugEnabled() || logger.isTraceEnabled()) {
            if (mgr.getNormalDistributionManagerIds().size() > 1) {
              // suppress this msg if we are the only member.
              if (logger.isTraceEnabled()) {
                logger.trace("<No Recipients> {}", this);
              }
            } else {
              // suppress this msg if we are the only member.
              if (logger.isDebugEnabled()) {
                logger.debug("<No Recipients> {}", this);
              }
            }
          }
        }
        if (reliableOp && !region.isNoDistributionOk()) {
          region.handleReliableDistribution(Collections.emptySet());
        }

        // compute local client routing before waiting for an ack only for a bucket
        if (region.isUsedForPartitionedRegionBucket()) {
          FilterInfo filterInfo = getLocalFilterRouting(filterRouting);
          event.setLocalFilterInfo(filterInfo);
        }

      } else {
        boolean directAck = false;
        boolean useMulticast = region.getMulticastEnabled()
            && region.getSystem().getConfig().getMcastPort() != 0 && supportsMulticast();
        boolean shouldAck = shouldAck();

        if (shouldAck) {
          if (supportsDirectAck() && adjunctRecipients.isEmpty()) {
            if (region.getSystem().threadOwnsResources()) {
              directAck = true;
            }
          }
        }
        // don't send to the sender of a remote-operation-message. Those messages send
        // their own response. fixes bug #45973
        if (entryEvent != null) {
          RemoteOperationMessage rmsg = entryEvent.getRemoteOperationMessage();
          if (rmsg != null) {
            recipients.remove(rmsg.getSender());
            useMulticast = false; // bug #45106: can't mcast or the sender of the one-hop op will
                                  // get it
          }
        }

        if (logger.isDebugEnabled()) {
          logger.debug("recipients for {}: {} with adjunct messages to: {}", this, recipients,
              adjunctRecipients);
        }

        if (shouldAck) {
          // adjunct messages are sent using the same reply processor, so
          // add them to the processor's membership set
          final Collection<InternalDistributedMember> waitForMembers;
          if (recipients.size() > 0 && adjunctRecipients.size() == 0 && cachelessNodes.isEmpty()) {
            // the common case
            waitForMembers = recipients;
          } else if (!cachelessNodes.isEmpty()) {
            waitForMembers = new HashSet<>(recipients);
            waitForMembers.addAll(cachelessNodes);
          } else {
            // note that we use a List instead of a Set for the responders
            // collection because partitioned regions sometimes send both a regular cache
            // operation and a partitioned-region notification message to the same recipient
            waitForMembers = new ArrayList<>(recipients);
            waitForMembers.addAll(adjunctRecipients);
            waitForMembers.addAll(needsOldValueInCacheOp);
            waitForMembers.addAll(cachelessNodes);
          }
          if (DistributedCacheOperation.LOSS_SIMULATION_RATIO != 0.0) {
            if (LOSS_SIMULATION_GENERATOR == null) {
              LOSS_SIMULATION_GENERATOR = new Random(hashCode());
            }
            if ((LOSS_SIMULATION_GENERATOR.nextInt(100) * 1.0 / 100.0) < LOSS_SIMULATION_RATIO) {
              if (logger.isDebugEnabled()) {
                logger.debug("loss simulation is inhibiting message transmission to {}",
                    recipients);
              }
              waitForMembers.removeAll(recipients);
              recipients = Collections.emptySet();
            }
          }
          if (reliableOp) {
            departedMembers = new HashSet<>();
            processor = new ReliableCacheReplyProcessor(region.getSystem(), waitForMembers,
                departedMembers);
          } else {
            processor = new CacheOperationReplyProcessor(region.getSystem(), waitForMembers);
          }
        }

        CacheOperationMessage msg = createMessage();
        initMessage(msg, processor);

        if (DistributedCacheOperation.internalBeforePutOutgoing != null) {
          DistributedCacheOperation.internalBeforePutOutgoing.run();
        }

        if (processor != null && msg.isSevereAlertCompatible()) {
          processor.enableSevereAlertProcessing();
          // if this message is distributing for a partitioned region message,
          // we can't wait as long as the full ack-severe-alert-threshold or
          // the sender might kick us out of the system before we can get an ack
          // back
          DistributedRegion r = getRegion();
          if (r.isUsedForPartitionedRegionBucket() && event.getOperation().isEntry()) {
            PartitionMessage pm = ((EntryEventImpl) event).getPartitionMessage();
            if (pm != null && pm.getSender() != null
                && !pm.getSender().equals(r.getDistributionManager().getDistributionManagerId())) {
              // PR message sent by another member
              ReplyProcessor21.setShortSevereAlertProcessing(true);
            }
          }
        }

        msg.setMulticast(useMulticast);
        msg.directAck = directAck;
        if (region.isUsedForPartitionedRegionBucket()) {
          if (!isPutAll && !isRemoveAll && filterRouting != null
              && filterRouting.hasMemberWithFilterInfo()) {
            if (logger.isDebugEnabled()) {
              logger.debug("Setting filter information for message to {}", filterRouting);
            }
            msg.filterRouting = filterRouting;
          }
        } else if (!routingComputed) {
          msg.needsRouting = true;
        }

        initProcessor(processor, msg);

        if (region.cache.isClosed() && !canBeSentDuringShutdown()) {
          throw region.cache.getCacheClosedException(
              "The cache has been closed",
              null);
        }

        msg.setRecipients(recipients);
        Set<InternalDistributedMember> failures = mgr.putOutgoing(msg);

        // distribute to members needing the old value now
        if (needsOldValueInCacheOp.size() > 0) {
          msg.appendOldValueToMessage((EntryEventImpl) event);
          msg.resetRecipients();
          msg.setRecipients(needsOldValueInCacheOp);
          Set<InternalDistributedMember> newFailures = mgr.putOutgoing(msg);
          if (newFailures != null) {
            if (logger.isDebugEnabled()) {
              logger.debug("Failed sending ({}) to {}", msg, newFailures);
            }
            if (failures != null && failures.size() > 0) {
              failures.addAll(newFailures);
            } else {
              failures = newFailures;
            }
          }
        }

        if (cachelessNodes.size() > 0) {
          cachelessNodes.removeAll(cachelessNodesWithNoCacheServer);
          if (cachelessNodes.size() > 0) {
            msg.resetRecipients();
            msg.setRecipients(cachelessNodes);
            msg.setSendDelta(false);
            Set<InternalDistributedMember> newFailures = mgr.putOutgoing(msg);
            if (newFailures != null) {
              if (failures != null && failures.size() > 0) {
                failures.addAll(newFailures);
              } else {
                failures = newFailures;
              }
            }
          }

          if (!cachelessNodesWithNoCacheServer.isEmpty()) {
            msg.resetRecipients();
            msg.setRecipients(cachelessNodesWithNoCacheServer);
            msg.setSendDelta(false);
            ((UpdateMessage) msg).setSendDeltaWithFullValue(false);
            Set<InternalDistributedMember> newFailures = mgr.putOutgoing(msg);
            if (newFailures != null) {
              if (failures != null && failures.size() > 0) {
                failures.addAll(newFailures);
              } else {
                failures = newFailures;
              }
            }
            // Add it back for size calculation ahead
            cachelessNodes.addAll(cachelessNodesWithNoCacheServer);
          }
        }

        if (failures != null && !failures.isEmpty() && logger.isDebugEnabled()) {
          logger.debug("Failed sending ({}) to {} while processing event:{}", msg, failures, event);
        }

        // send partitioned region listener notification messages now
        if (!adjunctRecipients.isEmpty()) {
          if (cachelessNodes.size() > 0) {
            // add non-delta recipients back into the set for adjunct
            // calculations
            if (recipients.isEmpty()) {
              recipients = cachelessNodes;
            } else {
              recipients.addAll(cachelessNodes);
            }
          }

          final Set<InternalDistributedMember> adjunctRecipientsWithNoCacheServer =
              new HashSet<>(adjunctRecipients);
          adviseCacheServers = ((Bucket) region).getPartitionedRegion()
              .getCacheDistributionAdvisor().adviseCacheServers();
          adjunctRecipientsWithNoCacheServer.removeAll(adviseCacheServers);

          if (isPutAll) {
            ((BucketRegion) region).performPutAllAdjunctMessaging((DistributedPutAllOperation) this,
                recipients, adjunctRecipients, filterRouting, processor);
          } else if (isRemoveAll) {
            ((BucketRegion) region).performRemoveAllAdjunctMessaging(
                (DistributedRemoveAllOperation) this, recipients, adjunctRecipients, filterRouting,
                processor);
          } else {
            boolean calculateDelta =
                adjunctRecipientsWithNoCacheServer.size() < adjunctRecipients.size();
            adjunctRecipients.removeAll(adjunctRecipientsWithNoCacheServer);
            if (!adjunctRecipients.isEmpty()) {
              ((BucketRegion) region).performAdjunctMessaging(getEvent(), recipients,
                  adjunctRecipients, filterRouting, processor, calculateDelta, true);
            }
            if (!adjunctRecipientsWithNoCacheServer.isEmpty()) {
              ((BucketRegion) region).performAdjunctMessaging(getEvent(), recipients,
                  adjunctRecipientsWithNoCacheServer, filterRouting, processor, calculateDelta,
                  false);
            }
          }
        }

        // compute local client routing before waiting for an ack only for a bucket
        if (region.isUsedForPartitionedRegionBucket()) {
          FilterInfo filterInfo = getLocalFilterRouting(filterRouting);
          event.setLocalFilterInfo(filterInfo);
        }

        waitForAckIfNeeded(persistentIds);

        if (/* msg != null && */reliableOp) {
          Set<InternalDistributedMember> successfulRecips = new HashSet<>(recipients);
          successfulRecips.addAll(cachelessNodes);
          successfulRecips.addAll(needsOldValueInCacheOp);
          if (failures != null && !failures.isEmpty()) {
            successfulRecips.removeAll(failures);
          }
          if (departedMembers != null) {
            successfulRecips.removeAll(departedMembers);
          }
          region.handleReliableDistribution(successfulRecips);
        }
      }

      if (region.isUsedForPartitionedRegionBucket() && filterRouting != null) {
        removeDestroyTokensFromCqResultKeys(filterRouting);
      }

    } catch (CancelException e) {
      if (logger.isDebugEnabled()) {
        logger.debug("distribution of message aborted by shutdown: {}", this);
      }
      throw e;
    } catch (RuntimeException e) {
      logger.info(String.format("Exception occurred while processing %s", this),
          e);
      throw e;
    } finally {
      ReplyProcessor21.setShortSevereAlertProcessing(false);
    }
  }

  /**
   * Cleanup destroyed events in CQ result cache for remote CQs. While maintaining the CQ results
   * key caching. the destroy event keys are marked as destroyed instead of removing them, this is
   * to take care, arrival of duplicate events. The key marked as destroyed are removed after the
   * event is placed in clients HAQueue or distributed to the peers.
   *
   * This is similar to CacheClientNotifier.removeDestroyTokensFromCqResultKeys() where the
   * destroyed events for local CQs are handled.
   */
  private void removeDestroyTokensFromCqResultKeys(FilterRoutingInfo filterRouting) {
    for (InternalDistributedMember m : filterRouting.getMembers()) {
      FilterInfo filterInfo = filterRouting.getFilterInfo(m);
      if (filterInfo == null || filterInfo.getCQs() == null) {
        continue;
      }

      CacheProfile cf = (CacheProfile) ((Bucket) getRegion()).getPartitionedRegion()
          .getCacheDistributionAdvisor().getProfile(m);

      if (cf == null || cf.filterProfile == null || cf.filterProfile.isLocalProfile()
          || cf.filterProfile.getCqMap().isEmpty()) {
        continue;
      }

      for (Object value : cf.filterProfile.getCqMap().values()) {
        ServerCQ cq = (ServerCQ) value;

        doRemoveDestroyTokensFromCqResultKeys(filterInfo, cq);
      }
    }
  }

  void doRemoveDestroyTokensFromCqResultKeys(FilterInfo filterInfo, ServerCQ cq) {
    for (Map.Entry<Long, MessageType> e : filterInfo.getCQs().entrySet()) {
      Long cqID = e.getKey();
      // For the CQs satisfying the event with destroy CQEvent, remove
      // the entry form CQ cache.
      if (cq != null && cq.getFilterID() != null && cq.getFilterID().equals(cqID)
          && e.getValue() != null && e.getValue().equals(MessageType.LOCAL_DESTROY)
          && ((EntryOperation<?, ?>) event).getKey() != null) {
        cq.removeFromCqResultKeys(((EntryOperation<?, ?>) event).getKey(), true);
      }
    }
  }

  /**
   * Get the adjunct receivers for a partitioned region operation
   *
   * @param br the PR bucket
   * @param cacheOpReceivers the receivers of the CacheOperationMessage for this op
   * @param twoMessages PR members that are creating the bucket and need both cache op and adjunct
   *        messages
   * @param routing client routing information
   */
  Set<InternalDistributedMember> getAdjunctReceivers(BucketRegion br,
      Set<InternalDistributedMember> cacheOpReceivers, Set<InternalDistributedMember> twoMessages,
      FilterRoutingInfo routing) {
    return br.getAdjunctReceivers(getEvent(), cacheOpReceivers, twoMessages, routing);
  }

  /**
   * perform any operation-specific initialization on the given reply processor
   */
  protected void initProcessor(CacheOperationReplyProcessor p, CacheOperationMessage msg) {
    // nothing to do here - see UpdateMessage
  }

  protected void waitForAckIfNeeded(
      Map<InternalDistributedMember, PersistentMemberID> persistentIds) {
    if (processor == null) {
      return;
    }
    try {
      // keep waiting even if interrupted
      try {
        processor.waitForRepliesUninterruptibly();
        Set<InternalDistributedMember> closedMembers = processor.closedMembers.getSnapshot();
        handleClosedMembers(closedMembers, persistentIds);
      } catch (ReplyException e) {
        if (this instanceof DestroyRegionOperation) {
          logger.fatal("waitForAckIfNeeded: exception", e);
        }
        e.handleCause();
      }
    } finally {
      processor = null;
    }
  }

  private void handleClosedMembers(Set<InternalDistributedMember> closedMembers,
      Map<InternalDistributedMember, PersistentMemberID> persistentIds) {
    if (persistentIds == null) {
      return;
    }

    for (InternalDistributedMember member : closedMembers) {
      PersistentMemberID persistentId = persistentIds.get(member);
      if (persistentId != null) {
        // Fix for bug 42142 - In order for recovery to work,
        // we must either
        // 1) persistent the region operation successfully on the peer
        // 2) record that the peer is offline
        // or
        // 3) fail the operation

        // if we have started to shutdown, we don't want to mark the peer
        // as offline, or we will think we have newer data when in fact we don't
        getRegion().getCancelCriterion().checkCancelInProgress(null);

        // Otherwise, mark the peer as offline, because it didn't complete
        // the operation.
        getRegion().getPersistenceAdvisor().markMemberOffline(member, persistentId);
      }
    }
  }

