code: public void track(AcknowledgableWatermark acknowledgableWatermark) {
      if (!_started.get() && _autoStart) {
        start();
      }
      maybeAbort();
      String source = acknowledgableWatermark.getCheckpointableWatermark().getSource();
      Deque<AcknowledgableWatermark> sourceWatermarks = _watermarksMap.get(source);
      if (sourceWatermarks == null) {
        sourceWatermarks = new ConcurrentLinkedDeque<>();
        _watermarksMap.put(source, sourceWatermarks);
      }
      sourceWatermarks.add(acknowledgableWatermark);
      _watermarksInserted.mark();
    }
docs: Track a watermark.
  Assumptions: Track is called sequentially from the same thread for watermarks that are
  progressively increasing.