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

Code before:

    @Test
    public void testPollWithFatalErrorShouldFailAllUnsentRequests() {
        CommitRequestManager commitRequestManager = create(true, 100);
        when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(mockedNode));

        commitRequestManager.fetchOffsets(Collections.singleton(new TopicPartition("test", 0)), 200);
        assertEquals(1, commitRequestManager.pendingRequests.unsentOffsetFetches.size());

        when(coordinatorRequestManager.coordinator()).thenReturn(Optional.empty());
        when(coordinatorRequestManager.fatalError())
                .thenReturn(Optional.of(new GroupAuthorizationException("Group authorization exception")));

        assertEquals(NetworkClientDelegate.PollResult.EMPTY, commitRequestManager.poll(200));

        assertEmptyPendingRequests(commitRequestManager);
    }

    private static void assertEmptyPendingRequests(CommitRequestManager commitRequestManager) {
        assertTrue(commitRequestManager.pendingRequests.inflightOffsetFetches.isEmpty());
        assertTrue(commitRequestManager.pendingRequests.unsentOffsetFetches.isEmpty());
        assertTrue(commitRequestManager.pendingRequests.unsentOffsetCommits.isEmpty());
    }

    @Test
    public void testPollWithFatalErrorDuringCoordinatorIsEmptyAndClosing() {
        CommitRequestManager commitRequestManager = create(true, 100);

        Map<TopicPartition, OffsetAndMetadata> offsets = Map.of(new TopicPartition("topic", 1),
                new OffsetAndMetadata(0));

        var commitFuture = commitRequestManager.commitAsync(offsets);

        commitRequestManager.signalClose();
        when(coordinatorRequestManager.coordinator()).thenReturn(Optional.empty());
        when(coordinatorRequestManager.fatalError())
                .thenReturn(Optional.of(new GroupAuthorizationException("Fatal error")));

        assertEquals(NetworkClientDelegate.PollResult.EMPTY, commitRequestManager.poll(200));

        assertTrue(commitFuture.isCompletedExceptionally());

        TestUtils.assertFutureThrows(GroupAuthorizationException.class, commitFuture, "Fatal error");
    }

    // Supplies (error, isRetriable)
    private static Stream<Arguments> partitionDataErrorSupplier() {
        return Stream.of(
            Arguments.of(Errors.UNSTABLE_OFFSET_COMMIT, true),
            Arguments.of(Errors.UNKNOWN_TOPIC_OR_PARTITION, false),
            Arguments.of(Errors.TOPIC_AUTHORIZATION_FAILED, false),

Review comment (bug):
This unit test only covers the path of `coordinatorRequestManager.coordinator().isEmpty() -> pendingRequests.maybeFailOnCoordinatorFatalError`. Could you please add another one to cover the case of `closing && pendingRequests.hasUnsentRequests()`?

Code after:

    @Test
    public void testPollWithFatalErrorShouldFailAllUnsentRequests() {
        CommitRequestManager commitRequestManager = create(true, 100);
        when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(mockedNode));

        commitRequestManager.fetchOffsets(Collections.singleton(new TopicPartition("test", 0)), 200);
        assertEquals(1, commitRequestManager.pendingRequests.unsentOffsetFetches.size());

        when(coordinatorRequestManager.coordinator()).thenReturn(Optional.empty());
        when(coordinatorRequestManager.fatalError())
                .thenReturn(Optional.of(new GroupAuthorizationException("Group authorization exception")));

        assertEquals(NetworkClientDelegate.PollResult.EMPTY, commitRequestManager.poll(200));

        assertEmptyPendingRequests(commitRequestManager);
    }

    private static void assertEmptyPendingRequests(CommitRequestManager commitRequestManager) {
        assertTrue(commitRequestManager.pendingRequests.inflightOffsetFetches.isEmpty());
        assertTrue(commitRequestManager.pendingRequests.unsentOffsetFetches.isEmpty());
        assertTrue(commitRequestManager.pendingRequests.unsentOffsetCommits.isEmpty());
    }

    @Test
    public void testPollWithFatalErrorDuringCoordinatorIsEmptyAndClosing() {
        CommitRequestManager commitRequestManager = create(true, 100);

        Map<TopicPartition, OffsetAndMetadata> offsets = Map.of(new TopicPartition("topic", 1),
                new OffsetAndMetadata(0));

        var commitFuture = commitRequestManager.commitAsync(offsets);

        commitRequestManager.signalClose();
        when(coordinatorRequestManager.coordinator()).thenReturn(Optional.empty());
        when(coordinatorRequestManager.fatalError())
                .thenReturn(Optional.of(new GroupAuthorizationException("Fatal error")));

        assertEquals(NetworkClientDelegate.PollResult.EMPTY, commitRequestManager.poll(time.milliseconds()));

        assertTrue(commitFuture.isCompletedExceptionally());

        TestUtils.assertFutureThrows(GroupAuthorizationException.class, commitFuture, "Fatal error");
    }

    @Test
    public void testPollWithClosingAndPendingRequests() {
        CommitRequestManager commitRequestManager = create(true, 100);

        Map<TopicPartition, OffsetAndMetadata> offsets = Map.of(new TopicPartition("topic", 1),
                new OffsetAndMetadata(0));