lucasbru commented on code in PR #17695:
URL: https://github.com/apache/kafka/pull/17695#discussion_r1831136347
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java:
##########
@@ -627,6 +626,99 @@ private List<TaskIds> convertTaskIdCollection(final
Set<StreamsAssignmentInterfa
.collect(Collectors.toList());
}
+ private List<StreamsGroupHeartbeatRequestData.Subtopology>
getTopologyFromStreams() {
+ final Map<String, StreamsAssignmentInterface.Subtopology>
subTopologyMap = streamsInterface.subtopologyMap();
+ final List<StreamsGroupHeartbeatRequestData.Subtopology>
subtopologies = new ArrayList<>(subTopologyMap.size());
+ for (final Map.Entry<String,
StreamsAssignmentInterface.Subtopology> subtopology :
subTopologyMap.entrySet()) {
+
subtopologies.add(getSubtopologyFromStreams(subtopology.getKey(),
subtopology.getValue()));
Review Comment:
How are we going to compute the topology ID on the client? I think the idea
is still to hash the topology metadata being sent here. So it needs to be
deterministic. So I'm preparing for the topology ID implementation. You are
right that we also need to sort the subtopologies.
--
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]
