@Test
public void testProcessingTimeTimersDontInterfere() throws Exception {

	TestOperator testOperator = new TestOperator();
	KeyedOneInputStreamOperatorTestHarness<Integer, Tuple2<Integer, String>, String> testHarness =
			new KeyedOneInputStreamOperatorTestHarness<>(testOperator, new TestKeySelector(), BasicTypeInfo.INT_TYPE_INFO);
	testHarness.open();
	testHarness.setProcessingTime(0L);
	testHarness.processElement(new Tuple2<>(1, "SET_PROC_TIME_TIMER:20"), 0);
	testHarness.processElement(new Tuple2<>(0, "SET_STATE:HELLO"), 0);
	testHarness.processElement(new Tuple2<>(1, "SET_STATE:CIAO"), 0);
	testHarness.processElement(new Tuple2<>(0, "SET_PROC_TIME_TIMER:10"), 0);
	testHarness.setProcessingTime(10L);
	assertThat(
			extractResult(testHarness),
			contains("ON_PROC_TIME:HELLO"));
	testHarness.setProcessingTime(20L);
	assertThat(
			extractResult(testHarness),
			contains("ON_PROC_TIME:CIAO"));
}