docs: 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));
  					}
  				}
  			}
  		}
  	}