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

Code before:
        assertFutureThrows(CommitFailedException.class, commitResult);
    }

    @Test
    public void testCommitSyncFailsWithCommitFailedExceptionOnStaleMemberEpoch() {
        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));

        // Send commit request expected to be retried on retriable errors
        CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> commitResult = commitRequestManager.commitSync(
            offsets, time.milliseconds() + defaultApiTimeoutMs);
        completeOffsetCommitRequestWithError(commitRequestManager, Errors.STALE_MEMBER_EPOCH);
        NetworkClientDelegate.PollResult res = commitRequestManager.poll(time.milliseconds());
        assertEquals(0, res.unsentRequests.size());

        // Commit should fail with CommitFailedException
        assertTrue(commitResult.isDone());
        assertFutureThrows(CommitFailedException.class, commitResult);
    }

    @Test
    public void testCommitSyncShouldSuccessWithTopicHasId() {
        subscriptionState = mock(SubscriptionState.class);
        TopicPartition tp = new TopicPartition("topic", 1);
        Uuid topicId = Uuid.randomUuid();
        when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(mockedNode));
        when(metadata.topicIds()).thenReturn(Map.of("topic", topicId));
        OffsetAndMetadata offsetAndMetadata = new OffsetAndMetadata(0, Optional.of(1), "");
        Map<TopicPartition, OffsetAndMetadata> offsets = Map.of(tp, offsetAndMetadata);

        CommitRequestManager commitRequestManager = create(false, 100);
        CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future = commitRequestManager.commitSync(
            offsets, time.milliseconds() + defaultApiTimeoutMs);
        assertEquals(1, commitRequestManager.unsentOffsetCommitRequests().size());
        List<NetworkClientDelegate.FutureCompletionHandler> pollResults = assertPoll(1, commitRequestManager);
        pollResults.forEach(v -> v.onComplete(mockOffsetCommitResponseWithTopicId(
            "topic",
            topicId,
            1,
            (short) 10,
            Errors.NONE)));

        verify(subscriptionState, never()).allConsumed();
        verify(metadata).updateLastSeenEpochIfNewer(tp, 1);
        Map<TopicPartition, OffsetAndMetadata> commitOffsets = assertDoesNotThrow(() -> future.get());
        assertTrue(future.isDone());
        assertEquals(offsets, commitOffsets);

Review comment (suggestion):
```suggestion
    public void testCommitSyncShouldSucceedWithTopicId() {
```

Code after:
        assertTrue(commitResult.isDone());
        assertFutureThrows(CommitFailedException.class, commitResult);
    }

    @Test
    public void testCommitSyncFailsWithCommitFailedExceptionOnStaleMemberEpoch() {
        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));

        // Send commit request expected to be retried on retriable errors
        CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> commitResult = commitRequestManager.commitSync(
            offsets, time.milliseconds() + defaultApiTimeoutMs);
        completeOffsetCommitRequestWithError(commitRequestManager, Errors.STALE_MEMBER_EPOCH);
        NetworkClientDelegate.PollResult res = commitRequestManager.poll(time.milliseconds());
        assertEquals(0, res.unsentRequests.size());

        // Commit should fail with CommitFailedException
        assertTrue(commitResult.isDone());
        assertFutureThrows(CommitFailedException.class, commitResult);
    }

    @Test
    public void testCommitSyncShouldSucceedWithTopicId() {
        subscriptionState = mock(SubscriptionState.class);
        TopicPartition tp = new TopicPartition("topic", 1);
        Uuid topicId = Uuid.randomUuid();
        when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(mockedNode));
        when(metadata.topicIds()).thenReturn(Map.of("topic", topicId));
        OffsetAndMetadata offsetAndMetadata = new OffsetAndMetadata(0, Optional.of(1), "");
        Map<TopicPartition, OffsetAndMetadata> offsets = Map.of(tp, offsetAndMetadata);

        CommitRequestManager commitRequestManager = create(false, 100);
        CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future = commitRequestManager.commitSync(
            offsets, time.milliseconds() + defaultApiTimeoutMs);
        assertEquals(1, commitRequestManager.unsentOffsetCommitRequests().size());
        List<NetworkClientDelegate.FutureCompletionHandler> pollResults = assertPoll(true, 1, commitRequestManager, true);
        pollResults.forEach(v -> v.onComplete(mockOffsetCommitResponseWithTopicId(
            topicId,
            1,
            (short) 10,
            Errors.NONE)));

        verify(subscriptionState, never()).allConsumed();
        verify(metadata).updateLastSeenEpochIfNewer(tp, 1);
        Map<TopicPartition, OffsetAndMetadata> commitOffsets = assertDoesNotThrow(() -> future.get());
        assertTrue(future.isDone());
        assertEquals(offsets, commitOffsets);