@VisibleForTesting
  void peekEventsFromIncompleteTransactions(List<GatewaySenderEventImpl> batch,
      PartitionedRegion prQ) {
    if (!mustGroupTransactionEvents()) {
      return;
    }

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

    int retries = 0;
    while (true) {
      for (Iterator<Map.Entry<TransactionId, Integer>> iter =
          incompleteTransactionIdsInBatch.entrySet().iterator(); iter.hasNext();) {
        Map.Entry<TransactionId, Integer> pendingTransaction = iter.next();
        TransactionId transactionId = pendingTransaction.getKey();
        int bucketId = pendingTransaction.getValue();
        List<Object> events = peekEventsWithTransactionId(prQ, bucketId, transactionId);
        for (Object object : events) {
          GatewaySenderEventImpl event = (GatewaySenderEventImpl) object;
          batch.add(event);
          peekedEvents.add(event);
          if (logger.isDebugEnabled()) {
            logger.debug(
                "Peeking extra event: {}, bucketId: {}, isLastEventInTransaction: {}, batch size: {}",
                event.getKey(), bucketId, event.isLastEventInTransaction(), batch.size());
          }
          if (event.isLastEventInTransaction()) {
            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();
    }
  }
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