Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/4649#discussion_r161791264
--- Diff:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraph.java
---
@@ -411,6 +413,13 @@ private void addEdgeInternal(Integer upStreamVertexID,
StreamNode upstreamNode =
getStreamNode(upStreamVertexID);
StreamNode downstreamNode =
getStreamNode(downStreamVertexID);
+ Tuple2<Integer, Integer> edgePair = new
Tuple2<>(upstreamNode.getId(), downstreamNode.getId());
+ if (!uniqueEdgeMap.containsKey(edgePair)) {
--- End diff --
```
uniqueEdgeMap.put(edgePair, uniqueEdgeMap.getOrDefault(edgePair, 0) + 1)
```
---
