@Override
    public PollResult poll(long currentTimeMs) {
        return pollInternal(
                prepareFetchRequests(),
                this::handleFetchSuccess,
                this::handleFetchFailure
        );
    }
@Test
    public void testFetchForgetTopicIdWhenUnassigned() {
        buildFetcher();

        TopicIdPartition foo = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("foo", 0));
        TopicIdPartition bar = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("bar", 0));

        // Assign foo and bar.
        subscriptions.assignFromUser(singleton(foo.topicPartition()));
        client.updateMetadata(RequestTestUtils.metadataUpdateWithIds(1, singleton(foo), tp -> validLeaderEpoch));
        subscriptions.seek(foo.topicPartition(), 0);

        assertEquals(1, sendFetches());

        // Fetch should use latest version.
        client.prepareResponse(
                fetchRequestMatcher(ApiKeys.FETCH.latestVersion(),
                        singletonMap(foo, new PartitionData(
                                foo.topicId(),
                                0,
                                FetchRequest.INVALID_LOG_START_OFFSET,
                                fetchSize,
                                Optional.of(validLeaderEpoch))
                        ),
                        emptyList()
                ),
                fullFetchResponse(1, foo, records, Errors.NONE, 100L, 0)
        );
        networkClientDelegate.poll(time.timer(0));
        assertTrue(fetcher.hasCompletedFetches());
        fetchRecords();

        // Assign bar and un-assign foo.
        subscriptions.assignFromUser(singleton(bar.topicPartition()));
        client.updateMetadata(RequestTestUtils.metadataUpdateWithIds(1, singleton(bar), tp -> validLeaderEpoch));
        subscriptions.seek(bar.topicPartition(), 0);

        // Fetch should use latest version.
        assertEquals(1, sendFetches());
        assertFalse(fetcher.hasCompletedFetches());

        client.prepareResponse(
                fetchRequestMatcher(ApiKeys.FETCH.latestVersion(),
                        singletonMap(bar, new PartitionData(
                                bar.topicId(),
                                0,
                                FetchRequest.INVALID_LOG_START_OFFSET,
                                fetchSize,
                                Optional.of(validLeaderEpoch))
                        ),
                        singletonList(foo)
                ),
                fullFetchResponse(1, bar, records, Errors.NONE, 100L, 0)
        );
        networkClientDelegate.poll(time.timer(0));
        assertTrue(fetcher.hasCompletedFetches());
        fetchRecords();
    }