FrankYang0529 commented on code in PR #19802:
URL: https://github.com/apache/kafka/pull/19802#discussion_r2105717603
##########
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:
To implement
[KIP-1101](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1101%3A+Trigger+rebalance+on+rack+topology+changes#KIP1101:Triggerrebalanceonracktopologychanges-StreamsGroupPartitionMetadataKey/StreamsGroupPartitionMetadataValue/ShareGroupPartitionMetadataKey/ShareGroupPartitionMetadataValue)
in stream group, we have to remove `StreamsGroupPartitionMetadataKey` and
`StreamsGroupPartitionMetadataValue`. This can save disk storage and the group
coordinator still can trigger a rebalance when metadata hash is changed.
In stream group, it tries to calculate configured topology and update group
state when `partitionMetadata` or `topology` data is updated. I wonder why it
doesn't save `configuredTopology` and `state` to records, so it can recover
when replaying records?
cc @lucasbru @bbejeck @chia7712
--
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
