@ParameterizedTest
    @EnumSource(value = GroupProtocol.class)
    public void testMissingOffsetNoResetPolicy(GroupProtocol groupProtocol) throws InterruptedException {
        SubscriptionState subscription = new SubscriptionState(new LogContext(), OffsetResetStrategy.NONE);
        ConsumerMetadata metadata = createMetadata(subscription);
        MockClient client = new MockClient(time, metadata);

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

        consumer = newConsumer(groupProtocol, time, client, subscription, metadata, assignor,
                true, groupId, groupInstanceId, false);
        consumer.assign(singletonList(tp0));

        client.prepareResponseFrom(FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId, node), node);
        Node coordinator = new Node(Integer.MAX_VALUE - node.id(), node.host(), node.port());

        // lookup committed offset and find nothing
        client.prepareResponseFrom(offsetResponse(Collections.singletonMap(tp0, -1L), Errors.NONE), coordinator);

        if (groupProtocol == GroupProtocol.CONSUMER) {
            // New consumer poll(ZERO) needs to wait for the offset fetch event added by a call to poll, to be processed
            // by the background thread, so it can realize there are no committed offsets and then
            // throw the NoOffsetForPartitionException
            assertPollEventuallyThrows(consumer, NoOffsetForPartitionException.class,
                    "Consumer was not able to update fetch positions on continuous calls with 0 timeout");
        } else {
            assertThrows(NoOffsetForPartitionException.class, () -> consumer.poll(Duration.ZERO));
        }
    }
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