private void addRecords(Map<TopicPartition, List<ConsumerRecord<K, V>>> records) {
        records.forEach((partition, partRecords) -> {
            this.numRecords += partRecords.size();
            List<ConsumerRecord<K, V>> currentRecords = this.records.get(partition);
            if (currentRecords == null) {
                this.records.put(partition, partRecords);
            } else {
                // this case shouldn't usually happen because we only send one fetch at a time per partition,
                // but it might conceivably happen in some rare cases (such as partition leader changes).
                // we have to copy to a new list because the old one may be immutable
                List<ConsumerRecord<K, V>> newRecords = new ArrayList<>(currentRecords.size() + partRecords.size());
                newRecords.addAll(currentRecords);
                newRecords.addAll(partRecords);
                this.records.put(partition, newRecords);
            }
        });
    }
@return {@code true} if and only if this fetch did not return any user-visible (i.e., non-control) records, and
did not cause the consumer position to advance for any topic partitions
clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetch.java