docs: public CloseableIterator<Entry<KeyType>> iterator()
    {
      if (!initialized) {
        throw new ISE("Grouper should be initialized first");
      }
  
      return new CloseableIterator<Entry<KeyType>>()
      {
        {
          // Wait for some data to be ready and initialize nextReadIndex.
          increaseReadIndexTo(0);
        }
  
        @Override
        public boolean hasNext()
        {
          // If setting finished happens-before the below check, curWriteIndex isn't changed anymore and thus remainig()
          // can be computed safely because nextReadIndex is changed only by the reading thread.
          // Otherwise, hasNext() always returns true.
          //
          // The below line can be executed between increasing curWriteIndex and setting finished in
          // StreamingMergeSortedGrouper.finish(), but it is also a valid case because there should be at least one slot
          // which is not read yet before finished is set.
          return !finished || remaining() > 0;
        }
  
        /**
         * Calculate the number of remaining items in the array.  Must be called only when
         * {@link StreamingMergeSortedGrouper#finished} is true.
         *
         * @return the number of remaining items
         */
        private int remaining()
        {
          if (curWriteIndex >= nextReadIndex) {
            return curWriteIndex - nextReadIndex;
          } else {
            return (maxNumSlots - nextReadIndex) + curWriteIndex;
          }
        }
  
        @Override
        public Entry<KeyType> next()
        {
          if (!hasNext()) {
            throw new NoSuchElementException();
          }
  
          // Here, nextReadIndex should be valid which means:
          // - a valid array index which should be >= 0 and < maxNumSlots
          // - an index of the array slot where the aggregation for the corresponding grouping key is done
          // - an index of the array slot which is not read yet
          final int recordOffset = recordSize * nextReadIndex;
          final KeyType key = keySerde.fromByteBuffer(buffer, recordOffset);
  
          final Object[] values = new Object[aggregators.length];
          for (int i = 0; i < aggregators.length; i++) {
            values[i] = aggregators[i].get(buffer, recordOffset + aggregatorOffsets[i]);
          }
  
          final int targetIndex = nextReadIndex == maxNumSlots - 1 ? 0 : nextReadIndex + 1;
          // Wait if the array is empty until at least one slot becomes available for read, and then increase
          // nextReadIndex.
          increaseReadIndexTo(targetIndex);
  
          return new Entry<>(key, values);
        }
  
        /**
         * Wait for {@link StreamingMergeSortedGrouper#curWriteIndex} to be moved if necessary and move
         * {@link StreamingMergeSortedGrouper#nextReadIndex}.
         *
         * @param target the target index {@link StreamingMergeSortedGrouper#nextReadIndex} will move to
         */
        private void increaseReadIndexTo(int target)
        {
          // Check that the array is empty and wait for at least one slot to become available.
          //
          // curWriteIndex is a volatile variable and the changes on it are continuously checked until they are seen in
          // the while loop. See the following links.
          // * http://docs.oracle.com/javase/specs/jls/se7/html/jls-8.html#jls-8.3.1.4
          // * http://docs.oracle.com/javase/specs/jls/se7/html/jls-17.html#jls-17.4.5
          // * https://stackoverflow.com/questions/11761552/detailed-semantics-of-volatile-regarding-timeliness-of-visibility
  
          final long startAtNs = System.nanoTime();
          final long queryTimeoutAtNs = getQueryTimeoutAtNs(startAtNs);
          final long spinTimeoutAtNs = startAtNs + SPIN_FOR_TIMEOUT_THRESHOLD_NS;
          long timeoutNs = queryTimeoutAtNs - startAtNs;
          long spinTimeoutNs = SPIN_FOR_TIMEOUT_THRESHOLD_NS;
  
          // The below condition is checked in a while loop instead of using a lock to avoid frequent thread park.
          while ((curWriteIndex == -1 || target == curWriteIndex) &&
                 !finished && !Thread.currentThread().isInterrupted()) {
            if (timeoutNs <= 0L) {
              throw new RuntimeException(new TimeoutException());
            }
            // Thread.yield() should not be called from the very beginning
            if (spinTimeoutNs <= 0L) {
              Thread.yield();
            }
            long now = System.nanoTime();
            timeoutNs = queryTimeoutAtNs - now;
            spinTimeoutNs = spinTimeoutAtNs - now;
          }
  
          // Changes on curWriteIndex happens-before changing nextReadIndex.
          nextReadIndex = target;
        }
  
        @Override
        public void close()
        {
          // do nothing
        }
      };
    }