public List<ConsumerRecord<K, V>> records(TopicPartition partition) {
        List<ConsumerRecord<K, V>> recs = this.records.get(partition);
        if (recs == null)
            return Collections.emptyList();
        else
            return Collections.unmodifiableList(recs);
    }
@Test
    public void testRecordsByTopic() {
        List<String> topics = Arrays.asList("topic1", "topic2", "topic3", "topic4");
        int recordSize = 3;
        int partitionSize = 10;
        int emptyPartitionIndex = 6;
        int expectedTotalRecordSizeOfEachTopic = recordSize * (partitionSize - 1);

        ConsumerRecords<Integer, String> consumerRecords = buildTopicTestRecords(recordSize, partitionSize, emptyPartitionIndex, topics);

        for (String topic : topics) {
            Iterable<ConsumerRecord<Integer, String>> records = consumerRecords.records(topic);
            int recordCount = 0;
            int partitionCount = 0;
            int currentPartition = -1;

            for (ConsumerRecord<Integer, String> record : records) {
                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);
            assertEquals(expectedTotalRecordSizeOfEachTopic, recordCount);
        }
    }