KAFKA-19369: Add group.share.assignors config and integration test

[FrankYang0529]

• Add group.share.assignors config to GroupCoordinatorConfig.
• Send rackId in share group heartbeat request if it's not null.
• Add integration test testShareConsumerWithRackAwareAssignor.

Reviewers: Lan Ding 53332773+DL1231@users.noreply.github.com, Andrew
Schofield aschofield@confluent.io

[FrankYang0529]

I will remove wip and rebase trunk after #19856 is merged.

[FrankYang0529]

The flaky test PlaintextConsumerTest#testCloseLeavesGroupOnInterrupt is not related to this PR.

https://issues.apache.org/jira/browse/KAFKA-18031

[DL1231]

Thanks for the PR, LGTM, left a small comment.

[DL1231]

Suggested change

	ShareGroupPartitionAssignor assignor;
	if (Objects.equals(kclass, SHARE_GROUP_ASSIGNORS_DEFAULT)) {
	assignor = SHARE_GROUP_BUILTIN_ASSIGNOR;
	} else {
	try {
	assignor = Utils.newInstance(kclass, ShareGroupPartitionAssignor.class);
	} catch (ClassNotFoundException e) {
	throw new KafkaException("Class " + kclass + " cannot be found", e);
	} catch (ClassCastException e) {
	throw new KafkaException(kclass + " is not an instance of " + ShareGroupPartitionAssignor.class.getName());
	}
	}
	ShareGroupPartitionAssignor assignor = SHARE_GROUP_BUILTIN_ASSIGNOR;
	if (!Objects.equals(kclass, SHARE_GROUP_ASSIGNORS_DEFAULT)) {
	try {
	assignor = Utils.newInstance(kclass, ShareGroupPartitionAssignor.class);
	} catch (ClassNotFoundException e) {
	throw new KafkaException("Class " + kclass + " cannot be found", e);
	} catch (ClassCastException e) {
	throw new KafkaException(kclass + " is not an instance of " + ShareGroupPartitionAssignor.class.getName());
	}
	}

[FrankYang0529]

Thanks for the review. Updated it.

[AndrewJSchofield]

Thanks for the PR. Just one comment but worth getting it right to start with I think. If you agree, I'll adjust the KIP before I merge.

[AndrewJSchofield]

I think you've done a slightly different thing here than it says in the KIP. I think it's an improvement so I want to adjust the KIP too. The doc should say "The server-side assignors as a list of either names for built-in assignors or full class names for custom assignors. The list must contain only a single entry which is used by all groups. The supported built-in assignors are: " + SHARE_GROUP_BUILTIN_ASSIGNOR.name() + ".". I don't think there's any need to include the sentence about future plans. That's more helpful as commentary in the KIP. In this way, I think we have a futureproof behaviour and accurate documentation.

[FrankYang0529]

Agree. I will update the PR later. Thanks.