flink的窗口时间属性TimeCharacteristic分为三种：ProcessingTime，IngestionTime，EventTime。
- ProcessingTime是处理时间，所有基于时间的操作(如时间窗口)将使用运行各自操作符的机器的系统时间，优点是简单，缺点是依赖每个节点的系统时间，如果数据流速不一样比如出现反压等会导致窗口数据不确定；
- EventTime是事件时间，是从数据里面提取出来的时间，EventTime依赖于数据，不依赖于系统时间，优点是能严格按照数据时间的发生顺序进行窗口统计，缺点是如果数据出现断流了，会导致watermark无法提高，从而无法导致窗口的触发
- IngestionTime是摄入时间，是数据从数据源取出的时候附带上系统时间作为watermark，作为ProcessingTime和EventTime的折中方案，会定时的往下游发送watermark，这个watermark是系统时间，不会因为数据断流导致watermark无法提高。适用于对数据延迟不大，对数据窗口统计要求不是很严格的场景。
下面接下来从源码分析IngestionTime，代码在StreamSourceContexts.java中：
switch (timeCharacteristic) {
case EventTime:
ctx = new ManualWatermarkContext<>(
output,
processingTimeService,
checkpointLock,
streamStatusMaintainer,
idleTimeout);
break;
case IngestionTime:
ctx = new AutomaticWatermarkContext<>(
output,
watermarkInterval,
processingTimeService,
checkpointLock,
streamStatusMaintainer,
idleTimeout);
break;
case ProcessingTime:
ctx = new NonTimestampContext<>(checkpointLock, output);
break;
default:
throw new IllegalArgumentException(String.valueOf(timeCharacteristic));
}
IngestionTime是通过AutomaticWatermarkContext类来实现逻辑的，继续看AutomaticWatermarkContext：
private AutomaticWatermarkContext(
final Output<StreamRecord<T>> output,
final long watermarkInterval,
final ProcessingTimeService timeService,
final Object checkpointLock,
final StreamStatusMaintainer streamStatusMaintainer,
final long idleTimeout) {
super(timeService, checkpointLock, streamStatusMaintainer, idleTimeout);
this.output = Preconditions.checkNotNull(output, "The output cannot be null.");
Preconditions.checkArgument(watermarkInterval >= 1L, "The watermark interval cannot be smaller than 1 ms.");
this.watermarkInterval = watermarkInterval;
this.reuse = new StreamRecord<>(null);
this.lastRecordTime = Long.MIN_VALUE;
long now = this.timeService.getCurrentProcessingTime();
//注册定时器，等到下一个watermarkInterval的时候触发
this.nextWatermarkTimer = this.timeService.registerTimer(now + watermarkInterval,
new WatermarkEmittingTask(this.timeService, checkpointLock, output));
}
可以看到，最后一行代码那里，注册了一个定时器，在下一个watermarkInterval时触发执行，再看触发的WatermarkEmittingTask里面的逻辑：
@Override
public void onProcessingTime(long timestamp) {
final long currentTime = timeService.getCurrentProcessingTime();
synchronized (lock) {
// we should continue to automatically emit watermarks if we are active
if (streamStatusMaintainer.getStreamStatus().isActive()) {
if (idleTimeout != -1 && currentTime - lastRecordTime > idleTimeout) {
// if we are configured to detect idleness, piggy-back the idle detection check on the
// watermark interval, so that we may possibly discover idle sources faster before waiting
// for the next idle check to fire
markAsTemporarilyIdle();
// no need to finish the next check, as we are now idle.
cancelNextIdleDetectionTask();
} else if (currentTime > nextWatermarkTime) {
// align the watermarks across all machines. this will ensure that we
// don't have watermarks that creep along at different intervals because
// the machine clocks are out of sync
// 这里是发送的watermark的值，取整处理
final long watermarkTime = currentTime - (currentTime % watermarkInterval);
output.emitWatermark(new Watermark(watermarkTime));
nextWatermarkTime = watermarkTime + watermarkInterval;
}
}
}
// 注册下一次定时器，下一次的执行时间又是间隔watermarkInterval
long nextWatermark = currentTime + watermarkInterval;
nextWatermarkTimer = this.timeService.registerTimer(
nextWatermark, new WatermarkEmittingTask(this.timeService, lock, output));
}
至此，逻辑已经很清楚了，IngestionTime是每经过watermarkInterval间隔发送一次watermark，watermark的值就是当前系统时间取整：currentTime - (currentTime % watermarkInterval)。IngestionTime并不会因为数据断流导致watermark无法提升，如果对数据延迟不大，对数据窗口统计要求不是很严格的场景，同时可能出现数据断流的情况下，IngestionTime比较适用。
