public Index createIndex(boolean remotelyOriginated, IndexType indexType, String indexName,
      String indexedExpression, String fromClause, String imports, boolean loadEntries,
      boolean sendMessage) throws ForceReattemptException, IndexCreationException,
      IndexNameConflictException, IndexExistsException {
    // Check if its remote request and this vm is an accessor.
    if (remotelyOriginated && dataStore == null) {
      // This check makes sure that for some region this vm cannot create
      // data store where as it should have.
      if (getLocalMaxMemory() != 0) {
        throw new IndexCreationException(
            String.format(
                "Data Store on this vm is null and the local max Memory is not zero, the data policy is %s and the localMaxMemeory is : %s",
                getDataPolicy(), (long) getLocalMaxMemory()));
      }
      // Not have to do anything since the region is just an Accessor and
      // does not store any data.
      logger.info("This is an accessor vm and doesnt contain data");
      return null;
    }

    // Create indexManager.
    if (indexManager == null) {
      indexManager = IndexUtils.getIndexManager(cache, this, true);
    }

    if (logger.isDebugEnabled()) {
      logger.debug(
          "Started creating index with Index Name :{} On PartitionedRegion {}, Indexfrom caluse={}, Remote Request: {}",
          indexName, getFullPath(), fromClause, remotelyOriginated);
    }
    IndexTask indexTask = new IndexTask(remotelyOriginated, indexType, indexName, indexedExpression,
        fromClause, imports, loadEntries);

    FutureTask<Index> indexFutureTask = new FutureTask<>(indexTask);

    // This will return either the Index FutureTask or Index itself, based
    // on whether the index creation is in process or completed.
    Object ind = indexes.putIfAbsent(indexTask, indexFutureTask);

    // Check if its instance of Index, in that the case throw index exists exception.
    if (ind instanceof Index) {
      if (remotelyOriginated) {
        return (Index) ind;
      }

      throw new IndexNameConflictException(
          String.format("Index named ' %s ' already exists.", indexName));
    }

    FutureTask<Index> oldIndexFutureTask = (FutureTask<Index>) ind;
    Index index = null;
    boolean interrupted = false;

    try {
      if (oldIndexFutureTask == null) {
        // Index doesn't exist, create index.
        indexFutureTask.run();
        index = indexFutureTask.get();
        if (index != null) {
          indexes.put(indexTask, index);
          PartitionedIndex prIndex = (PartitionedIndex) index;
          indexManager.addIndex(indexName, index);

          // Locally originated create index request.
          // Send create request to other PR nodes.
          if (!remotelyOriginated && sendMessage) {
            logger.info(
                "Created index locally, sending index creation message to all members, and will be waiting for response {}.",
                prIndex);
            HashSet<IndexCreationData> singleIndexDefinition = new HashSet<>();
            IndexCreationData icd = new IndexCreationData(indexName);
            icd.setIndexData(indexType, fromClause, indexedExpression, imports, loadEntries);
            singleIndexDefinition.add(icd);

            IndexCreationMsg.IndexCreationResponse response = null;
            try {
              response = (IndexCreationMsg.IndexCreationResponse) IndexCreationMsg.send(null,
                  this, singleIndexDefinition);
              if (response != null) {
                IndexCreationMsg.IndexCreationResult result = response.waitForResult();
                Map<String, Integer> indexBucketsMap = result.getIndexBucketsMap();
                if (indexBucketsMap != null && indexBucketsMap.size() > 0) {
                  prIndex.setRemoteBucketesIndexed(indexBucketsMap.values().iterator().next());
                }
              }
            } catch (UnsupportedOperationException ignore) {
              // if remote nodes are of older versions indexes will not be created there, so remove
              // index on this node as well.
              indexes.remove(index);
              indexManager.removeIndex(index);
              throw new IndexCreationException(
                  "Indexes should not be created when there are older versions of gemfire in the cluster.");
            }
          }
        }
      } else {
        // Some other thread is trying to create the same index.
        // Wait for index to be initialized from other thread.
        index = oldIndexFutureTask.get();
        // The Index is successfully created, throw appropriate error message from this thread.

        if (remotelyOriginated) {
          return index;
        }

        throw new IndexNameConflictException(
            String.format("Index named ' %s ' already exists.",
                indexName));
      }
    } catch (InterruptedException ignore) {
      interrupted = true;
    } catch (ExecutionException ee) {
      if (!remotelyOriginated) {
        Throwable cause = ee.getCause();
        if (cause instanceof IndexNameConflictException) {
          throw (IndexNameConflictException) cause;
        } else if (cause instanceof IndexExistsException) {
          throw (IndexExistsException) cause;
        }
        throw new IndexInvalidException(ee);
      }
    } finally {
      // If the index is not successfully created, remove IndexTask from the map.
      if (index == null) {
        ind = indexes.get(indexTask);
        if (ind != null && !(ind instanceof Index)) {
          indexes.remove(indexTask);
        }
      }

      if (interrupted) {
        Thread.currentThread().interrupt();
      }
    }
    if (logger.isDebugEnabled()) {
      logger.debug(
          "Completed creating index with Index Name :{} On PartitionedRegion {}, Remote Request: {}",
          indexName, getFullPath(), remotelyOriginated);
    }
    return index;
  }
Creates the actual index on this partitioned regions.

@param remotelyOriginated true if the index is created because of a remote index creation call
@param indexType the type of index created.
@param indexName the name for the index to be created
@param indexedExpression expression for index creation.
@param fromClause the from clause for index creation
@param imports class to be imported for fromClause.

@return Index an index created on this region.
@throws ForceReattemptException indicating the operation failed to create a remote index
@throws IndexCreationException if the index is not created properly
@throws IndexNameConflictException if an index exists with this name on this region
@throws IndexExistsException if and index already exists with the same properties as the one
        created
geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegion.java