@Test
    public void testCollectFetchInitializationWithUpdateLastStableOffsetOnNotAssignedPartition() {
        final TopicPartition topicPartition0 = new TopicPartition("topic", 0);
        final long fetchOffset = 42;
        final long highWatermark = 1000;
        final long logStartOffset = 10;
        final long lastStableOffset = 900;
        final SubscriptionState subscriptions = mock(SubscriptionState.class);
        when(subscriptions.hasValidPosition(topicPartition0)).thenReturn(true);
        when(subscriptions.positionOrNull(topicPartition0)).thenReturn(new SubscriptionState.FetchPosition(fetchOffset));
        when(subscriptions.tryUpdatingHighWatermark(topicPartition0, highWatermark)).thenReturn(true);
        when(subscriptions.tryUpdatingLogStartOffset(topicPartition0, logStartOffset)).thenReturn(true);
        when(subscriptions.tryUpdatingLastStableOffset(topicPartition0, lastStableOffset)).thenReturn(false);
        final FetchCollector<String, String> fetchCollector = createFetchCollector(subscriptions);
        final Records records = createRecords();
        FetchResponseData.PartitionData partitionData = new FetchResponseData.PartitionData()
            .setPartitionIndex(topicPartition0.partition())
            .setHighWatermark(highWatermark)
            .setRecords(records)
            .setLogStartOffset(logStartOffset)
            .setLastStableOffset(lastStableOffset);
        final CompletedFetch completedFetch = new CompletedFetchBuilder()
            .partitionData(partitionData)
            .partition(topicPartition0)
            .fetchOffset(fetchOffset).build();
        final FetchBuffer fetchBuffer = mock(FetchBuffer.class);
        when(fetchBuffer.nextInLineFetch()).thenReturn(null);
        when(fetchBuffer.peek()).thenReturn(completedFetch).thenReturn(null);

        final Fetch<String, String> fetch = fetchCollector.collectFetch(fetchBuffer);

        assertTrue(fetch.isEmpty());
        verify(fetchBuffer).setNextInLineFetch(null);
    }
This tests the {@link FetchCollector} functionality in addition to what {@link FetcherTest} tests during the course
of its tests.
clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchCollectorTest.java