@ VisibleForTesting protected void emitWatermark ( ) { LOG . debug ( "Evaluating watermark for subtask {} time {}" , indexOfThisConsumerSubtask , getCurrentTimeMillis ( ) ) ; long potentialWatermark = Long . MAX_VALUE ; long idleTime = ( shardIdleIntervalMillis > 0 ) ? getCurrentTimeMillis ( ) - shardIdleIntervalMillis : Long . MAX_VALUE ; for ( Map . Entry < Integer , ShardWatermarkState > e : shardWatermarks . entrySet ( ) ) { // consider only active shards, or those that would advance the watermark Watermark w = e . getValue ( ) . periodicWatermarkAssigner . getCurrentWatermark ( ) ; if ( w != null && ( e . getValue ( ) . lastUpdated >= idleTime || w . getTimestamp ( ) > lastWatermark ) ) { potentialWatermark = Math . min ( potentialWatermark , w . getTimestamp ( ) ) ; } } // advance watermark if possible (watermarks can only be ascending) if ( potentialWatermark == Long . MAX_VALUE ) { if ( shardWatermarks . isEmpty ( ) || shardIdleIntervalMillis > 0 ) { LOG . debug ( "No active shard for subtask {}, marking the source idle." , indexOfThisConsumerSubtask ) ; // no active shard, signal downstream operators to not wait for a watermark sourceContext . markAsTemporarilyIdle ( ) ; } } else if ( potentialWatermark > lastWatermark ) { LOG . debug ( "Emitting watermark {} from subtask {}" , potentialWatermark , indexOfThisConsumerSubtask ) ; sourceContext . emitWatermark ( new Watermark ( potentialWatermark ) ) ; lastWatermark = potentialWatermark ; } }
Called periodically to emit a watermark . Checks all shards for the current event time watermark and possibly emits the next watermark .