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

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

        // Assign foo with old topic id.
        subscriptions.assignFromUser(singleton(fooWithOldTopicId.topicPartition()));
        client.updateMetadata(RequestTestUtils.metadataUpdateWithIds(1, singleton(fooWithOldTopicId), tp -> validLeaderEpoch));
        subscriptions.seek(fooWithOldTopicId.topicPartition(), 0);

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

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

        // Replace foo with old topic id with foo with new topic id.
        subscriptions.assignFromUser(singleton(fooWithNewTopicId.topicPartition()));
        client.updateMetadata(RequestTestUtils.metadataUpdateWithIds(1, singleton(fooWithNewTopicId), tp -> validLeaderEpoch));
        subscriptions.seek(fooWithNewTopicId.topicPartition(), 0);

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

        // foo with old topic id should be removed from the session.
        client.prepareResponse(
                fetchRequestMatcher(ApiKeys.FETCH.latestVersion(),
                        singletonMap(fooWithNewTopicId, new PartitionData(
                                fooWithNewTopicId.topicId(),
                                0,
                                FetchRequest.INVALID_LOG_START_OFFSET,
                                fetchSize,
                                Optional.of(validLeaderEpoch))
                        ),
                        singletonList(fooWithOldTopicId)
                ),
                fullFetchResponse(1, fooWithNewTopicId, records, Errors.NONE, 100L, 0)
        );
        networkClientDelegate.poll(time.timer(0));
        assertTrue(fetcher.hasCompletedFetches());
        fetchRecords();
    }