public static MemoryRecords withRecords(byte magic, long initialOffset, Compression compression,
                                            TimestampType timestampType, long producerId, short producerEpoch,
                                            int baseSequence, int partitionLeaderEpoch, boolean isTransactional,
                                            SimpleRecord... records) {
        if (records.length == 0)
            return MemoryRecords.EMPTY;
        int sizeEstimate = AbstractRecords.estimateSizeInBytes(magic, compression.type(), Arrays.asList(records));
        ByteBufferOutputStream bufferStream = new ByteBufferOutputStream(sizeEstimate);
        long logAppendTime = RecordBatch.NO_TIMESTAMP;
        if (timestampType == TimestampType.LOG_APPEND_TIME)
            logAppendTime = System.currentTimeMillis();
        try (final MemoryRecordsBuilder builder = new MemoryRecordsBuilder(bufferStream, magic, compression, timestampType,
                initialOffset, logAppendTime, producerId, producerEpoch, baseSequence, isTransactional, false,
                partitionLeaderEpoch, sizeEstimate)) {
            for (SimpleRecord record : records)
                builder.append(record);
            return builder.build();
        }
    }
Check whether a record should be retained in the log. Note that {@link #checkBatchRetention(RecordBatch)}
is used prior to checking individual record retention. Only records from batches which were not
explicitly discarded with {@link BatchRetention#DELETE} will be considered.
clients/src/main/java/org/apache/kafka/common/record/MemoryRecords.java