@Test
    public void testDownconversionAfterMessageFormatDowngrade() throws IOException {
        // random bytes
        Random random = new Random();
        byte[] bytes = new byte[3000];
        random.nextBytes(bytes);

        // records
        GzipCompression compression = Compression.gzip().build();
        List<Long> offsets = asList(0L, 1L);
        List<Byte> magic = asList(RecordBatch.MAGIC_VALUE_V2, RecordBatch.MAGIC_VALUE_V1);  // downgrade message format from v2 to v1
        List<SimpleRecord> records = asList(
                new SimpleRecord(1L, "k1".getBytes(), bytes),
                new SimpleRecord(2L, "k2".getBytes(), bytes));
        byte toMagic = 1;

        // create MemoryRecords
        ByteBuffer buffer = ByteBuffer.allocate(8000);
        for (int i = 0; i < records.size(); i++) {
            MemoryRecordsBuilder builder = MemoryRecords.builder(buffer, magic.get(i), compression, TimestampType.CREATE_TIME, 0L);
            builder.appendWithOffset(offsets.get(i), records.get(i));
            builder.close();
        }
        buffer.flip();

        // create FileRecords, down-convert and verify
        try (FileRecords fileRecords = FileRecords.open(tempFile())) {
            fileRecords.append(MemoryRecords.readableRecords(buffer));
            fileRecords.flush();
            downConvertAndVerifyRecords(records, offsets, fileRecords, compression, toMagic, 0L, time);
        }
    }
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