import org.apache.flink.streaming.api.operators.AbstractStreamOperator; //导入依赖的package包/类
@Override
public void evaluateWindow(Collector<Type> out, TimeWindow window,
							AbstractStreamOperator<Type> operator) throws Exception {
	if (previousPanes.isEmpty()) {
		// optimized path for single pane case
		for (KeyMap.Entry<Key, Type> entry : latestPane) {
			out.collect(entry.getValue());
		}
	}
	else {
		// general code path for multi-pane case
		AggregatingTraversal<Key, Type> evaluator = new AggregatingTraversal<>(reducer, out, operator);
		traverseAllPanes(evaluator, evaluationPass);
	}

	evaluationPass++;
}
 

import org.apache.flink.streaming.api.operators.AbstractStreamOperator; //导入依赖的package包/类
@Override
public void evaluateWindow(Collector<Result> out, final TimeWindow window,
							AbstractStreamOperator<Result> operator) throws Exception {
	if (previousPanes.isEmpty()) {
		// optimized path for single pane case (tumbling window)
		for (KeyMap.Entry<Key, ArrayList<Type>> entry : latestPane) {
			Key key = entry.getKey();
			operator.setCurrentKey(key);
			context.globalState = operator.getKeyedStateStore();

			function.process(entry.getKey(), window, context, entry.getValue(), out);
		}
	}
	else {
		// general code path for multi-pane case
		WindowFunctionTraversal<Key, Type, Result> evaluator = new WindowFunctionTraversal<>(
				function, window, out, operator, context);
		traverseAllPanes(evaluator, evaluationPass);
	}

	evaluationPass++;
}
 

import org.apache.flink.streaming.api.operators.AbstractStreamOperator; //导入依赖的package包/类
/**
 * Users of the test harness can call this utility method to setup the stream config
 * if there will only be a single operator to be tested. The method will setup the
 * outgoing network connection for the operator.
 *
 * <p>For more advanced test cases such as testing chains of multiple operators with the harness,
 * please manually configure the stream config.
 */
public void setupOutputForSingletonOperatorChain() {
	Preconditions.checkState(!setupCalled, "This harness was already setup.");
	setupCalled = true;
	streamConfig.setChainStart();
	streamConfig.setBufferTimeout(0);
	streamConfig.setTimeCharacteristic(TimeCharacteristic.EventTime);
	streamConfig.setOutputSelectors(Collections.<OutputSelector<?>>emptyList());
	streamConfig.setNumberOfOutputs(1);
	streamConfig.setTypeSerializerOut(outputSerializer);
	streamConfig.setVertexID(0);
	streamConfig.setOperatorID(new OperatorID(4711L, 123L));

	StreamOperator<OUT> dummyOperator = new AbstractStreamOperator<OUT>() {
		private static final long serialVersionUID = 1L;
	};

	List<StreamEdge> outEdgesInOrder = new LinkedList<StreamEdge>();
	StreamNode sourceVertexDummy = new StreamNode(null, 0, "group", dummyOperator, "source dummy", new LinkedList<OutputSelector<?>>(), SourceStreamTask.class);
	StreamNode targetVertexDummy = new StreamNode(null, 1, "group", dummyOperator, "target dummy", new LinkedList<OutputSelector<?>>(), SourceStreamTask.class);

	outEdgesInOrder.add(new StreamEdge(sourceVertexDummy, targetVertexDummy, 0, new LinkedList<String>(), new BroadcastPartitioner<Object>(), null /* output tag */));

	streamConfig.setOutEdgesInOrder(outEdgesInOrder);
	streamConfig.setNonChainedOutputs(outEdgesInOrder);
}
 

import org.apache.flink.streaming.api.operators.AbstractStreamOperator; //导入依赖的package包/类
WindowFunctionTraversal(InternalWindowFunction<Iterable<Type>, Result, Key, Window> function, TimeWindow window,
						Collector<Result> out, AbstractStreamOperator<Result> contextOperator, AccumulatingKeyedTimePanesContext context) {
	this.function = function;
	this.out = out;
	this.unionIterator = new UnionIterator<>();
	this.window = window;
	this.contextOperator = contextOperator;
	this.context = context;
}
 

import org.apache.flink.streaming.api.operators.AbstractStreamOperator; //导入依赖的package包/类
private static <T extends Serializable> List<T> runRichSourceFunction(SourceFunction<T> sourceFunction) throws Exception {
	try (MockEnvironment environment = new MockEnvironment("MockTask", 3 * 1024 * 1024, new MockInputSplitProvider(), 1024)) {
		AbstractStreamOperator<?> operator = mock(AbstractStreamOperator.class);
		when(operator.getExecutionConfig()).thenReturn(new ExecutionConfig());

		RuntimeContext runtimeContext = new StreamingRuntimeContext(
			operator,
			environment,
			new HashMap<>());
		((RichFunction) sourceFunction).setRuntimeContext(runtimeContext);
		((RichFunction) sourceFunction).open(new Configuration());

		return runNonRichSourceFunction(sourceFunction);
	}
}
 

import org.apache.flink.streaming.api.operators.AbstractStreamOperator; //导入依赖的package包/类
@VisibleForTesting
public int numProcessingTimeTimers() {
	if (operator instanceof AbstractStreamOperator) {
		return ((AbstractStreamOperator) operator).numProcessingTimeTimers();
	} else {
		throw new UnsupportedOperationException();
	}
}
 

import org.apache.flink.streaming.api.operators.AbstractStreamOperator; //导入依赖的package包/类
@VisibleForTesting
public int numEventTimeTimers() {
	if (operator instanceof AbstractStreamOperator) {
		return ((AbstractStreamOperator) operator).numEventTimeTimers();
	} else {
		throw new UnsupportedOperationException();
	}
}
 

import org.apache.flink.streaming.api.operators.AbstractStreamOperator; //导入依赖的package包/类
AggregatingTraversal(ReduceFunction<Type> function, Collector<Type> out,
						AbstractStreamOperator<Type> operator) {
	this.function = function;
	this.out = out;
	this.operator = operator;
}
 

import org.apache.flink.streaming.api.operators.AbstractStreamOperator; //导入依赖的package包/类
@Test
public void testFailingCheckpointStreamOperator() throws Exception {
	final long checkpointId = 42L;
	final long timestamp = 1L;

	TaskInfo mockTaskInfo = mock(TaskInfo.class);
	when(mockTaskInfo.getTaskNameWithSubtasks()).thenReturn("foobar");
	when(mockTaskInfo.getIndexOfThisSubtask()).thenReturn(0);
	Environment mockEnvironment = mock(Environment.class);
	when(mockEnvironment.getTaskInfo()).thenReturn(mockTaskInfo);

	StreamTask<?, ?> streamTask = new EmptyStreamTask(mockEnvironment);
	CheckpointMetaData checkpointMetaData = new CheckpointMetaData(checkpointId, timestamp);

	// mock the operators
	StreamOperator<?> streamOperator1 = mock(StreamOperator.class);
	StreamOperator<?> streamOperator2 = mock(StreamOperator.class);
	StreamOperator<?> streamOperator3 = mock(StreamOperator.class);

	// mock the returned snapshots
	OperatorSnapshotResult operatorSnapshotResult1 = mock(OperatorSnapshotResult.class);
	OperatorSnapshotResult operatorSnapshotResult2 = mock(OperatorSnapshotResult.class);

	final Exception testException = new Exception("Test exception");

	when(streamOperator1.snapshotState(anyLong(), anyLong(), any(CheckpointOptions.class))).thenReturn(operatorSnapshotResult1);
	when(streamOperator2.snapshotState(anyLong(), anyLong(), any(CheckpointOptions.class))).thenReturn(operatorSnapshotResult2);
	when(streamOperator3.snapshotState(anyLong(), anyLong(), any(CheckpointOptions.class))).thenThrow(testException);

	OperatorID operatorID1 = new OperatorID();
	OperatorID operatorID2 = new OperatorID();
	OperatorID operatorID3 = new OperatorID();
	when(streamOperator1.getOperatorID()).thenReturn(operatorID1);
	when(streamOperator2.getOperatorID()).thenReturn(operatorID2);
	when(streamOperator3.getOperatorID()).thenReturn(operatorID3);

	// set up the task

	StreamOperator<?>[] streamOperators = {streamOperator1, streamOperator2, streamOperator3};

	OperatorChain<Void, AbstractStreamOperator<Void>> operatorChain = mock(OperatorChain.class);
	when(operatorChain.getAllOperators()).thenReturn(streamOperators);

	Whitebox.setInternalState(streamTask, "isRunning", true);
	Whitebox.setInternalState(streamTask, "lock", new Object());
	Whitebox.setInternalState(streamTask, "operatorChain", operatorChain);
	Whitebox.setInternalState(streamTask, "cancelables", new CloseableRegistry());
	Whitebox.setInternalState(streamTask, "configuration", new StreamConfig(new Configuration()));

	CheckpointExceptionHandlerFactory checkpointExceptionHandlerFactory = new CheckpointExceptionHandlerFactory();
	CheckpointExceptionHandler checkpointExceptionHandler =
		checkpointExceptionHandlerFactory.createCheckpointExceptionHandler(true, mockEnvironment);
	Whitebox.setInternalState(streamTask, "synchronousCheckpointExceptionHandler", checkpointExceptionHandler);

	StreamTask.AsyncCheckpointExceptionHandler asyncCheckpointExceptionHandler =
		new StreamTask.AsyncCheckpointExceptionHandler(streamTask);
	Whitebox.setInternalState(streamTask, "asynchronousCheckpointExceptionHandler", asyncCheckpointExceptionHandler);

	try {
		streamTask.triggerCheckpoint(checkpointMetaData, CheckpointOptions.forCheckpoint());
		fail("Expected test exception here.");
	} catch (Exception e) {
		assertEquals(testException, e.getCause());
	}

	verify(operatorSnapshotResult1).cancel();
	verify(operatorSnapshotResult2).cancel();
}
 

import org.apache.flink.streaming.api.operators.AbstractStreamOperator; //导入依赖的package包/类
/**
 * Tests that in case of a failing AsyncCheckpointRunnable all operator snapshot results are
 * cancelled and all non partitioned state handles are discarded.
 */
@Test
public void testFailingAsyncCheckpointRunnable() throws Exception {
	final long checkpointId = 42L;
	final long timestamp = 1L;

	TaskInfo mockTaskInfo = mock(TaskInfo.class);
	when(mockTaskInfo.getTaskNameWithSubtasks()).thenReturn("foobar");
	when(mockTaskInfo.getIndexOfThisSubtask()).thenReturn(0);
	Environment mockEnvironment = mock(Environment.class);
	when(mockEnvironment.getTaskInfo()).thenReturn(mockTaskInfo);

	StreamTask<?, ?> streamTask = spy(new EmptyStreamTask(mockEnvironment));
	CheckpointMetaData checkpointMetaData = new CheckpointMetaData(checkpointId, timestamp);

	// mock the operators
	StreamOperator<?> streamOperator1 = mock(StreamOperator.class);
	StreamOperator<?> streamOperator2 = mock(StreamOperator.class);
	StreamOperator<?> streamOperator3 = mock(StreamOperator.class);

	// mock the new state operator snapshots
	OperatorSnapshotResult operatorSnapshotResult1 = mock(OperatorSnapshotResult.class);
	OperatorSnapshotResult operatorSnapshotResult2 = mock(OperatorSnapshotResult.class);
	OperatorSnapshotResult operatorSnapshotResult3 = mock(OperatorSnapshotResult.class);

	RunnableFuture<OperatorStateHandle> failingFuture = mock(RunnableFuture.class);
	when(failingFuture.get()).thenThrow(new ExecutionException(new Exception("Test exception")));

	when(operatorSnapshotResult3.getOperatorStateRawFuture()).thenReturn(failingFuture);

	when(streamOperator1.snapshotState(anyLong(), anyLong(), any(CheckpointOptions.class))).thenReturn(operatorSnapshotResult1);
	when(streamOperator2.snapshotState(anyLong(), anyLong(), any(CheckpointOptions.class))).thenReturn(operatorSnapshotResult2);
	when(streamOperator3.snapshotState(anyLong(), anyLong(), any(CheckpointOptions.class))).thenReturn(operatorSnapshotResult3);

	OperatorID operatorID1 = new OperatorID();
	OperatorID operatorID2 = new OperatorID();
	OperatorID operatorID3 = new OperatorID();
	when(streamOperator1.getOperatorID()).thenReturn(operatorID1);
	when(streamOperator2.getOperatorID()).thenReturn(operatorID2);
	when(streamOperator3.getOperatorID()).thenReturn(operatorID3);

	StreamOperator<?>[] streamOperators = {streamOperator1, streamOperator2, streamOperator3};

	OperatorChain<Void, AbstractStreamOperator<Void>> operatorChain = mock(OperatorChain.class);
	when(operatorChain.getAllOperators()).thenReturn(streamOperators);

	Whitebox.setInternalState(streamTask, "isRunning", true);
	Whitebox.setInternalState(streamTask, "lock", new Object());
	Whitebox.setInternalState(streamTask, "operatorChain", operatorChain);
	Whitebox.setInternalState(streamTask, "cancelables", new CloseableRegistry());
	Whitebox.setInternalState(streamTask, "asyncOperationsThreadPool", new DirectExecutorService());
	Whitebox.setInternalState(streamTask, "configuration", new StreamConfig(new Configuration()));

	CheckpointExceptionHandlerFactory checkpointExceptionHandlerFactory = new CheckpointExceptionHandlerFactory();
	CheckpointExceptionHandler checkpointExceptionHandler =
		checkpointExceptionHandlerFactory.createCheckpointExceptionHandler(true, mockEnvironment);
	Whitebox.setInternalState(streamTask, "synchronousCheckpointExceptionHandler", checkpointExceptionHandler);

	StreamTask.AsyncCheckpointExceptionHandler asyncCheckpointExceptionHandler =
		new StreamTask.AsyncCheckpointExceptionHandler(streamTask);
	Whitebox.setInternalState(streamTask, "asynchronousCheckpointExceptionHandler", asyncCheckpointExceptionHandler);

	streamTask.triggerCheckpoint(checkpointMetaData, CheckpointOptions.forCheckpoint());

	verify(streamTask).handleAsyncException(anyString(), any(Throwable.class));

	verify(operatorSnapshotResult1).cancel();
	verify(operatorSnapshotResult2).cancel();
	verify(operatorSnapshotResult3).cancel();
}
 

import org.apache.flink.streaming.api.operators.AbstractStreamOperator; //导入依赖的package包/类
/**
 * FLINK-5985
 *
 * <p>This test ensures that empty snapshots (no op/keyed stated whatsoever) will be reported as stateless tasks. This
 * happens by translating an empty {@link SubtaskState} into reporting 'null' to #acknowledgeCheckpoint.
 */
@Test
public void testEmptySubtaskStateLeadsToStatelessAcknowledgment() throws Exception {
	final long checkpointId = 42L;
	final long timestamp = 1L;

	TaskInfo mockTaskInfo = mock(TaskInfo.class);

	when(mockTaskInfo.getTaskNameWithSubtasks()).thenReturn("foobar");
	when(mockTaskInfo.getIndexOfThisSubtask()).thenReturn(0);

	Environment mockEnvironment = mock(Environment.class);

	// latch blocks until the async checkpoint thread acknowledges
	final OneShotLatch checkpointCompletedLatch = new OneShotLatch();
	final List<SubtaskState> checkpointResult = new ArrayList<>(1);

	// we remember what is acknowledged (expected to be null as our task will snapshot empty states).
	doAnswer(new Answer() {
		@Override
		public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
			SubtaskState subtaskState = invocationOnMock.getArgumentAt(2, SubtaskState.class);
			checkpointResult.add(subtaskState);
			checkpointCompletedLatch.trigger();
			return null;
		}
	}).when(mockEnvironment).acknowledgeCheckpoint(anyLong(), any(CheckpointMetrics.class), any(TaskStateSnapshot.class));

	when(mockEnvironment.getTaskInfo()).thenReturn(mockTaskInfo);

	StreamTask<?, ?> streamTask = new EmptyStreamTask(mockEnvironment);
	CheckpointMetaData checkpointMetaData = new CheckpointMetaData(checkpointId, timestamp);

	// mock the operators
	StreamOperator<?> statelessOperator =
			mock(StreamOperator.class);

	final OperatorID operatorID = new OperatorID();
	when(statelessOperator.getOperatorID()).thenReturn(operatorID);

	// mock the returned empty snapshot result (all state handles are null)
	OperatorSnapshotResult statelessOperatorSnapshotResult = new OperatorSnapshotResult();
	when(statelessOperator.snapshotState(anyLong(), anyLong(), any(CheckpointOptions.class)))
			.thenReturn(statelessOperatorSnapshotResult);

	// set up the task
	StreamOperator<?>[] streamOperators = {statelessOperator};
	OperatorChain<Void, AbstractStreamOperator<Void>> operatorChain = mock(OperatorChain.class);
	when(operatorChain.getAllOperators()).thenReturn(streamOperators);

	Whitebox.setInternalState(streamTask, "isRunning", true);
	Whitebox.setInternalState(streamTask, "lock", new Object());
	Whitebox.setInternalState(streamTask, "operatorChain", operatorChain);
	Whitebox.setInternalState(streamTask, "cancelables", new CloseableRegistry());
	Whitebox.setInternalState(streamTask, "configuration", new StreamConfig(new Configuration()));
	Whitebox.setInternalState(streamTask, "asyncOperationsThreadPool", Executors.newCachedThreadPool());

	streamTask.triggerCheckpoint(checkpointMetaData, CheckpointOptions.forCheckpoint());
	checkpointCompletedLatch.await(30, TimeUnit.SECONDS);
	streamTask.cancel();

	// ensure that 'null' was acknowledged as subtask state
	Assert.assertNull(checkpointResult.get(0));
}
 

import org.apache.flink.streaming.api.operators.AbstractStreamOperator; //导入依赖的package包/类
public abstract void evaluateWindow(Collector<Result> out, TimeWindow window, AbstractStreamOperator<Result> operator) throws Exception;