private void invokeCqsConnected(Pool pool, boolean connected) {
    String poolName = pool.getName();
    // Check to see if we are already connected/disconnected.
    // If state has not changed, do not invoke another connected/disconnected
    synchronized (cqPoolsConnected) {
      // don't repeatedly send same connect/disconnect message to cq's on repeated fails of
      // RedundancySatisfier
      if (cqPoolsConnected.containsKey(poolName) && connected == cqPoolsConnected.get(poolName)) {
        return;
      }
      cqPoolsConnected.put(poolName, connected);

      Collection<? extends InternalCqQuery> cqs = getAllCqs();
      String cqName;
      final boolean isDebugEnabled = logger.isDebugEnabled();
      for (InternalCqQuery query : cqs) {
        try {
          if (query == null) {
            continue;
          }

          cqName = query.getName();
          ClientCQImpl cQuery = (ClientCQImpl) getCq(cqName);

          // Check cq pool to determine if the pool matches, if not continue.
          // Also if the connected state is already the same, we do not have to send status again.
          if (cQuery == null || cQuery.getCQProxy() == null) {
            continue;
          }
          Pool cqPool = cQuery.getCQProxy().getPool();
          if (cQuery.isConnected() == connected || !cqPool.getName().equals(poolName)) {
            continue;
          }

          if ((!cQuery.isRunning() && cQuery.getQueuedEvents() == null)) {
            if (isDebugEnabled) {
              logger.debug("Unable to invoke CqListener, CQ is Not running, CqName : {}", cqName);
            }
            continue;
          }

          invokeCqConnectedListeners(cqName, cQuery, connected);
        } catch (VirtualMachineError e) {
          SystemFailure.initiateFailure(e);
          throw e;
        } catch (Throwable t) {
          SystemFailure.checkFailure();
          logger.warn("Error while sending connection status to cq listeners", t);
        }
      }
    }
  }
Let cq listeners know that they are connected or disconnected
geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/CqServiceImpl.java