@Test
public void testOperatorChainWithProcessingTime() throws Exception {

	JobVertex chainedVertex = createChainedVertex(false);
	final OneInputStreamTask<Integer, Integer> task = new OneInputStreamTask<>();
	final OneInputStreamTaskTestHarness<Integer, Integer> testHarness =
			new OneInputStreamTaskTestHarness<>(task, 1, 1, BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO);
	testHarness.setupOutputForSingletonOperatorChain();
	testHarness.taskConfig = chainedVertex.getConfiguration();
	final StreamConfig streamConfig = testHarness.getStreamConfig();
	final StreamConfig operatorChainStreamConfig = new StreamConfig(chainedVertex.getConfiguration());
	final AsyncWaitOperator<Integer, Integer> headOperator =
			operatorChainStreamConfig.getStreamOperator(AsyncWaitOperatorTest.class.getClassLoader());
	streamConfig.setStreamOperator(headOperator);
	testHarness.invoke();
	testHarness.waitForTaskRunning();
	long initialTimestamp = 0L;
	testHarness.processElement(new StreamRecord<>(5, initialTimestamp));
	testHarness.processElement(new StreamRecord<>(6, initialTimestamp + 1L));
	testHarness.processElement(new StreamRecord<>(7, initialTimestamp + 2L));
	testHarness.processElement(new StreamRecord<>(8, initialTimestamp + 3L));
	testHarness.processElement(new StreamRecord<>(9, initialTimestamp + 4L));
	testHarness.endInput();
	testHarness.waitForTaskCompletion();
	ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();
	expectedOutput.add(new StreamRecord<>(22, initialTimestamp));
	expectedOutput.add(new StreamRecord<>(26, initialTimestamp + 1L));
	expectedOutput.add(new StreamRecord<>(30, initialTimestamp + 2L));
	expectedOutput.add(new StreamRecord<>(34, initialTimestamp + 3L));
	expectedOutput.add(new StreamRecord<>(38, initialTimestamp + 4L));
	TestHarnessUtil.assertOutputEqualsSorted(
			"Test for chained operator with AsyncWaitOperator failed",
			expectedOutput,
			testHarness.getOutput(),
			new StreamRecordComparator());
}