You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by sr...@apache.org on 2018/11/13 15:45:04 UTC
[flink] branch master updated: [FLINK-10863][tests] Assign UIDs to
all operators in DataStreamAllroundTestProgram
This is an automated email from the ASF dual-hosted git repository.
srichter pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 31176d5 [FLINK-10863][tests] Assign UIDs to all operators in DataStreamAllroundTestProgram
31176d5 is described below
commit 31176d5f027f30e74e89aaf2a177d25b1692dc1b
Author: Stefan Richter <s....@data-artisans.com>
AuthorDate: Tue Nov 13 12:32:56 2018 +0100
[FLINK-10863][tests] Assign UIDs to all operators in DataStreamAllroundTestProgram
This closes #7085.
---
.../streaming/tests/DataStreamAllroundTestProgram.java | 17 +++++++++--------
1 file changed, 9 insertions(+), 8 deletions(-)
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 b14e2af..3c406c7 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 class DataStreamAllroundTestProgram {
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 class DataStreamAllroundTestProgram {
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 class DataStreamAllroundTestProgram {
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 class DataStreamAllroundTestProgram {
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");
}