一文搞懂 flink 处理水印全过程

		FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<String>(topics, new SimpleStringSchema(), properties);
		consumer.setStartFromLatest();
		consumer.assignTimestampsAndWatermarks(new AscendingTimestampExtractor<String>() {
			@Override
			public long extractAscendingTimestamp(String element) {
				String locTime = "";
				try {
					Map<String, Object> map = Json2Others.json2map(element);
					locTime = map.get("locTime").toString();
				} catch (IOException e) {
				}
				LocalDateTime startDateTime =
					LocalDateTime.parse(locTime, DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
				long milli = startDateTime.toInstant(OffsetDateTime.now().getOffset()).toEpochMilli();
				return milli;
			}
		});

// if we have periodic watermarks, kick off the interval scheduler
		// 在构建 fetcher 的时候创建 PeriodicWatermarkEmitter 并启动，以周期性发送
		if (timestampWatermarkMode == PERIODIC_WATERMARKS) {
			@SuppressWarnings("unchecked")
			PeriodicWatermarkEmitter periodicEmitter = new PeriodicWatermarkEmitter(
					subscribedPartitionStates,
					sourceContext,
					processingTimeProvider,
					autoWatermarkInterval);

			periodicEmitter.start();
		}

@Override
		//每隔 interval 时间会调用此方法
		public void onProcessingTime(long timestamp) throws Exception {

			long minAcrossAll = Long.MAX_VALUE;
			boolean isEffectiveMinAggregation = false;
			for (KafkaTopicPartitionState<?> state : allPartitions) {

				// we access the current watermark for the periodic assigners under the state
				// lock, to prevent concurrent modification to any internal variables
				final long curr;
				//noinspection SynchronizationOnLocalVariableOrMethodParameter
				synchronized (state) {
					curr = ((KafkaTopicPartitionStateWithPeriodicWatermarks<?, ?>) state).getCurrentWatermarkTimestamp();
				}

				minAcrossAll = Math.min(minAcrossAll, curr);
				isEffectiveMinAggregation = true;
			}

			// emit next watermark, if there is one
			// 每隔 interval 对 watermark 进行合并取其最小的 watermark 发送至下游算子，并且保持单调递增
			if (isEffectiveMinAggregation && minAcrossAll > lastWatermarkTimestamp) {
				lastWatermarkTimestamp = minAcrossAll;
				emitter.emitWatermark(new Watermark(minAcrossAll));// StreamSourceContexts.ManualWatermarkContext,watermark 与 record 的发送路径是分开的
			}

			// schedule the next watermark
			timerService.registerTimer(timerService.getCurrentProcessingTime() + interval, this);
		}

......
					// 如果元素是 watermark，就准备更新当前 channel 的 watermark 值（并不是简单赋值，因为有乱序存在）
					if (recordOrMark.isWatermark()) {
						// handle watermark
						statusWatermarkValve.inputWatermark(recordOrMark.asWatermark(), currentChannel);
						continue;
						// 如果元素是 status，就进行相应处理。可以看作是一个 flag，标志着当前 stream 接下来即将没有元素输入（idle），
						// 或者当前即将由空闲状态转为有元素状态（active）。同时，StreamStatus 还对如何处理 watermark 有影响。
						// 通过发送 status，上游的 operator 可以很方便的通知下游当前的数据流的状态。
					} else if (recordOrMark.isStreamStatus()) {
						// handle stream status 把对应的 channelStatuse 改为 空闲，
						// 然后如果所有的 channelStatuse 都是 idle 则找到最大的 watermark 并处理，否则找到最小的 watermark 并处理
						statusWatermarkValve.inputStreamStatus(recordOrMark.asStreamStatus(), currentChannel);
						continue;
					} 
					......

//当水印输入时的处理操作
	public void inputWatermark(Watermark watermark, int channelIndex) {
		// ignore the input watermark if its input channel, or all input channels are idle (i.e. overall the valve is idle).
		// streamStatus 和 channelStatus 都是 active 才继续往下计算
		if (lastOutputStreamStatus.isActive() && channelStatuses[channelIndex].streamStatus.isActive()) {
			long watermarkMillis = watermark.getTimestamp();

			// if the input watermark's value is less than the last received watermark for its input channel, ignore it also.
			// ignore 小于已经接收到的 watermark 的 watermark，保持其单调性
			if (watermarkMillis > channelStatuses[channelIndex].watermark) {
				channelStatuses[channelIndex].watermark = watermarkMillis;

				// previously unaligned input channels are now aligned if its watermark has caught up
				// 如果之前未对齐的，现在对齐。
				if (!channelStatuses[channelIndex].isWatermarkAligned && watermarkMillis >= lastOutputWatermark) {
					channelStatuses[channelIndex].isWatermarkAligned = true;
				}

				// ow, attempt to find a new min watermark across all aligned channels
				findAndOutputNewMinWatermarkAcrossAlignedChannels();
			}
		}
	}

@Override
		public void handleWatermark(Watermark watermark) {
			try {
				synchronized (lock) {
					watermarkGauge.setCurrentWatermark(watermark.getTimestamp());//gauge
					//处理 watermark 的入口
					operator.processWatermark(watermark);
				}
			} catch (Exception e) {
				throw new RuntimeException("Exception occurred while processing valve output watermark: ", e);
			}
		}

public void processWatermark(Watermark mark) throws Exception {//operator.processWatermark(mark)
		if (timeServiceManager != null) {//有 timeService 则不为 null 如 window   InternalTimeServiceManager
			//timeService
			timeServiceManager.advanceWatermark(mark);
		}
		//处理结束之后继续往下游发送依次循环。
		output.emitWatermark(mark);
	}

	public void advanceWatermark(Watermark watermark) throws Exception {
		for (InternalTimerServiceImpl<?, ?> service : timerServices.values()) {
			service.advanceWatermark(watermark.getTimestamp());//处理 watermark  event time 对于 trigger 的调用逻辑
		}
	}

	public void advanceWatermark(long time) throws Exception {//watermark timestamp
		currentWatermark = time;

		InternalTimer<K, N> timer;
		
		while ((timer = eventTimeTimersQueue.peek()) != null && timer.getTimestamp() <= time) {//自定义，触发一系列满足条件的 key
			eventTimeTimersQueue.poll();
			keyContext.setCurrentKey(timer.getKey());//
			// 触发 triggerTarget.onEventTime
			triggerTarget.onEventTime(timer);
		}
	}

// 这个是通过 timer 来调用的
	// window processElement 的时候 registerCleanupTimer(window)   window.maxTimestamp()+allowedLateness
	// 和 eventTrigger onElement  registerEventTimeTimer(window.maxTimestamp()) 会创建相应的 timer
	// 所以当 allowedLateness 不为 0 的时候，同一个 window.maxTimestamp() 对应的 eventWindow 会触发两次,
	// 而且默认 windowAssigner.isEventTime() && isCleanupTime(triggerContext.window, timer.getTimestamp()) 才会清除 window state
	public void onEventTime(InternalTimer<K, W> timer) throws Exception {
		triggerContext.key = timer.getKey();
		triggerContext.window = timer.getNamespace();

		MergingWindowSet<W> mergingWindows;

		if (windowAssigner instanceof MergingWindowAssigner) {
			mergingWindows = getMergingWindowSet();
			W stateWindow = mergingWindows.getStateWindow(triggerContext.window);
			if (stateWindow == null) {
				// Timer firing for non-existent window, this can only happen if a
				// trigger did not clean up timers. We have already cleared the merging
				// window and therefore the Trigger state, however, so nothing to do.
				return;
			} else {
				windowState.setCurrentNamespace(stateWindow);
			}
		} else {
			windowState.setCurrentNamespace(triggerContext.window);
			mergingWindows = null;
		}

		TriggerResult triggerResult = triggerContext.onEventTime(timer.getTimestamp());

		if (triggerResult.isFire()) {
			ACC contents = windowState.get();
			if (contents != null) {
				emitWindowContents(triggerContext.window, contents);
			}
		}

		if (triggerResult.isPurge()) {
			windowState.clear();
		}

		if (windowAssigner.isEventTime() && isCleanupTime(triggerContext.window, timer.getTimestamp())) {
			clearAllState(triggerContext.window, windowState, mergingWindows);
		}

		if (mergingWindows != null) {
			// need to make sure to update the merging state in state
			mergingWindows.persist();
		}
	}

	1. 水印在满足单调递增的情况下，要么直接发往下游( OneInputStreamOperator,像 keyby、filter、flatMap ),
	要么取最小值然后发往下游( TwoInputStreamOperator,像 co系列 coFlatMap、IntervalJoinOperator、TemporalJoin)
	2. 设置水印时间为当前 StreamRecord 中的时间戳，此时间戳是<= watermark ，因为 watermark 是单调递增的，而 StreamRecord 的时间戳就是提取出来的时间戳