### Question:
AbstractStreamOperator implements StreamOperator<OUT>, Serializable { @Override public void open() throws Exception {}  @Override void setup(StreamTask<?, ?> containingTask, StreamConfig config, Output<StreamRecord<OUT>> output); @Override MetricGroup getMetricGroup(); @Override final void initializeState(); @Override void open(); @Override void close(); @Override void dispose(); @Override void prepareSnapshotPreBarrier(long checkpointId); @Override final OperatorSnapshotFutures snapshotState(long checkpointId, long timestamp, CheckpointOptions checkpointOptions,
			CheckpointStreamFactory factory); void snapshotState(StateSnapshotContext context); void initializeState(StateInitializationContext context); @Override void notifyCheckpointComplete(long checkpointId); ExecutionConfig getExecutionConfig(); StreamConfig getOperatorConfig(); StreamTask<?, ?> getContainingTask(); ClassLoader getUserCodeClassloader(); StreamingRuntimeContext getRuntimeContext(); @SuppressWarnings("unchecked") KeyedStateBackend<K> getKeyedStateBackend(); OperatorStateBackend getOperatorStateBackend(); @Override @SuppressWarnings({"unchecked", "rawtypes"}) void setKeyContextElement1(StreamRecord record); @Override @SuppressWarnings({"unchecked", "rawtypes"}) void setKeyContextElement2(StreamRecord record); @SuppressWarnings({"unchecked", "rawtypes"}) void setCurrentKey(Object key); @SuppressWarnings({"unchecked", "rawtypes"}) Object getCurrentKey(); KeyedStateStore getKeyedStateStore(); @Override final void setChainingStrategy(ChainingStrategy strategy); @Override final ChainingStrategy getChainingStrategy(); void processLatencyMarker(LatencyMarker latencyMarker); void processLatencyMarker1(LatencyMarker latencyMarker); void processLatencyMarker2(LatencyMarker latencyMarker); InternalTimerService<N> getInternalTimerService(
			String name,
			TypeSerializer<N> namespaceSerializer,
			Triggerable<K, N> triggerable); void processWatermark(Watermark mark); void processWatermark1(Watermark mark); void processWatermark2(Watermark mark); @Override OperatorID getOperatorID(); @VisibleForTesting int numProcessingTimeTimers(); @VisibleForTesting int numEventTimeTimers();  }### Answer:
@Test public void testStateDoesNotInterfere() 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.processElement(new Tuple2<>(0, "SET_STATE:HELLO"), 0); testHarness.processElement(new Tuple2<>(1, "SET_STATE:CIAO"), 0); testHarness.processElement(new Tuple2<>(1, "EMIT_STATE"), 0); testHarness.processElement(new Tuple2<>(0, "EMIT_STATE"), 0); assertThat( extractResult(testHarness), contains("ON_ELEMENT:1:CIAO", "ON_ELEMENT:0:HELLO")); }
@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")); }