code: @VisibleForTesting
    synchronized int sweep() {
      long startTime = System.nanoTime();
      int swept = 0;
      for (Map.Entry<String, Deque<AcknowledgableWatermark>> entry : _watermarksMap.entrySet()) {
        Deque<AcknowledgableWatermark> watermarks = entry.getValue();
        /**
         * Keep popping acked elements from the front as long as their next element is also acked.
         * So: Acked_A -> Acked_B -> Not-Acked_C -> ...  becomes
         *     Acked_B -> Not-Acked_C -> ...
         *
         * We keep the acked element around because that represents the highest contiguous acked watermark.
         */
        boolean continueIteration = true;
        while (continueIteration) {
          Iterator<AcknowledgableWatermark> iter = watermarks.iterator();
          if (!iter.hasNext()) {  // null
            continueIteration = false;
            continue;
          }
  
          AcknowledgableWatermark first = iter.next();
          if (first.isAcked()) {
            if (!iter.hasNext()) { // Acked_A -> null
              continueIteration = false;
              continue;
            }
            AcknowledgableWatermark second = iter.next();
            if ((second != null) && second.isAcked()) { // Acked_A -> Acked_B -> ...
              watermarks.pop();
              swept++;
            } else { // Acked_A -> Not_Acked_B
              continueIteration = false;
            }
          } else { // Not_Acked_A -> ..
            continueIteration = false;
          }
        }
      }
      long duration = (System.nanoTime() - startTime)/ MILLIS_TO_NANOS;
      log.debug("Swept {} watermarks in {} millis", swept, duration);
      _watermarksSwept.mark(swept);
      return swept;
    }
docs: A helper method to garbage collect acknowledged watermarks
  @return number of elements collected