[GitHub] asfgit closed pull request #7085: [FLINK-10863][tests] Assign UIDs to all operators in DataStreamAllrou…

[GitBox <gi...@apache.org>]

asfgit closed pull request #7085: [FLINK-10863][tests] Assign UIDs to all operators in DataStreamAllrou…
URL: https://github.com/apache/flink/pull/7085
 
 
   

This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:

As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):

diff --git a/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/DataStreamAllroundTestProgram.java b/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/DataStreamAllroundTestProgram.java
index b14e2af1b52..3c406c7598d 100644
--- a/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/DataStreamAllroundTestProgram.java
+++ b/flink-end-to-end-tests/flink-datastream-allround-test/src/main/java/org/apache/flink/streaming/tests/DataStreamAllroundTestProgram.java
@@ -78,7 +78,7 @@ public static void main(String[] args) throws Exception {
 		setupEnvironment(env, pt);
 
 		// add a keyed stateful map operator, which uses Kryo for state serialization
-		DataStream<Event> eventStream = env.addSource(createEventSource(pt))
+		DataStream<Event> eventStream = env.addSource(createEventSource(pt)).uid("0001")
 			.assignTimestampsAndWatermarks(createTimestampExtractor(pt))
 			.keyBy(Event::getKey)
 			.map(createArtificialKeyedStateMapper(
@@ -97,7 +97,7 @@ public static void main(String[] args) throws Exception {
 						new StatefulComplexPayloadSerializer()), // custom stateful serializer
 					Collections.singletonList(ComplexPayload.class) // KryoSerializer via type extraction
 				)
-			).returns(Event.class).name(KEYED_STATE_OPER_NAME + "_Kryo_and_Custom_Stateful");
+			).returns(Event.class).name(KEYED_STATE_OPER_NAME + "_Kryo_and_Custom_Stateful").uid("0002");
 
 		// add a keyed stateful map operator, which uses Avro for state serialization
 		eventStream = eventStream
@@ -124,12 +124,12 @@ public static void main(String[] args) throws Exception {
 						new AvroSerializer<>(ComplexPayloadAvro.class)), // custom AvroSerializer
 					Collections.singletonList(ComplexPayloadAvro.class) // AvroSerializer via type extraction
 				)
-			).returns(Event.class).name(KEYED_STATE_OPER_NAME + "_Avro");
+			).returns(Event.class).name(KEYED_STATE_OPER_NAME + "_Avro").uid("0003");
 
 		DataStream<Event> eventStream2 = eventStream
 			.map(createArtificialOperatorStateMapper((MapFunction<Event, Event>) in -> in))
-			.name(OPERATOR_STATE_OPER_NAME)
-			.returns(Event.class);
+			.returns(Event.class)
+			.name(OPERATOR_STATE_OPER_NAME).uid("0004");
 
 		// apply a tumbling window that simply passes forward window elements;
 		// this allows the job to cover timers state
@@ -141,19 +141,20 @@ public void apply(Integer integer, TimeWindow window, Iterable<Event> input, Col
 						out.collect(e);
 					}
 				}
-			}).name(TIME_WINDOW_OPER_NAME);
+			}).name(TIME_WINDOW_OPER_NAME).uid("0005");
 
 		if (isSimulateFailures(pt)) {
 			eventStream3 = eventStream3
 				.map(createFailureMapper(pt))
 				.setParallelism(1)
-				.name(FAILURE_MAPPER_NAME);
+				.name(FAILURE_MAPPER_NAME).uid("0006");
 		}
 
 		eventStream3.keyBy(Event::getKey)
 			.flatMap(createSemanticsCheckMapper(pt))
 			.name(SEMANTICS_CHECK_MAPPER_NAME)
-			.addSink(new PrintSinkFunction<>());
+			.uid("0007")
+			.addSink(new PrintSinkFunction<>()).uid("0008");
 
 		env.execute("General purpose test job");
 	}


 

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services