Page History

...

Following table compares cpu / memory / disk usage of different strategies:

If using the second strategy, it still uses lot of disk size to store redundant data. For example, a group subscribes a same topic list. Every time a subscribed topic change, the group needs to store a new subscription topic metadata map with a single entry changed. If using the third strategy, it wastes too much CPU resource to recalculate the hash when the metadata image is expired.

...

ConsumerGroupMetadataValue

Add a new field "AllTopicHashMetadataHash". Set it as tagged field for backward compatibility.

{
  "type": "data",
  "name": "ConsumerGroupMetadataValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "Epoch", "versions": "0+", "type": "int32",
      "about": "The group epoch." },
    { "name": "AllTopicHashMetadataHash", "versions": "0+", "type": "int64",
      "default": 0, "taggedVersions": "0+", "tag": 0,
      "about": "The hash of all topics in the group." } <-- new field
  ]
}

ShareGroupMetadataValue

Add a new field "AllTopicHashMetadataHash".

{
  "type": "data",
  "name": "ShareGroupMetadataValue",
  "validVersions": "0",
  "flexibleVersions": "0+",
  "fields": [
    { "name": "Epoch", "type": "int32", "versions": "0+",
      "about": "The group epoch." },
    { "name": "AllTopicHashMetadataHash", "versions": "0+", "type": "int64",
      "about": "The hash of all topics in the group." } <-- new field
  ]
}

StreamsGroupMetadataValue

Add a new field "MetadataHash".

{
  "type": "data",
  "name": "StreamsGroupMetadataValue",
  "validVersions": "0",
  "defaultflexibleVersions": "0+",
  "fields": [
    { "name": "Epoch", "taggedVersionsversions": "0+", "type": "int32",
      "about": "The group epoch." },
    { "name": "MetadataHash", "tagversions": "0+", "type": "int64",
      "about": "The hash of all topics in the group." } <-- new field
  ]
}

...

The rack of a partition is from "broker.rack" configuration. This config can only be changed after the broker reboots. When the broker is stopped, the broker id is removed from PartitionRegistration#replicasPartitionRegistration#isrMetadataHash. This is reflected in TopicDelta#partitionChanges. The renew process is similar to above. A new MetadataDelta contains the topic in TopicsDelta#changedTopics and the coordinator remove the topic hash after it receives the change. In the next group heartbeat, the coordinator re-compute the value.

...

A topic hash represents topic id, name, number of partitions, and partition racks. To avoid useless rebalance, the hash function should return same value for same data, even if it runs on different JDKs or partition racks have different order. For different JDKs, the KIP will use Murmur3 to compute the hash value. The function is already in org.apache.kafka.streams.state.internals. We will move it to org.apache.kafka.common.hash for common usage. For partition racks with different order, we will compute hash for each value in it and sum as a result. We will also set the first byte as magic byte to represent hash version.

public static long topicHash(String topicName, int numPartitions, Map<Integer, List<Integer>> partitionIdToSortedRacks)

SubscribedTopicDescriberImpl

...