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 testFetchMaxPollRecordsUnaligned() {
        final int maxPollRecords = 2;
        buildFetcher(maxPollRecords);

        Set<TopicPartition> tps = new HashSet<>();
        tps.add(tp0);
        tps.add(tp1);
        assignFromUser(tps);
        subscriptions.seek(tp0, 1);
        subscriptions.seek(tp1, 6);

        client.prepareResponse(fetchResponse2(tidp0, records, 100L, tidp1, moreRecords, 100L));
        client.prepareResponse(fullFetchResponse(tidp0, emptyRecords, Errors.NONE, 100L, 0));

        // Send fetch request because we do not have pending fetch responses to process.
        // The first fetch response will return 3 records for tp0 and 3 more for tp1.
        assertEquals(1, sendFetches());
        // The poll returns 2 records from one of the topic-partitions (non-deterministic).
        // This leaves 1 record pending from that topic-partition, and the remaining 3 from the other.
        pollAndValidateMaxPollRecordsNotExceeded(maxPollRecords);

        // See if we need to send another fetch, which we do not because we have records in hand.
        assertEquals(0, sendFetches());
        // The poll returns 2 more records, 1 from the topic-partition we've already been
        // processing, and 1 more from the other topic-partition. This means we have processed
        // all records from the former, and 2 remain from the latter.
        pollAndValidateMaxPollRecordsNotExceeded(maxPollRecords);

        // See if we need to send another fetch, which we do because we've processed all of the records
        // from one of the topic-partitions. The fetch response does not contain any more records.
        assertEquals(1, sendFetches());
        // The poll returns the final 2 records.
        pollAndValidateMaxPollRecordsNotExceeded(maxPollRecords);
    }