public static ByteBuffer serializeSubscription(final Subscription subscription, short version) {
        version = checkSubscriptionVersion(version);

        ConsumerProtocolSubscription data = new ConsumerProtocolSubscription();

        List<String> topics = new ArrayList<>(subscription.topics());
        Collections.sort(topics);
        data.setTopics(topics);

        data.setUserData(subscription.userData() != null ? subscription.userData().duplicate() : null);

        List<TopicPartition> ownedPartitions = new ArrayList<>(subscription.ownedPartitions());
        ownedPartitions.sort(Comparator.comparing(TopicPartition::topic).thenComparing(TopicPartition::partition));
        ConsumerProtocolSubscription.TopicPartition partition = null;
        for (TopicPartition tp : ownedPartitions) {
            if (partition == null || !partition.topic().equals(tp.topic())) {
                partition = new ConsumerProtocolSubscription.TopicPartition().setTopic(tp.topic());
                data.ownedPartitions().add(partition);
            }
            partition.partitions().add(tp.partition());
        }
        subscription.rackId().ifPresent(data::setRackId);

        data.setGenerationId(subscription.generationId().orElse(-1));
        return MessageUtil.toVersionPrefixedByteBuffer(version, data);
    }
ConsumerProtocol contains the schemas for consumer subscriptions and assignments for use with
Kafka's generalized group management protocol.

The current implementation assumes that future versions will not break compatibility. When
it encounters a newer version, it parses it using the current format. This basically means
that new versions cannot remove or reorder any of the existing fields.
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerProtocol.java