diff --git a/rm-automation/rm-automation-community-rest-api/src/main/java/org/alfresco/rest/v0/service/DispositionScheduleService.java b/rm-automation/rm-automation-community-rest-api/src/main/java/org/alfresco/rest/v0/service/DispositionScheduleService.java
index ba38536179..67fb23a68e 100644
--- a/rm-automation/rm-automation-community-rest-api/src/main/java/org/alfresco/rest/v0/service/DispositionScheduleService.java
+++ b/rm-automation/rm-automation-community-rest-api/src/main/java/org/alfresco/rest/v0/service/DispositionScheduleService.java
@@ -30,7 +30,6 @@ package org.alfresco.rest.v0.service;
 import java.util.HashMap;
 
 import org.alfresco.rest.core.v0.BaseAPI;
-import org.alfresco.rest.rm.community.model.recordcategory.RecordCategory;
 import org.alfresco.rest.v0.RecordCategoriesAPI;
 import org.alfresco.utility.data.DataUser;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -56,16 +55,15 @@ public class DispositionScheduleService extends BaseAPI
      *
      * @param categoryName the category in whose schedule the step will be added
      * @param period
-     * @return
      */
     public void addRetainAfterPeriodStep(String categoryName, String period)
     {
-        HashMap<RETENTION_SCHEDULE, String> cutOffStep = new HashMap<>();
-        cutOffStep.put(RETENTION_SCHEDULE.NAME, "retain");
-        cutOffStep.put(RETENTION_SCHEDULE.RETENTION_PERIOD, period);
-        cutOffStep.put(RETENTION_SCHEDULE.DESCRIPTION, "Retain after a period step");
+        HashMap<RETENTION_SCHEDULE, String> retainStep = new HashMap<>();
+        retainStep.put(RETENTION_SCHEDULE.NAME, "retain");
+        retainStep.put(RETENTION_SCHEDULE.RETENTION_PERIOD, period);
+        retainStep.put(RETENTION_SCHEDULE.DESCRIPTION, "Retain after a period step");
         recordCategoriesAPI.addDispositionScheduleSteps(dataUser.getAdminUser().getUsername(),
-                dataUser.getAdminUser().getPassword(), categoryName, cutOffStep);
+                dataUser.getAdminUser().getPassword(), categoryName, retainStep);
     }
 
     /**
@@ -73,7 +71,6 @@ public class DispositionScheduleService extends BaseAPI
      *
      * @param categoryName the category in whose schedule the step will be added
      * @param period
-     * @return
      */
     public void addCutOffAfterPeriodStep(String categoryName, String period)
     {
@@ -85,6 +82,23 @@ public class DispositionScheduleService extends BaseAPI
                 dataUser.getAdminUser().getPassword(), categoryName, cutOffStep);
     }
 
+    /**
+     * Helper method for adding a destroy with ghosting after period 
+     *
+     * @param categoryName the category in whose schedule the step will be added
+     * @param period
+     */
+    public void addDestroyWithGhostingAfterPeriodStep(String categoryName, String period)
+    {
+        HashMap<RETENTION_SCHEDULE, String> destroyStep = new HashMap<>();
+        destroyStep.put(RETENTION_SCHEDULE.NAME, "destroy");
+        destroyStep.put(RETENTION_SCHEDULE.RETENTION_PERIOD, period);
+        destroyStep.put(RETENTION_SCHEDULE.DESCRIPTION, "Destroy after a period step");
+        destroyStep.put(RETENTION_SCHEDULE.RETENTION_GHOST, "on");
+        recordCategoriesAPI.addDispositionScheduleSteps(dataUser.getAdminUser().getUsername(),
+                dataUser.getAdminUser().getPassword(), categoryName, destroyStep);
+    }
+
     /**
      * Helper method for adding a cut off after an event occurs step
      *
diff --git a/rm-automation/rm-automation-community-rest-api/src/test/java/org/alfresco/rest/rm/community/records/DeleteRecordTests.java b/rm-automation/rm-automation-community-rest-api/src/test/java/org/alfresco/rest/rm/community/records/DeleteRecordTests.java
index 2a4a5cf51e..ef63360a39 100644
--- a/rm-automation/rm-automation-community-rest-api/src/test/java/org/alfresco/rest/rm/community/records/DeleteRecordTests.java
+++ b/rm-automation/rm-automation-community-rest-api/src/test/java/org/alfresco/rest/rm/community/records/DeleteRecordTests.java
@@ -35,34 +35,45 @@ import static org.alfresco.rest.rm.community.model.user.UserPermissions.PERMISSI
 import static org.alfresco.rest.rm.community.model.user.UserRoles.ROLE_RM_POWER_USER;
 import static org.alfresco.rest.rm.community.utils.FilePlanComponentsUtil.IMAGE_FILE;
 import static org.alfresco.rest.rm.community.utils.FilePlanComponentsUtil.createElectronicRecordModel;
-import static org.alfresco.rest.rm.community.utils.FilePlanComponentsUtil.createNonElectronicRecordModel;
 import static org.alfresco.rest.rm.community.utils.FilePlanComponentsUtil.createElectronicUnfiledContainerChildModel;
+import static org.alfresco.rest.rm.community.utils.FilePlanComponentsUtil.createNonElectronicRecordModel;
 import static org.alfresco.rest.rm.community.utils.FilePlanComponentsUtil.createNonElectronicUnfiledContainerChildModel;
 import static org.alfresco.rest.rm.community.utils.FilePlanComponentsUtil.getFile;
 import static org.alfresco.utility.constants.UserRole.SiteCollaborator;
+import static org.alfresco.utility.data.RandomData.getRandomName;
+import static org.alfresco.utility.report.log.Step.STEP;
 import static org.springframework.http.HttpStatus.CREATED;
 import static org.springframework.http.HttpStatus.FORBIDDEN;
 import static org.springframework.http.HttpStatus.NOT_FOUND;
 import static org.springframework.http.HttpStatus.NO_CONTENT;
 import static org.springframework.http.HttpStatus.OK;
 
+import org.alfresco.dataprep.CMISUtil;
+import org.alfresco.rest.core.JsonBodyGenerator;
 import org.alfresco.rest.core.RestResponse;
+import org.alfresco.rest.core.v0.BaseAPI.RM_ACTIONS;
 import org.alfresco.rest.model.RestNodeBodyMoveCopyModel;
 import org.alfresco.rest.model.RestNodeModel;
 import org.alfresco.rest.requests.Node;
 import org.alfresco.rest.rm.community.base.BaseRMRestTest;
 import org.alfresco.rest.rm.community.model.record.Record;
+import org.alfresco.rest.rm.community.model.record.RecordBodyFile;
+import org.alfresco.rest.rm.community.model.recordcategory.RecordCategory;
 import org.alfresco.rest.rm.community.model.recordcategory.RecordCategoryChild;
 import org.alfresco.rest.rm.community.model.unfiledcontainer.UnfiledContainerChild;
 import org.alfresco.rest.rm.community.requests.gscore.api.RecordCategoryAPI;
 import org.alfresco.rest.rm.community.requests.gscore.api.RecordFolderAPI;
 import org.alfresco.rest.rm.community.requests.gscore.api.RecordsAPI;
+import org.alfresco.rest.v0.RMRolesAndActionsAPI;
+import org.alfresco.rest.v0.service.DispositionScheduleService;
 import org.alfresco.test.AlfrescoTest;
 import org.alfresco.utility.data.RandomData;
+import org.alfresco.utility.model.FileModel;
+import org.alfresco.utility.model.FolderModel;
 import org.alfresco.utility.model.RepoTestModel;
 import org.alfresco.utility.model.SiteModel;
 import org.alfresco.utility.model.UserModel;
-import org.alfresco.utility.report.log.Step;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.testng.annotations.Test;
 
 /**
@@ -73,6 +84,13 @@ import org.testng.annotations.Test;
  */
 public class DeleteRecordTests extends BaseRMRestTest
 {
+    @Autowired
+    private DispositionScheduleService dispositionScheduleService;
+    @Autowired
+    private RMRolesAndActionsAPI rmRolesAndActionsAPI;
+    @Autowired
+    private org.alfresco.rest.v0.RecordsAPI recordsAPI;
+
     /**
      * <pre>
      * Given an electronic record
@@ -263,28 +281,158 @@ public class DeleteRecordTests extends BaseRMRestTest
     @AlfrescoTest(jira="MNT-18806")
     public void deleteCopyOfRecord()
     {
-        Step.STEP("Create two record categories and folders.");
+        STEP("Create two record categories and folders.");
         RecordCategoryChild recordFolderA = createCategoryFolderInFilePlan();
         RecordCategoryChild recordFolderB = createCategoryFolderInFilePlan();
 
-        Step.STEP("Create a record in folder A and copy it into folder B.");
+        STEP("Create a record in folder A and copy it into folder B.");
         String recordId = getRestAPIFactory().getRecordFolderAPI()
                     .createRecord(createElectronicRecordModel(), recordFolderA.getId(), getFile(IMAGE_FILE)).getId();
-        String copyId = copyRecord(recordId, recordFolderB.getId()).getId();
+        String copyId = copyNode(recordId, recordFolderB.getId()).getId();
         assertStatusCode(CREATED);
 
-        Step.STEP("Check that it's possible to load the original content.");
+        STEP("Check that it's possible to load the original content.");
         getNodeContent(recordId);
         assertStatusCode(OK);
 
-        Step.STEP("Delete the copy.");
+        STEP("Delete the copy.");
         deleteAndVerify(copyId);
 
-        Step.STEP("Check that the original record node and content still exist.");
+        STEP("Check that the original record node and content still exist.");
         checkNodeExists(recordId);
         getNodeContent(recordId);
     }
 
+
+    /**
+     * <pre>
+     * Given a file that has a copy
+     * And the original file is declared as record
+     * When I delete the original
+     * Then it is still possible to view the content of the copy
+     * </pre>
+     */
+    @Test (description = "Deleting record doesn't delete the content for the copies")
+    @AlfrescoTest (jira = "MNT-20145")
+    public void deleteOriginOfRecord() throws Exception
+    {
+        STEP("Create a file.");
+        testSite = dataSite.usingAdmin().createPublicRandomSite();
+        FileModel testFile = dataContent.usingSite(testSite).createContent(CMISUtil.DocumentType.TEXT_PLAIN);
+
+        STEP("Create a copy of the file.");
+        RestNodeModel copyOfTestFile = copyNode(testFile.getNodeRefWithoutVersion(), testSite.getGuid());
+
+        STEP("Declare original file as record");
+        getRestAPIFactory().getFilesAPI().declareAsRecord(testFile.getNodeRefWithoutVersion());
+        assertStatusCode(CREATED);
+
+        STEP("Delete the record.");
+        deleteAndVerify(testFile.getNodeRefWithoutVersion());
+
+        STEP("Check that it's possible to load the copy content.");
+        getNodeContent(copyOfTestFile.getId());
+        assertStatusCode(OK);
+
+        STEP("Clean up.");
+        dataSite.deleteSite(testSite);
+    }
+
+    /**
+     * <pre>
+     * Given a file that has a copy
+     * And the original file is declared as record
+     * And the record becomes part of a disposition schedule with a destroy step
+     * When the record is destroyed
+     * Then it is still possible to view the content of the copy
+     * </pre>
+     */
+    @Test (description = "Destroying record doesn't delete the content for the associated copy")
+    @AlfrescoTest (jira = "MNT-20145")
+    public void destroyOfRecord() throws Exception
+    {
+        STEP("Create a file.");
+        testSite = dataSite.usingAdmin().createPublicRandomSite();
+        FileModel testFile = dataContent.usingSite(testSite).createContent(CMISUtil.DocumentType.TEXT_PLAIN);
+        FolderModel folderModel = dataContent.usingSite(testSite).createFolder();
+
+        STEP("Create a copy of the file.");
+        RestNodeModel copy = copyNode(testFile.getNodeRefWithoutVersion(), folderModel.getNodeRefWithoutVersion());
+        assertStatusCode(CREATED);
+
+        STEP("Declare the file as record.");
+        getRestAPIFactory().getFilesAPI().declareAsRecord(testFile.getNodeRefWithoutVersion());
+        assertStatusCode(CREATED);
+
+        STEP("Create a record category with a disposition schedule.");
+        RecordCategory recordCategory = createRootCategory(getRandomName("Category with disposition"));
+        dispositionScheduleService.createCategoryRetentionSchedule(recordCategory.getName(), true);
+
+        STEP("Add retention schedule cut off and destroy step with immediate period.");
+        dispositionScheduleService.addCutOffAfterPeriodStep(recordCategory.getName(), "immediately");
+        dispositionScheduleService.addDestroyWithGhostingAfterPeriodStep(recordCategory.getName(), "immediately");
+
+        STEP("Create a record folder and file the record");
+        RecordCategoryChild recFolder = createFolder(recordCategory.getId(), getRandomName("recFolder"));
+        RecordBodyFile recordBodyFile = RecordBodyFile.builder().targetParentId(recFolder.getId()).build();
+        Record recordFiled = getRestAPIFactory().getRecordsAPI().fileRecord(recordBodyFile, testFile.getNodeRefWithoutVersion());
+        getRestAPIFactory().getRecordsAPI().completeRecord(recordFiled.getId());
+        assertStatusCode(CREATED);
+
+        STEP("Execute the disposition schedule steps.");
+        rmRolesAndActionsAPI.executeAction(getAdminUser().getUsername(), getAdminUser().getUsername(), recordFiled.getName(),
+                RM_ACTIONS.CUT_OFF);
+        rmRolesAndActionsAPI.executeAction(getAdminUser().getUsername(), getAdminUser().getUsername(), recordFiled.getName(),
+                RM_ACTIONS.DESTROY);
+
+        STEP("Check that it's possible to load the copy content.");
+        getNodeContent(copy.getId());
+        assertStatusCode(OK);
+
+        STEP("Clean up.");
+        dataSite.deleteSite(testSite);
+
+    }
+
+    /**
+     * <pre>
+     * Given a file that has version declared as record
+     * When the record is deleted
+     * Then it is still possible to view the content of the file
+     * </pre>
+     */
+    @Test (description = "Deleting record made from version doesn't delete the content for the file")
+    @AlfrescoTest (jira = "MNT-20145")
+    public void deleteVersionDeclaredAsRecord() throws Exception
+    {
+        STEP("Create a file.");
+        testSite = dataSite.usingAdmin().createPublicRandomSite();
+        FileModel testFile = dataContent.usingSite(testSite).createContent(CMISUtil.DocumentType.TEXT_PLAIN);
+
+        STEP("Declare file version as record.");
+        recordsAPI.declareDocumentVersionAsRecord(getAdminUser().getUsername(), getAdminUser().getPassword(), testSite.getId(),
+                testFile.getName());
+        UnfiledContainerChild unfiledContainerChild = getRestAPIFactory().getUnfiledContainersAPI()
+                                                                       .getUnfiledContainerChildren(UNFILED_RECORDS_CONTAINER_ALIAS)
+                                                                       .getEntries().stream()
+                                                                       .filter(child -> child.getEntry().getName()
+                                                                                             .startsWith(testFile.getName().substring(0, testFile.getName().indexOf("."))))
+                                                                       .findFirst()
+                                                                       .get().getEntry();
+
+        STEP("Delete the record.");
+        deleteAndVerify(unfiledContainerChild.getId());
+
+        STEP("Check that it's possible to load the file declared version as record.");
+        getNodeContent(testFile.getNodeRefWithoutVersion());
+        assertStatusCode(OK);
+
+        STEP("Clean up.");
+        dataSite.deleteSite(testSite);
+
+    }
+
+
     /**
      * Utility method to delete a record and verify successful deletion
      *
@@ -299,20 +447,20 @@ public class DeleteRecordTests extends BaseRMRestTest
         assertStatusCode(NO_CONTENT);
 
         // Try to get deleted record
-        recordsAPI.deleteRecord(recordId);
+        recordsAPI.getRecord(recordId);
         assertStatusCode(NOT_FOUND);
     }
 
     /**
-     * Copy a record to a record folder.
+     * Copy a node to a folder.
      *
-     * @param recordId The id of the record to copy.
-     * @param destinationFolder The id of the record folder to copy it to.
+     * @param nodeId The id of the node to copy.
+     * @param destinationFolder The id of the folder to copy it to.
      * @return The model returned by the copy API.
      */
-    private RestNodeModel copyRecord(String recordId, String destinationFolder)
+    private RestNodeModel copyNode(String nodeId, String destinationFolder)
     {
-        Node node = getNode(recordId);
+        Node node = getNode(nodeId);
         RestNodeBodyMoveCopyModel copyBody = new RestNodeBodyMoveCopyModel();
         copyBody.setTargetParentId(destinationFolder);
         try
diff --git a/rm-community/rm-community-repo/config/alfresco/module/org_alfresco_module_rm/rm-model-context.xml b/rm-community/rm-community-repo/config/alfresco/module/org_alfresco_module_rm/rm-model-context.xml
index 4cb71d28e0..ddd303c982 100644
--- a/rm-community/rm-community-repo/config/alfresco/module/org_alfresco_module_rm/rm-model-context.xml
+++ b/rm-community/rm-community-repo/config/alfresco/module/org_alfresco_module_rm/rm-model-context.xml
@@ -143,7 +143,6 @@
         <property name="recordService" ref="RecordService" />
         <property name="dispositionService" ref="DispositionService" />
         <property name="quickShareService" ref="QuickShareService"/>
-        <property name="fileFolderService" ref="FileFolderService"/>
    </bean>
 
    <bean id="rma.recordComponentIdentifier" class="org.alfresco.module.org_alfresco_module_rm.model.rma.aspect.RecordComponentIdentifierAspect" parent="rm.baseBehaviour">
diff --git a/rm-community/rm-community-repo/config/alfresco/module/org_alfresco_module_rm/rm-service-context.xml b/rm-community/rm-community-repo/config/alfresco/module/org_alfresco_module_rm/rm-service-context.xml
index 1d2f094680..32ea68a220 100644
--- a/rm-community/rm-community-repo/config/alfresco/module/org_alfresco_module_rm/rm-service-context.xml
+++ b/rm-community/rm-community-repo/config/alfresco/module/org_alfresco_module_rm/rm-service-context.xml
@@ -45,6 +45,7 @@
         <property name="authenticationUtil" ref="rm.authenticationUtil"/>
         <property name="transactionalResourceHelper" ref="rm.transactionalResourceHelper" />
         <property name="renditionService" ref="RenditionService" />
+        <property name="contentService" ref="ContentService" />
      </bean>
 
     <!-- Records Management Service Registry -->
diff --git a/rm-community/rm-community-repo/source/java/org/alfresco/module/org_alfresco_module_rm/model/rma/aspect/RecordAspect.java b/rm-community/rm-community-repo/source/java/org/alfresco/module/org_alfresco_module_rm/model/rma/aspect/RecordAspect.java
index b59181191b..441af631bb 100644
--- a/rm-community/rm-community-repo/source/java/org/alfresco/module/org_alfresco_module_rm/model/rma/aspect/RecordAspect.java
+++ b/rm-community/rm-community-repo/source/java/org/alfresco/module/org_alfresco_module_rm/model/rma/aspect/RecordAspect.java
@@ -51,13 +51,9 @@ import org.alfresco.repo.policy.annotation.BehaviourBean;
 import org.alfresco.repo.policy.annotation.BehaviourKind;
 import org.alfresco.repo.security.authentication.AuthenticationUtil;
 import org.alfresco.repo.security.authentication.AuthenticationUtil.RunAsWork;
-import org.alfresco.service.cmr.model.FileFolderService;
 import org.alfresco.service.cmr.quickshare.QuickShareService;
 import org.alfresco.service.cmr.repository.ChildAssociationRef;
 import org.alfresco.service.cmr.repository.ContentData;
-import org.alfresco.service.cmr.repository.ContentReader;
-import org.alfresco.service.cmr.repository.ContentService;
-import org.alfresco.service.cmr.repository.ContentWriter;
 import org.alfresco.service.cmr.repository.NodeRef;
 import org.alfresco.service.cmr.repository.ScriptService;
 import org.alfresco.service.namespace.QName;
@@ -98,9 +94,6 @@ public class RecordAspect extends    AbstractDisposableItem
     /** quickShare service */
     private QuickShareService quickShareService;
 
-    /** File folder service */
-    private FileFolderService fileFolderService;
-
     /** I18N */
     private static final String MSG_CANNOT_UPDATE_RECORD_CONTENT = "rm.service.update-record-content";
 
@@ -137,15 +130,6 @@ public class RecordAspect extends    AbstractDisposableItem
         this.quickShareService = quickShareService;
     }
 
-    /**
-     *
-     * @param fileFolderService file folder service
-     */
-    public void setFileFolderService(FileFolderService fileFolderService)
-    {
-        this.fileFolderService = fileFolderService;
-    }
-
     /**
      * Behaviour to ensure renditions have the appropriate extended security.
      *
@@ -351,11 +335,7 @@ public class RecordAspect extends    AbstractDisposableItem
     /**
      * On copy complete behaviour for record aspect.
      *
-     * @param classRef
-     * @param sourceNodeRef
-     * @param targetNodeRef
-     * @param copyToNewNode
-     * @param copyMap
+     * @see org.alfresco.repo.copy.CopyServicePolicies.OnCopyCompletePolicy#onCopyComplete(QName, NodeRef, NodeRef, boolean, Map)
      */
     @Override
     @Behaviour
@@ -376,12 +356,7 @@ public class RecordAspect extends    AbstractDisposableItem
             extendedSecurityService.remove(targetNodeRef);
 
             //create a new content URL for the copy
-            ContentReader reader = fileFolderService.getReader(targetNodeRef);
-            if (reader != null)
-            {
-                ContentWriter writer = fileFolderService.getWriter(targetNodeRef);
-                writer.putContent(reader);
-            }
+            createNewContentURL(targetNodeRef);
         }
     }
 
@@ -402,6 +377,7 @@ public class RecordAspect extends    AbstractDisposableItem
 
     /**
      * Behaviour to remove the shared link before declare a record
+     * and to create new bin if the node is a copy or has copies
      *
      * @see org.alfresco.repo.node.NodeServicePolicies.BeforeAddAspectPolicy#beforeAddAspect(org.alfresco.service.cmr.repository.NodeRef,
      *      org.alfresco.service.namespace.QName)
@@ -421,6 +397,26 @@ public class RecordAspect extends    AbstractDisposableItem
                     quickShareService.unshareContent(sharedId);
                 }
 
+                // if the node has a copy or is a copy of an existing node
+                if (!nodeService.getTargetAssocs(nodeRef, ContentModel.ASSOC_ORIGINAL).isEmpty() ||
+                        !nodeService.getSourceAssocs(nodeRef, ContentModel.ASSOC_ORIGINAL).isEmpty())
+                {
+                    //disable versioning and auditing
+                    behaviourFilter.disableBehaviour(ContentModel.ASPECT_AUDITABLE);
+                    behaviourFilter.disableBehaviour(ContentModel.ASPECT_VERSIONABLE);
+                    try
+                    {
+                        //create a new content URL for the copy/original node
+                        createNewContentURL(nodeRef);
+                    }
+                    finally
+                    {
+                        //enable versioning and auditing
+                        behaviourFilter.enableBehaviour(ContentModel.ASPECT_AUDITABLE);
+                        behaviourFilter.enableBehaviour(ContentModel.ASPECT_VERSIONABLE);
+                    }
+                }
+
                 return null;
             }
         }, AuthenticationUtil.getSystemUserName());
diff --git a/rm-community/rm-community-repo/source/java/org/alfresco/module/org_alfresco_module_rm/model/rma/aspect/VersionRecordAspect.java b/rm-community/rm-community-repo/source/java/org/alfresco/module/org_alfresco_module_rm/model/rma/aspect/VersionRecordAspect.java
index 9437c6c6ef..f4712a8995 100644
--- a/rm-community/rm-community-repo/source/java/org/alfresco/module/org_alfresco_module_rm/model/rma/aspect/VersionRecordAspect.java
+++ b/rm-community/rm-community-repo/source/java/org/alfresco/module/org_alfresco_module_rm/model/rma/aspect/VersionRecordAspect.java
@@ -34,12 +34,14 @@ import org.alfresco.module.org_alfresco_module_rm.relationship.Relationship;
 import org.alfresco.module.org_alfresco_module_rm.relationship.RelationshipService;
 import org.alfresco.module.org_alfresco_module_rm.version.RecordableVersionService;
 import org.alfresco.repo.node.NodeServicePolicies;
+import org.alfresco.repo.policy.Behaviour.NotificationFrequency;
 import org.alfresco.repo.policy.annotation.Behaviour;
 import org.alfresco.repo.policy.annotation.BehaviourBean;
 import org.alfresco.repo.policy.annotation.BehaviourKind;
 import org.alfresco.repo.security.authentication.AuthenticationUtil.RunAsWork;
 import org.alfresco.service.cmr.repository.NodeRef;
 import org.alfresco.service.cmr.version.Version;
+import org.alfresco.service.namespace.QName;
 
 /**
  * rmv:versionRecord behaviour bean
@@ -52,14 +54,15 @@ import org.alfresco.service.cmr.version.Version;
    defaultType = "rmv:versionRecord"
 )
 public class VersionRecordAspect extends    BaseBehaviourBean
-                                 implements NodeServicePolicies.BeforeDeleteNodePolicy
+                                 implements NodeServicePolicies.BeforeAddAspectPolicy,
+                                            NodeServicePolicies.BeforeDeleteNodePolicy
 {
     /** recordable version service */
     private RecordableVersionService recordableVersionService;
     
     /** relationship service */
     private RelationshipService relationshipService;
-    
+
     /**
      * @param recordableVersionService  recordable version service
      */
@@ -75,7 +78,7 @@ public class VersionRecordAspect extends    BaseBehaviourBean
     {
         this.relationshipService = relationshipService;
     }
-    
+
     /**
      * If the record is a version record then delete the associated version entry
      * 
@@ -129,4 +132,18 @@ public class VersionRecordAspect extends    BaseBehaviourBean
             });
         }         
     }
+
+    /**
+     * Behaviour to duplicate the bin before declaring a version record
+     *
+     * @see org.alfresco.repo.node.NodeServicePolicies.BeforeAddAspectPolicy#beforeAddAspect(org.alfresco.service.cmr.repository.NodeRef,
+     *      org.alfresco.service.namespace.QName)
+     */
+    @Override
+    @Behaviour(kind = BehaviourKind.CLASS, notificationFrequency = NotificationFrequency.FIRST_EVENT)
+    public void beforeAddAspect(NodeRef nodeRef, QName qName)
+    {
+        //create a new content URL for the version record
+        createNewContentURL(nodeRef);
+    }
 }
diff --git a/rm-community/rm-community-repo/source/java/org/alfresco/module/org_alfresco_module_rm/util/ServiceBaseImpl.java b/rm-community/rm-community-repo/source/java/org/alfresco/module/org_alfresco_module_rm/util/ServiceBaseImpl.java
index 8f45c69441..8fc5094b05 100644
--- a/rm-community/rm-community-repo/source/java/org/alfresco/module/org_alfresco_module_rm/util/ServiceBaseImpl.java
+++ b/rm-community/rm-community-repo/source/java/org/alfresco/module/org_alfresco_module_rm/util/ServiceBaseImpl.java
@@ -31,6 +31,7 @@ import java.util.Map;
 import java.util.Set;
 import java.util.WeakHashMap;
 
+import org.alfresco.model.ContentModel;
 import org.alfresco.module.org_alfresco_module_rm.fileplan.FilePlanComponentKind;
 import org.alfresco.module.org_alfresco_module_rm.fileplan.FilePlanService;
 import org.alfresco.module.org_alfresco_module_rm.hold.HoldService;
@@ -38,6 +39,9 @@ import org.alfresco.module.org_alfresco_module_rm.model.RecordsManagementModel;
 import org.alfresco.service.cmr.dictionary.DictionaryService;
 import org.alfresco.service.cmr.rendition.RenditionService;
 import org.alfresco.service.cmr.repository.ChildAssociationRef;
+import org.alfresco.service.cmr.repository.ContentReader;
+import org.alfresco.service.cmr.repository.ContentService;
+import org.alfresco.service.cmr.repository.ContentWriter;
 import org.alfresco.service.cmr.repository.NodeRef;
 import org.alfresco.service.cmr.repository.NodeService;
 import org.alfresco.service.namespace.QName;
@@ -71,9 +75,12 @@ public class ServiceBaseImpl implements RecordsManagementModel, ApplicationConte
 
     /** authentication helper */
     protected AuthenticationUtil authenticationUtil;
-    
+
     /** transactional resource helper */
-    protected TransactionalResourceHelper transactionalResourceHelper; 
+    protected TransactionalResourceHelper transactionalResourceHelper;
+
+    /** Content service */
+    protected ContentService contentService;
 
     /**
      * @see org.springframework.context.ApplicationContextAware#setApplicationContext(org.springframework.context.ApplicationContext)
@@ -124,6 +131,16 @@ public class ServiceBaseImpl implements RecordsManagementModel, ApplicationConte
         this.transactionalResourceHelper = transactionalResourceHelper;
     }
 
+    /**
+     * Set the content service
+     *
+     * @param contentService content service
+     */
+    public void setContentService(ContentService contentService)
+    {
+        this.contentService = contentService;
+    }
+
     /**
      * Helper to get internal node service.
      * <p>
@@ -537,4 +554,30 @@ public class ServiceBaseImpl implements RecordsManagementModel, ApplicationConte
         result.add(nodeService.getType(nodeRef));
         return result;
     }
+
+    /**
+     * Helper to update the given content property for the node
+     *
+     * @param nodeRef the node
+     * @param contentProperty the property to be updated
+     */
+    protected void updateContentProperty(NodeRef nodeRef, QName contentProperty)
+    {
+        ContentReader reader = contentService.getReader(nodeRef, contentProperty);
+        if (reader != null)
+        {
+            ContentWriter writer = contentService.getWriter(nodeRef, contentProperty, true);
+            writer.putContent(reader);
+        }
+    }
+
+    /**
+     * Helper to create a new content URL for the node
+     *
+     * @param nodeRef the node
+     */
+    protected void createNewContentURL(NodeRef nodeRef)
+    {
+        updateContentProperty(nodeRef, ContentModel.PROP_CONTENT);
+    }
 }
diff --git a/rm-community/rm-community-repo/test/java/org/alfresco/module/org_alfresco_module_rm/test/integration/version/RecordableVersionsBaseTest.java b/rm-community/rm-community-repo/test/java/org/alfresco/module/org_alfresco_module_rm/test/integration/version/RecordableVersionsBaseTest.java
index c75997e0e1..2727c0b730 100644
--- a/rm-community/rm-community-repo/test/java/org/alfresco/module/org_alfresco_module_rm/test/integration/version/RecordableVersionsBaseTest.java
+++ b/rm-community/rm-community-repo/test/java/org/alfresco/module/org_alfresco_module_rm/test/integration/version/RecordableVersionsBaseTest.java
@@ -238,7 +238,16 @@ public abstract class RecordableVersionsBaseTest extends BaseRMTestCase implemen
             if (frozenProperties.containsKey(beforePropertyName))
             {
                 Serializable frozenValue = frozenProperties.get(beforePropertyName);
-                assertEquals("Frozen property " + beforePropertyName.getLocalName() + " value is incorrect.", entry.getValue(), frozenValue);
+                if(beforePropertyName.equals(ContentModel.PROP_CONTENT))
+                {
+                    assertTrue("Content property value should be different.",
+                            entry.getValue() != frozenValue);
+                }
+                else
+                {
+                    assertEquals("Frozen property " + beforePropertyName.getLocalName() + " value is incorrect.",
+                            entry.getValue(), frozenValue);
+                }
                 cloneFrozenProperties.remove(beforePropertyName);               
             }
             else if (!PROP_FILE_PLAN.equals(beforePropertyName) &&
diff --git a/rm-community/rm-community-repo/unit-test/java/org/alfresco/module/org_alfresco_module_rm/model/rma/aspect/RecordAspectUnitTest.java b/rm-community/rm-community-repo/unit-test/java/org/alfresco/module/org_alfresco_module_rm/model/rma/aspect/RecordAspectUnitTest.java
new file mode 100644
index 0000000000..730da904f3
--- /dev/null
+++ b/rm-community/rm-community-repo/unit-test/java/org/alfresco/module/org_alfresco_module_rm/model/rma/aspect/RecordAspectUnitTest.java
@@ -0,0 +1,176 @@
+/*
+ * #%L
+ * Alfresco Records Management Module
+ * %%
+ * Copyright (C) 2005 - 2019 Alfresco Software Limited
+ * %%
+ * This file is part of the Alfresco software.
+ * -
+ * If the software was purchased under a paid Alfresco license, the terms of
+ * the paid license agreement will prevail.  Otherwise, the software is
+ * provided under the following open source license terms:
+ * -
+ * Alfresco is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ * -
+ * Alfresco is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Lesser General Public License for more details.
+ * -
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with Alfresco. If not, see <http://www.gnu.org/licenses/>.
+ * #L%
+ */
+package org.alfresco.module.org_alfresco_module_rm.model.rma.aspect;
+
+import static java.util.Arrays.asList;
+import static java.util.Collections.emptyList;
+
+import static org.alfresco.module.org_alfresco_module_rm.model.RecordsManagementModel.ASPECT_RECORD;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.mockito.MockitoAnnotations.initMocks;
+
+import org.alfresco.model.ContentModel;
+import org.alfresco.module.org_alfresco_module_rm.security.ExtendedSecurityService;
+import org.alfresco.repo.policy.BehaviourFilter;
+import org.alfresco.service.cmr.repository.AssociationRef;
+import org.alfresco.service.cmr.repository.ContentReader;
+import org.alfresco.service.cmr.repository.ContentService;
+import org.alfresco.service.cmr.repository.ContentWriter;
+import org.alfresco.service.cmr.repository.NodeRef;
+import org.alfresco.service.cmr.repository.NodeService;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+
+/**
+ * Unit tests for the {@link RecordAspect}.
+ *
+ * @author Claudia Agache
+ */
+public class RecordAspectUnitTest
+{
+    private static final NodeRef NODE_REF = new NodeRef("node://Ref/");
+    private static final NodeRef COPY_REF = new NodeRef("node://Copy/");
+    private static final AssociationRef SOURCE_ASSOC_REF = new AssociationRef(COPY_REF, ContentModel.ASSOC_ORIGINAL,
+            NODE_REF);
+    private static final AssociationRef TARGET_ASSOC_REF = new AssociationRef(NODE_REF, ContentModel.ASSOC_ORIGINAL,
+            COPY_REF);
+
+    @InjectMocks
+    private RecordAspect recordAspect;
+    @Mock
+    private NodeService mockNodeService;
+    @Mock
+    private BehaviourFilter mockBehaviorFilter;
+    @Mock
+    private ContentService mockContentService;
+    @Mock
+    private ContentReader mockContentReader;
+    @Mock
+    private ContentWriter mockContentWriter;
+    @Mock
+    private ExtendedSecurityService mockExtendedSecurityService;
+
+    @Before
+    public void setUp()
+    {
+        initMocks(this);
+    }
+
+    /** Check that the bin is duplicated before adding the aspect if the file has a copy. */
+    @Test
+    public void testDuplicateBinBeforeAddingAspectForFileWithCopy()
+    {
+        when(mockNodeService.getSourceAssocs(NODE_REF, ContentModel.ASSOC_ORIGINAL)).thenReturn(asList(SOURCE_ASSOC_REF));
+        when(mockContentService.getReader(NODE_REF, ContentModel.PROP_CONTENT)).thenReturn(mockContentReader);
+        when(mockContentService.getWriter(NODE_REF, ContentModel.PROP_CONTENT, true)).thenReturn(mockContentWriter);
+
+        recordAspect.beforeAddAspect(NODE_REF, ASPECT_RECORD);
+
+         verifyBeforeAddAspectMethodsInvocations(1);
+    }
+
+    /** Check that the bin is duplicated before adding the aspect if the file is a copy. */
+    @Test
+    public void testDuplicateBinBeforeAddingAspectForCopy()
+    {
+        when(mockNodeService.getTargetAssocs(NODE_REF, ContentModel.ASSOC_ORIGINAL)).thenReturn(asList(TARGET_ASSOC_REF));
+        when(mockContentService.getReader(NODE_REF, ContentModel.PROP_CONTENT)).thenReturn(mockContentReader);
+        when(mockContentService.getWriter(NODE_REF, ContentModel.PROP_CONTENT, true)).thenReturn(mockContentWriter);
+
+        recordAspect.beforeAddAspect(NODE_REF, ASPECT_RECORD);
+
+         verifyBeforeAddAspectMethodsInvocations(1);
+    }
+
+    /** Check that no content bin is created if the file does not have content. */
+    @Test
+    public void testBeforeAddAspectOnFileWithNoContent()
+    {
+        when(mockNodeService.getTargetAssocs(NODE_REF, ContentModel.ASSOC_ORIGINAL)).thenReturn(asList(TARGET_ASSOC_REF));
+        when(mockContentService.getReader(NODE_REF, ContentModel.PROP_CONTENT)).thenReturn(null);
+
+        recordAspect.beforeAddAspect(NODE_REF, ASPECT_RECORD);
+
+        verify(mockBehaviorFilter, times(1)).disableBehaviour(eq(ContentModel.ASPECT_AUDITABLE));
+        verify(mockBehaviorFilter, times(1)).disableBehaviour(eq(ContentModel.ASPECT_VERSIONABLE));
+        verify(mockContentService, times(1)).getReader(NODE_REF, ContentModel.PROP_CONTENT);
+        verify(mockContentService, never()).getWriter(NODE_REF, ContentModel.PROP_CONTENT, true);
+        verify(mockBehaviorFilter, times(1)).enableBehaviour(eq(ContentModel.ASPECT_AUDITABLE));
+        verify(mockBehaviorFilter, times(1)).enableBehaviour(eq(ContentModel.ASPECT_VERSIONABLE));
+    }
+
+    /** Check that the bin is not duplicated before adding the aspect if the node has no copies. */
+    @Test
+    public void testNotDuplicateBinForFileWithNoCopies()
+    {
+        when(mockNodeService.getSourceAssocs(NODE_REF, ContentModel.ASSOC_ORIGINAL)).thenReturn(emptyList());
+        when(mockNodeService.getTargetAssocs(NODE_REF, ContentModel.ASSOC_ORIGINAL)).thenReturn(emptyList());
+
+        recordAspect.beforeAddAspect(NODE_REF, ASPECT_RECORD);
+
+        verifyBeforeAddAspectMethodsInvocations(0);
+    }
+
+    /** Check that the bin is duplicated when copying a record. */
+    @Test
+    public void testDuplicateBinWhenCopyingRecord()
+    {
+        when(mockNodeService.exists(COPY_REF)).thenReturn(true);
+        when(mockNodeService.hasAspect(COPY_REF, ASPECT_RECORD)).thenReturn(true);
+        when(mockContentService.getReader(COPY_REF, ContentModel.PROP_CONTENT)).thenReturn(mockContentReader);
+        when(mockContentService.getWriter(COPY_REF, ContentModel.PROP_CONTENT, true)).thenReturn(mockContentWriter);
+
+        recordAspect.onCopyComplete(null, NODE_REF, COPY_REF, true, null);
+
+        verify(mockExtendedSecurityService, times(1)).remove(COPY_REF);
+        verify(mockContentService, times(1)).getReader(COPY_REF, ContentModel.PROP_CONTENT);
+        verify(mockContentService, times(1)).getWriter(COPY_REF, ContentModel.PROP_CONTENT, true);
+        verify(mockContentWriter, times(1)).putContent(mockContentReader);
+    }
+
+    /**
+     * Helper to verify beforeAddAspect methods invocations
+     *
+     * @param wantedNumberOfInvocations wanted number of invocations for each method
+     */
+    private void verifyBeforeAddAspectMethodsInvocations(int wantedNumberOfInvocations)
+    {
+        verify(mockBehaviorFilter, times(wantedNumberOfInvocations)).disableBehaviour(eq(ContentModel.ASPECT_AUDITABLE));
+        verify(mockBehaviorFilter, times(wantedNumberOfInvocations)).disableBehaviour(eq(ContentModel.ASPECT_VERSIONABLE));
+        verify(mockContentService, times(wantedNumberOfInvocations)).getReader(NODE_REF, ContentModel.PROP_CONTENT);
+        verify(mockContentService, times(wantedNumberOfInvocations)).getWriter(NODE_REF, ContentModel.PROP_CONTENT, true);
+        verify(mockContentWriter, times(wantedNumberOfInvocations)).putContent(mockContentReader);
+        verify(mockBehaviorFilter, times(wantedNumberOfInvocations)).enableBehaviour(eq(ContentModel.ASPECT_AUDITABLE));
+        verify(mockBehaviorFilter, times(wantedNumberOfInvocations)).enableBehaviour(eq(ContentModel.ASPECT_VERSIONABLE));
+    }
+}