private void updateMinPunctuatedWatermark ( Watermark nextWatermark ) { if ( nextWatermark . getTimestamp ( ) > maxWatermarkSoFar ) { long newMin = Long . MAX_VALUE ; for ( KafkaTopicPartitionState < ? > state : subscribedPartitionStates ) { @ SuppressWarnings ( "unchecked" ) final KafkaTopicPartitionStateWithPunctuatedWatermarks < T , KPH > withWatermarksState = ( KafkaTopicPartitionStateWithPunctuatedWatermarks < T , KPH > ) state ; newMin = Math . min ( newMin , withWatermarksState . getCurrentPartitionWatermark ( ) ) ; } // double-check locking pattern if ( newMin > maxWatermarkSoFar ) { synchronized ( checkpointLock ) { if ( newMin > maxWatermarkSoFar ) { maxWatermarkSoFar = newMin ; sourceContext . emitWatermark ( new Watermark ( newMin ) ) ; } } } } }
Checks whether a new per - partition watermark is also a new cross - partition watermark .