public static void flushTo(Set<InternalDistributedMember> targets, DistributedRegion region) {
    DistributionManager dm = region.getDistributionManager();
    boolean initialized = region.isInitialized();
    if (initialized) {
      // force a new "view" so we can track current ops
      region.getDistributionAdvisor().forceNewMembershipVersion();
      try {
        region.getDistributionAdvisor().waitForCurrentOperations();
      } catch (RegionDestroyedException ignore) {
        return;
      }
    }
    // send all state-flush messages and then wait for replies
    Set<ReplyProcessor21> processors = new HashSet<>();
    for (InternalDistributedMember target : targets) {
      StateStabilizationMessage gr = new StateStabilizationMessage();
      gr.isSingleFlushTo = true; // new for flushTo operation
      gr.requestingMember = dm.getDistributionManagerId();
      gr.setRecipient(target);
      ReplyProcessor21 processor = new ReplyProcessor21(dm, target);
      gr.processorId = processor.getProcessorId();
      gr.channelState = dm.getDistribution().getMessageState(target, false);
      if (logger.isTraceEnabled(LogMarker.STATE_FLUSH_OP_VERBOSE)
          && ((gr.channelState != null) && (gr.channelState.size() > 0))) {
        logger.trace(LogMarker.STATE_FLUSH_OP_VERBOSE, "channel states: {}",
            gr.channelStateDescription(gr.channelState));
      }
      if (logger.isTraceEnabled(LogMarker.STATE_FLUSH_OP_VERBOSE)) {
        logger.trace(LogMarker.STATE_FLUSH_OP_VERBOSE, "Sending {}", gr);
      }
      dm.putOutgoing(gr);
      processors.add(processor);
    }

    if (region.getRegionMap().getARMLockTestHook() != null) {
      region.getRegionMap().getARMLockTestHook().beforeStateFlushWait();
    }

    for (ReplyProcessor21 processor : processors) {
      try {
        processor.waitForReplies();
      } catch (InterruptedException ignore) {
        Thread.currentThread().interrupt();
        return;
      }
    }
  }
flush current ops to the given members for the given region
geode-core/src/main/java/org/apache/geode/internal/cache/StateFlushOperation.java