@Override
    public Iterator<ConsumerRecord<K, V>> iterator() {
        return new ConcatenatedIterable<>(records.values()).iterator();
    }
@Test
    public void testIterator() {
        String topic = "topic";
        int recordSize = 10;
        int partitionSize = 15;
        int emptyPartitionIndex = 3;
        ConsumerRecords<Integer, String> records = buildTopicTestRecords(recordSize, partitionSize, emptyPartitionIndex, Collections.singleton(topic));
        Iterator<ConsumerRecord<Integer, String>> iterator = records.iterator();

        int recordCount = 0;
        int partitionCount = 0;
        int currentPartition = -1;

        while (iterator.hasNext()) {
            ConsumerRecord<Integer, String> record = iterator.next();
            validateEmptyPartition(record, emptyPartitionIndex);

            // Check if we have moved to a new partition
            if (currentPartition != record.partition()) {
                // Increment the partition count as we have encountered a new partition
                partitionCount++;
                // Update the current partition to the new partition
                currentPartition = record.partition();
            }

            validateRecordPayload(topic, record, currentPartition, recordCount, recordSize);
            recordCount++;
        }

        // Including empty partition
        assertEquals(partitionSize, partitionCount + 1);
    }