@Override
  protected boolean operateOnPartitionedRegion(ClusterDistributionManager dm, PartitionedRegion pr,
      long startTime) throws CacheException, ForceReattemptException {
    // region exists
    ReplyException replyEx = null;
    boolean result = false;
    List<Index> indexes = null;
    List<String> failedIndexNames = new ArrayList<>();

    if (logger.isDebugEnabled()) {
      StringBuilder sb = new StringBuilder();
      for (IndexCreationData icd : indexDefinitions) {
        sb.append(icd.getIndexName()).append(" ");
      }
      logger.debug(
          "Processing index creation message on this remote partitioned region vm for indexes: {}",
          sb);
    }

    try {
      indexes = pr.createIndexes(true, indexDefinitions);
    } catch (IndexCreationException e1) {
      replyEx = new ReplyException(
          "Remote Index Creation Failed", e1);
    } catch (MultiIndexCreationException exx) {
      failedIndexNames.addAll(exx.getExceptionsMap().keySet());

      if (logger.isDebugEnabled()) {
        StringBuilder exceptionMsgs = new StringBuilder();
        for (Exception ex : exx.getExceptionsMap().values()) {
          exceptionMsgs.append(ex.getMessage()).append("\n");
        }
        logger.debug("Got an MultiIndexCreationException with \n: {}", exceptionMsgs);
        logger.debug("{} indexes were created successfully", failedIndexNames.size());
      }
      replyEx = new ReplyException(
          "Remote Index Creation Failed", exx);
    }

    if (null == replyEx) {
      result = true;
    }

    if (result) {
      Map<String, Integer> indexBucketsMap = new HashMap<>();
      for (Index index : indexes) {
        PartitionedIndex prIndex = (PartitionedIndex) index;
        indexBucketsMap.put(prIndex.getName(), prIndex.getNumberOfIndexedBuckets());
      }
      sendReply(getSender(), getProcessorId(), dm, replyEx, result, indexBucketsMap,
          pr.getDataStore().getAllLocalBuckets().size());
    } else {
      // add the indexes that were successfully created to the map
      Map<String, Integer> indexBucketsMap = new HashMap<>();
      for (IndexCreationData icd : indexDefinitions) {
        // if the index was successfully created
        if (!failedIndexNames.contains(icd.getIndexName())) {
          PartitionedIndex prIndex = pr.getIndex(icd.getIndexName());
          indexBucketsMap.put(icd.getIndexName(), prIndex.getNumberOfIndexedBuckets());
        }
      }
      sendReply(getSender(), getProcessorId(), dm, replyEx, result, indexBucketsMap,
          pr.getDataStore().getAllLocalBuckets().size());
    }

    if (logger.isDebugEnabled()) {
      logger.debug(
          "Multi Index creation completed on remote host and has sent the reply to the originating vm.");
    }
    return false;
  }
This method actually operates on the partitioned region and creates given list of indexes from
a index creation message.

@param dm distribution manager.
@param pr partitioned region on which to create an index.
@throws CacheException indicating a cache level error
@throws ForceReattemptException if the peer is no longer available
geode-core/src/main/java/org/apache/geode/internal/cache/partitioned/IndexCreationMsg.java