/*
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
 * agreements. See the NOTICE file distributed with this work for additional information regarding
 * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance with the License. You may obtain a
 * copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */
package org.apache.geode.internal.cache.wan.serial;

import static org.apache.geode.cache.wan.GatewaySender.DEFAULT_BATCH_SIZE;
import static org.apache.geode.cache.wan.GatewaySender.GET_TRANSACTION_EVENTS_FROM_QUEUE_WAIT_TIME_MS;
import static org.apache.geode.util.internal.UncheckedUtils.uncheckedCast;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Deque;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Predicate;

import org.apache.logging.log4j.Logger;

import org.apache.geode.CancelException;
import org.apache.geode.SystemFailure;
import org.apache.geode.annotations.Immutable;
import org.apache.geode.annotations.VisibleForTesting;
import org.apache.geode.cache.AttributesMutator;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheException;
import org.apache.geode.cache.CacheListener;
import org.apache.geode.cache.CacheWriterException;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.EntryNotFoundException;
import org.apache.geode.cache.EvictionAction;
import org.apache.geode.cache.EvictionAttributes;
import org.apache.geode.cache.Operation;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionAttributes;
import org.apache.geode.cache.RegionDestroyedException;
import org.apache.geode.cache.RegionShortcut;
import org.apache.geode.cache.Scope;
import org.apache.geode.cache.TimeoutException;
import org.apache.geode.cache.TransactionId;
import org.apache.geode.cache.asyncqueue.AsyncEvent;
import org.apache.geode.cache.asyncqueue.internal.AsyncEventQueueImpl;
import org.apache.geode.internal.cache.CachedDeserializable;
import org.apache.geode.internal.cache.Conflatable;
import org.apache.geode.internal.cache.DistributedRegion;
import org.apache.geode.internal.cache.EntryEventImpl;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.InternalRegionArguments;
import org.apache.geode.internal.cache.InternalRegionFactory;
import org.apache.geode.internal.cache.LocalRegion;
import org.apache.geode.internal.cache.RegionQueue;
import org.apache.geode.internal.cache.Token;
import org.apache.geode.internal.cache.event.EventTracker;
import org.apache.geode.internal.cache.event.NonDistributedEventTracker;
import org.apache.geode.internal.cache.versions.RegionVersionVector;
import org.apache.geode.internal.cache.versions.VersionSource;
import org.apache.geode.internal.cache.wan.AbstractGatewaySender;
import org.apache.geode.internal.cache.wan.GatewaySenderEventImpl;
import org.apache.geode.internal.cache.wan.GatewaySenderStats;
import org.apache.geode.internal.offheap.OffHeapClearRequired;
import org.apache.geode.internal.statistics.StatisticsClock;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.management.ManagementService;
import org.apache.geode.management.internal.beans.AsyncEventQueueMBean;
import org.apache.geode.management.internal.beans.GatewaySenderMBean;
import org.apache.geode.pdx.internal.PeerTypeRegistration;
import org.apache.geode.util.internal.GeodeGlossary;

/**
 * @since GemFire 7.0
 */
public class SerialGatewaySenderQueue implements RegionQueue {

  private static final Logger logger = LogService.getLogger();

  /**
   * The key into the <code>Region</code> used when taking entries from the queue. This value is
   * either set when the queue is instantiated or read from the <code>Region</code> in the case
   * where this queue takes over where a previous one left off.
   */
  private long headKey;

  /**
   * The key into the <code>Region</code> used when putting entries onto the queue. This value is
   * either set when the queue is instantiated or read from the <code>Region</code> in the case
   * where this queue takes over where a previous one left off.
   */
  private final AtomicLong tailKey = new AtomicLong();

  /**
   * Last key peeked from the queue excluding the keys peeked
   * to complete transactions when group-transaction-events is enabled.
   */
  private final AtomicLong lastPeekedId = new AtomicLong(-1);

  private final Deque<Long> peekedIds = new LinkedBlockingDeque<>();

  /**
   * Contains the set of peekedIds that were peeked to complete a transaction
   * inside a batch when groupTransactionEvents is set.
   */
  private final Set<Long> extraPeekedIds = ConcurrentHashMap.newKeySet();

  /**
   * Contains the set of peekedIds that were peeked to complete a transaction
   * inside a batch when groupTransactionEvents is set and whose event has been
   * removed from the queue because an ack has been received from the receiver.
   * Elements from this set are deleted when the event with the previous id
   * is removed.
   */
  private final Set<Long> extraPeekedIdsRemovedButPreviousIdNotRemoved =
      ConcurrentHashMap.newKeySet();

  /**
   * The name of the <code>Region</code> backing this queue
   */
  private final String regionName;

  /**
   * The <code>Region</code> backing this queue
   */
  private Region<Long, AsyncEvent<?, ?>> region;

  /**
   * The name of the <code>DiskStore</code> to overflow this queue
   */
  private final String diskStoreName;

  /**
   * The maximum amount of memory (MB) to allow in the queue before overflowing entries to disk
   */
  private final int maximumQueueMemory;

  /**
   * Whether conflation is enabled for this queue.
   */
  private final boolean enableConflation;

  /**
   * Whether persistence is enabled for this queue.
   */
  private final boolean enablePersistence;

  private final boolean cleanQueues;

  /**
   * Whether write to disk is synchronous.
   */
  private final boolean isDiskSynchronous;

  /**
   * The <code>Map</code> mapping the regionName->key to the queue key. This index allows fast
   * updating of entries in the queue for conflation.
   */
  private final Map<String, Map<Object, Long>> indexes;

  private final GatewaySenderStats stats;

  /**
   * The maximum allowed key before the keys are rolled over
   */
  private static final long MAXIMUM_KEY = Long.MAX_VALUE;

  /**
   * Whether the <code>Gateway</code> queue should be no-ack instead of ack.
   */
  private static final boolean NO_ACK =
      Boolean.getBoolean(GeodeGlossary.GEMFIRE_PREFIX + "gateway-queue-no-ack");

  private volatile long lastDispatchedKey = -1;

  private volatile long lastDestroyedKey = -1;

  public static final int DEFAULT_MESSAGE_SYNC_INTERVAL = 1;

  @Immutable
  private static final int messageSyncInterval = DEFAULT_MESSAGE_SYNC_INTERVAL;

  private final BatchRemovalThread removalThread;

  private final AbstractGatewaySender sender;

  private final MetaRegionFactory metaRegionFactory;

  public SerialGatewaySenderQueue(AbstractGatewaySender abstractSender, String regionName,
      CacheListener<Long, AsyncEvent<?, ?>> listener, boolean cleanQueues) {
    this(abstractSender, regionName, listener, cleanQueues, new MetaRegionFactory());
  }

  public SerialGatewaySenderQueue(AbstractGatewaySender abstractSender, String regionName,
      CacheListener<Long, AsyncEvent<?, ?>> listener, boolean cleanQueues,
      MetaRegionFactory metaRegionFactory) {
    // The queue starts out with headKey and tailKey equal to -1 to force
    // them to be initialized from the region.
    this.regionName = regionName;
    this.cleanQueues = cleanQueues;
    this.metaRegionFactory = metaRegionFactory;
    headKey = -1;
    tailKey.set(-1);
    indexes = new HashMap<>();
    enableConflation = abstractSender.isBatchConflationEnabled();
    diskStoreName = abstractSender.getDiskStoreName();
    enablePersistence = abstractSender.isPersistenceEnabled();
    if (enablePersistence) {
      isDiskSynchronous = abstractSender.isDiskSynchronous();
    } else {
      isDiskSynchronous = false;
    }
    maximumQueueMemory = abstractSender.getMaximumMemeoryPerDispatcherQueue();
    stats = abstractSender.getStatistics();
    initializeRegion(abstractSender, listener);
    // Increment queue size. Fix for bug 51988.
    stats.incQueueSize(region.size());
    removalThread = new BatchRemovalThread(abstractSender.getCache());
    removalThread.start();
    sender = abstractSender;
    if (logger.isDebugEnabled()) {
      logger.debug("{}: Contains {} elements", this, size());
    }
  }

  @Override
  public Region<Long, AsyncEvent<?, ?>> getRegion() {
    return region;
  }

  public void destroy() {
    getRegion().localDestroyRegion();
  }

  @Override
  public synchronized boolean put(Object event) throws CacheException {
    GatewaySenderEventImpl eventImpl = (GatewaySenderEventImpl) event;
    final Region<?, ?> r = eventImpl.getRegion();
    final boolean isPDXRegion =
        (r instanceof DistributedRegion && r.getName().equals(PeerTypeRegistration.REGION_NAME));
    final boolean isWbcl =
        regionName.startsWith(AsyncEventQueueImpl.ASYNC_EVENT_QUEUE_PREFIX);
    if (!(isPDXRegion && isWbcl)) {
      putAndGetKey(event);
      return true;
    }
    return false;
  }

  private void putAndGetKey(Object object) throws CacheException {
    // Get the tail key
    Long key = getTailKey();
    // Put the object into the region at that key
    region.put(key, (AsyncEvent<?, ?>) object);

    // Increment the tail key
    // It is important that we increment the tail
    // key after putting in the region, this is the
    // signal that a new object is available.
    incrementTailKey();

    if (logger.isDebugEnabled()) {
      logger.debug("{}: Inserted {} -> {}", this, key, object);
    }
    if (object instanceof Conflatable) {
      removeOldEntry((Conflatable) object, key);
    }
  }


  @Override
  public AsyncEvent<?, ?> take() throws CacheException {
    // Unsupported since we have no callers.
    // If we do want to support it then each caller needs
    // to call freeOffHeapResources and the returned GatewaySenderEventImpl
    throw new UnsupportedOperationException();
  }

  @Override
  public List<AsyncEvent<?, ?>> take(int batchSize) throws CacheException {
    // This method has no callers.
    // If we do want to support it then the callers
    // need to call freeOffHeapResources on each returned GatewaySenderEventImpl
    throw new UnsupportedOperationException();
  }

  /**
   * This method removes the last entry. However, it will only let the user remove entries that they
   * have peeked. If the entry was not peeked, this method will silently return.
   */
  @Override
  public synchronized void remove() throws CacheException {
    if (peekedIds.isEmpty()) {
      return;
    }
    boolean wasEmpty = lastDispatchedKey == lastDestroyedKey;
    Long key = peekedIds.remove();
    boolean isExtraPeekedId = extraPeekedIds.contains(key);
    if (!isExtraPeekedId) {
      updateHeadKey(key);
      lastDispatchedKey = key;
    } else {
      extraPeekedIdsRemovedButPreviousIdNotRemoved.add(key);
    }
    removeIndex(key);
    // Remove the entry at that key with a callback arg signifying it is
    // a WAN queue so that AbstractRegionEntry.destroy can get the value
    // even if it has been evicted to disk. In the normal case, the
    // AbstractRegionEntry.destroy only gets the value in the VM.
    try {
      region.localDestroy(key, WAN_QUEUE_TOKEN);
      stats.decQueueSize();
    } catch (EntryNotFoundException ok) {
      // this is acceptable because the conflation can remove entries
      // out from underneath us.
      if (logger.isDebugEnabled()) {
        logger.debug(
            "{}: Did not destroy entry at {} it was not there. It should have been removed by conflation.",
            this, key);
      }
    }

    // For those extraPeekedIds removed that are consecutive to lastDispatchedKey:
    // - Update lastDispatchedKey with them so that they are removed
    // by the batch removal thread.
    // - Update the head key with them.
    // - Remove them from extraPeekedIds.
    long tmpKey = lastDispatchedKey;
    while (extraPeekedIdsRemovedButPreviousIdNotRemoved.contains(tmpKey = inc(tmpKey))) {
      extraPeekedIdsRemovedButPreviousIdNotRemoved.remove(tmpKey);
      extraPeekedIds.remove(tmpKey);
      updateHeadKey(tmpKey);
      lastDispatchedKey = tmpKey;
    }

    if (wasEmpty) {
      synchronized (this) {
        notifyAll();
      }
    }

    if (logger.isDebugEnabled()) {
      logger.debug(
          "{}: Destroyed entry at key {} setting the lastDispatched Key to {}. The last destroyed entry was {}",
          this, key, lastDispatchedKey, lastDestroyedKey);
    }
  }

  /**
   * This method removes batchSize entries from the queue. It will only remove entries that were
   * previously peeked.
   *
   * @param size the number of entries to remove
   */
  @Override
  public void remove(int size) throws CacheException {
    for (int i = 0; i < size; i++) {
      remove();
    }
    if (logger.isTraceEnabled()) {
      logger.trace("{}: Removed a batch of {} entries", this, size);
    }
  }

  @Override
  public Object peek() throws CacheException {
    KeyAndEventPair object = peekAhead();
    if (logger.isTraceEnabled()) {
      logger.trace("{}: Peeked {} -> {}", this, peekedIds, object);
    }

    return object.event;
    // OFFHEAP returned object only used to see if queue is empty
    // so no need to worry about off-heap refCount.
  }

  @Override
  public List<AsyncEvent<?, ?>> peek(int size) throws CacheException {
    return peek(size, -1);
  }

  @Override
  public List<AsyncEvent<?, ?>> peek(int size, int timeToWait) throws CacheException {
    final boolean isTraceEnabled = logger.isTraceEnabled();

    long start = System.currentTimeMillis();
    long end = start + timeToWait;
    if (isTraceEnabled) {
      logger.trace("{}: Peek start time={} end time={} time to wait={}", this, start, end,
          timeToWait);
    }

    List<AsyncEvent<?, ?>> batch =
        new ArrayList<>(size == BATCH_BASED_ON_TIME_ONLY ? DEFAULT_BATCH_SIZE : size);
    long lastKey = -1;
    while (size == BATCH_BASED_ON_TIME_ONLY || batch.size() < size) {
      KeyAndEventPair pair = peekAhead();
      // Conflate here
      if (pair != null) {
        AsyncEvent<?, ?> object = pair.event;
        lastKey = pair.key;
        batch.add(object);
      } else {
        // If time to wait is -1 (don't wait) or time interval has elapsed
        long currentTime = System.currentTimeMillis();
        if (isTraceEnabled) {
          logger.trace("{}: Peek current time: {}", this, currentTime);
        }
        if (timeToWait == -1 || (end <= currentTime)) {
          if (isTraceEnabled) {
            logger.trace("{}: Peek breaking", this);
          }
          break;
        }

        if (isTraceEnabled) {
          logger.trace("{}: Peek continuing", this);
        }
        // Sleep a bit before trying again.
        try {
          Thread.sleep(50);
        } catch (InterruptedException e) {
          Thread.currentThread().interrupt();
          break;
        }
      }
    }
    if (batch.size() > 0) {
      peekEventsFromIncompleteTransactions(batch, lastKey);
    }

    if (isTraceEnabled) {
      logger.trace("{}: Peeked a batch of {} entries", this, batch.size());
    }
    return batch;
    // OFFHEAP: all returned AsyncEvent end up being removed from queue after the batch is sent
    // so no need to worry about off-heap refCount.
  }

  @VisibleForTesting
  void peekEventsFromIncompleteTransactions(List<AsyncEvent<?, ?>> batch, long lastKey) {
    if (!mustGroupTransactionEvents()) {
      return;
    }

    Set<TransactionId> incompleteTransactionIdsInBatch = getIncompleteTransactionsInBatch(batch);
    if (incompleteTransactionIdsInBatch.size() == 0) {
      return;
    }

    int retries = 0;
    while (true) {
      for (Iterator<TransactionId> iter = incompleteTransactionIdsInBatch.iterator(); iter
          .hasNext();) {
        TransactionId transactionId = iter.next();
        List<KeyAndEventPair> keyAndEventPairs =
            peekEventsWithTransactionId(transactionId, lastKey);
        if (keyAndEventPairs.size() > 0
            && ((GatewaySenderEventImpl) (keyAndEventPairs.get(keyAndEventPairs.size() - 1)).event)
                .isLastEventInTransaction()) {
          for (KeyAndEventPair object : keyAndEventPairs) {
            GatewaySenderEventImpl event = (GatewaySenderEventImpl) object.event;
            batch.add(event);
            peekedIds.add(object.key);
            extraPeekedIds.add(object.key);
            if (logger.isDebugEnabled()) {
              logger.debug(
                  "Peeking extra event: {}, isLastEventInTransaction: {}, batch size: {}",
                  event.getKey(), event.isLastEventInTransaction(), batch.size());
            }
          }
          iter.remove();
        }
      }
      if (incompleteTransactionIdsInBatch.size() == 0 ||
          retries >= sender.getRetriesToGetTransactionEventsFromQueue()) {
        break;
      }
      retries++;
      try {
        Thread.sleep(GET_TRANSACTION_EVENTS_FROM_QUEUE_WAIT_TIME_MS);
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
      }
    }
    if (incompleteTransactionIdsInBatch.size() > 0) {
      logger.warn("Not able to retrieve all events for transactions: {} after {} retries of {}ms",
          incompleteTransactionIdsInBatch, retries, GET_TRANSACTION_EVENTS_FROM_QUEUE_WAIT_TIME_MS);
      stats.incBatchesWithIncompleteTransactions();
    }
  }

  protected boolean mustGroupTransactionEvents() {
    return sender.mustGroupTransactionEvents();
  }

  private Set<TransactionId> getIncompleteTransactionsInBatch(List<AsyncEvent<?, ?>> batch) {
    Set<TransactionId> incompleteTransactionsInBatch = new HashSet<>();
    for (Object object : batch) {
      if (object instanceof GatewaySenderEventImpl) {
        GatewaySenderEventImpl event = (GatewaySenderEventImpl) object;
        if (event.getTransactionId() != null) {
          if (event.isLastEventInTransaction()) {
            incompleteTransactionsInBatch.remove(event.getTransactionId());
          } else {
            incompleteTransactionsInBatch.add(event.getTransactionId());
          }
        }
      }
    }
    return incompleteTransactionsInBatch;
  }

  @Override
  public String toString() {
    return "SerialGatewaySender queue :" + regionName;
  }

  @Override
  public int size() {
    int size = ((LocalRegion) region).entryCount();
    return size + sender.getTmpQueuedEventSize();
  }

  @Override
  public void addCacheListener(@SuppressWarnings("rawtypes") CacheListener listener) {
    AttributesMutator<Long, AsyncEvent<?, ?>> mutator = region.getAttributesMutator();
    mutator.addCacheListener(uncheckedCast(listener));
  }

  @Override
  public void removeCacheListener() {
    AttributesMutator<Long, AsyncEvent<?, ?>> mutator = region.getAttributesMutator();
    CacheListener<Long, AsyncEvent<?, ?>>[] listeners = region.getAttributes().getCacheListeners();
    for (CacheListener<Long, AsyncEvent<?, ?>> listener : listeners) {
      if (listener instanceof SerialSecondaryGatewayListener) {
        mutator.removeCacheListener(listener);
        break;
      }
    }
  }

  private void removeOldEntry(Conflatable object, Long tailKey) throws CacheException {
    final boolean isDebugEnabled = logger.isDebugEnabled();

    // Determine whether conflation is enabled for this queue and object
    // Conflation is enabled iff:
    // - this queue has conflation enabled
    // - the object can be conflated
    if (enableConflation && object.shouldBeConflated()) {
      if (isDebugEnabled) {
        logger.debug("{}: Conflating {} at queue index={} queue size={} head={} tail={}", this,
            object, tailKey, size(), headKey, tailKey);
      }

      // Determine whether this region / key combination is already indexed.
      // If so, it is already in the queue. Update the value in the queue and
      // set the shouldAddToQueue flag accordingly.
      String rName = object.getRegionToConflate();
      Object key = object.getKeyToConflate();
      Long previousIndex;

      synchronized (this) {
        Map<Object, Long> latestIndexesForRegion =
            indexes.computeIfAbsent(rName, k -> new HashMap<>());

        previousIndex = latestIndexesForRegion.put(key, tailKey);
      }

      if (isDebugEnabled) {
        logger.debug("{}: Adding index key={}->index={} for {} head={} tail={}", this, key, tailKey,
            object, headKey, tailKey);
      }
      // Test if the key is contained in the latest indexes map. If the key is
      // not contained in the latest indexes map, then it should be added to
      // the queue.
      //
      // It no longer matters if we remove an entry that is going out in the
      // current
      // batch, because we already put the latest value on the tail of the
      // queue, and
      // peekedIds list prevents us from removing an entry that was not peeked.
      boolean keepOldEntry;
      if (previousIndex != null) {
        if (isDebugEnabled) {
          logger.debug(
              "{}: Indexes contains index={} for key={} head={} tail={} and it can be used.", this,
              previousIndex, key, headKey, tailKey);
        }
        keepOldEntry = false;
      } else {
        if (isDebugEnabled) {
          logger.debug("{}: No old entry for key={} head={} tail={} not removing old entry.", this,
              key, headKey, tailKey);
        }
        stats.incConflationIndexesMapSize();
        keepOldEntry = true;
      }

      // Remove the old entry from the queue if necessary. The new entry is already present
      // in the queue region.
      if (!keepOldEntry) {
        Conflatable previous = (Conflatable) region.remove(previousIndex);
        stats.decQueueSize(1);
        if (isDebugEnabled) {
          logger.debug("{}: Previous conflatable at key={} head={} tail={}: {}", this,
              previousIndex, headKey, tailKey, previous);
          logger.debug("{}: Current conflatable at key={} head={} tail={}: {}", this, tailKey,
              headKey, tailKey, object);
          if (previous != null) {
            logger.debug(
                "{}: Removed {} and added {} for key={} head={} tail={} in queue for region={} old event={}",
                this, previous.getValueToConflate(), object.getValueToConflate(), key, headKey,
                tailKey, rName, previous);
          }
        }
      }
    } else {
      if (isDebugEnabled) {
        logger.debug("{}: Not conflating {} queue size: {} head={} tail={}", this, object, size(),
            headKey, tailKey);
      }
    }
  }

  /**
   * Does a get that gets the value without fault values in from disk.
   */
  private AsyncEvent<?, ?> optimalGet(Long k) {
    // Get the object at that key (to remove the index).
    LocalRegion lr = (LocalRegion) region;
    Object o = null;
    try {
      o = lr.getValueInVMOrDiskWithoutFaultIn(k);
      if (o instanceof CachedDeserializable) {
        o = ((CachedDeserializable) o).getDeserializedValue(lr, lr.getRegionEntry(k));
      }
    } catch (EntryNotFoundException ok) {
      // just return null;
    }
    // bug #46023 do not return a destroyed entry marker
    if (o == Token.TOMBSTONE) {
      o = null;
    }
    return (AsyncEvent<?, ?>) o;
  }

  /*
   * this must be invoked under synchronization
   */
  private void removeIndex(Long qkey) {
    // Determine whether conflation is enabled for this queue and object
    if (enableConflation) {
      // only call get after checking enableConflation for bug 40508
      Object o = optimalGet(qkey);
      if (o instanceof Conflatable) {
        Conflatable object = (Conflatable) o;
        if (object.shouldBeConflated()) {
          // Otherwise, remove the index from the indexes map.
          String rName = object.getRegionToConflate();
          Object key = object.getKeyToConflate();

          Map<Object, Long> latestIndexesForRegion = indexes.get(rName);
          if (latestIndexesForRegion != null) {
            // Remove the index.
            Long index = latestIndexesForRegion.remove(key);
            // Decrement the size if something was removed. This check is for
            // the case where failover has occurred and the entry was not put
            // into the map initially.
            if (index != null) {
              stats.decConflationIndexesMapSize();
            }
            if (logger.isDebugEnabled()) {
              if (index != null) {
                logger.debug("{}: Removed index {} for {}", this, index, object);
              }
            }
          }
        }
      }
    }
  }

  /**
   * returns true if key a is before key b. This test handles keys that have wrapped around
   *
   */
  private boolean before(long a, long b) {
    // a is before b if a < b or a>b and a MAXIMUM_KEY/2 larger than b
    // (indicating we have wrapped)
    return a < b ^ a - b > (MAXIMUM_KEY / 2);
  }

  private long inc(long value) {
    long val = value + 1;
    val = val == MAXIMUM_KEY ? 0 : val;
    return val;
  }


  /**
   * Clear the list of peeked keys. The next peek will start again at the head key.
   */
  public void resetLastPeeked() {
    peekedIds.clear();
    extraPeekedIds.clear();
    lastPeekedId.set(-1);
  }

  /**
   * Finds the next object after the last key peeked
   *
   */
  private Long getCurrentKey() {
    long currentKey;
    if (lastPeekedId.get() == -1) {
      currentKey = getHeadKey();
    } else {
      currentKey = inc(lastPeekedId.get());
    }
    return currentKey;
  }

  private AsyncEvent<?, ?> getObjectInSerialSenderQueue(Long currentKey) {
    AsyncEvent<?, ?> object = optimalGet(currentKey);
    if ((null != object) && logger.isDebugEnabled()) {
      logger.debug("{}: Peeked {}->{}", this, currentKey, object);
    }
    if (object instanceof GatewaySenderEventImpl) {
      GatewaySenderEventImpl copy = ((GatewaySenderEventImpl) object).makeHeapCopyIfOffHeap();
      if (copy == null) {
        logger.debug(
            "Unable to make heap copy and will not be added to peekedIds for object" + " : {} ",
            object.toString());
      }
      object = copy;
    }
    return object;
  }

  @VisibleForTesting
  static class KeyAndEventPair {
    public final long key;
    public final AsyncEvent<?, ?> event;

    KeyAndEventPair(Long key, AsyncEvent<?, ?> event) {
      this.key = key;
      this.event = event;
    }
  }

  @VisibleForTesting
  public KeyAndEventPair peekAhead() throws CacheException {
    AsyncEvent<?, ?> object = null;
    Long currentKey = getCurrentKey();
    if (currentKey == null) {
      return null;
    }

    // It's important here that we check where the current key
    // is in relation to the tail key before we check to see if the
    // object exists. The reason is that the tail key is basically
    // the synchronization between this thread and the putter thread.
    // The tail key will not be incremented until the object is put in the
    // region
    // If we check for the object, and then check the tail key, we could
    // skip objects.

    // TODO: don't do a get which updates the lru, instead just get the value
    // w/o modifying the LRU.
    // Note: getting the serialized form here (if it has overflowed to disk)
    // does not save anything since GatewayBatchOp needs to GatewayEventImpl
    // in object form.
    while (before(currentKey, getTailKey())) {
      if (!extraPeekedIds.contains(currentKey)) {
        object = getObjectInSerialSenderQueue(currentKey);
        if (object != null) {
          break;
        }
      }
      if (logger.isTraceEnabled()) {
        logger.trace("{}: Trying head key + offset: {}", this, currentKey);
      }
      currentKey = inc(currentKey);
      // When mustGroupTransactionEvents is true, conflation cannot be enabled.
      // Therefore, if we reach here, it would not be due to a conflated event
      // but rather to an extra peeked event already sent.
      if (!mustGroupTransactionEvents() && stats != null) {
        stats.incEventsNotQueuedConflated();
      }
    }

    if (logger.isDebugEnabled()) {
      logger.debug("{}: Peeked {}->{}", this, currentKey, object);
    }

    if (object != null) {
      peekedIds.add(currentKey);
      lastPeekedId.set(currentKey);
      return new KeyAndEventPair(currentKey, object);
    }
    return null;
  }

  private List<KeyAndEventPair> peekEventsWithTransactionId(TransactionId transactionId,
      long lastKey) {
    Predicate<GatewaySenderEventImpl> hasTransactionIdPredicate =
        x -> transactionId.equals(x.getTransactionId());
    Predicate<GatewaySenderEventImpl> isLastEventInTransactionPredicate =
        GatewaySenderEventImpl::isLastEventInTransaction;

    return getElementsMatching(hasTransactionIdPredicate, isLastEventInTransactionPredicate,
        lastKey);
  }

  /**
   * This method returns a list of objects that fulfill the matchingPredicate
   * If a matching object also fulfills the endPredicate then the method
   * stops looking for more matching objects.
   */
  List<KeyAndEventPair> getElementsMatching(Predicate<GatewaySenderEventImpl> condition,
      Predicate<GatewaySenderEventImpl> stopCondition,
      long lastKey) {
    GatewaySenderEventImpl event;
    List<KeyAndEventPair> elementsMatching = new ArrayList<>();

    long currentKey = lastKey;

    while ((currentKey = inc(currentKey)) != getTailKey()) {
      if (extraPeekedIds.contains(currentKey)) {
        continue;
      }
      event = (GatewaySenderEventImpl) optimalGet(currentKey);
      if (event == null) {
        continue;
      }

      if (condition.test(event)) {
        elementsMatching.add(new KeyAndEventPair(currentKey, event));

        if (stopCondition.test(event)) {
          break;
        }
      }
    }

    return elementsMatching;
  }

  /**
   * Returns the value of the tail key. The tail key points to an empty where the next queue entry
   * will be stored.
   *
   * @return the value of the tail key
   */
  private long getTailKey() throws CacheException {
    long tlKey;
    // Test whether tailKey = -1. If so, the queue has just been created.
    // Go into the region to get the value of TAIL_KEY. If it is null, then
    // this is the first attempt to access this queue. Set the tailKey and
    // tailKey appropriately (to 0). If there is a value in the region, then
    // this queue has been accessed before and this instance is taking up where
    // a previous one left off. Set the tailKey to the value in the region.
    // From now on, this queue will use the value of tailKey in the VM to
    // determine the tailKey. If the tailKey != -1, set the tailKey
    // to the value of the tailKey.
    initializeKeys();

    tlKey = tailKey.get();
    if (logger.isTraceEnabled()) {
      logger.trace("{}: Determined tail key: {}", this, tlKey);
    }
    return tlKey;
  }

  /**
   * Increments the value of the tail key by one.
   *
   */
  private void incrementTailKey() throws CacheException {
    tailKey.set(inc(tailKey.get()));
    if (logger.isTraceEnabled()) {
      logger.trace("{}: Incremented TAIL_KEY for region {} to {}", this, region.getName(),
          tailKey);
    }
  }

  /**
   * If the keys are not yet initialized, initialize them from the region .
   *
   * TODO - We could initialize the indexes maps at the time here. However, that would require
   * iterating over the values of the region rather than the keys, which could be much more
   * expensive if the region has overflowed to disk.
   *
   * We do iterate over the values of the region in SerialGatewaySender at the time of failover. see
