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);
    }
  }
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.
geode-core/src/main/java/org/apache/geode/internal/cache/DistributedCacheOperation.java