@Override
    protected MemberData memberData(Subscription subscription) {
        // In ConsumerProtocolSubscription v2 or higher, we can take member data from fields directly
        if (subscription.generationId().isPresent()) {
            return new MemberData(subscription.ownedPartitions(), subscription.generationId());
        }

        ByteBuffer buffer = subscription.userData();
        Optional<Integer> encodedGeneration;
        if (buffer == null) {
            encodedGeneration = Optional.empty();
        } else {
            try {
                Struct struct = COOPERATIVE_STICKY_ASSIGNOR_USER_DATA_V0.read(buffer);
                encodedGeneration = Optional.of(struct.getInt(GENERATION_KEY_NAME));
            } catch (Exception e) {
                encodedGeneration = Optional.of(DEFAULT_GENERATION);
            }
        }
        return new MemberData(subscription.ownedPartitions(), encodedGeneration, subscription.rackId());
    }
@Test
    public void testMemberDataWithEmptyPartitionsAndHigherGeneration() {
        List<String> topics = topics(topic);
        List<TopicPartition> ownedPartitions = partitions(tp(topic1, 0), tp(topic2, 1));

        // subscription containing empty owned partitions and a higher generation id, and non-empty owned partition in user data,
        // member data should honor the one in subscription since generation id is higher
        Subscription subscription = new Subscription(topics, generateUserData(topics, ownedPartitions, generationId - 1), Collections.emptyList(), generationId, Optional.empty());

        AbstractStickyAssignor.MemberData memberData = memberData(subscription);
        assertEquals(Collections.emptyList(), memberData.partitions, "subscription: " + subscription + " doesn't have expected owned partition");
        assertEquals(generationId, memberData.generation.orElse(-1), "subscription: " + subscription + " doesn't have expected generation id");
    }