chickenchickenlove commented on code in PR #23855:
URL: https://github.com/apache/kafka/pull/23855#discussion_r4411082799
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsTopology.java:
##########
@@ -128,6 +131,178 @@ public static StreamsTopology
fromHeartbeatRequest(StreamsGroupHeartbeatRequestD
return new StreamsTopology(topology.epoch(), subtopologyMap);
}
+ /**
+ * Checks whether the request describes this topology without converting
or modifying it.
+ * Subtopologies are matched by ID. Copartition groups and their topic
indexes are compared
+ * without regard to order, but duplicate occurrences must still match.
All other lists
+ * retain their order-sensitive comparison, including the topic lists
referenced by indexes.
+ *
+ * @param topology The topology supplied in the heartbeat request.
+ * @return Whether the request matches this topology.
+ */
+ public boolean matches(StreamsGroupHeartbeatRequestData.Topology topology)
{
+ if (topologyEpoch != topology.epoch() || subtopologies.size() !=
topology.subtopologies().size()) {
+ return false;
+ }
+ for (int i = 0; i < topology.subtopologies().size(); i++) {
+ StreamsGroupHeartbeatRequestData.Subtopology requested =
topology.subtopologies().get(i);
+ Subtopology current = subtopologies.get(requested.subtopologyId());
+ if (current == null || !matchesSubtopology(current, requested)) {
+ return false;
+ }
+
+ // Do not let repeated IDs match one stored subtopology while
omitting another.
+ for (int j = 0; j < i; j++) {
+ if
(requested.subtopologyId().equals(topology.subtopologies().get(j).subtopologyId()))
{
+ return false;
+ }
+ }
+ }
+ return true;
+ }
+
+ // Keep these comparisons in sync with the generated fields guarded by
StreamsTopologyTest.
+ private static boolean matchesSubtopology(Subtopology current,
StreamsGroupHeartbeatRequestData.Subtopology requested) {
+ if (!current.subtopologyId().equals(requested.subtopologyId())) {
+ return false;
+ }
+
+ return current.sourceTopics().equals(requested.sourceTopics())
+ && current.sourceTopicRegex().equals(requested.sourceTopicRegex())
+ &&
current.repartitionSinkTopics().equals(requested.repartitionSinkTopics())
+ && matchesTopicInfos(current.repartitionSourceTopics(),
requested.repartitionSourceTopics())
+ && matchesTopicInfos(current.stateChangelogTopics(),
requested.stateChangelogTopics())
+ && matchesCopartitionGroups(current.copartitionGroups(),
requested.copartitionGroups());
+ }
+
+ private static boolean matchesTopicInfos(List<TopicInfo> current,
List<StreamsGroupHeartbeatRequestData.TopicInfo> requested) {
+ if (current.size() != requested.size()) {
+ return false;
+ }
+ for (int i = 0; i < current.size(); i++) {
+ TopicInfo fromCurrent = current.get(i);
+ StreamsGroupHeartbeatRequestData.TopicInfo fromRequested =
requested.get(i);
+ if (!fromCurrent.name().equals(fromRequested.name())
+ || fromCurrent.partitions() != fromRequested.partitions()
+ || fromCurrent.replicationFactor() !=
fromRequested.replicationFactor()
+ || !matchesTopicConfigs(fromCurrent.topicConfigs(),
fromRequested.topicConfigs())) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static boolean matchesTopicConfigs(
+ List<StreamsGroupTopologyValue.TopicConfig> current,
+ List<StreamsGroupHeartbeatRequestData.KeyValue> requested
+ ) {
+ // Preserve the null handling of convertToTopicInfo for directly
constructed requests.
+ if (current == null || requested == null) {
+ return current == null && requested == null;
+ }
+ if (current.size() != requested.size()) {
+ return false;
+ }
+ for (int i = 0; i < current.size(); i++) {
+ if (!Objects.equals(current.get(i).key(), requested.get(i).key())
+ || !Objects.equals(current.get(i).value(),
requested.get(i).value())) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private static boolean matchesCopartitionGroups(
Review Comment:
I agree.
In that case, performance degradation is expected.
I've addressed it!
--
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]
