code: 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;
    }
docs: Invokes the trigger all pending windows up to the
  watermark timestamp. The end ts of the window is set
  in the eviction policy context so that the events falling
  within that window can be processed.