@Test
    public void testCloseShouldBeIdempotent() {
        buildFetcher();

        fetcher.close();
        fetcher.close();
        fetcher.close();

        verify(fetcher, times(1)).closeInternal(any(Timer.class));
    }
This validates that no records are fetched for partitions that are waiting for the
onPartitionsAssigned callback to complete. Note that this markPendingOnAssignedCallback
functionality is only used from the {@link AsyncKafkaConsumer}.
clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchRequestManagerTest.java