KAFKA-17747: [6/N] Replace subscription metadata with metadata hash in share group

[FrankYang0529]

• Use metadata hash to replace subscription metadata.
• Remove ShareGroupPartitionMetadataKey and
  ShareGroupPartitionMetadataValue.
• Use subscriptionTopicNames and metadataImage to replace
  subscriptionMetadata in subscribedTopicsChangeMap function.

Reviewers: Chia-Ping Tsai [email protected], David Jacot
[email protected], Andrew Schofield [email protected]

[dajac]

This one should basically be a copy of #19761. Let's merge that one first and then we can review this one.

[AndrewJSchofield]

Thanks for the PR. I have only just started reviewing and will have more comments.

Because you are removing ShareGroupPartitionMetadata, you should make some changes to CoordinatorRecordType.java also I think. Essentially, ordinal 9 in the enum is no longer used (and could be used in the future).

[FrankYang0529]

@AndrewJSchofield The CoordinatorRecordType.java is automatically generated. Do you mean that we can change ShareGroupMemberMetadata to 9, SHARE_GROUP_METADATA to 10, and so on?

[AndrewJSchofield]

@AndrewJSchofield The CoordinatorRecordType.java is automatically generated. Do you mean that we can change ShareGroupMemberMetadata to 9, SHARE_GROUP_METADATA to 10, and so on?

Ah, yes, I see. It's fine. One day, someone might notice that 9 is unused and then assign a new record type to it. The way that you've got it is fine. I just had to clone your branch and build it :)

[chia7712]

@FrankYang0529 thanks for this patch. two small comments are left. PTAL

[chia7712]

Should we calculate the metadataHash if it is not defined? That is analogous to subscriptionMetadata

[FrankYang0529]

Thanks for the suggestion. The subscriptionMetadata can be null, so we can determine whether a developer sets the value. However, the metadataHash is long and cannot be null. I would prefer to let the developer to set it manually.

[chia7712]

Could you consider using alreadyInitializedPartSet to filter out the partitions from partitionSet?

                    Set<Integer> partitionSet = IntStream.range(0, topicImage.partitions().size()).boxed()
                            .filter(p -> !alreadyInitializedPartSet.contains(p)).collect(Collectors.toSet());

It could be a small optimization, I guess :)

[AndrewJSchofield]

Thanks for the PR. Apart from one tiny nit, this looks like a faithful copy of the equivalent changes for consumer groups. I'll download the patch and try it on my local environment, but I expect we'll be good to merge for 4.1.

[AndrewJSchofield]

nit: This log includes the metadata hash in consumer groups. Seems sensible here too.

[FrankYang0529]

Updated it. Thanks.

[FrankYang0529]

@AndrewJSchofield, There is no assignment function, so we can not test it like consumer group in #19856. A workable way may need to use rebalance-total metric to check whether rebalance happen. Do you have any suggestion for this? I can add it in the next PR.

[dajac]

@AndrewJSchofield, There is no assignment function, so we can not test it like consumer group in #19856. A workable way may need to use rebalance-total metric to check whether rebalance happen. Do you have any suggestion for this? I can add it in the next PR.

@FrankYang0529 Could we use the share group describe method from the admin client?

[AndrewJSchofield]

@FrankYang0529 I think you are planning a follow-on PR with additional testing when rebalance occurs. Apart from that future work, this looks ready to merge.

[FrankYang0529]

@AndrewJSchofield I tried to add an integration test. However, the GroupMetadataManager doesn't support to use other assignors for share groups. Currently, there is no usage of withShareGroupAssignor in production code, so it can only use SimpleAssignor.

The consumer group uses group.consumer.assignors to define assignors. Do we want to add a new config group.share.assignors for share group? I can add this in the next PR. Thanks.

kafka/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java

Lines 351 to 354 in b1ea280

	Builder withShareGroupAssignor(ShareGroupPartitionAssignor shareGroupAssignor) {
	this.shareGroupAssignor = shareGroupAssignor;
	return this;
	}

kafka/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java

Lines 374 to 375 in b1ea280

	if (shareGroupAssignor == null)
	shareGroupAssignor = new SimpleAssignor();

[AndrewJSchofield]

@AndrewJSchofield I tried to add an integration test. However, the GroupMetadataManager doesn't support to use other assignors for share groups. Currently, there is no usage of withShareGroupAssignor in production code, so it can only use SimpleAssignor.

The consumer group uses group.consumer.assignors to define assignors. Do we want to add a new config group.share.assignors for share group? I can add this in the next PR. Thanks.

kafka/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java

Lines 351 to 354 in b1ea280

	Builder withShareGroupAssignor(ShareGroupPartitionAssignor shareGroupAssignor) {
	this.shareGroupAssignor = shareGroupAssignor;
	return this;
	}

kafka/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java

Lines 374 to 375 in b1ea280

	if (shareGroupAssignor == null)
	shareGroupAssignor = new SimpleAssignor();

Ah, yes, good catch. We missed this config. I have created a specific issue https://issues.apache.org/jira/browse/KAFKA-19369. Given that KIP-932 does not permit a list of assignors, the only thing a user could do is implement their own assignor and replace the simple one. We've essentially hard-coded the SimpleAssignor with no way to override it. When we add more assignors in the future, the plan was to enhance the support for this config to support a list, just as for consumer groups.