private void processEntryEvent(CacheEvent event, Profile localProfile, Profile[] profiles,
      FilterRoutingInfo frInfo) throws CqException {
    final boolean isDebugEnabled = logger.isDebugEnabled();
    HashSet<Object> cqUnfilteredEventsSet_newValue = new HashSet<>();
    HashSet<Object> cqUnfilteredEventsSet_oldValue = new HashSet<>();
    boolean b_cqResults_newValue;
    boolean b_cqResults_oldValue;
    boolean queryOldValue;
    EntryEvent entryEvent = (EntryEvent) event;
    Object eventKey = entryEvent.getKey();

    boolean isDupEvent = ((EntryEventImpl) event).isPossibleDuplicate();
    // The CQ query needs to be applied when the op is update, destroy
    // invalidate and in case when op is create and its an duplicate
    // event, the reason for this is when peer sends a duplicate event
    // it marks it as create and sends it, so that the receiving node
    // applies it (see DR.virtualPut()).
    boolean opRequiringQueryOnOldValue = (event.getOperation().isUpdate()
        || event.getOperation().isDestroy() || event.getOperation().isInvalidate()
        || (event.getOperation().isCreate() && isDupEvent));

    HashMap<String, MessageType> matchedCqs = new HashMap<>();
    long executionStartTime;
    for (int i = -1; i < profiles.length; i++) {
      CacheProfile cf;
      if (i < 0) {
        cf = (CacheProfile) localProfile;
        if (cf == null) {
          continue;
        }
      } else {
        cf = (CacheProfile) profiles[i];
      }
      FilterProfile pf = cf.filterProfile;
      if (pf == null || pf.getCqMap().isEmpty()) {
        continue;
      }

      Map<?, ServerCQImpl> cqs = pf.getCqMap();

      if (isDebugEnabled) {
        logger.debug("Profile for {} processing {} CQs", cf.peerMemberId, cqs.size());
      }

      if (cqs.isEmpty()) {
        continue;
      }

      // Get new value. If its not retrieved.
      if (cqUnfilteredEventsSet_newValue.isEmpty()
          && (event.getOperation().isCreate() || event.getOperation().isUpdate())) {
        Object newValue = entryEvent.getNewValue();
        if (newValue != null) {
          // We have a new value to run the query on
          cqUnfilteredEventsSet_newValue.add(newValue);
        }
      }

      HashMap<Long, MessageType> cqInfo = new HashMap<>();

      for (Map.Entry<?, ServerCQImpl> cqEntry : cqs.entrySet()) {
        ServerCQImpl cQuery = cqEntry.getValue();
        b_cqResults_newValue = false;
        b_cqResults_oldValue = false;
        queryOldValue = false;
        if (cQuery == null) {
          continue;
        }
        String cqName = cQuery.getServerCqName();
        Long filterID = cQuery.getFilterID();

        if (isDebugEnabled) {
          logger.debug("Processing CQ : {} Key: {}", cqName, eventKey);
        }

        MessageType cqEvent = null;
        if (matchedCqs.containsKey(cqName)) {
          cqEvent = matchedCqs.get(cqName);
          if (isDebugEnabled) {
            logger.debug("query {} has already been processed and returned {}", cqName, cqEvent);
          }
          if (cqEvent == null) {
            continue;
          }
          // Update the Cache Results for this CQ.
          if (cqEvent == LOCAL_CREATE
              || cqEvent == LOCAL_UPDATE) {
            cQuery.addToCqResultKeys(eventKey);
          } else if (cqEvent == LOCAL_DESTROY) {
            cQuery.markAsDestroyedInCqResultKeys(eventKey);
          }
        } else {
          boolean error = false;
          {
            try {
              // Apply query on new value.
              if (!cqUnfilteredEventsSet_newValue.isEmpty()) {
                executionStartTime = stats.startCqQueryExecution();

                synchronized (cQuery) {
                  b_cqResults_newValue =
                      evaluateQuery(cQuery, new Object[] {cqUnfilteredEventsSet_newValue});
                }

                stats.endCqQueryExecution(executionStartTime);
              }

              // In case of Update, destroy and invalidate.
              // Apply query on oldValue.
              if (opRequiringQueryOnOldValue) {
                // Check if CQ Result is cached, if not apply query on old
                // value. Currently the CQ Results are not cached for the
                // Partitioned Regions. Once this is added remove the check
                // with PR region.
                if (cQuery.isCqResultsCacheInitialized()) {
                  b_cqResults_oldValue =
                      (cQuery.isPartOfCqResult(eventKey) && !cQuery.isKeyDestroyed(eventKey));
                  // For PR if not found in cache, apply the query on old value.
                  // Also apply if the query was not executed during cq execute
                  if ((cQuery.isPR || !CqServiceImpl.EXECUTE_QUERY_DURING_INIT)
                      && b_cqResults_oldValue == false) {
                    queryOldValue = true;
                  }
                  if (isDebugEnabled && !cQuery.isPR && !b_cqResults_oldValue) {
                    logger.debug(
                        "Event Key not found in the CQ Result Queue. EventKey : {} CQ Name : {}",
                        eventKey, cqName);
                  }
                } else {
                  queryOldValue = true;
                }

                if (queryOldValue) {
                  if (cqUnfilteredEventsSet_oldValue.isEmpty()) {
                    Object oldValue = entryEvent.getOldValue();
                    if (oldValue != null) {
                      cqUnfilteredEventsSet_oldValue.add(oldValue);
                    }
                  }

                  // Apply query on old value.
                  if (!cqUnfilteredEventsSet_oldValue.isEmpty()) {
                    executionStartTime = stats.startCqQueryExecution();

                    synchronized (cQuery) {
                      b_cqResults_oldValue =
                          evaluateQuery(cQuery, new Object[] {cqUnfilteredEventsSet_oldValue});
                    }

                    stats.endCqQueryExecution(executionStartTime);
                  } else {
                    if (isDebugEnabled) {
                      logger.debug(
                          "old value for event with key {} is null - query execution not performed",
                          eventKey);
                    }
                  }
                } // Query oldValue
              }
            } catch (Exception ex) {
              // Any exception in running the query should be caught here and
              // buried because this code is running in-line with the message
              // processing code and we don't want to kill that thread
              error = true;
              // CHANGE LOG MESSAGE:
              logger.info("Error while processing CQ on the event, key : {} CqName: {}, Error: {}",
                  new Object[] {((EntryEvent) event).getKey(), cQuery.getName(),
                      ex.getLocalizedMessage()});
            }

            if (error) {
              cqEvent = EXCEPTION;
            } else {
              if (b_cqResults_newValue) {
                if (b_cqResults_oldValue) {
                  cqEvent = LOCAL_UPDATE;
                } else {
                  cqEvent = LOCAL_CREATE;
                }
                // If its create and caching is enabled, cache the key
                // for this CQ.
                cQuery.addToCqResultKeys(eventKey);
              } else if (b_cqResults_oldValue) {
                // Base invalidate operation is treated as destroy.
                // When the invalidate comes through, the entry will no longer
                // satisfy the query and will need to be deleted.
                cqEvent = LOCAL_DESTROY;
                // If caching is enabled, mark this event's key as removed
                // from the CQ cache.
                cQuery.markAsDestroyedInCqResultKeys(eventKey);
              }
            }
          }

          // Get the matching CQs if any.
          // synchronized (this.matchingCqMap){
          String query = cQuery.getQueryString();
          Set matchingCqs = matchingCqMap.get(query);
          if (matchingCqs != null) {
            for (Object matchingCq : matchingCqs) {
              String matchingCqName = (String) matchingCq;
              if (!matchingCqName.equals(cqName)) {
                matchedCqs.put(matchingCqName, cqEvent);
                if (isDebugEnabled) {
                  logger.debug("Adding CQ into Matching CQ Map: {} Event is: {}", matchingCqName,
                      cqEvent);
                }
              }
            }
          }
        }

        if (cqEvent != null && cQuery.isRunning()) {
          if (isDebugEnabled) {
            logger.debug("Added event to CQ with client-side name: {} key: {} operation : {}",
                cQuery.cqName, eventKey, cqEvent);
          }
          cqInfo.put(filterID, cqEvent);
          CqQueryVsdStats stats = cQuery.getVsdStats();
          if (stats != null) {
            stats.updateStats(cqEvent);
          }
        }
      }
      if (cqInfo.size() > 0) {
        if (pf.isLocalProfile()) {
          if (isDebugEnabled) {
            logger.debug("Setting local CQ matches to {}", cqInfo);
          }
          frInfo.setLocalCqInfo(cqInfo);
        } else {
          if (isDebugEnabled) {
            logger.debug("Setting CQ matches for {} to {}", cf.getDistributedMember(), cqInfo);
          }
          frInfo.setCqRoutingInfo(cf.getDistributedMember(), cqInfo);
        }
      }
    } // iteration over Profiles.
  }
Returns the Operation for the given EnumListenerEvent type.
geode-cq/src/main/java/org/apache/geode/cache/query/cq/internal/CqServiceImpl.java