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 ; }
Computes the min ts across all streams .