@ParameterizedTest
    @EnumSource(value = GroupProtocol.class, names = "CLASSIC")
    public void verifyNoCoordinatorLookupForManualAssignmentWithOffsetCommit(GroupProtocol groupProtocol) {
        ConsumerMetadata metadata = createMetadata(subscription);
        MockClient client = new MockClient(time, metadata);

        initMetadata(client, Collections.singletonMap(topic, 1));
        Node node = metadata.fetch().nodes().get(0);

        // create a consumer with groupID with manual assignment
        consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor, true, groupInstanceId);
        consumer.assign(singleton(tp0));

        // 1st coordinator error should cause coordinator unknown
        client.prepareResponseFrom(FindCoordinatorResponse.prepareResponse(Errors.COORDINATOR_NOT_AVAILABLE, groupId, node), node);
        consumer.poll(Duration.ofMillis(0));

        // 2nd coordinator error should find the correct coordinator and clear the findCoordinatorFuture
        client.prepareResponseFrom(FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId, node), node);

        client.prepareResponse(offsetResponse(Collections.singletonMap(tp0, 50L), Errors.NONE));
        client.prepareResponse(fetchResponse(tp0, 50L, 5));

        @SuppressWarnings("unchecked")
        ConsumerRecords<String, String> records = (ConsumerRecords<String, String>) consumer.poll(Duration.ofMillis(0));
        assertEquals(5, records.count());
        assertEquals(55L, consumer.position(tp0));

        // after coordinator found, consumer should be able to commit the offset successfully
        client.prepareResponse(offsetCommitResponse(Collections.singletonMap(tp0, Errors.NONE)));
        consumer.commitSync(Collections.singletonMap(tp0, new OffsetAndMetadata(55L)));

        // verify the offset is committed
        client.prepareResponse(offsetResponse(Collections.singletonMap(tp0, 55L), Errors.NONE));
        assertEquals(55, consumer.committed(Collections.singleton(tp0), Duration.ZERO).get(tp0).offset());
    }
Note to future authors in this class. If you close the consumer, close with DURATION.ZERO to reduce the duration of
the test.
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java