KAFKA-17247: Revised share group record schemas

[AndrewJSchofield]

In KIP-932, the group coordinator does not persist assignments for share groups. While this sounds like a good idea in terms of minimising overhead for data which doesn't strictly need to be recoverable, it significantly adds to the complexity of working with the coordinator framework.

This PR revises the definitions of the share group record schemas following more closely the schemas used for consumer groups, and eliminating the need to maintain soft state alongside the group coordinator's timeline structure.

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[apoorvmittal10]

Thanks for the PR @AndrewJSchofield. LGTM.

[chia7712]

LGTM. the failed streams tests are fixed by #16857 and one small question is left.

BTW, there are many similar code. I guess that is the side effect of reducing the complexity for coordinator framework :(

[chia7712]

Pardon me, why ShareGroupStatePartitionMetadataValue (version 15) does not join this party? or it is a future work.

[AndrewJSchofield]

That's in a later PR (possibly AK 4.1). That record type will be used to keep track of which partitions are being tracked by the share coordinator. I'll update KIP-932 with these revised definitions soon.

[junrao]

@AndrewJSchofield : Thanks for the PR. Left a couple of comments.

[junrao]

This matches ConsumerGroupPartitionMetadataValue for tracking the last know topic/partition metadata. However, InitializedTopics and DeletingTopics were designed for keeping track of the pending state needed for the Persister. How is the functionality achieved now?

[AndrewJSchofield]

That's because the previous ShareGroupPartitionMetadata in the KIP has become the new ShareGroupStatePartitionMetadata. This is because the pattern is that every record in KIP-848 which has the name ConsumerGroupXXXX has an equivalent ShareGroupXXXX, and ConsumerGroupPartitionMetadata needed an equivalent called ShareGroupPartitionMetadata. Thus, I needed a new name for the previous ShareGroupPartitionMetadata, but the way it works has not changed. I will update the KIP shortly.

[chia7712]

the failed tests are traced by https://issues.apache.org/jira/browse/KAFKA-15071 https://issues.apache.org/jira/browse/KAFKA-17265 https://issues.apache.org/jira/browse/KAFKA-10725 https://issues.apache.org/jira/browse/KAFKA-16634 https://issues.apache.org/jira/browse/KAFKA-16024 https://issues.apache.org/jira/browse/KAFKA-17334

I run the following command to test the failed tests on my local.

./gradlew cleanTest :streams:test --tests StoreQueryIntegrationTest.shouldQuerySpecificStalePartitionStoresMultiStreamThreads :tools:test --tests LeaderElectionCommandTest.testPreferredReplicaElection --tests TopicCommandIntegrationTest.testDescribeAtMinIsrPartitions :metadata:test --tests QuorumControllerTest.testBootstrapZkMigrationRecord --tests QuorumControllerTest.testFenceMultipleBrokers :core:test --tests SaslMultiMechanismConsumerTest.testCoordinatorFailover --tests SaslPlainPlaintextConsumerTest.testCoordinatorFailover --tests ZkMigrationFailoverTest.testDriverSkipsEventsFromOlderEpoch --tests SslConsumerTest.testCoordinatorFailover :clients:test --tests MemoryRecordsBuilderTest.testBuffersDereferencedOnClose

all pass. will merge this PR