protected void processQueue() {
    final boolean isDebugEnabled = logger.isDebugEnabled();
    final boolean isTraceEnabled = logger.isTraceEnabled();

    final GatewaySenderStats statistics = sender.getStatistics();

    if (isDebugEnabled) {
      logger.debug("STARTED processQueue {}", getId());
    }
    // list of the events peeked from queue
    List<GatewaySenderEventImpl> events = null;
    // list of the PDX events which are peeked from pDX region and needs to go acrossthe site
    List<GatewaySenderEventImpl> pdxEventsToBeDispatched = new ArrayList<>();
    // list of filteredList + pdxEventsToBeDispatched events
    List<GatewaySenderEventImpl> eventsToBeDispatched = new ArrayList<>();

    for (;;) {
      if (stopped()) {
        resetLastPeekedEvents = true;
        break;
      }

      try {
        // Check if paused. If so, wait for resumption
        if (isPaused) {
          waitForResumption();
        }

        // Peek a batch
        if (isDebugEnabled) {
          logger.debug("Attempting to peek a batch of {} events", batchSize);
        }
        for (;;) {
          // check before sleeping
          if (stopped()) {
            resetLastPeekedEvents = true;
            if (isDebugEnabled) {
              logger.debug(
                  "GatewaySenderEventProcessor is stopped. Returning without peeking events.");
            }
            break;
          }

          // Check if paused. If so, wait for resumption
          if (isPaused) {
            waitForResumption();
          }

          // sleep a little bit, look for events
          boolean interrupted = Thread.interrupted();
          try {
            if (resetLastPeekedEvents) {
              notifyPossibleDuplicate(RESET_BATCH, pendingEventsInBatches());
              resetLastPeekedEvents();
              resetLastPeekedEvents = false;
            }

            {
              // Below code was added to consider the case of queue region is
              // destroyed due to userPRs localdestroy or destroy operation.
              // In this case we were waiting for queue region to get created
              // and then only peek from the region queue.
              // With latest change of multiple PR with single ParalleSender, we
              // cant wait for particular regionqueue to get recreated as there
              // will be other region queue from which events can be picked

              /*
               * // Check if paused. If so, wait for resumption if (this.isPaused) {
               * waitForResumption(); }
               *
               * synchronized (this.getQueue()) { // its quite possible that the queue region is //
               * destroyed(userRegion // localdestroy destroys shadow region locally). In this case
               * // better to // wait for shadows region to get recreated instead of keep loop //
               * for peeking events if (this.getQueue().getRegion() == null ||
               * this.getQueue().getRegion().isDestroyed()) { try { this.getQueue().wait();
               * continue; // this continue is important to recheck the // conditions of stop/ pause
               * after the wait of 1 sec } catch (InterruptedException e1) {
               * Thread.currentThread().interrupt(); } } }
               */
            }
            events = queue.peek(batchSize, batchTimeInterval);
          } catch (InterruptedException e) {
            interrupted = true;
            sender.getCancelCriterion().checkCancelInProgress(e);
            continue; // keep trying
          } finally {
            if (interrupted) {
              Thread.currentThread().interrupt();
            }
          }
          if (events.isEmpty()) {
            continue; // nothing to do!
          }

          beforeExecute();
          try {
            // this list is access by ack reader thread so create new every time. #50220

            List<GatewaySenderEventImpl> filteredList = new ArrayList<>(events);

            // If the exception has been set and its cause is an IllegalStateExcetption,
            // remove all events whose serialized value is no longer available
            if (exception != null && exception.getCause() != null
                && exception.getCause() instanceof IllegalStateException) {
              filteredList.removeIf(GatewaySenderEventImpl::isSerializedValueNotAvailable);
              exception = null;
            }

            // Filter the events
            for (GatewayEventFilter filter : sender.getGatewayEventFilters()) {
              Iterator<GatewaySenderEventImpl> itr = filteredList.iterator();
              while (itr.hasNext()) {
                GatewayQueueEvent<?, ?> event = itr.next();

                // This seems right place to prevent transmission of UPDATE_VERSION events if
                // receiver's version is < 7.0.1, especially to prevent another loop over events.
                if (event.getOperation() == Operation.UPDATE_VERSION_STAMP) {
                  if (isTraceEnabled) {
                    logger.trace(
                        "Update Event Version event: {} removed from Gateway Sender queue: {}",
                        event, sender);
                  }

                  itr.remove();
                  statistics.incEventsNotQueued();
                  continue;
                }

                boolean transmit = filter.beforeTransmit(event);
                if (!transmit) {
                  if (isDebugEnabled) {
                    logger.debug("{}: Did not transmit event due to filtering: {}", sender.getId(),
                        event);
                  }
                  itr.remove();
                  statistics.incEventsFiltered();
                }
              }
            }

            // filter out the events with CME
            Iterator<GatewaySenderEventImpl> cmeItr = filteredList.iterator();
            while (cmeItr.hasNext()) {
              GatewaySenderEventImpl event = cmeItr.next();
              if (event.isConcurrencyConflict()) {
                cmeItr.remove();
                logger.debug("The CME event: {} is removed from Gateway Sender queue: {}", event,
                    sender);
                statistics.incEventsNotQueued();
              }
            }

            /*
             * if (filteredList.isEmpty()) { eventQueueRemove(events.size()); continue; }
             */

            // if the bucket becomes secondary after the event is picked from it,
            // check again before dispatching the event. Do this only for
            // AsyncEventQueue since possibleDuplicate flag is not used in WAN.
            if (getSender().isParallel()
                && (getDispatcher() instanceof GatewaySenderEventCallbackDispatcher)) {
              for (final GatewaySenderEventImpl event : filteredList) {
                final PartitionedRegion qpr;
                if (getQueue() instanceof ConcurrentParallelGatewaySenderQueue) {
                  qpr = ((ConcurrentParallelGatewaySenderQueue) getQueue())
                      .getRegion(event.getRegionPath());
                } else {
                  qpr = ((ParallelGatewaySenderQueue) getQueue())
                      .getRegion(event.getRegionPath());
                }
                int bucketId = event.getBucketId();
                // if the bucket from which the event has been picked is no longer
                // primary, then set possibleDuplicate to true on the event
                if (qpr != null) {
                  BucketRegion bucket = qpr.getDataStore().getLocalBucketById(bucketId);
                  if (bucket == null || !bucket.getBucketAdvisor().isPrimary()) {
                    event.setPossibleDuplicate(true);
                    if (isDebugEnabled) {
                      logger.debug(
                          "Bucket id: {} is no longer primary on this node. The event: {} will be dispatched from this node with possibleDuplicate set to true.",
                          bucketId, event);
                    }
                  }
                }
              }
            }

            eventsToBeDispatched.clear();
            if (!(dispatcher instanceof GatewaySenderEventCallbackDispatcher)) {
              // store the batch before dispatching so it can be retrieved by the ack thread.
              List<GatewaySenderEventImpl>[] eventsArr = uncheckedCast(new List[2]);
              eventsArr[0] = events;
              eventsArr[1] = filteredList;
              batchIdToEventsMap.put(getBatchId(), eventsArr);
              // find out PDX event and append it in front of the list
              pdxEventsToBeDispatched = addPDXEvent();
              eventsToBeDispatched.addAll(pdxEventsToBeDispatched);
              if (!pdxEventsToBeDispatched.isEmpty()) {
                batchIdToPDXEventsMap.put(getBatchId(), pdxEventsToBeDispatched);
              }
            }

            eventsToBeDispatched.addAll(filteredList);

            // Conflate the batch. Event conflation only occurs on the queue.
            // Once an event has been peeked into a batch, it won't be
            // conflated. So if events go through the queue quickly (as in the
            // no-ack case), then multiple events for the same key may end up in
            // the batch.
            List<GatewaySenderEventImpl> conflatedEventsToBeDispatched =
                conflate(eventsToBeDispatched);

            if (isDebugEnabled) {
              logBatchFine("During normal processing, dispatching the following ",
                  conflatedEventsToBeDispatched);
            }

            boolean success = dispatcher.dispatchBatch(conflatedEventsToBeDispatched,
                sender.isRemoveFromQueueOnException(), false);
            if (success) {
              if (isDebugEnabled) {
                logger.debug(
                    "During normal processing, successfully dispatched {} events (batch #{})",
                    conflatedEventsToBeDispatched.size(), getBatchId());
              }
              removeEventFromFailureMap(getBatchId());
            } else {
              if (!skipFailureLogging(getBatchId())) {
                logger.warn(
                    "During normal processing, unsuccessfully dispatched {} events (batch #{})",
                    new Object[] {filteredList.size(), getBatchId()});
              }
            }
            // check again, don't do post-processing if we're stopped.
            if (stopped()) {
              resetLastPeekedEvents = true;
              break;
            }

            // If the batch is successfully processed, remove it from the queue.
            if (success) {
              if (dispatcher instanceof GatewaySenderEventCallbackDispatcher) {
                handleSuccessfulBatchDispatch(conflatedEventsToBeDispatched, events);
              } else {
                incrementBatchId();
              }

              // pdx related gateway sender events needs to be updated for
              // isDispatched
              for (GatewaySenderEventImpl pdxGatewaySenderEvent : pdxEventsToBeDispatched) {
                pdxGatewaySenderEvent.isDispatched = true;
              }

              increaseNumEventsDispatched(conflatedEventsToBeDispatched.size());
            } // successful batch
            else { // The batch was unsuccessful.
              if (dispatcher instanceof GatewaySenderEventCallbackDispatcher) {
                handleUnSuccessfulBatchDispatch(events);
                resetLastPeekedEvents = true;
              } else {
                handleUnSuccessfulBatchDispatch(events);
                if (!resetLastPeekedEvents) {
                  while (!dispatcher.dispatchBatch(conflatedEventsToBeDispatched,
                      sender.isRemoveFromQueueOnException(), true)) {
                    if (isDebugEnabled) {
                      logger.debug(
                          "During normal processing, unsuccessfully dispatched {} events (batch #{})",
                          conflatedEventsToBeDispatched.size(), getBatchId());
                    }
                    if (stopped()) {
                      resetLastPeekedEvents = true;
                      break;
                    }
                    if (resetLastPeekedEvents) {
                      break;
                    }
                    try {
                      if (threadMonitoring != null) {
                        threadMonitoring.updateThreadStatus();
                      }
                      Thread.sleep(100);
                    } catch (InterruptedException ie) {
                      Thread.currentThread().interrupt();
                    }
                  }
                  if (!resetLastPeekedEvents) {
                    incrementBatchId();
                  }
                }
              }
            } // unsuccessful batch
            if (logger.isDebugEnabled()) {
              logger.debug("Finished processing events (batch #{})", (getBatchId() - 1));
            }
          } finally {
            afterExecute();
          }
        } // for
      } catch (RegionDestroyedException e) {
        // setting this flag will ensure that already peeked events will make
        // it to the next batch before new events are peeked (fix for #48784)
        resetLastPeekedEvents = true;
        // most possible case is ParallelWan when user PR is locally destroyed
        // shadow PR is also locally destroyed
        if (logger.isDebugEnabled()) {
          logger.debug("Observed RegionDestroyedException on Queue's region.");
        }
      } catch (CancelException e) {
        logger.debug("Caught cancel exception", e);
        setIsStopped(true);
      } 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 e) {
        // 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();

        // Well, OK. Some strange nonfatal thing.
        if (stopped()) {
          return; // don't complain, just exit.
        }

        if (events != null) {
          handleUnSuccessfulBatchDispatch(events);
        }
        resetLastPeekedEvents = true;

        if (e instanceof GatewaySenderException) {
          Throwable cause = e.getCause();
          if (cause instanceof IOException || e instanceof GatewaySenderConfigurationException) {
            continue;
          }
        }

        // We'll log it but continue on with the next batch.
        logger.warn("An Exception occurred. The dispatcher will continue.",
            e);
      }
    } // for
  }
After a successful batch execution remove from failure map if present (i.e. if the event had
failed on a previous try).
geode-core/src/main/java/org/apache/geode/internal/cache/wan/AbstractGatewaySenderEventProcessor.java