一文搞懂 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();		}	}