flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/DataStreamTest.java
/**
	 * Verify that a {@link DataStream#process(ProcessFunction)} call is correctly translated to
	 * an operator.
	 */
Verify that a  DataStream#process(ProcessFunction) call is correctly translated to
an operator.
Verify that a DataStream#process ( ProcessFunction ) call is correctly translated to an operator .
@Test
	public void testProcessTranslation() {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		DataStreamSource<Long> src = env.generateSequence(0, 0);
		ProcessFunction<Long, Integer> processFunction = new ProcessFunction<Long, Integer>() {
			private static final long serialVersionUID = 1L;
			@Override
			public void processElement(
					Long value,
					Context ctx,
					Collector<Integer> out) throws Exception {
			}
			@Override
			public void onTimer(
					long timestamp,
					OnTimerContext ctx,
					Collector<Integer> out) throws Exception {
			}
		};
		DataStream<Integer> processed = src
				.process(processFunction);
		processed.addSink(new DiscardingSink<Integer>());
		assertEquals(processFunction, getFunctionForDataStream(processed));
		assertTrue(getOperatorForDataStream(processed) instanceof ProcessOperator);
	}
@ Test public void testProcessTranslation ( ) { StreamExecutionEnvironment env = StreamExecutionEnvironment . getExecutionEnvironment ( ) ; DataStreamSource < Long > src = env . generateSequence ( 0 , 0 ) ; ProcessFunction < Long , Integer > processFunction = new ProcessFunction < Long , Integer > ( ) { private static final long serialVersionUID = 1L ; @ Override public void processElement ( Long value , Context ctx , Collector < Integer > out ) throws Exception { } @ Override public void onTimer ( long timestamp , OnTimerContext ctx , Collector < Integer > out ) throws Exception { } } ; DataStream < Integer > processed = src . process ( processFunction ) ; processed . addSink ( new DiscardingSink < Integer > ( ) ) ; assertEquals ( processFunction , getFunctionForDataStream ( processed ) ) ; assertTrue ( getOperatorForDataStream ( processed ) instanceof ProcessOperator ) ; }
Verify that a {@link DataStream#process(ProcessFunction)} call is correctly translated to
 an operator.
Verify that a { @link DataStream#process ( ProcessFunction ) } call is correctly translated to an operator .