rkhachatryan commented on a change in pull request #15013:
URL: https://github.com/apache/flink/pull/15013#discussion_r582769787
##########
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
##########
@@ -347,7 +348,13 @@ private void setChaining(Map<Integer, byte[]> hashes,
List<Map<Integer, byte[]>>
final Map<Integer, OperatorChainInfo> chainEntryPoints =
buildChainedInputsAndGetHeadInputs(hashes, legacyHashes);
final Collection<OperatorChainInfo> initialEntryPoints =
- new ArrayList<>(chainEntryPoints.values());
+ chainEntryPoints.values().stream()
+ .sorted(
+ Comparator.comparing(
+ operatorChainInfo ->
+
hashes.get(operatorChainInfo.getStartNodeId()),
+
StreamingJobGraphGenerator::compareHashes))
Review comment:
Then shouldn't hashes be compared first (before nodeId)?
Edit:
1. Then why do we need to compare nodeIds at all?
2. I'm not sure that node ids are re-generated upon restart unless the
topology has changed. I think that the generated graph is re-used.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
