code: @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;
  		}
  	}
docs: Called periodically to emit a watermark. Checks all shards for the current event time
  watermark, and possibly emits the next watermark.
  
  <p>Shards that have not received an update for a certain interval are considered inactive so as
  to not hold back the watermark indefinitely. When all shards are inactive, the subtask will be
  marked as temporarily idle to not block downstream operators.