@Override
    protected void process(ClusterDistributionManager dm) {
      try {
        CacheDistributionAdvisee r = findRegion(dm);
        if (r == null) {
          if (logger.isDebugEnabled()) {
            logger.debug("Region not found, so ignoring filter profile update: {}", this);
          }
          return;
        }
        // we only need to record the delta if this is a partitioned region
        if (!(r instanceof PartitionedRegion)) {
          return;
        }

        CacheDistributionAdvisor cda = (CacheDistributionAdvisor) r.getDistributionAdvisor();
        CacheDistributionAdvisor.CacheProfile cp;

        // prevent adding the message to queue after we have processed the queue
        // in CreateRegionReplyProcessor.process
        synchronized (cda) {
          cp = (CacheDistributionAdvisor.CacheProfile) cda.getProfile(getSender());
          if (cp == null) {
            // only need to hold the lock if cda doesn't have the profile yet. This makes sure
            // we add the message to the queue before they are processed
            if (logger.isDebugEnabled()) {
              logger.debug(
                  "No cache profile to update, adding filter profile message to queue. Message :{}",
                  this);
            }
            FilterProfile localFP = ((LocalRegion) r).getFilterProfile();
            localFP.addToFilterProfileQueue(getSender(), this);
            dm.getCancelCriterion().checkCancelInProgress(null);
          }
        }

        if (cp != null) {
          cp.hasCacheServer = true;
          FilterProfile fp = cp.filterProfile;
          if (fp == null) { // PR accessors do not keep filter profiles around
            if (logger.isDebugEnabled()) {
              logger.debug("No filter profile to update: {}", this);
            }
          } else {
            if (logger.isDebugEnabled()) {
              logger.debug("Processing the filter profile request for : {}", this);
            }
            fp.region = (LocalRegion) r;
            processRequest(fp);
          }
        }
      } catch (RuntimeException e) {
        logger.warn("Exception thrown while processing profile update", e);
      } finally {
        ReplyMessage reply = new ReplyMessage();
        reply.setProcessorId(processorId);
        reply.setRecipient(getSender());
        try {
          dm.putOutgoing(reply);
        } catch (CancelException ignore) {
          // can't send a reply, so ignore the exception
        }
      }
    }
OperationMessage synchronously propagates a change in the profile to another member. It is a
serial message so that there is no chance of out-of-order execution.
geode-core/src/main/java/org/apache/geode/internal/cache/FilterProfile.java