public boolean flush(Set recipients, DistributedMember target, int processorType,
      boolean flushNewOps) throws InterruptedException {

    Set recips = recipients; // do not use recipients parameter past this point
    if (Thread.interrupted()) {
      throw new InterruptedException();
    }

    InternalDistributedMember myId = dm.getDistributionManagerId();

    if (!recips.contains(target) && !myId.equals(target)) {
      recips = new HashSet(recipients);
      recips.add(target);
    }
    // partial fix for bug 38773 - ensures that this cache will get both
    // a cache op and an adjunct message when creating a bucket region
    // if (recips.size() < 2 && !myId.equals(target)) {
    // return true; // no state to flush to a single holder of the region
    // }
    StateMarkerMessage smm = new StateMarkerMessage();
    smm.relayRecipient = target;
    smm.processorType = processorType;
    smm.flushNewOps = flushNewOps;
    if (region == null) {
      smm.allRegions = true;
    } else {
      smm.regionPath = region.getFullPath();
    }
    smm.setRecipients(recips);

    StateFlushReplyProcessor gfprocessor = new StateFlushReplyProcessor(dm, recips, target);
    smm.processorId = gfprocessor.getProcessorId();
    if (region != null && region.isUsedForPartitionedRegionBucket()
        && region.getDistributionConfig().getAckSevereAlertThreshold() > 0) {
      smm.severeAlertEnabled = true;
      gfprocessor.enableSevereAlertProcessing();
    }
    if (logger.isTraceEnabled(LogMarker.STATE_FLUSH_OP_VERBOSE)) {
      logger.trace(LogMarker.STATE_FLUSH_OP_VERBOSE, "Sending {} with processor {}", smm,
          gfprocessor);
    }
    Set failures = dm.putOutgoing(smm);
    if (failures != null) {
      if (failures.contains(target)) {
        if (logger.isTraceEnabled(LogMarker.STATE_FLUSH_OP_VERBOSE)) {
          logger.trace(LogMarker.STATE_FLUSH_OP_VERBOSE,
              "failed to send StateMarkerMessage to target {}; returning from flush without waiting for replies",
              target);
        }
        return false;
      }
      gfprocessor.messageNotSentTo(failures);
    }

    try {
      gfprocessor.waitForReplies();
      if (logger.isTraceEnabled(LogMarker.STATE_FLUSH_OP_VERBOSE)) {
        logger.trace(LogMarker.STATE_FLUSH_OP_VERBOSE, "Finished processing {}", smm);
      }
    } catch (ReplyException re) {
      logger.warn("state flush terminated with exception", re);
      return false;
    }
    return true;
  }
flush state to the given target

@param recipients The members who may be making state changes to the region. This is typically
       taken from a CacheDistributionAdvisor membership set
@param target The member who should have all state flushed to it
@param processorType The execution processor type for the marker message that is sent to all
       members using the given region
@param flushNewOps normally only ops that were started before region profile exchange are
       flushed. Setting this to true causes the flush to wait for any started after the profile
       exchange as well.
@throws InterruptedException If the operation is interrupted, usually for shutdown, an
        InterruptedException will be thrown
@return true if the state was flushed, false if not
geode-core/src/main/java/org/apache/geode/internal/cache/StateFlushOperation.java