[FLINK-25072][streaming] Introduce description on Transformation and StreamGraph

This closes apache#17924.

Author: 龙三

flink-core/src/main/java/org/apache/flink/api/dag/Transformation.java

@@ -119,6 +119,8 @@ public static int getNewNodeId() {
 
     protected String name;
 
+    protected String description;
+
     protected TypeInformation<T> outputType;
     // This is used to handle MissingTypeInfo. As long as the outputType has not been queried
     // it can still be changed using setOutputType(). Afterwards an exception is thrown when
@@ -203,6 +205,16 @@ public String getName() {
         return name;
     }
 
+    /** Changes the description of this {@code Transformation}. */
+    public void setDescription(String description) {
+        this.description = Preconditions.checkNotNull(description);
+    }
+
+    /** Returns the description of this {@code Transformation}. */
+    public String getDescription() {
+        return description;
+    }
+
     /** Returns the parallelism of this {@code Transformation}. */
     public int getParallelism() {
         return parallelism;

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/DataStreamSink.java

@@ -146,6 +146,24 @@ public DataStreamSink<T> setParallelism(int parallelism) {
         return this;
     }
 
+    /**
+     * Sets the description for this sink.
+     *
+     * <p>Description is used in json plan and web ui, but not in logging and metrics where only
+     * name is available. Description is expected to provide detailed information about the sink,
+     * while name is expected to be more simple, providing summary information only, so that we can
+     * have more user-friendly logging messages and metric tags without losing useful messages for
+     * debugging.
+     *
+     * @param description The description for this sink.
+     * @return The sink with new description.
+     */
+    @PublicEvolving
+    public DataStreamSink<T> setDescription(String description) {
+        transformation.setDescription(description);
+        return this;
+    }
+
     //	---------------------------------------------------------------------------
     //	 Fine-grained resource profiles are an incomplete work-in-progress feature
     //	 The setters are hence private at this point.

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/SingleOutputStreamOperator.java

@@ -419,4 +419,22 @@ public <X> DataStream<X> getSideOutput(OutputTag<X> sideOutputTag) {
                 new SideOutputTransformation<>(this.getTransformation(), sideOutputTag);
         return new DataStream<>(this.getExecutionEnvironment(), sideOutputTransformation);
     }
+
+    /**
+     * Sets the description for this operation.
+     *
+     * <p>Description is used in json plan and web ui, but not in logging and metrics where only
+     * name is available. Description is expected to provide detailed information about the sink,
+     * while name is expected to be more simple, providing summary information only, so that we can
+     * have more user-friendly logging messages and metric tags without losing useful messages for
+     * debugging.
+     *
+     * @param description The description for this operation.
+     * @return The operation with new description.
+     */
+    @PublicEvolving
+    public SingleOutputStreamOperator<T> setDescription(String description) {
+        transformation.setDescription(description);
+        return this;
+    }
 }

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/JSONGenerator.java

@@ -182,7 +182,7 @@ private void decorateNode(Integer vertexID, ObjectNode node) {
             node.put(PACT, "Operator");
         }
 
-        node.put(CONTENTS, vertex.getOperatorName());
+        node.put(CONTENTS, vertex.getOperatorDescription());
 
         node.put(PARALLELISM, streamGraph.getStreamNode(vertexID).getParallelism());
     }

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/SimpleTransformationTranslator.java

@@ -124,6 +124,9 @@ private void configure(final T transformation, final Context context) {
             streamNode.setManagedMemoryUseCaseWeights(
                     transformation.getManagedMemoryOperatorScopeUseCaseWeights(),
                     transformation.getManagedMemorySlotScopeUseCases());
+            if (null != transformation.getDescription()) {
+                streamNode.setOperatorDescription(transformation.getDescription());
+            }
         }
     }
 }

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamNode.java

@@ -65,6 +65,7 @@ public class StreamNode {
     private final Set<ManagedMemoryUseCase> managedMemorySlotScopeUseCases = new HashSet<>();
     private long bufferTimeout;
     private final String operatorName;
+    private String operatorDescription;
     private @Nullable String slotSharingGroup;
     private @Nullable String coLocationGroup;
     private KeySelector<?, ?>[] statePartitioners = new KeySelector[0];
@@ -113,6 +114,7 @@ public StreamNode(
             Class<? extends TaskInvokable> jobVertexClass) {
         this.id = id;
         this.operatorName = operatorName;
+        this.operatorDescription = operatorName;
         this.operatorFactory = operatorFactory;
         this.jobVertexClass = jobVertexClass;
         this.slotSharingGroup = slotSharingGroup;
@@ -242,6 +244,14 @@ public String getOperatorName() {
         return operatorName;
     }
 
+    public String getOperatorDescription() {
+        return operatorDescription;
+    }
+
+    public void setOperatorDescription(String operatorDescription) {
+        this.operatorDescription = operatorDescription;
+    }
+
     public void setSerializersIn(TypeSerializer<?>... typeSerializersIn) {
         checkArgument(typeSerializersIn.length > 0);
         this.typeSerializersIn = typeSerializersIn;

flink-streaming-java/src/test/java/org/apache/flink/streaming/api/DataStreamTest.java

@@ -1104,6 +1104,84 @@ public void processElement(
                 });
     }
 
+    /**
+     * Tests {@link SingleOutputStreamOperator#setDescription(String)} functionality.
+     *
+     * @throws Exception
+     */
+    @Test
+    public void testUserDefinedDescription() {
+        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+        DataStream<Long> dataStream1 =
+                env.generateSequence(0, 0)
+                        .name("testSource1")
+                        .setDescription("this is test source 1")
+                        .map(
+                                new MapFunction<Long, Long>() {
+                                    @Override
+                                    public Long map(Long value) throws Exception {
+                                        return null;
+                                    }
+                                })
+                        .name("testMap")
+                        .setDescription("this is test map 1");
+
+        DataStream<Long> dataStream2 =
+                env.generateSequence(0, 0)
+                        .name("testSource2")
+                        .setDescription("this is test source 2")
+                        .map(
+                                new MapFunction<Long, Long>() {
+                                    @Override
+                                    public Long map(Long value) throws Exception {
+                                        return null;
+                                    }
+                                })
+                        .name("testMap")
+                        .setDescription("this is test map 2");
+
+        dataStream1
+                .connect(dataStream2)
+                .flatMap(
+                        new CoFlatMapFunction<Long, Long, Long>() {
+
+                            @Override
+                            public void flatMap1(Long value, Collector<Long> out)
+                                    throws Exception {}
+
+                            @Override
+                            public void flatMap2(Long value, Collector<Long> out)
+                                    throws Exception {}
+                        })
+                .name("testCoFlatMap")
+                .setDescription("this is test co flat map")
+                .windowAll(GlobalWindows.create())
+                .trigger(PurgingTrigger.of(CountTrigger.of(10)))
+                .reduce(
+                        new ReduceFunction<Long>() {
+                            private static final long serialVersionUID = 1L;
+
+                            @Override
+                            public Long reduce(Long value1, Long value2) throws Exception {
+                                return null;
+                            }
+                        })
+                .name("testWindowReduce")
+                .setDescription("this is test window reduce")
+                .print();
+
+        // test functionality through the operator names in the execution plan
+        String plan = env.getExecutionPlan();
+
+        assertTrue(plan.contains("this is test source 1"));
+        assertTrue(plan.contains("this is test source 2"));
+        assertTrue(plan.contains("this is test map 1"));
+        assertTrue(plan.contains("this is test map 2"));
+        assertTrue(plan.contains("this is test co flat map"));
+        assertTrue(plan.contains("this is test window reduce"));
+    }
+
     private abstract static class CustomWmEmitter<T>
             implements AssignerWithPunctuatedWatermarks<T> {
 

flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/DataStream.scala

@@ -1225,4 +1225,22 @@ class DataStream[T](stream: JavaStream[T]) {
       operator: OneInputStreamOperator[T, R]): DataStream[R] = {
     asScalaStream(stream.transform(operatorName, implicitly[TypeInformation[R]], operator))
   }
+
+  /**
+   * Sets the description of this data stream.
+   *
+   * <p>Description is used in json plan and web ui, but not in logging and metrics where only
+   * name is available. Description is expected to provide detailed information about
+   * this operation, while name is expected to be more simple, providing summary information only,
+   * so that we can have more user-friendly logging messages and metric tags
+   * without losing useful messages for debugging.
+   *
+   * @return The operator with new description
+   */
+  @PublicEvolving
+  def setDescription(description: String) : DataStream[T] = stream match {
+    case stream : SingleOutputStreamOperator[T] => asScalaStream(stream.setDescription(description))
+    case _ => throw new UnsupportedOperationException("Only supported for operators.")
+      this
+  }
 }

flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/DataStreamTest.scala

@@ -97,6 +97,38 @@ class DataStreamTest extends AbstractTestBase {
     assert(plan contains "testWindowReduce")
   }
 
+  @Test
+  def testUserDefinedDescription(): Unit = {
+    val env = StreamExecutionEnvironment.getExecutionEnvironment
+    val dataStream1 = env.generateSequence(0, 0)
+      .setDescription("this is test source 1")
+      .map(x => x)
+      .setDescription("this is test map 1")
+    val dataStream2 = env.generateSequence(0, 0)
+      .setDescription("this is test source 2")
+      .map(x => x)
+      .setDescription("this is test map 2")
+
+    val func: (((Long, Long), (Long, Long)) => (Long, Long)) =
+      (x: (Long, Long), y: (Long, Long)) => (0L, 0L)
+    dataStream1.connect(dataStream2)
+      .flatMap({ (in, out: Collector[(Long, Long)]) => }, { (in, out: Collector[(Long, Long)]) => })
+      .setDescription("this is test co flat map")
+      .windowAll(GlobalWindows.create)
+      .trigger(PurgingTrigger.of(CountTrigger.of[GlobalWindow](10)))
+      .reduce(func)
+      .setDescription("this is test window reduce")
+      .print
+    // test functionality through the operator names in the execution plan
+    val plan = env.getExecutionPlan
+    assertTrue(plan.contains("this is test source 1"))
+    assertTrue(plan.contains("this is test source 2"))
+    assertTrue(plan.contains("this is test map 1"))
+    assertTrue(plan.contains("this is test map 2"))
+    assertTrue(plan.contains("this is test co flat map"))
+    assertTrue(plan.contains("this is test window reduce"))
+  }
+
   /**
    * Tests that [[DataStream.keyBy]] and [[DataStream.partitionCustom]] result in
    * different and correct topologies. Does the some for the [[ConnectedStreams]].