lucasbru commented on code in PR #19802:
URL: https://github.com/apache/kafka/pull/19802#discussion_r2106982543
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/StreamsGroup.java:
##########
@@ -582,54 +592,44 @@ public Set<String> currentWarmupTaskProcessIds(
}
/**
- * @return An immutable map of partition metadata for each topic that are
inputs for this streams group.
+ * @return The metadata hash.
*/
- public Map<String, TopicMetadata> partitionMetadata() {
- return Collections.unmodifiableMap(partitionMetadata);
+ public long metadataHash() {
+ return metadataHash.get();
}
/**
- * Updates the partition metadata. This replaces the previous one.
+ * Updates the metadata hash.
*
- * @param partitionMetadata The new partition metadata.
+ * @param metadataHash The new metadata hash.
*/
- public void setPartitionMetadata(
- Map<String, TopicMetadata> partitionMetadata
- ) {
- this.partitionMetadata.clear();
- this.partitionMetadata.putAll(partitionMetadata);
- maybeUpdateConfiguredTopology();
- maybeUpdateGroupState();
Review Comment:
Could you instead store the group metadata hash along with / inside
ConfiguredTopology, and rebuild configuredTopology not only when it is empty,
but also when the group metadata hash does not match the current group metadata
hash?
Also, that would mean, when we call `setTopology` or `setPartitionMetadata`
inside `StreamsGroup`, we will not call `maybeUpdateConfiguredTopology`, since
it will be only be created in one place - from the `streamsGroupHeartbeat`.
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
