public int numRecords() {
        int numRecords = 0;
        if (!batches.isEmpty()) {
            Iterator<Map.Entry<TopicIdPartition, ShareInFlightBatch<K, V>>> iterator = batches.entrySet().iterator();
            while (iterator.hasNext()) {
                Map.Entry<TopicIdPartition, ShareInFlightBatch<K, V>> entry = iterator.next();
                ShareInFlightBatch<K, V> batch = entry.getValue();
                if (batch.isEmpty()) {
                    iterator.remove();
                } else {
                    numRecords += batch.numRecords();
                }
            }
        }

        return numRecords;
    }
@return the total number of non-control messages for this fetch, across all partitions
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareFetch.java