@Override
  public void registerCq(ClientProxyMembershipID p_clientProxyId,
      CacheClientNotifier p_cacheClientNotifier,
      int p_cqState) throws CqException, RegionNotFoundException {

    CacheClientProxy clientProxy = null;
    clientProxyId = p_clientProxyId;

    if (p_cacheClientNotifier != null) {
      cacheClientNotifier = p_cacheClientNotifier;
      clientProxy = p_cacheClientNotifier.getClientProxy(p_clientProxyId, true);
    }

    validateCq();

    final boolean isDebugEnabled = logger.isDebugEnabled();
    String msg = "%s";
    Throwable t = null;
    try {
      query = constructServerSideQuery();
      if (isDebugEnabled) {
        logger.debug("Server side query for the cq: {} is: {}", cqName,
            query.getQueryString());
      }
    } catch (Exception ex) {
      t = ex;
      if (ex instanceof ClassNotFoundException) {
        msg =
            "Class not found exception. The antlr.jar or the spcified class may be missing from server side classpath. Error : %s";
      } else {
        msg = "Error while parsing the query. Error : %s";
      }
    } finally {
      if (t != null) {
        String s = String.format(msg, t);
        if (isDebugEnabled) {
          logger.debug(s, t);
        }

        throw new CqException(s);
      }
    }

    // Update Regions Book keeping.
    // TODO replace getRegion() with getRegionByPathForProcessing() so this doesn't block
    // if the region is still being initialized
    cqBaseRegion = (LocalRegion) cqService.getCache().getRegion(regionName);
    if (cqBaseRegion == null) {
      throw new RegionNotFoundException(
          String.format("Region : %s specified with cq not found. CqName: %s", regionName,
              cqName));
    }

    // Make sure that the region is partitioned or
    // replicated with distributed ack or global.
    DataPolicy dp = cqBaseRegion.getDataPolicy();
    isPR = dp.withPartitioning();
    if (!(isPR || dp.withReplication())) {
      String errMsg;
      // replicated regions with eviction set to local destroy get turned into preloaded
      if (dp.withPreloaded() && cqBaseRegion.getAttributes().getEvictionAttributes() != null
          && cqBaseRegion.getAttributes().getEvictionAttributes().getAction()
              .equals(EvictionAction.LOCAL_DESTROY)) {
        errMsg =
            String.format("CQ is not supported for replicated region: %s with eviction action: %s",
                regionName, cqBaseRegion.getAttributes().getEvictionAttributes().getAction());
      } else {
        errMsg = "The region " + regionName
            + "  specified in CQ creation is neither replicated nor partitioned; "
            + "only replicated or partitioned regions are allowed in CQ creation.";
      }
      if (isDebugEnabled) {
        logger.debug(errMsg);
      }
      throw new CqException(errMsg);
    }
    if ((dp.withReplication() && (!(cqBaseRegion.getAttributes().getScope().isDistributedAck()
        || cqBaseRegion.getAttributes().getScope().isGlobal())))) {
      String errMsg = "The replicated region " + regionName
          + " specified in CQ creation does not have scope supported by CQ."
          + " The CQ supported scopes are DISTRIBUTED_ACK and GLOBAL.";
      if (isDebugEnabled) {
        logger.debug(errMsg);
      }
      throw new CqException(errMsg);
    }

    // Can be null by the time we are here
    if (clientProxy != null) {
      clientProxy.incCqCount();
      if (clientProxy.hasOneCq()) {
        cqService.stats().incClientsWithCqs();
      }
      if (isDebugEnabled) {
        logger.debug("Added CQ to the base region: {} With key as: {}", cqBaseRegion.getFullPath(),
            serverCqName);
      }
    }

    updateCqCreateStats();

    // Initialize the state of CQ.
    if (cqState.getState() != p_cqState || cacheClientNotifier == null) {
      setCqState(p_cqState);
    }

    // Register is called from both filter profile and cqService
    // In either case, if we are trying to start/run the cq, we need to add
    // it to other matching cqs for performance reasons
    if (p_cqState == CqStateImpl.RUNNING) {
      // Add to the matchedCqMap.
      cqService.addToMatchingCqMap(this);
    }

    // Initialize CQ results (key) cache.
    if (CqServiceProvider.MAINTAIN_KEYS) {
      // Currently the CQ Result keys are not cached for the Partitioned
      // Regions. Supporting this with PR needs more work like forcing
      // query execution on primary buckets only; and handling the bucket
      // re-balancing. Once this is added remove the check with PR region.
      // Only the events which are seen during event processing is
      // added to the results cache (not from the CQ Results).
      if (isPR) {
        serverCQResultsCache = new ServerCQResultsCachePartitionRegionImpl();
        setCqResultsCacheInitialized();
      } else {
        serverCQResultsCache = new ServerCQResultsCacheReplicateRegionImpl();
      }
    }

    if (p_cacheClientNotifier != null) {
      try {
        cqService.addToCqMap(this);
      } catch (CqExistsException cqe) {
        // Should not happen.
        throw new CqException(
            String.format("Unable to create cq %s Error : %s", cqName, cqe.getMessage()));
      }

      cqBaseRegion.getFilterProfile().registerCq(this);
    }
  }
identifier assigned to this query for FilterRoutingInfos
geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/ServerCQImpl.java