void drain() {
        if (!isConsumed) {
            maybeCloseRecordStream();
            cachedRecordException = null;
            this.isConsumed = true;
            recordAggregatedMetrics(bytesRead, recordsRead);

            // we move the partition to the end if we received some bytes. This way, it's more likely that partitions
            // for the same topic can remain together (allowing for more efficient serialization).
            if (bytesRead > 0)
                subscriptions.movePartitionToEnd(partition);
        }
    }
Draining a {@link CompletedFetch} will signal that the data has been consumed and the underlying resources
are closed. This is somewhat analogous to {@link Closeable#close() closing}, though no error will result if a
caller invokes {@link #fetchRecords(FetchConfig, Deserializers, int)}; an empty {@link List list} will be
returned instead.
clients/src/main/java/org/apache/kafka/clients/consumer/internals/CompletedFetch.java