public static Subscription deserializeSubscription(final ByteBuffer buffer, short version) {
        version = checkSubscriptionVersion(version);

        try {
            ConsumerProtocolSubscription data =
                new ConsumerProtocolSubscription(new ByteBufferAccessor(buffer), version);

            List<TopicPartition> ownedPartitions = new ArrayList<>();
            for (ConsumerProtocolSubscription.TopicPartition tp : data.ownedPartitions()) {
                for (Integer partition : tp.partitions()) {
                    ownedPartitions.add(new TopicPartition(tp.topic(), partition));
                }
            }

            return new Subscription(
                data.topics(),
                data.userData() != null ? data.userData().duplicate() : null,
                ownedPartitions,
                data.generationId(),
                data.rackId() == null || data.rackId().isEmpty() ? Optional.empty() : Optional.of(data.rackId()));
        } catch (BufferUnderflowException e) {
            throw new SchemaException("Buffer underflow while parsing consumer protocol's subscription", e);
        }
    }
@Test
    public void deserializeFutureSubscriptionVersion() {
        ByteBuffer buffer = generateFutureSubscriptionVersionData();

        Subscription subscription = ConsumerProtocol.deserializeSubscription(buffer);
        subscription.setGroupInstanceId(groupInstanceId);
        assertEquals(Collections.singleton("topic"), toSet(subscription.topics()));
        assertEquals(Collections.singleton(tp2), toSet(subscription.ownedPartitions()));
        assertEquals(groupInstanceId, subscription.groupInstanceId());
        assertEquals(generationId, subscription.generationId().orElse(DEFAULT_GENERATION));
        assertEquals(rackId, subscription.rackId());
    }