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

Code before:
        assertTrue(commitRequestManager.pendingRequests.unsentOffsetCommits.isEmpty());
    }

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

        Map<TopicPartition, OffsetAndMetadata> offsets = Collections.singletonMap(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());

        ExecutionException exception = assertThrows(ExecutionException.class, commitFuture::get);

        assertInstanceOf(GroupAuthorizationException.class, exception.getCause());
        assertEquals("Fatal error", exception.getCause().getMessage());
    }

    // 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),
            Arguments.of(Errors.UNKNOWN_SERVER_ERROR, false));
    }

    private List<CompletableFuture<Map<TopicPartition, OffsetAndMetadata>>> sendAndVerifyDuplicatedOffsetFetchRequests(
            final CommitRequestManager commitRequestManager,
            final Set<TopicPartition> partitions,
            int numRequest,
            final Errors error) {
        List<CompletableFuture<Map<TopicPartition, OffsetAndMetadata>>> futures = new ArrayList<>();
        long deadlineMs = time.milliseconds() + defaultApiTimeoutMs;
        for (int i = 0; i < numRequest; i++) {
            futures.add(commitRequestManager.fetchOffsets(partitions, deadlineMs));
        }

        NetworkClientDelegate.PollResult res = commitRequestManager.poll(time.milliseconds());
        assertEquals(1, res.unsentRequests.size());
        res.unsentRequests.get(0).handler().onComplete(buildOffsetFetchClientResponse(res.unsentRequests.get(0),

Review comment (suggestion):
You might consider using this method.
`TestUtils.assertFutureThrows`

Code after:
        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));

        var commitFuture = commitRequestManager.commitAsync(offsets);

        commitRequestManager.signalClose();
        when(coordinatorRequestManager.coordinator()).thenReturn(Optional.empty());

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

        assertTrue(commitFuture.isCompletedExceptionally());

        TestUtils.assertFutureThrows(CommitFailedException.class, commitFuture,
                "Failed to commit offsets: Coordinator unknown and consumer is closing");
    }

    // 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),