KAFKA-17747: [5/N] Replace subscription metadata with metadata hash in stream group (wip)

[FrankYang0529]

• Use metadata hash to replace subscription metadata.
• Remove StreamsGroupPartitionMetadataKey and
  StreamsGroupPartitionMetadataValue.
• Check whether configuredTopology is empty. If it's, call
  InternalTopicManager.configureTopics and set the result to the group.

[FrankYang0529]

To implement KIP-1101 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

[lucasbru]

ConfiguredTopology can be derived completely from the records for PartitionMetadata and Topology. So we just need to cache it, and can recreate it upon the first heartbeat after failover. We could consider persisting it, but that would just mean storing somewhat duplicate data. state is also derived from the other records.

[FrankYang0529]

Thanks for the reply. I will keep working on this PR and change streamsGroupHeartbeat to rebuild these two fields if they are empty.

[lucasbru]

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.

[FrankYang0529]

We may not be able to set metadata hash in ConfiguredTopology. When replaying StreamsGroupMetadataValue, we need to set metadataHash back. If we add it to ConfiguredTopology, that means there will have a non-empty ConfiguredTopology in the group and the metadataHash may be the same as latest computed value. Then we don't have a value to compare whether need to compute a new ConfiguredTopology.

[lucasbru]

Ah, I did not mean as a replacement for the TimelineLong inside, StreamsGroup. We need an additional hash to be stored inside ConfiguredTopology, to remember for which state of the topics it was calculated.