bowenli86 commented on code in PR #28459:
URL: https://github.com/apache/flink/pull/28459#discussion_r3539681959
##########
flink-runtime/src/main/java/org/apache/flink/streaming/api/graph/AdaptiveGraphManager.java:
##########
@@ -473,11 +525,17 @@ private void createOperatorChainInfos(
final Map<Integer, OperatorChainInfo> chainEntryPoints =
buildAndGetChainEntryPoints(streamNodes,
jobVertexBuildContext);
+ chainEntryPoints.values().stream()
+ .filter(
+ chainInfo ->
+ streamGraph
+
.getStreamNode(chainInfo.getStartNodeId())
+ .getTransformationUID()
+ != null)
+ .forEach(chainInfo ->
generateHashesByStreamNodeId(chainInfo.getStartNodeId()));
Review Comment:
Good catch, thanks. Addressed in 6ae21e0c5eb: I removed the eager
generateHashesByStreamNodeId call. The UID-backed sort key is now derived
directly from StreamGraphHasherV2.generateUserSpecifiedHash(transformationUid),
so sorting no longer mutates the shared hashes map or shifts non-UID hash
positions. I also added testMixedUidAndNonUidSourceIdsMatchStreamingGenerator
to verify adaptive and regular generation keep the non-UID source JobVertexID
aligned.
##########
flink-runtime/src/main/java/org/apache/flink/streaming/api/graph/AdaptiveGraphManager.java:
##########
@@ -447,6 +450,55 @@ private void
connectToFinishedUpStreamVertex(JobVertexBuildContext jobVertexBuil
}
}
+ private List<StreamEdge> getTransitiveInEdgesInOrder(
+ List<StreamEdge> transitiveInEdges, JobVertexBuildContext
jobVertexBuildContext) {
+ final List<StreamEdge> transitiveInEdgesInOrder =
+ transitiveInEdges.stream()
+ .sorted(
+ Comparator.comparing(
+ inEdge ->
getStartNodeId(inEdge.getSourceId())))
+ .collect(Collectors.toList());
+ final List<StreamEdge> uidTransitiveInEdges =
+ transitiveInEdgesInOrder.stream()
+ .filter(this::hasUidBackedUpstream)
+ .sorted(
+ Comparator.comparing(
+ inEdge ->
+ getStartNodeJobVertexId(
+ inEdge,
jobVertexBuildContext)))
+ .collect(Collectors.toList());
+
+ if (uidTransitiveInEdges.size() < 2) {
+ return transitiveInEdgesInOrder;
Review Comment:
Agreed, thanks. Addressed in 6ae21e0c5eb: the reconnect path now starts from
a copy of the original transitiveInEdges list and only reorders UID-backed
entries in their existing slots. This preserves declaration and insertion order
when UID reordering does not apply, including non-UID and uidHash-only inputs.
I added testNonUidReconnectKeepsUnionDeclarationOrder and
testUidHashReconnectKeepsUnionDeclarationOrder for the source-a created first
but sourceB.union(sourceA) case.
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
