@Override
  public void process(final ClusterDistributionManager dm) {

    final boolean isDebugEnabled = logger.isDebugEnabled();

    Throwable thr = null;
    boolean sendReply = true;
    PartitionedRegion pr = null;

    try {
      if (isDebugEnabled) {
        logger.debug("Trying to get pr with id: {}", regionId);
      }
      try {
        if (isDebugEnabled) {
          logger.debug("Again trying to get pr with id : {}", regionId);
        }
        pr = PartitionedRegion.getPRFromId(regionId);
        if (isDebugEnabled) {
          logger.debug("Index creation message got the pr {}", pr);
        }
        if (null == pr) {
          boolean wait = true;
          int attempts = 0;
          while (wait && attempts < 30) { // max 30 seconds of wait.
            dm.getCancelCriterion().checkCancelInProgress(null);
            if (isDebugEnabled) {
              logger.debug(
                  "Waiting for Partitioned Region to be intialized with id {}for processing index creation messages",
                  regionId);
            }
            try {
              boolean interrupted = Thread.interrupted();
              try {
                Thread.sleep(500);
              } catch (InterruptedException e) {
                interrupted = true;
                dm.getCancelCriterion().checkCancelInProgress(e);
              } finally {
                if (interrupted) {
                  Thread.currentThread().interrupt();
                }
              }

              pr = PartitionedRegion.getPRFromId(regionId);
              if (null != pr) {
                wait = false;
                if (isDebugEnabled) {
                  logger.debug("Indexcreation message got the pr {}", pr);
                }
              }
              attempts++;
            } catch (CancelException ignorAndLoopWait) {
              if (isDebugEnabled) {
                logger.debug(
                    "IndexCreationMsg waiting for pr to be properly created with prId : {}",
                    regionId);
              }
            }
          }

        }
      } catch (CancelException letPRInitialized) {
        // Not sure if the CacheClosedException is still thrown in response
        // to the PR being initialized.
        if (logger.isDebugEnabled()) {
          logger.debug("Waiting for notification from pr being properly created on {}",
              regionId);
        }

        boolean wait = true;
        while (wait) {
          dm.getCancelCriterion().checkCancelInProgress(null);
          try {
            boolean interrupted = Thread.interrupted();
            try {
              Thread.sleep(500);
            } catch (InterruptedException e) {
              interrupted = true;
              dm.getCancelCriterion().checkCancelInProgress(e);
            } finally {
              if (interrupted) {
                Thread.currentThread().interrupt();
              }
            }
            pr = PartitionedRegion.getPRFromId(regionId);
            wait = false;
            if (logger.isDebugEnabled()) {
              logger.debug("Indexcreation message got the pr {}", pr);
            }
          } catch (CancelException ignorAndLoopWait) {
            if (logger.isDebugEnabled()) {
              logger.debug("IndexCreationMsg waiting for pr to be properly created with prId : {}",
                  regionId);
            }
          }
        }

      }

      if (pr == null /* && failIfRegionMissing() */) {
        String msg =
            String.format(
                "Could not get Partitioned Region from Id %s for message %s received on member= %s map= %s",
                regionId, this, dm.getId(),
                PartitionedRegion.dumpPRId());
        throw new PartitionedRegionException(msg, new RegionNotFoundException(msg));
      }
      sendReply = operateOnPartitionedRegion(dm, pr, 0);

    } catch (PRLocallyDestroyedException pre) {
      if (isDebugEnabled) {
        logger.debug("Region is locally Destroyed ");
      }
      thr = pre;
    } 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 t) {
      // 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();
      // log the exception at fine level if there is no reply to the message
      if (processorId == 0) {
        logger.debug("{} exception while processing message:{}", this, t.getMessage(), t);
      } else if (logger.isDebugEnabled(LogMarker.DM_VERBOSE) && (t instanceof RuntimeException)) {
        logger.debug(LogMarker.DM_VERBOSE, "Exception caught while processing message: {}",
            t.getMessage(), t);
      }
      if (t instanceof RegionDestroyedException && pr != null) {
        if (pr.isClosed) {
          logger.info("Region is locally destroyed, throwing RegionDestroyedException for {}",
              pr);
          thr = new RegionDestroyedException(
              String.format("Region is locally destroyed on %s",
                  dm.getId()),
              pr.getFullPath());
        }
      } else {
        thr = t;
      }
    } finally {
      if (sendReply && processorId != 0) {
        ReplyException rex = null;
        if (thr != null) {
          rex = new ReplyException(thr);
        }
        sendReply(getSender(), processorId, dm, rex, pr, 0);
      }
    }

  }
Process this index creation message on the receiver.
geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/IndexCreationMsg.java