private void verifyRebalanceWithMetadataChange(Optional<String> rackId,
                                                   MockPartitionAssignor partitionAssignor,
                                                   MetadataResponse metadataResponse1,
                                                   MetadataResponse metadataResponse2,
                                                   boolean expectRebalance) {
        final String consumerId = "leader";
        final List<String> topics = Arrays.asList(topic1, topic2);
        final List<TopicPartition> partitions = metadataResponse1.topicMetadata().stream()
                .flatMap(t -> t.partitionMetadata().stream().map(p -> new TopicPartition(t.topic(), p.partition())))
                .collect(Collectors.toList());
        subscriptions.subscribe(toSet(topics), Optional.of(rebalanceListener));
        client.updateMetadata(metadataResponse1);
        coordinator.maybeUpdateSubscriptionMetadata();

        client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE));
        coordinator.ensureCoordinatorReady(time.timer(Long.MAX_VALUE));

        Map<String, List<String>> initialSubscription = singletonMap(consumerId, topics);
        partitionAssignor.prepare(singletonMap(consumerId, partitions));

        client.prepareResponse(joinGroupLeaderResponse(1, consumerId, initialSubscription, false, Errors.NONE, rackId));
        client.prepareResponse(syncGroupResponse(partitions, Errors.NONE));
        coordinator.poll(time.timer(Long.MAX_VALUE));

        // rejoin will only be set in the next poll call
        assertFalse(coordinator.rejoinNeededOrPending());
        assertEquals(toSet(topics), subscriptions.subscription());
        assertEquals(toSet(partitions), subscriptions.assignedPartitions());
        assertEquals(0, rebalanceListener.revokedCount);
        assertNull(rebalanceListener.revoked);
        assertEquals(1, rebalanceListener.assignedCount);

        // Change metadata to trigger rebalance.
        client.updateMetadata(metadataResponse2);
        coordinator.poll(time.timer(0));

        if (!expectRebalance) {
            assertEquals(0, client.requests().size());
            return;
        }
        assertEquals(1, client.requests().size());

        // Revert metadata to original value. Fail pending JoinGroup. Another
        // JoinGroup should be sent, which will be completed successfully.
        client.updateMetadata(metadataResponse1);
        client.respond(joinGroupFollowerResponse(1, consumerId, "leader", Errors.NOT_COORDINATOR));
        client.prepareResponse(groupCoordinatorResponse(node, Errors.NONE));
        assertFalse(client.hasInFlightRequests());
        coordinator.poll(time.timer(1));
        assertTrue(coordinator.rejoinNeededOrPending());

        client.respond(request -> {
            if (!(request instanceof JoinGroupRequest)) {
                return false;
            } else {
                JoinGroupRequest joinRequest = (JoinGroupRequest) request;
                return consumerId.equals(joinRequest.data().memberId());
            }
        }, joinGroupLeaderResponse(2, consumerId, initialSubscription, false, Errors.NONE, rackId));
        client.prepareResponse(syncGroupResponse(partitions, Errors.NONE));
        coordinator.poll(time.timer(Long.MAX_VALUE));

        assertFalse(coordinator.rejoinNeededOrPending());
        Collection<TopicPartition> revoked = getRevoked(partitions, partitions);
        assertEquals(revoked.isEmpty() ? 0 : 1, rebalanceListener.revokedCount);
        assertEquals(revoked.isEmpty() ? null : revoked, rebalanceListener.revoked);
        // No partitions have been lost since the rebalance failure was not fatal
        assertEquals(0, rebalanceListener.lostCount);
        assertNull(rebalanceListener.lost);

        Collection<TopicPartition> added = getAdded(partitions, partitions);
        assertEquals(2, rebalanceListener.assignedCount);
        assertEquals(added.isEmpty() ? Collections.emptySet() : toSet(partitions), rebalanceListener.assigned);
        assertEquals(toSet(partitions), subscriptions.assignedPartitions());
    }
Verifies that the consumer re-joins after a metadata change. If JoinGroup fails
and metadata reverts to its original value, the consumer should still retry JoinGroup.
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinatorTest.java