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

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

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

        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();
        Map<TopicPartition, OffsetAndMetadata> commitOffsets = assertDoesNotThrow(() -> future.get());
        assertTrue(future.isDone());
        assertEquals(offsets, commitOffsets);
    }

Review comment (suggestion):
we can verify that we never `updateLastSeenEpochIfNewer` on this path (opposed to the test above where we do expect a call to it)

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

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

        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(
            topicId,
            1,
            (short) 10,
            Errors.NONE)));

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