public synchronized int sendFetches() {
        final Map<Node, FetchSessionHandler.FetchRequestData> fetchRequests = prepareFetchRequests();
        sendFetchesInternal(
                fetchRequests,
                (fetchTarget, data, clientResponse) -> {
                    synchronized (Fetcher.this) {
                        handleFetchSuccess(fetchTarget, data, clientResponse);
                    }
                },
                (fetchTarget, data, error) -> {
                    synchronized (Fetcher.this) {
                        handleFetchFailure(fetchTarget, data, error);
                    }
                });
        return fetchRequests.size();
    }
@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)
        );
        consumerClient.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)
        );
        consumerClient.poll(time.timer(0));
        assertTrue(fetcher.hasCompletedFetches());
        fetchRecords();
    }