private JoinGroupResponse joinGroupLeaderResponse(
        int generationId,
        String memberId,
        Map<String, List<String>> subscriptions,
        boolean skipAssignment,
        Errors error,
        Optional<String> rackId
    ) {
        List<JoinGroupResponseData.JoinGroupResponseMember> metadata = new ArrayList<>();
        for (Map.Entry<String, List<String>> subscriptionEntry : subscriptions.entrySet()) {
            ConsumerPartitionAssignor.Subscription subscription = new ConsumerPartitionAssignor.Subscription(subscriptionEntry.getValue(),
                    null, Collections.emptyList(), DEFAULT_GENERATION, rackId);
            ByteBuffer buf = ConsumerProtocol.serializeSubscription(subscription);
            metadata.add(new JoinGroupResponseData.JoinGroupResponseMember()
                    .setMemberId(subscriptionEntry.getKey())
                    .setMetadata(buf.array()));
        }

        return new JoinGroupResponse(
                new JoinGroupResponseData()
                        .setErrorCode(error.code())
                        .setGenerationId(generationId)
                        .setProtocolName(partitionAssignor.name())
                        .setLeader(memberId)
                        .setSkipAssignment(skipAssignment)
                        .setMemberId(memberId)
                        .setMembers(metadata),
                ApiKeys.JOIN_GROUP.latestVersion()
        );
    }
Verifies that subscription change updates SubscriptionState correctly even after JoinGroup failures
that don't re-invoke onJoinPrepare.
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java