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

        TopicIdPartition fooWithoutId = new TopicIdPartition(Uuid.ZERO_UUID, new TopicPartition("foo", 0));

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

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

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

        // Upgrade.
        TopicIdPartition fooWithId = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("foo", 0));
        subscriptions.assignFromUser(singleton(fooWithId.topicPartition()));
        client.updateMetadata(RequestTestUtils.metadataUpdateWithIds(1, singleton(fooWithId), tp -> validLeaderEpoch));
        subscriptions.seek(fooWithId.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(fooWithId, new PartitionData(
                                fooWithId.topicId(),
                                0,
                                FetchRequest.INVALID_LOG_START_OFFSET,
                                fetchSize,
                                Optional.of(validLeaderEpoch))
                        ),
                        emptyList()
                ),
                fullFetchResponse(1, fooWithId, records, Errors.NONE, 100L, 0)
        );
        networkClientDelegate.poll(time.timer(0));
        assertTrue(fetcher.hasCompletedFetches());
        fetchRecords();

        // Downgrade.
        subscriptions.assignFromUser(singleton(fooWithoutId.topicPartition()));
        client.updateMetadata(RequestTestUtils.metadataUpdateWithIds(1, singleton(fooWithoutId), tp -> validLeaderEpoch));
        subscriptions.seek(fooWithoutId.topicPartition(), 0);

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

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