@Override
  public void remove() throws InterruptedException {
    List peekedIds = (List) HARegionQueue.peekedEventsContext.get();

    if (peekedIds == null) {
      if (logger.isDebugEnabled()) {
        logger.debug("Remove() called before peek(), nothing to remove.");
      }
      return;
    }

    if (!checkPrevAcks()) {
      return;
    }

    Map groupedThreadIDs = new HashMap<>();

    for (Object peekedId : peekedIds) {
      Long counter = (Long) peekedId;

      Conflatable event = (Conflatable) region.get(counter);
      if (event != null) {
        EventID eventid = event.getEventId();
        long sequenceId = eventid.getSequenceID();
        ThreadIdentifier threadid = getThreadIdentifier(eventid);

        if (!checkEventForRemoval(counter, threadid, sequenceId)) {
          continue;
        }

        Object key = null;
        String r = null;
        if (shouldBeConflated(event)) {
          key = event.getKeyToConflate();
          r = event.getRegionToConflate();
        }
        RemovedEventInfo info = new RemovedEventInfo(counter, r, key);

        List countersList;
        if ((countersList = (List) groupedThreadIDs.get(threadid)) != null) {
          countersList.add(info);
          countersList.set(0, sequenceId);
        } else {
          countersList = new ArrayList();
          countersList.add(sequenceId);
          countersList.add(info);
          groupedThreadIDs.put(threadid, countersList);
        }
      } else {
        stats.incNumVoidRemovals();
      }
    }

    for (Object o : groupedThreadIDs.entrySet()) {
      Map.Entry<ThreadIdentifier, List> element = (Map.Entry<ThreadIdentifier, List>) o;
      ThreadIdentifier tid = element.getKey();
      List removedEvents = element.getValue();
      long lastDispatchedId = (Long) removedEvents.remove(0);
      DispatchedAndCurrentEvents dace = (DispatchedAndCurrentEvents) eventsMap.get(tid);
      if (dace != null && dace.lastDispatchedSequenceId < lastDispatchedId) {
        try {
          dace.setLastDispatchedIDAndRemoveEvents(removedEvents, lastDispatchedId);
        } catch (CacheException e) {
          // ignore and log
          logger.error("Exception occurred while trying to set the last dispatched id",
              e);
        }
      }

      // Periodic ack from the client will add to the addDispatchMessage Map.
      // This method gets called from cacheClientNotifier upon receiving the ack from client.
      // addDispatchedMessage(tid, lastDispatchedId);
    }
    // removed the events from queue, now clear the peekedEventsContext

    setPeekedEvents();
  }
Removes the events that were peeked by this thread. The events are destroyed from the queue and
conflation map and DispatchedAndCurrentEvents are updated accordingly.
geode-core/src/main/java/org/apache/geode/internal/cache/ha/HARegionQueue.java