@Override
  public List peek(int batchSize, int timeToWait) throws InterruptedException, CacheException {
    final boolean isDebugEnabled = logger.isDebugEnabled();

    PartitionedRegion prQ = getRandomShadowPR();
    if (prQ == null || prQ.getLocalMaxMemory() == 0) {
      try {
        Thread.sleep(50);
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
      }
      blockProcessorThreadIfRequired();
      return Collections.EMPTY_LIST;
    }
    List<GatewaySenderEventImpl> batch =
        new ArrayList<>(batchSize == BATCH_BASED_ON_TIME_ONLY ? DEFAULT_BATCH_SIZE : batchSize);

    long start = System.currentTimeMillis();
    long end = start + timeToWait;

    // Add peeked events
    addPeekedEvents(batch, batchSize == BATCH_BASED_ON_TIME_ONLY ? DEFAULT_BATCH_SIZE : batchSize);

    int bId;
    while (batchSize == BATCH_BASED_ON_TIME_ONLY || batch.size() < batchSize) {
      if (areLocalBucketQueueRegionsPresent() && ((bId = getRandomPrimaryBucket(prQ)) != -1)) {
        GatewaySenderEventImpl object = (GatewaySenderEventImpl) peekAhead(prQ, bId);
        if (object != null) {
          GatewaySenderEventImpl copy = object.makeHeapCopyIfOffHeap();
          if (copy == null) {
            if (stats != null) {
              stats.incEventsNotQueuedConflated();
            }
            continue;
          }
          object = copy;
        }
        // Conflate here
        if (object != null) {
          if (isDebugEnabled) {
            logger.debug("The gatewayEventImpl in peek is {}", object);
          }
          batch.add(object);
          peekedEvents.add(object);

        } else {
          if (stopPeekingDueToTime(timeToWait, end)) {
            break;
          }
          if (isDebugEnabled) {
            logger.debug("{}: Peeked object was null. Peek continuing", this);
          }
        }
      } else {
        if (stopPeekingDueToTime(timeToWait, end)) {
          break;
        }
        if (isDebugEnabled) {
          logger.debug("{}: Peek continuing", this);
        }
        // Sleep a bit before trying again.
        long currentTime = System.currentTimeMillis();
        try {
          Thread.sleep(calculateTimeToSleep(end - currentTime));
        } catch (InterruptedException e) {
          Thread.currentThread().interrupt();
          break;
        }
      }
    }

    if (batch.size() > 0) {
      peekEventsFromIncompleteTransactions(batch, prQ);
    }


    if (isDebugEnabled) {
      logger.debug("{}: Peeked a batch of {} entries. The size of the queue is {}. localSize is {}",
          this, batch.size(), size(), localSize());
    }

    if (batch.size() == 0) {
      blockProcessorThreadIfRequired();
    }
    return batch;
  }
Take will choose a random BucketRegionQueue which is primary and will take the head element
from it.
geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderQueue.java