public ShareFetch<K, V> collect(final ShareFetchBuffer fetchBuffer) {
        ShareFetch<K, V> fetch = ShareFetch.empty();
        int recordsRemaining = fetchConfig.maxPollRecords;

        try {
            while (recordsRemaining > 0) {
                final ShareCompletedFetch nextInLineFetch = fetchBuffer.nextInLineFetch();

                if (nextInLineFetch == null || nextInLineFetch.isConsumed()) {
                    final ShareCompletedFetch completedFetch = fetchBuffer.peek();

                    if (completedFetch == null) {
                        break;
                    }

                    if (!completedFetch.isInitialized()) {
                        try {
                            fetchBuffer.setNextInLineFetch(initialize(completedFetch));
                        } catch (Exception e) {
                            if (fetch.isEmpty()) {
                                fetchBuffer.poll();
                            }
                            throw e;
                        }
                    } else {
                        fetchBuffer.setNextInLineFetch(completedFetch);
                    }

                    fetchBuffer.poll();
                } else {
                    final TopicIdPartition tp = nextInLineFetch.partition;

                    ShareInFlightBatch<K, V> batch = nextInLineFetch.fetchRecords(
                            deserializers,
                            recordsRemaining,
                            fetchConfig.checkCrcs);

                    if (batch.isEmpty()) {
                        nextInLineFetch.drain();
                    }

                    recordsRemaining -= batch.numRecords();
                    fetch.add(tp, batch);

                    if (batch.getException() != null) {
                        throw batch.getException();
                    } else if (batch.hasCachedException()) {
                        break;
                    }
                }
            }
        } catch (KafkaException e) {
            if (fetch.isEmpty()) {
                throw e;
            }
        }

        return fetch;
    }
@Test
    public void testFetchNormal() {
        int recordCount = DEFAULT_MAX_POLL_RECORDS;
        buildDependencies();
        subscribeAndAssign(topicAPartition0);

        ShareCompletedFetch completedFetch = completedFetchBuilder
                .recordCount(recordCount)
                .build();

        // Validate that the buffer is empty until after we add the fetch data.
        assertTrue(fetchBuffer.isEmpty());
        fetchBuffer.add(completedFetch);
        assertFalse(fetchBuffer.isEmpty());

        // Validate that the completed fetch isn't initialized just because we add it to the buffer.
        assertFalse(completedFetch.isInitialized());

        // Fetch the data and validate that we get all the records we want back.
        ShareFetch<String, String> fetch = fetchCollector.collect(fetchBuffer);
        assertFalse(fetch.isEmpty());
        assertEquals(recordCount, fetch.numRecords());

        // When we collected the data from the buffer, this will cause the completed fetch to get initialized.
        assertTrue(completedFetch.isInitialized());

        // However, even though we've collected the data, it isn't (completely) consumed yet.
        assertFalse(completedFetch.isConsumed());

        // The buffer is now considered "empty" because our queue is empty.
        assertTrue(fetchBuffer.isEmpty());
        assertNull(fetchBuffer.peek());
        assertNull(fetchBuffer.poll());

        // However, while the queue is "empty", the next-in-line fetch is actually still in the buffer.
        assertNotNull(fetchBuffer.nextInLineFetch());

        // Now attempt to collect more records from the fetch buffer.
        fetch = fetchCollector.collect(fetchBuffer);
        assertEquals(0, fetch.numRecords());
        assertTrue(fetch.isEmpty());

        // However, once we read *past* the end of the records in the ShareCompletedFetch, then we will call
        // drain on it, and it will be considered all consumed.
        assertTrue(completedFetch.isConsumed());
    }