@Test
    public void testFetchAfterPartitionWithFetchedRecordsIsUnassigned() {
        buildFetcher(2);

        assignFromUser(singleton(tp0));
        subscriptions.seek(tp0, 1);

        // Returns 3 records while `max.poll.records` is configured to 2
        client.prepareResponse(matchesOffset(tidp0, 1), fullFetchResponse(tidp0, records, Errors.NONE, 100L, 0));

        assertEquals(1, sendFetches());
        consumerClient.poll(time.timer(0));
        Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> recordsByPartition = fetchRecords();
        List<ConsumerRecord<byte[], byte[]>> recordsToTest = recordsByPartition.get(tp0);
        assertEquals(2, recordsToTest.size());
        assertEquals(3L, subscriptions.position(tp0).offset);
        assertEquals(1, recordsToTest.get(0).offset());
        assertEquals(2, recordsToTest.get(1).offset());

        assignFromUser(singleton(tp1));
        client.prepareResponse(matchesOffset(tidp1, 4), fullFetchResponse(tidp1, nextRecords, Errors.NONE, 100L, 0));
        subscriptions.seek(tp1, 4);

        assertEquals(1, sendFetches());
        consumerClient.poll(time.timer(0));
        Map<TopicPartition, List<ConsumerRecord<byte[], byte[]>>> fetchedRecords = fetchRecords();
        assertNull(fetchedRecords.get(tp0));
        recordsToTest = fetchedRecords.get(tp1);
        assertEquals(2, recordsToTest.size());
        assertEquals(6L, subscriptions.position(tp1).offset);
        assertEquals(4, recordsToTest.get(0).offset());
        assertEquals(5, recordsToTest.get(1).offset());
    }
Test the scenario where a partition with fetched but not consumed records (i.e. max.poll.records is
less than the number of fetched records) is unassigned and a different partition is assigned. This is a
pattern used by Streams state restoration and KAFKA-5097 would have been caught by this test.
clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java