@Override
    protected void process(ClusterDistributionManager dm) {
      logger.trace(LogMarker.STATE_FLUSH_OP_VERBOSE, "Processing {}", this);
      if (dm.getDistributionManagerId().equals(relayRecipient)) {
        try {
          // wait for inflight operations to the aeqs even if the recipient is the primary
          Set<DistributedRegion> regions = getRegions(dm);
          for (DistributedRegion r : regions) {
            if (r != null) {
              if (allRegions && r.doesNotDistribute()) {
                // no need to flush a region that does no distribution
                continue;
              }
              waitForCurrentOperations(r, r.isInitialized());
            }
          }
        } catch (CancelException ignore) {
          // cache is closed - no distribution advisor available for the region so nothing to do but
          // send the stabilization message
        } catch (Exception e) {
          logger.fatal(String.format("%s Exception caught while determining channel state",
              this),
              e);
        } finally {
          // no need to send a relay request to this process - just send the
          // ack back to the sender
          StateStabilizedMessage ga = new StateStabilizedMessage();
          ga.sendingMember = relayRecipient;
          ga.setRecipient(getSender());
          ga.setProcessorId(processorId);
          if (logger.isTraceEnabled(LogMarker.STATE_FLUSH_OP_VERBOSE)) {
            logger.trace(LogMarker.STATE_FLUSH_OP_VERBOSE, "Sending {}", ga);
          }
          dm.putOutgoing(ga);
        }
      } else {
        // 1) wait for all messages based on the membership version (or older)
        // at which the sender "joined" this region to be put on the pipe
        // 2) record the state of all communication channels from this process
        // to the relay point
        // 3) send a stabilization message to the relay point that holds the
        // communication channel state information
        StateStabilizationMessage gr = new StateStabilizationMessage();
        gr.setRecipient((InternalDistributedMember) relayRecipient);
        gr.requestingMember = getSender();
        gr.processorId = processorId;
        try {
          Set<DistributedRegion> regions = getRegions(dm);
          for (DistributedRegion r : regions) {
            if (r == null) {
              if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) {
                logger.trace(LogMarker.DM_VERBOSE,
                    "Region not found - skipping channel state assessment");
              }
            }
            if (r != null) {
              if (allRegions && r.doesNotDistribute()) {
                // no need to flush a region that does no distribution
                continue;
              }
              boolean initialized = r.isInitialized();
              waitForCurrentOperations(r, initialized);
              boolean useMulticast =
                  r.getMulticastEnabled() && r.getSystem().getConfig().getMcastPort() != 0;
              if (initialized) {
                Map channelStates =
                    dm.getDistribution().getMessageState(relayRecipient, useMulticast);
                if (gr.channelState != null) {
                  gr.channelState.putAll(channelStates);
                } else {
                  gr.channelState = channelStates;
                }
                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));
                }
              }
            }
          }
        } catch (CancelException ignore) {
          // cache is closed - no distribution advisor available for the region so nothing to do but
          // send the stabilization message
        } catch (Exception e) {
          logger.fatal(String.format("%s Exception caught while determining channel state",
              this),
              e);
        } finally {
          if (logger.isTraceEnabled(LogMarker.STATE_FLUSH_OP_VERBOSE)) {
            logger.trace(LogMarker.STATE_FLUSH_OP_VERBOSE, "Sending {}", gr);
          }
          dm.putOutgoing(gr);
        }
      }
    }
returns a set of all DistributedRegions for allRegions processing
geode-core/src/main/java/org/apache/geode/internal/cache/StateFlushOperation.java