private Object executeCqOnRedundantsAndPrimary(boolean executeWithInitialResults)
      throws CqClosedException, RegionNotFoundException, CqException {

    Object initialResults = null;

    synchronized (cqState) {
      if (isClosed()) {
        throw new CqClosedException(
            String.format("CQ is closed, CqName : %s", cqName));
      }
      if (isRunning()) {
        throw new IllegalStateException(String.format("CQ is in running state, CqName : %s",
            cqName));
      }

      if (logger.isDebugEnabled()) {
        logger.debug("Performing Execute {} request for CQ. CqName: {}",
            ((executeWithInitialResults) ? "WithInitialResult" : ""), cqName);
      }
      cqBaseRegion = (LocalRegion) cqService.getCache().getRegion(regionName);

      // If not server send the request to server.
      if (!cqService.isServer()) {
        // Send execute request to the server.

        // If CqService is initialized using the pool, the connection proxy is set using the
        // pool that initializes the CQ. Else its set using the Region proxy.
        if (cqProxy == null) {
          initConnectionProxy();
        }

        boolean success = false;
        try {
          if (proxyCache != null) {
            if (proxyCache.isClosed()) {
              throw proxyCache.getCacheClosedException("Cache is closed for this user.");
            }
            UserAttributes.userAttributes.set(proxyCache.getUserAttributes());
          }
          if (executeWithInitialResults) {
            initialResults = cqProxy.createWithIR(this);
            if (initialResults == null) {
              String errMsg = "Failed to execute the CQ.  CqName: " + cqName
                  + ", Query String is: " + queryString;
              throw new CqException(errMsg);
            }
          } else {
            cqProxy.create(this);
          }
          success = true;
        } catch (Exception ex) {
          // Check for system shutdown.
          if (shutdownInProgress()) {
            throw new CqException("System shutdown in progress.");
          }
          if (ex.getCause() instanceof GemFireSecurityException) {
            if (securityLogWriter.warningEnabled()) {
              securityLogWriter.warning(
                  String.format("Exception while executing cq Exception: %s", ex));
            }
            throw new CqException(ex.getCause().getMessage(), ex.getCause());
          } else if (ex instanceof CqException) {
            throw (CqException) ex;
          } else {
            String errMsg =
                String.format(
                    "Failed to execute the CQ. CqName: %s, Query String is: %s, Error from last server: %s",
                    cqName, queryString, ex.getLocalizedMessage());
            if (logger.isDebugEnabled()) {
              logger.debug(errMsg, ex);
            }
            throw new CqException(errMsg, ex);
          }
        } finally {
          if (!success && !shutdownInProgress()) {
            try {
              cqProxy.close(this);
            } catch (Exception e) {
              if (logger.isDebugEnabled()) {
                logger.debug("Exception cleaning up failed cq", e);
              }
              UserAttributes.userAttributes.set(null);
            }
          }
          UserAttributes.userAttributes.set(null);
        }
      }
      cqState.setState(CqStateImpl.RUNNING);
    }
    // If client side, alert listeners that a cqs have been connected
    if (!cqService.isServer()) {
      connected = true;
      CqListener[] cqListeners = getCqAttributes().getCqListeners();
      for (CqListener cqListener : cqListeners) {
        if (cqListener != null) {
          if (cqListener instanceof CqStatusListener) {
            CqStatusListener listener = (CqStatusListener) cqListener;
            listener.onCqConnected();
          }
        }
      }
    }
    // Update CQ-base region for book keeping.
    cqService.stats().incCqsActive();
    cqService.stats().decCqsStopped();
    return initialResults;
  }
This executes the CQ first on the redundant server and then on the primary server. This is
required to keep the redundancy behavior in accordance with the HAQueue expectation (wherein
the events are delivered only from the primary).

@param executeWithInitialResults boolean
@return Object SelectResults in case of executeWithInitialResults
geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/ClientCQImpl.java