@Override
  public <E> CqResults<E> executeWithInitialResults()
      throws CqClosedException, RegionNotFoundException, CqException {

    synchronized (queuedEventsSynchObject) {
      // Prevents multiple calls to executeWithInitialResults from squishing
      // each others queuedEvents. Forces the second call to wait
      // until first call is completed.
      while (queuedEvents != null) {
        try {
          queuedEventsSynchObject.wait();
        } catch (InterruptedException e) {
          Thread.currentThread().interrupt();
        }
      }
      // At this point we know queuedEvents is null and no one is adding to queuedEvents yet.
      queuedEvents = new ConcurrentLinkedQueue<>();
    }

    if (CqQueryImpl.testHook != null) {
      testHook.pauseUntilReady();
    }
    // Send CQ request to servers.
    // If an exception is thrown, we need to clean up the queuedEvents
    // or else client will hang on next executeWithInitialResults
    CqResults<E> initialResults;
    try {
      initialResults = (CqResults<E>) executeCqOnRedundantsAndPrimary(true);
    } catch (RegionNotFoundException | CqException | RuntimeException e) {
      queuedEvents = null;
      throw e;
    }

    // lock was released earlier so that events that are received while executing
    // initial results can be added to the queue.
    synchronized (queuedEventsSynchObject) {
      // Check if any CQ Events are received during query execution.
      // Invoke the CQ Listeners with the received CQ Events.
      try {
        if (!queuedEvents.isEmpty()) {
          try {
            Thread thread = new LoggingThread("CQEventHandler For " + cqName, () -> {
              Object[] eventArray;
              if (CqQueryImpl.testHook != null) {
                testHook.setEventCount(queuedEvents.size());
              }
              // Synchronization for the executer thread.
              synchronized (queuedEventsSynchObject) {
                try {
                  eventArray = queuedEvents.toArray();

                  // Process through the events
                  for (Object cqEvent : eventArray) {
                    cqService.invokeListeners(cqName, this, (CqEventImpl) cqEvent);
                    stats.decQueuedCqListenerEvents();
                  }
                } finally {
                  // Make sure that we notify waiting threads or else possible dead lock
                  queuedEvents.clear();
                  queuedEvents = null;
                  queuedEventsSynchObject.notify();
                }
              }
            });
            thread.start();
          } catch (Exception ex) {
            if (logger.isDebugEnabled()) {
              logger.debug("Exception while invoking the CQ Listener with queued events.", ex);
            }
          }
        } else {
          queuedEvents = null;
        }
      } finally {
        queuedEventsSynchObject.notify();
      }
      return initialResults;
    }
  }
Start or resume executing the query. Gets or updates the CQ results and returns them.
geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/ClientCQImpl.java