private void addMemberSubscriptions() {
        // In the rebalance case, we will add the last member as a trigger.
        // This is done to keep the final members count consistent with the input.
        int numberOfMembers = assignmentType.equals(AssignmentType.INCREMENTAL) ? memberCount - 1 : memberCount;

        // When subscriptions are homogeneous, all members are assigned all topics.
        if (subscriptionType == HOMOGENEOUS) {
            for (int i = 0; i < numberOfMembers; i++) {
                String memberName = "member" + i;
                subscriptions.put(memberName, subscription(allTopicNames, i));
            }
        } else {
            // Adjust bucket count based on member count when member count < max bucket count.
            int bucketCount = Math.min(MAX_BUCKET_COUNT, numberOfMembers);

            // Check minimum topics requirement
            if (topicCount < bucketCount) {
                throw new IllegalArgumentException("At least " + bucketCount + " topics are recommended for effective bucketing.");
            }

            int bucketSizeTopics = (int) Math.ceil((double) topicCount / bucketCount);
            int bucketSizeMembers = (int) Math.ceil((double) numberOfMembers / bucketCount);

            // Define buckets for each member and assign topics from the same bucket
            for (int bucket = 0; bucket < bucketCount; bucket++) {
                int memberStartIndex = bucket * bucketSizeMembers;
                int memberEndIndex = Math.min((bucket + 1) * bucketSizeMembers, numberOfMembers);

                int topicStartIndex = bucket * bucketSizeTopics;
                int topicEndIndex = Math.min((bucket + 1) * bucketSizeTopics, topicCount);

                List<String> bucketTopics = allTopicNames.subList(topicStartIndex, topicEndIndex);

                // Assign topics to each member in the current bucket
                for (int i = memberStartIndex; i < memberEndIndex; i++) {
                    String memberName = "member" + i;
                    subscriptions.put(memberName, subscription(bucketTopics, i));
                }
            }
        }

        groupSubscription = new ConsumerPartitionAssignor.GroupSubscription(subscriptions);
    }
The assignment type is decided based on whether all the members are assigned partitions
for the first time (full), or incrementally when a rebalance is triggered.
jmh-benchmarks/src/main/java/org/apache/kafka/jmh/assignor/ClientSideAssignorBenchmark.java