import org.apache.flink.streaming.api.operators.InternalTimer; //导入依赖的package包/类
/**
 * Occurs when an event-time timer fires due to watermark progression.
 *
 * @param timer the timer details.
 */
@Override
public void onEventTime(InternalTimer<K, VoidNamespace> timer) throws Exception {

    long currentWatermark = internalTimerService.currentWatermark();

    PriorityQueue<Long> sortedTimestamps = getSortedTimestamps();
    while (!sortedTimestamps.isEmpty() && sortedTimestamps.peek() <= currentWatermark) {
        long timestamp = sortedTimestamps.poll();
        for (T event : elementQueueState.get(timestamp)) {
            output.collect(new StreamRecord<>(event, timestamp));
        }
        elementQueueState.remove(timestamp);
    }

    if (sortedTimestamps.isEmpty()) {
        elementQueueState.clear();
    }

    if (!sortedTimestamps.isEmpty()) {
        saveRegisterWatermarkTimer();
    }
}
 

import org.apache.flink.streaming.api.operators.InternalTimer; //导入依赖的package包/类
@Override
public void fireTimer(InternalTimer<?, TimerData> timer) {
  doFnRunner.processElement(WindowedValue.valueInGlobalWindow(
      KeyedWorkItems.<K, InputT>timersWorkItem(
          (K) keyedStateInternals.getKey(),
          Collections.singletonList(timer.getNamespace()))));
}
 

import org.apache.flink.streaming.api.operators.InternalTimer; //导入依赖的package包/类
@Override
public void fireTimer(InternalTimer<?, TimerInternals.TimerData> timer) {
  doFnRunner.processElement(WindowedValue.valueInGlobalWindow(
      KeyedWorkItems.<String, KV<InputT, RestrictionT>>timersWorkItem(
          (String) keyedStateInternals.getKey(),
          Collections.singletonList(timer.getNamespace()))));
}
 

import org.apache.flink.streaming.api.operators.InternalTimer; //导入依赖的package包/类
public void fireTimer(InternalTimer<?, TimerData> timer) {
  TimerInternals.TimerData timerData = timer.getNamespace();
  StateNamespace namespace = timerData.getNamespace();
  // This is a user timer, so namespace must be WindowNamespace
  checkArgument(namespace instanceof WindowNamespace);
  BoundedWindow window = ((WindowNamespace) namespace).getWindow();
  pushbackDoFnRunner.onTimer(timerData.getTimerId(), window,
      timerData.getTimestamp(), timerData.getDomain());
}
 

import org.apache.flink.streaming.api.operators.InternalTimer; //导入依赖的package包/类
@Override
public void onEventTime(InternalTimer<Long, Long> timer) throws Exception {
	ValueState<Long> state = getKeyedStateBackend().getPartitionedState(
			timer.getNamespace(),
			LongSerializer.INSTANCE,
			stateDescriptor);

	assertEquals(state.value(), timer.getNamespace());
	getRuntimeContext().getAccumulator(SUCCESSFUL_EVENT_TIME_CHECK_ACCUMULATOR).add(1);
}
 

import org.apache.flink.streaming.api.operators.InternalTimer; //导入依赖的package包/类
@Override
public void onProcessingTime(InternalTimer<Long, Long> timer) throws Exception {
	ValueState<Long> state = getKeyedStateBackend().getPartitionedState(
			timer.getNamespace(),
			LongSerializer.INSTANCE,
			stateDescriptor);

	assertEquals(state.value(), timer.getNamespace());
	getRuntimeContext().getAccumulator(SUCCESSFUL_PROCESSING_TIME_CHECK_ACCUMULATOR).add(1);
}
 

import org.apache.flink.streaming.api.operators.InternalTimer; //导入依赖的package包/类
@Override
public void onEventTime(InternalTimer<KEY, VoidNamespace> timer) throws Exception {

	// 1) get the queue of pending elements for the key and the corresponding NFA,
	// 2) process the pending elements in event time order and custom comparator if exists
	//		by feeding them in the NFA
	// 3) advance the time to the current watermark, so that expired patterns are discarded.
	// 4) update the stored state for the key, by only storing the new NFA and MapState iff they
	//		have state to be used later.
	// 5) update the last seen watermark.

	// STEP 1
	PriorityQueue<Long> sortedTimestamps = getSortedTimestamps();
	NFA<IN> nfa = getNFA();

	// STEP 2
	while (!sortedTimestamps.isEmpty() && sortedTimestamps.peek() <= timerService.currentWatermark()) {
		long timestamp = sortedTimestamps.poll();
		sort(elementQueueState.get(timestamp)).forEachOrdered(
			event -> processEvent(nfa, event, timestamp)
		);
		elementQueueState.remove(timestamp);
	}

	// STEP 3
	advanceTime(nfa, timerService.currentWatermark());

	// STEP 4
	if (sortedTimestamps.isEmpty()) {
		elementQueueState.clear();
	}
	updateNFA(nfa);

	if (!sortedTimestamps.isEmpty() || !nfa.isEmpty()) {
		saveRegisterWatermarkTimer();
	}

	// STEP 5
	updateLastSeenWatermark(timerService.currentWatermark());
}
 

import org.apache.flink.streaming.api.operators.InternalTimer; //导入依赖的package包/类
@Override
public void onProcessingTime(InternalTimer<KEY, VoidNamespace> timer) throws Exception {
	// 1) get the queue of pending elements for the key and the corresponding NFA,
	// 2) process the pending elements in process time order and custom comparator if exists
	//		by feeding them in the NFA
	// 3) update the stored state for the key, by only storing the new NFA and MapState iff they
	//		have state to be used later.

	// STEP 1
	PriorityQueue<Long> sortedTimestamps = getSortedTimestamps();
	NFA<IN> nfa = getNFA();

	// STEP 2
	while (!sortedTimestamps.isEmpty()) {
		long timestamp = sortedTimestamps.poll();
		sort(elementQueueState.get(timestamp)).forEachOrdered(
			event -> processEvent(nfa, event, timestamp)
		);
		elementQueueState.remove(timestamp);
	}

	// STEP 3
	if (sortedTimestamps.isEmpty()) {
		elementQueueState.clear();
	}
	updateNFA(nfa);
}
 

import org.apache.flink.streaming.api.operators.InternalTimer; //导入依赖的package包/类
@Override
public void onEventTime(InternalTimer<K, VoidNamespace> timer) throws Exception {
	collector.setAbsoluteTimestamp(timer.getTimestamp());
	onTimerContext.timeDomain = TimeDomain.EVENT_TIME;
	onTimerContext.timer = timer;
	userFunction.onTimer(timer.getTimestamp(), onTimerContext, collector);
	onTimerContext.timeDomain = null;
	onTimerContext.timer = null;
}
 

import org.apache.flink.streaming.api.operators.InternalTimer; //导入依赖的package包/类
@Override
public void onProcessingTime(InternalTimer<K, VoidNamespace> timer) throws Exception {
	collector.eraseTimestamp();
	onTimerContext.timeDomain = TimeDomain.PROCESSING_TIME;
	onTimerContext.timer = timer;
	userFunction.onTimer(timer.getTimestamp(), onTimerContext, collector);
	onTimerContext.timeDomain = null;
	onTimerContext.timer = null;
}
 

import org.apache.flink.streaming.api.operators.InternalTimer; //导入依赖的package包/类
@Override
public void onProcessingTime(InternalTimer<K, VoidNamespace> timer) throws Exception {
	collector.setAbsoluteTimestamp(timer.getTimestamp());
	onTimerContext.timeDomain = TimeDomain.PROCESSING_TIME;
	onTimerContext.timer = timer;
	userFunction.onTimer(timer.getTimestamp(), onTimerContext, collector);
	onTimerContext.timeDomain = null;
	onTimerContext.timer = null;
}
 

import org.apache.flink.streaming.api.operators.InternalTimer; //导入依赖的package包/类
@Override
public void onEventTime(InternalTimer<KEY, VoidNamespace> internalTimer) throws Exception {
    // not supported yet
}
 

import org.apache.flink.streaming.api.operators.InternalTimer; //导入依赖的package包/类
@Override
public void onProcessingTime(InternalTimer<KEY, VoidNamespace> internalTimer) throws Exception {
    EPServiceProvider epServiceProvider = getServiceProvider(this.hashCode() + "");
    epServiceProvider.getEPRuntime().sendEvent(new CurrentTimeSpanEvent(internalTimer.getTimestamp()));
    this.engineState.update(epServiceProvider);
}
 

import org.apache.flink.streaming.api.operators.InternalTimer; //导入依赖的package包/类
@Override
public void onProcessingTime(InternalTimer<K, VoidNamespace> timer) throws Exception {
}
 

import org.apache.flink.streaming.api.operators.InternalTimer; //导入依赖的package包/类
@Override
public void onEventTime(InternalTimer<Object, TimerData> timer) throws Exception {
  // We don't have to cal checkInvokeStartBundle() because it's already called in
  // processWatermark*().
  fireTimer(timer);
}
 

import org.apache.flink.streaming.api.operators.InternalTimer; //导入依赖的package包/类
@Override
public void onProcessingTime(InternalTimer<Object, TimerData> timer) throws Exception {
  checkInvokeStartBundle();
  fireTimer(timer);
}
 

import org.apache.flink.streaming.api.operators.InternalTimer; //导入依赖的package包/类
@Override
public void onEventTime(InternalTimer<K, W> timer) throws Exception {

	triggerContext.key = timer.getKey();
	triggerContext.window = timer.getNamespace();
	evictorContext.key = timer.getKey();
	evictorContext.window = timer.getNamespace();

	MergingWindowSet<W> mergingWindows = null;

	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 {
			evictingWindowState.setCurrentNamespace(stateWindow);
		}
	} else {
		evictingWindowState.setCurrentNamespace(triggerContext.window);
	}

	Iterable<StreamRecord<IN>> contents = evictingWindowState.get();

	if (contents != null) {
		TriggerResult triggerResult = triggerContext.onEventTime(timer.getTimestamp());
		if (triggerResult.isFire()) {
			emitWindowContents(triggerContext.window, contents, evictingWindowState);
		}
		if (triggerResult.isPurge()) {
			evictingWindowState.clear();
		}
	}

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

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

import org.apache.flink.streaming.api.operators.InternalTimer; //导入依赖的package包/类
@Override
public void onProcessingTime(InternalTimer<K, W> timer) throws Exception {
	triggerContext.key = timer.getKey();
	triggerContext.window = timer.getNamespace();
	evictorContext.key = timer.getKey();
	evictorContext.window = timer.getNamespace();

	MergingWindowSet<W> mergingWindows = null;

	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 {
			evictingWindowState.setCurrentNamespace(stateWindow);
		}
	} else {
		evictingWindowState.setCurrentNamespace(triggerContext.window);
	}

	Iterable<StreamRecord<IN>> contents = evictingWindowState.get();

	if (contents != null) {
		TriggerResult triggerResult = triggerContext.onProcessingTime(timer.getTimestamp());
		if (triggerResult.isFire()) {
			emitWindowContents(triggerContext.window, contents, evictingWindowState);
		}
		if (triggerResult.isPurge()) {
			evictingWindowState.clear();
		}
	}

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

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

import org.apache.flink.streaming.api.operators.InternalTimer; //导入依赖的package包/类
@Override
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;
	}

	ACC contents = null;
	if (windowState != null) {
		contents = windowState.get();
	}

	if (contents != null) {
		TriggerResult triggerResult = triggerContext.onEventTime(timer.getTimestamp());
		if (triggerResult.isFire()) {
			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();
	}
}
 

import org.apache.flink.streaming.api.operators.InternalTimer; //导入依赖的package包/类
@Override
public void onProcessingTime(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;
	}

	ACC contents = null;
	if (windowState != null) {
		contents = windowState.get();
	}

	if (contents != null) {
		TriggerResult triggerResult = triggerContext.onProcessingTime(timer.getTimestamp());
		if (triggerResult.isFire()) {
			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();
	}
}