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