private void verifyConvertedRecords(List<SimpleRecord> initialRecords,
                                        List<Long> initialOffsets,
                                        List<Records> convertedRecordsList,
                                        Compression compression,
                                        byte magicByte) {
        int i = 0;

        for (Records convertedRecords : convertedRecordsList) {
            for (RecordBatch batch : convertedRecords.batches()) {
                assertTrue(batch.magic() <= magicByte, "Magic byte should be lower than or equal to " + magicByte);
                if (batch.magic() == RecordBatch.MAGIC_VALUE_V0)
                    assertEquals(TimestampType.NO_TIMESTAMP_TYPE, batch.timestampType());
                else
                    assertEquals(TimestampType.CREATE_TIME, batch.timestampType());
                assertEquals(compression.type(), batch.compressionType(), "Compression type should not be affected by conversion");
                for (Record record : batch) {
                    assertTrue(record.hasMagic(batch.magic()), "Inner record should have magic " + magicByte);
                    assertEquals(initialOffsets.get(i).longValue(), record.offset(), "Offset should not change");
                    assertEquals(utf8(initialRecords.get(i).key()), utf8(record.key()), "Key should not change");
                    assertEquals(utf8(initialRecords.get(i).value()), utf8(record.value()), "Value should not change");
                    assertFalse(record.hasTimestampType(TimestampType.LOG_APPEND_TIME));
                    if (batch.magic() == RecordBatch.MAGIC_VALUE_V0) {
                        assertEquals(RecordBatch.NO_TIMESTAMP, record.timestamp());
                        assertFalse(record.hasTimestampType(TimestampType.CREATE_TIME));
                        assertTrue(record.hasTimestampType(TimestampType.NO_TIMESTAMP_TYPE));
                    } else if (batch.magic() == RecordBatch.MAGIC_VALUE_V1) {
                        assertEquals(initialRecords.get(i).timestamp(), record.timestamp(), "Timestamp should not change");
                        assertTrue(record.hasTimestampType(TimestampType.CREATE_TIME));
                        assertFalse(record.hasTimestampType(TimestampType.NO_TIMESTAMP_TYPE));
                    } else {
                        assertEquals(initialRecords.get(i).timestamp(), record.timestamp(), "Timestamp should not change");
                        assertFalse(record.hasTimestampType(TimestampType.CREATE_TIME));
                        assertFalse(record.hasTimestampType(TimestampType.NO_TIMESTAMP_TYPE));
                        assertArrayEquals(initialRecords.get(i).headers(), record.headers(), "Headers should not change");
                    }
                    i += 1;
                }
            }
        }
        assertEquals(initialOffsets.size(), i);
    }
Test the new FileRecords with pre allocate as true and file has been clearly shut down, the file will be truncate to end of valid data.
clients/src/test/java/org/apache/kafka/common/record/FileRecordsTest.java