code: private long computeWaterMarkTs() {
          long ts = 0;
          // only if some data has arrived on each input stream
          if (streamToTs.size() >= inputStreams.size()) {
              ts = Long.MAX_VALUE;
              for (Map.Entry<GlobalStreamId, Long> entry : streamToTs.entrySet()) {
                  ts = Math.min(ts, entry.getValue());
              }
          }
          return ts - eventTsLag;
      }
docs: Computes the min ts across all streams.