KAFKA-20066: Implement KIP-1251: Assignment epochs for consumer groups [1/N]

[lucliu1108]

Summary

This PR moves assignedPartitions out of ModernConsumerMember
interface, add it as independent properties for ShareGroupMember and
ConsumerGroupMember.

Reason for the change

In an upcoming PR, the structure of
ConsumerGroupMember#assignedPartitions and
ConsumerGroupMember#partitionsPendingRevocation will be changed to
include epoch information as

Map<Uuid, Map<Integer, Integer>>

This differs from the ShareGroupMember#assignedPartitions structure,
which remains Map<Uuid, Set<Integer>>. Therefore, it is no longer
appropriate to have this as a shared field in the base class.

Reviewers: Sean Quah squah@confluent.io, Lucas Brutschy
lbrutschy@confluent.io

[squah-confluent]

Thanks for filing the PR!

[squah-confluent]

For symmetry, could we also qualify the hasAssignedPartitionsChanged call in consumerGroupHeartbeat?

[squah-confluent]

I think this declaration belongs in between the builder and constructor, like in ConsumerGroupMember.

[squah-confluent]

Thanks, lgtm!

[lucasbru]

LGTM, thanks!

[lucasbru]

nit: @lucliu1108 if you want to prefill the Reviewers header, use the committer-tools/reviewers.py tool to generate it

[lucliu1108]

Thanks for the review!
@squah-confluent @lucasbru I notice there's also change need in AssignorBenchmarkUtils#createGroupSpec method. In this PR since we are still keeping the same data structure for assignedPartitions for both SharedGroupMember and ConsumerGroupMember, instanceof conversion could work. However, later on when the ConsumerGroupMember#assignedPartition is modified to include epochs, the current fix won't work.

Considering that createGroupSpec is only used for empty member now, does it make sense to simply pass in an empty assignment in this case?

memberSpecs.put(memberId, new MemberSubscriptionAndAssignmentImpl(
                Optional.ofNullable(member.rackId()),
                Optional.ofNullable(member.instanceId()),
                new TopicIds(member.subscribedTopicNames(), topicResolver),
                new Assignment(Map.of())
            ));

[squah-confluent]

@lucliu1108 That's unfortunate. I would probably duplicate createGroupSpec and have one version for consumer group members and another for share group members.

[lucliu1108]

Thanks @squah-confluent
Right now i spitted them into 2 different methods and rename separately, since they have the same set of parameters and will be marked as name clash.
In the next PR, since the consumerGroupMember assignedPartitions is restructured, I'll rename them back.

[squah-confluent]

Thanks!