Repository: apache/kafka
File: clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java (Java)

Code before:
            final ConsumerMetadata metadata = createMetadata(subscription);
            final MockClient client = new MockClient(time, metadata);

            initMetadata(client, Map.of(topic, 1));

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

            // poll once to update with the current metadata
            consumer.poll(Duration.ofMillis(0));
            TestUtils.waitForCondition(
                () -> requestGenerated(client, ApiKeys.FIND_COORDINATOR),
                "No FIND_COORDINATOR request sent within allotted timeout"
            );
            client.respond(FindCoordinatorResponse.prepareResponse(Errors.NONE, groupId, metadata.fetch().nodes().get(0)));

            // Validate the state of the endOffsetRequested flag. It should be unset before the call to currentLag(),
            // then set immediately afterward.
            assertFalse(subscription.partitionEndOffsetRequested(tp0));
            assertEquals(OptionalLong.empty(), consumer.currentLag(tp0));
            assertTrue(subscription.partitionEndOffsetRequested(tp0));
            assertLogEmitted(
                appender,
                "^Requesting the log end offset for " + tp0 + " in order to compute lag$"
            );

            if (groupProtocol == GroupProtocol.CLASSIC) {
                // Classic consumer does not send the LIST_OFFSETS right away (requires an explicit poll),
                // different from the new async consumer, that will send the LIST_OFFSETS request in the background
                // thread on the next background thread poll.
                consumer.poll(Duration.ofMillis(0));
            }

            TestUtils.waitForCondition(
                () -> requestGenerated(client, ApiKeys.LIST_OFFSETS),
                "No LIST_OFFSETS request sent within allotted timeout"
            );

            // Validate the state of the endOffsetRequested flag. It should still be set before the call to
            // currentLag(), because the previous LIST_OFFSETS call has not received a response.
            assertTrue(subscription.partitionEndOffsetRequested(tp0));
            assertEquals(OptionalLong.empty(), consumer.currentLag(tp0));
            assertTrue(subscription.partitionEndOffsetRequested(tp0));
            assertLogEmitted(
                appender,
                "^Not requesting the log end offset for " + tp0 + " to compute lag as an outstanding request already exists$"
            );

            // Now respond to the LIST_OFFSETS request with an error in the partition.
            ClientRequest listOffsetRequest = findRequest(client, ApiKeys.LIST_OFFSETS);

Review comment (suggestion):
this log is printed when there is a call to `subscriptionState.requestPartitionEndOffset`, should we better `verify(subscriptionState).requestPartitionEndOffset(tp)`? (we truly care about the func being called, and these log checks easily get flaky) 

Code after:
        );
    }

    // TODO: this test validate that the consumer clears the endOffsetRequested flag, but this is not yet implemented
    //       in the CONSUMER group protocol (see KAFKA-20187).
    //       Once it is implemented, this should use both group protocols.
    @ParameterizedTest
    @EnumSource(value = GroupProtocol.class, names = "CLASSIC")
    public void testCurrentLagClearsFlagOnFatalPartitionError(GroupProtocol groupProtocol) throws InterruptedException {
        final ConsumerMetadata metadata = createMetadata(subscription);
        final MockClient client = new MockClient(time, metadata);
        consumer = setUpConsumerForCurrentLag(groupProtocol, client, metadata);

        // Validate the state of the endOffsetRequested flag. It should be unset before the call to currentLag(),
        // then set immediately afterward.
        assertFalse(subscription.partitionEndOffsetRequested(tp0));
        assertEquals(OptionalLong.empty(), consumer.currentLag(tp0));
        assertTrue(subscription.partitionEndOffsetRequested(tp0));
        verify(subscription).requestPartitionEndOffset(tp0);

        if (groupProtocol == GroupProtocol.CLASSIC) {
            // Classic consumer does not send the LIST_OFFSETS right away (requires an explicit poll),
            // different from the new async consumer, that will send the LIST_OFFSETS request in the background
            // thread on the next background thread poll.
            consumer.poll(Duration.ofMillis(0));
        }

        TestUtils.waitForCondition(
            () -> requestGenerated(client, ApiKeys.LIST_OFFSETS),
            "No LIST_OFFSETS request sent within allotted timeout"
        );

        clearInvocations(subscription);

        // Validate the state of the endOffsetRequested flag. It should still be set before the call to
        // currentLag(), because the previous LIST_OFFSETS call has not received a response. In this case,
        // the SubscriptionState.requestPartitionEndOffset() method should *not* have been invoked.
        assertTrue(subscription.partitionEndOffsetRequested(tp0));
        assertEquals(OptionalLong.empty(), consumer.currentLag(tp0));
        assertTrue(subscription.partitionEndOffsetRequested(tp0));
        verify(subscription, never()).requestPartitionEndOffset(tp0);

        // Now respond to the LIST_OFFSETS request with an error in the partition.
        ClientRequest listOffsetRequest = findRequest(client, ApiKeys.LIST_OFFSETS);
        client.respondToRequest(listOffsetRequest, listOffsetsResponse(Map.of(), Map.of(tp0, Errors.TOPIC_AUTHORIZATION_FAILED)));

        if (groupProtocol == GroupProtocol.CLASSIC) {
            // Classic consumer does not send the LIST_OFFSETS right away (requires an explicit poll),
            // different from the new async consumer, that will send the LIST_OFFSETS request in the background
            // thread on the next background thread poll.
            assertThrows(TopicAuthorizationException.class, () -> consumer.poll(Duration.ofMillis(0)));