docs: private void handleWaterMarkEvent(Event<T> event) {
      long watermarkTs = event.getTimestamp();
      long windowEndTs = nextWindowEndTs;
      LOG.fine(String.format("Window end ts %d Watermark ts %d", windowEndTs, watermarkTs));
      while (windowEndTs <= watermarkTs) {
        long currentCount = windowManager.getEventCount(windowEndTs);
        evictionPolicy.setContext(new DefaultEvictionContext(windowEndTs, currentCount));
        if (handler.onTrigger()) {
          windowEndTs += slidingIntervalMs;
        } else {
                  /*
                   * No events were found in the previous window interval.
                   * Scan through the events in the queue to find the next
                   * window intervals based on event ts.
                   */
          long ts = getNextAlignedWindowTs(windowEndTs, watermarkTs);
          LOG.fine(String.format("Next aligned window end ts %d", ts));
          if (ts == Long.MAX_VALUE) {
            LOG.fine(String.format("No events to process between %d and watermark ts %d",
                windowEndTs, watermarkTs));
            break;
          }
          windowEndTs = ts;
        }
      }
      nextWindowEndTs = windowEndTs;
    }