@Override
    protected void process(final ClusterDistributionManager dm) {
      // though this message must be transmitted on an ordered connection to
      // ensure that datagram channnels are flushed, we need to execute
      // in the waiting pool to avoid blocking those connections
      dm.getExecutors().getWaitingThreadPool().execute(new Runnable() {
        @Override
        public void run() {
          if (logger.isTraceEnabled(LogMarker.STATE_FLUSH_OP_VERBOSE)) {
            logger.trace(LogMarker.STATE_FLUSH_OP_VERBOSE, "Processing {}", this);
          }
          try {
            if (channelState != null) {
              if (logger.isTraceEnabled(LogMarker.STATE_FLUSH_OP_VERBOSE)
                  && (channelState.size() > 0)) {
                logger.trace(LogMarker.STATE_FLUSH_OP_VERBOSE, "Waiting for channel states:  {}",
                    channelStateDescription(channelState));
              }
              for (;;) {
                dm.getCancelCriterion().checkCancelInProgress(null);
                boolean interrupted = Thread.interrupted();
                try {
                  dm.getDistribution().waitForMessageState(getSender(), channelState);
                  break;
                } catch (InterruptedException ignore) {
                  interrupted = true;
                } finally {
                  if (interrupted) {
                    Thread.currentThread().interrupt();
                  }
                }
              } // for
            }
          } catch (ThreadDeath td) {
            throw td;
          } 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 e) {
            // 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();
            logger.fatal("Exception caught while waiting for channel state",
                e);
          } finally {
            StateStabilizedMessage ga = new StateStabilizedMessage();
            ga.setRecipient((InternalDistributedMember) requestingMember);
            if (isSingleFlushTo) {
              // not a proxied message but a simple request-response
              ga.sendingMember = dm.getDistributionManagerId();
            } else {
              ga.sendingMember = getSender();
            }
            ga.setProcessorId(processorId);
            if (logger.isTraceEnabled(LogMarker.STATE_FLUSH_OP_VERBOSE)) {
              logger.trace(LogMarker.STATE_FLUSH_OP_VERBOSE, "Sending {}", ga);
            }
            if (requestingMember.equals(dm.getDistributionManagerId())) {
              ga.dmProcess(dm);
            } else {
              dm.putOutgoing(ga);
            }
          }
        }
      });
    }
whether this is a simple request/response two-party flush or (false) a proxied flush
geode-core/src/main/java/org/apache/geode/internal/cache/StateFlushOperation.java