@Override
  public void close(boolean sendRequestToServer) throws CqClosedException, CqException {
    final boolean isDebugEnabled = logger.isDebugEnabled();
    if (isDebugEnabled) {
      logger.debug("Started closing CQ CqName: {} SendRequestToServer: {}", cqName,
          sendRequestToServer);
    }
    // Synchronize with stop and execute CQ commands
    synchronized (cqState) {
      // Check if the cq is already closed.
      if (isClosed()) {
        // throw new CqClosedException("CQ is already closed, CqName : " + this.cqName);
        if (isDebugEnabled) {
          logger.debug("CQ is already closed, CqName: {}", cqName);
        }
        return;
      }

      int stateBeforeClosing = cqState.getState();
      cqState.setState(CqStateImpl.CLOSING);
      boolean isClosed = false;

      // Client Close. Proxy is null in case of server.
      // Check if this has been sent to server, if so send
      // Close request to server.
      Exception exception = null;
      if (cqProxy != null && sendRequestToServer) {
        try {
          if (proxyCache != null) {
            if (proxyCache.isClosed()) {
              throw proxyCache.getCacheClosedException("Cache is closed for this user.");
            }
            UserAttributes.userAttributes.set(proxyCache.getUserAttributes());
          }
          cqProxy.close(this);
          isClosed = true;
        } catch (CancelException e) {
          throw e;
        } catch (Exception ex) {
          if (shutdownInProgress()) {
            return;
          }
          exception = ex;
        } finally {
          UserAttributes.userAttributes.set(null);
        }
      }

      // Cleanup the resource used by cq.
      removeFromCqMap();

      if (cqProxy == null || !sendRequestToServer || isClosed) {
        // Stat update.
        if (stateBeforeClosing == CqStateImpl.RUNNING) {
          cqService.stats().decCqsActive();
        } else if (stateBeforeClosing == CqStateImpl.STOPPED) {
          cqService.stats().decCqsStopped();
        }

        // Set the state to close, and update stats
        cqState.setState(CqStateImpl.CLOSED);
        cqService.stats().incCqsClosed();
        cqService.stats().decCqsOnClient();
        if (stats != null) {
          stats.close();
        }
      } else {
        if (shutdownInProgress()) {
          return;
        }
        // Hasn't able to send close request to any server.
        if (exception != null) {
          throw new CqException(
              String.format("Failed to close the cq. CqName: %s. Error from last endpoint: %s",
                  cqName, exception.getLocalizedMessage()),
              exception.getCause());
        } else {
          throw new CqException(
              String.format(
                  "Failed to close the cq. CqName: %s. The server endpoints on which this cq was registered were not found.",
                  cqName));
        }
      }
    }

    // Invoke close on Listeners if any.
    if (cqAttributes != null) {
      CqListener[] cqListeners = getCqAttributes().getCqListeners();

      if (cqListeners != null) {
        if (isDebugEnabled) {
          logger.debug(
              "Invoking CqListeners close() api for the CQ, CqName: {} Number of CqListeners: {}",
              cqName, cqListeners.length);
        }
        for (CqListener cqListener : cqListeners) {
          try {
            cqListener.close();
            // Handle client side exceptions.
          } catch (Exception ex) {
            logger.warn("Exception occurred in the CqListener of the CQ, CqName : {} Error : {}",
                new Object[] {cqName, ex.getLocalizedMessage()});
            if (isDebugEnabled) {
              logger.debug(ex.getMessage(), ex);
            }
          } catch (VirtualMachineError err) {
            SystemFailure.initiateFailure(err);
            // If this ever returns, rethrow the error. We're poisoned
            // now, so don't let this thread continue.
            throw err;
          } catch (Throwable t) {
            // Whenever you catch Error or Throwable, you must also
            // catch VirtualMachineError (see above). However, there is
            // _still_ a possibility that you are dealing with a cascading
            // error condition, so you also need to check to see if the JVM
            // is still usable:
            SystemFailure.checkFailure();
            logger.warn(
                "RuntimeException occurred in the CqListener of the CQ, CqName : {} Error : {}",
                new Object[] {cqName, t.getLocalizedMessage()});
            if (isDebugEnabled) {
              logger.debug(t.getMessage(), t);
            }
          }
        }
      }
    }
    if (isDebugEnabled) {
      logger.debug("Successfully closed the CQ. {}", cqName);
    }
  }
Initializes the connection using the pool from the client region. Also sets the cqBaseRegion
value of this CQ.
geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/ClientCQImpl.java