Allow migration to be performed in stages in ACM intermediary

Allow migration to be performed in stages,
and replace newElement and removedElement with an Enum
in CompositionElementDto and InstanceElementDto.

Issue-ID: POLICY-5091
Change-Id: I2d66abc453776fd708fc18fd9019ca248f8d2eee
Signed-off-by: FrancescoFioraEst <francesco.fiora@est.tech>

9 files changed, 277 insertions, 101 deletions

diff --git a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/api/CompositionElementDto.java b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/api/CompositionElementDto.java
index 51800cef1..50699e22c 100644
--- a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/api/CompositionElementDto.java
+++ b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/api/CompositionElementDto.java
@@ -26,11 +26,11 @@ import org.onap.policy.models.tosca.authorative.concepts.ToscaConceptIdentifier;
 
 public record CompositionElementDto(UUID compositionId, ToscaConceptIdentifier elementDefinitionId,
                                     Map<String, Object> inProperties, Map<String, Object> outProperties,
-                                    boolean newElement, boolean removedElement) {
+                                    ElementState state) {
 
     public CompositionElementDto(UUID compositionId, ToscaConceptIdentifier elementDefinitionId,
                                  Map<String, Object> inProperties, Map<String, Object> outProperties) {
-        this(compositionId, elementDefinitionId, inProperties, outProperties, false, false);
+        this(compositionId, elementDefinitionId, inProperties, outProperties, ElementState.PRESENT);
 
     }
 }
diff --git a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/api/ElementState.java b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/api/ElementState.java
new file mode 100644
index 000000000..afb01d014
--- /dev/null
+++ b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/api/ElementState.java
@@ -0,0 +1,28 @@
+/*-
+ * ============LICENSE_START=======================================================
+ *  Copyright (C) 2024 Nordix Foundation.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.clamp.acm.participant.intermediary.api;
+
+public enum ElementState {
+    PRESENT,
+    NOT_PRESENT,
+    REMOVED,
+    NEW
+}
diff --git a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/api/InstanceElementDto.java b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/api/InstanceElementDto.java
index 65a83e222..b4fdefbf3 100644
--- a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/api/InstanceElementDto.java
+++ b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/api/InstanceElementDto.java
@@ -26,10 +26,10 @@ import org.onap.policy.models.tosca.authorative.concepts.ToscaServiceTemplate;
 
 public record InstanceElementDto(UUID instanceId, UUID elementId, ToscaServiceTemplate toscaServiceTemplateFragment,
                                  Map<String, Object> inProperties, Map<String, Object> outProperties,
-                                 boolean newElement, boolean removedElement) {
+                                 ElementState state) {
 
     public InstanceElementDto(UUID instanceId, UUID elementId, ToscaServiceTemplate toscaServiceTemplateFragment,
                               Map<String, Object> inProperties, Map<String, Object> outProperties) {
-        this(instanceId, elementId, toscaServiceTemplateFragment, inProperties, outProperties, false, false);
+        this(instanceId, elementId, toscaServiceTemplateFragment, inProperties, outProperties, ElementState.PRESENT);
     }
 }
diff --git a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/api/impl/ParticipantIntermediaryApiImpl.java b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/api/impl/ParticipantIntermediaryApiImpl.java
index e1bd06465..233b55926 100644
--- a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/api/impl/ParticipantIntermediaryApiImpl.java
+++ b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/api/impl/ParticipantIntermediaryApiImpl.java
@@ -59,7 +59,8 @@ public class ParticipantIntermediaryApiImpl implements ParticipantIntermediaryAp
     @Override
     public void updateAutomationCompositionElementStage(UUID instance, UUID elementId,
             StateChangeResult stateChangeResult, int stage, String message) {
-        //
+        automationCompositionHandler.updateAutomationCompositionElementStage(instance, elementId, stateChangeResult,
+            stage, message);
     }
 
     @Override
diff --git a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/handler/AcSubStateHandler.java b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/handler/AcSubStateHandler.java
index 3044080af..1dbf2c935 100644
--- a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/handler/AcSubStateHandler.java
+++ b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/handler/AcSubStateHandler.java
@@ -21,16 +21,21 @@
 package org.onap.policy.clamp.acm.participant.intermediary.handler;
 
 import java.util.List;
+import java.util.Map;
 import java.util.UUID;
 import lombok.RequiredArgsConstructor;
+import org.onap.policy.clamp.acm.participant.intermediary.api.CompositionElementDto;
+import org.onap.policy.clamp.acm.participant.intermediary.api.ElementState;
 import org.onap.policy.clamp.acm.participant.intermediary.api.InstanceElementDto;
 import org.onap.policy.clamp.models.acm.concepts.AcElementDeploy;
 import org.onap.policy.clamp.models.acm.concepts.AutomationComposition;
+import org.onap.policy.clamp.models.acm.concepts.AutomationCompositionElement;
 import org.onap.policy.clamp.models.acm.concepts.DeployState;
 import org.onap.policy.clamp.models.acm.concepts.SubState;
 import org.onap.policy.clamp.models.acm.messages.kafka.participant.AutomationCompositionMigration;
 import org.onap.policy.clamp.models.acm.messages.kafka.participant.AutomationCompositionPrepare;
 import org.onap.policy.clamp.models.acm.utils.AcmUtils;
+import org.onap.policy.models.tosca.authorative.concepts.ToscaServiceTemplate;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Component;
@@ -77,24 +82,64 @@ public class AcSubStateHandler {
         var acElementList = automationComposition.getElements();
         for (var acElement : acElements) {
             var element = acElementList.get(acElement.getId());
-            element.setSubState(SubState.MIGRATION_PRECHECKING);
+            if (element != null) {
+                element.setSubState(SubState.MIGRATION_PRECHECKING);
+            }
         }
         var acCopyMigrateTo = new AutomationComposition(automationComposition);
-        acElementList = acCopyMigrateTo.getElements();
+        var acElementCopyList = acCopyMigrateTo.getElements();
         for (var acElement : acElements) {
-            var element = acElementList.get(acElement.getId());
-            AcmUtils.recursiveMerge(element.getProperties(), acElement.getProperties());
-            element.setDefinition(acElement.getDefinition());
+            var element = acElementCopyList.get(acElement.getId());
+            if (element != null) {
+                AcmUtils.recursiveMerge(element.getProperties(), acElement.getProperties());
+                element.setDefinition(acElement.getDefinition());
+            } else {
+                element = CacheProvider.createAutomationCompositionElement(acElement);
+                element.setSubState(SubState.MIGRATION_PRECHECKING);
+                acElementCopyList.put(element.getId(), element);
+            }
         }
+        var toDelete = acElementCopyList.values().stream()
+                .filter(el -> !SubState.MIGRATION_PRECHECKING.equals(el.getSubState()))
+                .map(AutomationCompositionElement::getId)
+                .toList();
+        toDelete.forEach(acElementCopyList::remove);
 
         var compositionElementTargetMap = cacheProvider.getCompositionElementDtoMap(acCopyMigrateTo,
             compositionTargetId);
         var instanceElementMigrateMap = cacheProvider.getInstanceElementDtoMap(acCopyMigrateTo);
 
         for (var acElement : acElements) {
-            listener.migratePrecheck(messageId, compositionElementMap.get(acElement.getId()),
-                compositionElementTargetMap.get(acElement.getId()),
-                instanceElementMap.get(acElement.getId()), instanceElementMigrateMap.get(acElement.getId()));
+            var compositionElement = compositionElementMap.get(acElement.getId());
+            var compositionElementTarget = compositionElementTargetMap.get(acElement.getId());
+            var instanceElement = instanceElementMap.get(acElement.getId());
+            var instanceElementMigrate = instanceElementMigrateMap.get(acElement.getId());
+
+            if (instanceElement == null) {
+                // new element scenario
+                compositionElement = new CompositionElementDto(automationComposition.getCompositionId(),
+                        acElement.getDefinition(), Map.of(), Map.of(), ElementState.NOT_PRESENT);
+                instanceElement = new InstanceElementDto(automationComposition.getInstanceId(), acElement.getId(),
+                        new ToscaServiceTemplate(), Map.of(), Map.of(), ElementState.NOT_PRESENT);
+                compositionElementTarget = CacheProvider.changeStateToNew(compositionElementTarget);
+                instanceElementMigrate = CacheProvider.changeStateToNew(instanceElementMigrate);
+            }
+
+            listener.migratePrecheck(messageId, compositionElement, compositionElementTarget,
+                    instanceElement, instanceElementMigrate);
+        }
+
+        for (var elementId : toDelete) {
+            var compositionDtoTarget =
+                    new CompositionElementDto(compositionTargetId,
+                            automationComposition.getElements().get(elementId).getDefinition(),
+                            Map.of(), Map.of(), ElementState.REMOVED);
+            var instanceDtoTarget =
+                    new InstanceElementDto(automationComposition.getInstanceId(), elementId,
+                            null, Map.of(), Map.of(), ElementState.REMOVED);
+
+            listener.migratePrecheck(messageId, compositionElementMap.get(elementId), compositionDtoTarget,
+                    instanceElementMap.get(elementId), instanceDtoTarget);
         }
     }
 
diff --git a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/handler/AutomationCompositionHandler.java b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/handler/AutomationCompositionHandler.java
index 6c560e7fa..6d94efb0f 100644
--- a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/handler/AutomationCompositionHandler.java
+++ b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/handler/AutomationCompositionHandler.java
@@ -21,13 +21,13 @@
 
 package org.onap.policy.clamp.acm.participant.intermediary.handler;
 
-
-import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
+import java.util.stream.Collectors;
 import lombok.RequiredArgsConstructor;
 import org.onap.policy.clamp.acm.participant.intermediary.api.CompositionElementDto;
+import org.onap.policy.clamp.acm.participant.intermediary.api.ElementState;
 import org.onap.policy.clamp.acm.participant.intermediary.api.InstanceElementDto;
 import org.onap.policy.clamp.acm.participant.intermediary.comm.ParticipantMessagePublisher;
 import org.onap.policy.clamp.models.acm.concepts.AcElementDeploy;
@@ -47,7 +47,6 @@ import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantMe
 import org.onap.policy.clamp.models.acm.messages.kafka.participant.PropertiesUpdate;
 import org.onap.policy.clamp.models.acm.messages.rest.instantiation.DeployOrder;
 import org.onap.policy.clamp.models.acm.utils.AcmUtils;
-import org.onap.policy.models.tosca.authorative.concepts.ToscaServiceTemplate;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Component;
@@ -124,8 +123,7 @@ public class AutomationCompositionHandler {
                         updateMsg.getAutomationCompositionId());
                 automationComposition.setDeployState(DeployState.UPDATING);
                 var acCopy = new AutomationComposition(automationComposition);
-                updateExistingElementsOnThisParticipant(updateMsg.getAutomationCompositionId(), participantDeploy,
-                        DeployState.UPDATING);
+                updateExistingElementsOnThisParticipant(updateMsg.getAutomationCompositionId(), participantDeploy);
 
                 callParticipantUpdateProperty(updateMsg.getMessageId(), participantDeploy.getAcElementList(), acCopy);
             }
@@ -188,56 +186,56 @@ public class AutomationCompositionHandler {
         }
     }
 
-    private void updateExistingElementsOnThisParticipant(UUID instanceId, ParticipantDeploy participantDeploy,
-        DeployState deployState) {
-        var acElementList = cacheProvider.getAutomationComposition(instanceId).getElements();
+    private void migrateExistingElementsOnThisParticipant(UUID instanceId, UUID compositionTargetId,
+        ParticipantDeploy participantDeploy, int stage) {
+        var automationComposition = cacheProvider.getAutomationComposition(instanceId);
+        var acElementList = automationComposition.getElements();
         for (var element : participantDeploy.getAcElementList()) {
-            var acElement = acElementList.get(element.getId());
-            if (acElement == null && deployState.equals(DeployState.MIGRATING)) {
-                var newElement = new AutomationCompositionElement();
-                newElement.setId(element.getId());
-                newElement.setParticipantId(participantDeploy.getParticipantId());
-                newElement.setDefinition(element.getDefinition());
-                newElement.setDeployState(deployState);
-                newElement.setSubState(SubState.NONE);
-                newElement.setLockState(LockState.LOCKED);
-                newElement.setProperties(element.getProperties());
+            var compositionInProperties =
+                    cacheProvider.getCommonProperties(compositionTargetId, element.getDefinition());
+            var stageSet = ParticipantUtils.findStageSet(compositionInProperties);
+            if (stageSet.contains(stage)) {
+                var acElement = acElementList.get(element.getId());
+                if (acElement == null) {
+                    var newElement = CacheProvider.createAutomationCompositionElement(element);
+                    newElement.setParticipantId(participantDeploy.getParticipantId());
+                    newElement.setDeployState(DeployState.MIGRATING);
+                    newElement.setLockState(LockState.LOCKED);
+                    newElement.setStage(stage);
 
-                acElementList.put(element.getId(), newElement);
-                LOGGER.info("New Ac Element with id {} is added in Migration", element.getId());
-            } else if (acElement != null) {
-                AcmUtils.recursiveMerge(acElement.getProperties(), element.getProperties());
-                acElement.setDeployState(deployState);
-                acElement.setSubState(SubState.NONE);
-                acElement.setDefinition(element.getDefinition());
+                    acElementList.put(element.getId(), newElement);
+                    LOGGER.info("New Ac Element with id {} is added in Migration", element.getId());
+                } else {
+                    AcmUtils.recursiveMerge(acElement.getProperties(), element.getProperties());
+                    acElement.setDeployState(DeployState.MIGRATING);
+                    acElement.setStage(stage);
+                    acElement.setDefinition(element.getDefinition());
+                }
             }
         }
-        if (deployState.equals(DeployState.MIGRATING)) {
-            // Check for missing elements and remove them from cache
-            List<UUID> elementsToRemove = findElementsToRemove(participantDeploy.getAcElementList(), acElementList);
-            for (UUID key : elementsToRemove) {
-                acElementList.remove(key);
-                LOGGER.info("Element with id {} is removed in Migration", key);
-            }
+        // Check for missing elements and remove them from cache
+        var elementsToRemove = findElementsToRemove(participantDeploy.getAcElementList(), acElementList);
+        for (var key : elementsToRemove) {
+            acElementList.remove(key);
+            LOGGER.info("Element with id {} is removed in Migration", key);
+        }
+    }
+
+    private void updateExistingElementsOnThisParticipant(UUID instanceId, ParticipantDeploy participantDeploy) {
+        var acElementList = cacheProvider.getAutomationComposition(instanceId).getElements();
+        for (var element : participantDeploy.getAcElementList()) {
+            var acElement = acElementList.get(element.getId());
+            AcmUtils.recursiveMerge(acElement.getProperties(), element.getProperties());
+            acElement.setDeployState(DeployState.UPDATING);
+            acElement.setSubState(SubState.NONE);
+            acElement.setDefinition(element.getDefinition());
         }
     }
 
     private List<UUID> findElementsToRemove(List<AcElementDeploy> acElementDeployList, Map<UUID,
             AutomationCompositionElement> acElementList) {
-        List<UUID> elementsToRemove = new ArrayList<>();
-        for (var elementInCache : acElementList.entrySet()) {
-            boolean found = false;
-            for (var element : acElementDeployList) {
-                if (element.getId().equals(elementInCache.getValue().getId())) {
-                    found = true;
-                    break;
-                }
-            }
-            if (!found) {
-                elementsToRemove.add(elementInCache.getKey());
-            }
-        }
-        return elementsToRemove;
+        var acElementDeploySet = acElementDeployList.stream().map(AcElementDeploy::getId).collect(Collectors.toSet());
+        return acElementList.keySet().stream().filter(id -> !acElementDeploySet.contains(id)).toList();
     }
 
     /**
@@ -311,17 +309,17 @@ public class AutomationCompositionHandler {
         for (var participantDeploy : migrationMsg.getParticipantUpdatesList()) {
             if (cacheProvider.getParticipantId().equals(participantDeploy.getParticipantId())) {
 
-                updateExistingElementsOnThisParticipant(migrationMsg.getAutomationCompositionId(), participantDeploy,
-                    DeployState.MIGRATING);
+                migrateExistingElementsOnThisParticipant(migrationMsg.getAutomationCompositionId(),
+                        migrationMsg.getCompositionTargetId(), participantDeploy, migrationMsg.getStage());
 
                 callParticipantMigrate(migrationMsg.getMessageId(), participantDeploy.getAcElementList(),
-                    acCopy, migrationMsg.getCompositionTargetId());
+                    acCopy, migrationMsg.getCompositionTargetId(), migrationMsg.getStage());
             }
         }
     }
 
     private void callParticipantMigrate(UUID messageId, List<AcElementDeploy> acElements,
-            AutomationComposition acCopy, UUID compositionTargetId) {
+            AutomationComposition acCopy, UUID compositionTargetId, int stage) {
         var compositionElementMap = cacheProvider.getCompositionElementDtoMap(acCopy);
         var instanceElementMap = cacheProvider.getInstanceElementDtoMap(acCopy);
         var automationComposition = cacheProvider.getAutomationComposition(acCopy.getInstanceId());
@@ -331,32 +329,46 @@ public class AutomationCompositionHandler {
 
         // Call migrate for newly added and updated elements
         for (var acElement : acElements) {
-            if (instanceElementMap.get(acElement.getId()) == null) {
-                var compositionDto = new CompositionElementDto(acElement.getId(), acElement.getDefinition(),
-                        Map.of(), Map.of(), true, false);
-                var instanceDto = new InstanceElementDto(acCopy.getInstanceId(), acElement.getId(),
-                        new ToscaServiceTemplate(), Map.of(), Map.of(), true, false);
+            var compositionInProperties = cacheProvider
+                    .getCommonProperties(compositionTargetId, acElement.getDefinition());
+            var stageSet = ParticipantUtils.findStageSet(compositionInProperties);
+            if (stageSet.contains(stage)) {
+                if (instanceElementMap.get(acElement.getId()) == null) {
+                    var compositionElementDto =
+                            new CompositionElementDto(acCopy.getCompositionId(), acElement.getDefinition(),
+                                    Map.of(), Map.of(), ElementState.NOT_PRESENT);
+                    var instanceElementDto = new InstanceElementDto(acCopy.getInstanceId(), acElement.getId(),
+                            null, Map.of(), Map.of(), ElementState.NOT_PRESENT);
+                    var compositionElementTargetDto = CacheProvider.changeStateToNew(
+                            compositionElementTargetMap.get(acElement.getId()));
+                    var instanceElementMigrateDto = CacheProvider
+                            .changeStateToNew(instanceElementMigrateMap.get(acElement.getId()));
 
-                listener.migrate(messageId, compositionDto,
-                        compositionElementTargetMap.get(acElement.getId()),
-                        instanceDto, instanceElementMigrateMap.get(acElement.getId()));
-            } else {
-                listener.migrate(messageId, compositionElementMap.get(acElement.getId()),
-                        compositionElementTargetMap.get(acElement.getId()),
-                        instanceElementMap.get(acElement.getId()), instanceElementMigrateMap.get(acElement.getId()));
+                    listener.migrate(messageId, compositionElementDto, compositionElementTargetDto, instanceElementDto,
+                            instanceElementMigrateDto, stage);
+                } else {
+                    listener.migrate(messageId, compositionElementMap.get(acElement.getId()),
+                            compositionElementTargetMap.get(acElement.getId()),
+                            instanceElementMap.get(acElement.getId()), instanceElementMigrateMap.get(acElement.getId()),
+                            stage);
+                }
             }
         }
-        // Call migrate for removed elements
-        List<UUID> removedElements = findElementsToRemove(acElements, acCopy.getElements());
-        for (var elementId : removedElements) {
-            var compositionDtoTarget = new CompositionElementDto(elementId, acCopy.getElements().get(elementId)
-                    .getDefinition(), Map.of(), Map.of(), false, true);
-            var instanceDtoTarget = new InstanceElementDto(acCopy.getInstanceId(), elementId,
-                    new ToscaServiceTemplate(), Map.of(), Map.of(), false, true);
+        if (stage == 0) {
+            // Call migrate for removed elements
+            List<UUID> removedElements = findElementsToRemove(acElements, acCopy.getElements());
+            for (var elementId : removedElements) {
+                var compositionDtoTarget =
+                        new CompositionElementDto(compositionTargetId,
+                                acCopy.getElements().get(elementId).getDefinition(),
+                                Map.of(), Map.of(), ElementState.REMOVED);
+                var instanceDtoTarget =
+                        new InstanceElementDto(acCopy.getInstanceId(), elementId, null, Map.of(),
+                                Map.of(), ElementState.REMOVED);
 
-            listener.migrate(messageId, compositionElementMap.get(elementId),
-                    compositionDtoTarget,
-                    instanceElementMap.get(elementId), instanceDtoTarget);
+                listener.migrate(messageId, compositionElementMap.get(elementId), compositionDtoTarget,
+                        instanceElementMap.get(elementId), instanceDtoTarget, 0);
+            }
         }
     }
 }
diff --git a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/handler/AutomationCompositionOutHandler.java b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/handler/AutomationCompositionOutHandler.java
index 06b11bbca..77bcb19ef 100644
--- a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/handler/AutomationCompositionOutHandler.java
+++ b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/handler/AutomationCompositionOutHandler.java
@@ -56,27 +56,77 @@ public class AutomationCompositionOutHandler {
     private final CacheProvider cacheProvider;
 
     /**
+     * Handle a automation composition element stage change message.
+     *
+     * @param instance the automationComposition Id
+     * @param elementId the automationComposition Element Id
+     * @param stage the next stage
+     * @param message the message
+     * @param stateChangeResult the indicator if error occurs
+     */
+    public void updateAutomationCompositionElementStage(UUID instance, UUID elementId,
+        StateChangeResult stateChangeResult, int stage, String message) {
+
+        if (instance == null || elementId == null) {
+            LOGGER.error("Cannot update Automation composition element stage, id is null");
+            return;
+        }
+
+        var automationComposition = cacheProvider.getAutomationComposition(instance);
+        if (automationComposition == null) {
+            LOGGER.error("Cannot update Automation composition element stage, Automation composition id {} not present",
+                instance);
+            return;
+        }
+
+        var element = automationComposition.getElements().get(elementId);
+        if (element == null) {
+            var msg = "Cannot update Automation composition element stage, AC Element id {} not present";
+            LOGGER.error(msg, elementId);
+            return;
+        }
+
+        element.setRestarting(null);
+
+        var automationCompositionStateChangeAck =
+            new AutomationCompositionDeployAck(ParticipantMessageType.AUTOMATION_COMPOSITION_STATECHANGE_ACK);
+        automationCompositionStateChangeAck.setParticipantId(cacheProvider.getParticipantId());
+        automationCompositionStateChangeAck.setMessage(message);
+        automationCompositionStateChangeAck.setResponseTo(cacheProvider.getMsgIdentification().get(element.getId()));
+        automationCompositionStateChangeAck.setStateChangeResult(stateChangeResult);
+        automationCompositionStateChangeAck.setStage(stage);
+        automationCompositionStateChangeAck.setAutomationCompositionId(instance);
+        automationCompositionStateChangeAck.getAutomationCompositionResultMap().put(element.getId(),
+            new AcElementDeployAck(element.getDeployState(), element.getLockState(), element.getOperationalState(),
+                element.getUseState(), element.getOutProperties(), true, message));
+        LOGGER.debug("Automation composition element {} stage changed to {}", elementId, stage);
+        automationCompositionStateChangeAck.setResult(true);
+        publisher.sendAutomationCompositionAck(automationCompositionStateChangeAck);
+        cacheProvider.getMsgIdentification().remove(element.getId());
+    }
+
+    /**
      * Handle a automation composition element state change message.
      *
-     * @param automationCompositionId the automationComposition Id
+     * @param instance the automationComposition Id
      * @param elementId the automationComposition Element Id
      * @param deployState the DeployState state
      * @param lockState the LockState state
      * @param message the message
      * @param stateChangeResult the indicator if error occurs
      */
-    public void updateAutomationCompositionElementState(UUID automationCompositionId, UUID elementId,
+    public void updateAutomationCompositionElementState(UUID instance, UUID elementId,
             DeployState deployState, LockState lockState, StateChangeResult stateChangeResult, String message) {
 
-        if (automationCompositionId == null || elementId == null) {
+        if (instance == null || elementId == null) {
             LOGGER.error("Cannot update Automation composition element state, id is null");
             return;
         }
 
-        var automationComposition = cacheProvider.getAutomationComposition(automationCompositionId);
+        var automationComposition = cacheProvider.getAutomationComposition(instance);
         if (automationComposition == null) {
             LOGGER.error("Cannot update Automation composition element state, Automation composition id {} not present",
-                    automationCompositionId);
+                instance);
             return;
         }
 
@@ -114,7 +164,7 @@ public class AutomationCompositionOutHandler {
         automationCompositionStateChangeAck.setMessage(message);
         automationCompositionStateChangeAck.setResponseTo(cacheProvider.getMsgIdentification().get(element.getId()));
         automationCompositionStateChangeAck.setStateChangeResult(stateChangeResult);
-        automationCompositionStateChangeAck.setAutomationCompositionId(automationCompositionId);
+        automationCompositionStateChangeAck.setAutomationCompositionId(instance);
         automationCompositionStateChangeAck.getAutomationCompositionResultMap().put(element.getId(),
                 new AcElementDeployAck(element.getDeployState(), element.getLockState(), element.getOperationalState(),
                         element.getUseState(), element.getOutProperties(), true, message));
diff --git a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/handler/CacheProvider.java b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/handler/CacheProvider.java
index f56fabf3f..7aa06badb 100644
--- a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/handler/CacheProvider.java
+++ b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/handler/CacheProvider.java
@@ -30,8 +30,10 @@ import lombok.Getter;
 import lombok.NonNull;
 import lombok.Setter;
 import org.onap.policy.clamp.acm.participant.intermediary.api.CompositionElementDto;
+import org.onap.policy.clamp.acm.participant.intermediary.api.ElementState;
 import org.onap.policy.clamp.acm.participant.intermediary.api.InstanceElementDto;
 import org.onap.policy.clamp.acm.participant.intermediary.parameters.ParticipantParameters;
+import org.onap.policy.clamp.models.acm.concepts.AcElementDeploy;
 import org.onap.policy.clamp.models.acm.concepts.AutomationComposition;
 import org.onap.policy.clamp.models.acm.concepts.AutomationCompositionElement;
 import org.onap.policy.clamp.models.acm.concepts.AutomationCompositionElementDefinition;
@@ -181,14 +183,10 @@ public class CacheProvider {
         var acLast = automationCompositions.get(instanceId);
         Map<UUID, AutomationCompositionElement> acElementMap = new LinkedHashMap<>();
         for (var element : participantDeploy.getAcElementList()) {
-            var acElement = new AutomationCompositionElement();
-            acElement.setId(element.getId());
+            var acElement = createAutomationCompositionElement(element);
             acElement.setParticipantId(getParticipantId());
-            acElement.setDefinition(element.getDefinition());
             acElement.setDeployState(deployState);
-            acElement.setLockState(LockState.NONE);
             acElement.setSubState(subState);
-            acElement.setProperties(element.getProperties());
             var acElementLast = acLast != null ? acLast.getElements().get(element.getId()) : null;
             if (acElementLast != null) {
                 acElement.setOutProperties(acElementLast.getOutProperties());
@@ -249,6 +247,22 @@ public class CacheProvider {
     }
 
     /**
+     * Create AutomationCompositionElement to save in memory.
+     *
+     * @param element AcElementDeploy
+     * @return a new AutomationCompositionElement
+     */
+    public static AutomationCompositionElement createAutomationCompositionElement(AcElementDeploy element) {
+        var acElement = new AutomationCompositionElement();
+        acElement.setId(element.getId());
+        acElement.setDefinition(element.getDefinition());
+        acElement.setProperties(element.getProperties());
+        acElement.setSubState(SubState.NONE);
+        acElement.setLockState(LockState.LOCKED);
+        return acElement;
+    }
+
+    /**
      * Create CompositionElementDto.
      *
      * @param compositionId the composition Id
@@ -302,4 +316,27 @@ public class CacheProvider {
         }
         return map;
     }
+
+    /**
+     * Create a new InstanceElementDto record with state New.
+     *
+     * @param instanceElement the InstanceElementDto
+     * @return a new InstanceElementDto
+     */
+    public static InstanceElementDto changeStateToNew(InstanceElementDto instanceElement) {
+        return new InstanceElementDto(instanceElement.instanceId(), instanceElement.elementId(),
+                instanceElement.toscaServiceTemplateFragment(),
+                instanceElement.inProperties(), instanceElement.outProperties(), ElementState.NEW);
+    }
+
+    /**
+     * Create a new CompositionElementDto record with state New.
+     *
+     * @param compositionElement the CompositionElementDto
+     * @return a new CompositionElementDto
+     */
+    public static CompositionElementDto changeStateToNew(CompositionElementDto compositionElement) {
+        return new CompositionElementDto(compositionElement.compositionId(), compositionElement.elementDefinitionId(),
+                compositionElement.inProperties(), compositionElement.outProperties(), ElementState.NEW);
+    }
 }
diff --git a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/handler/ThreadHandler.java b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/handler/ThreadHandler.java
index 24b7f3adc..9a43bf4c3 100644
--- a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/handler/ThreadHandler.java
+++ b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/handler/ThreadHandler.java
@@ -53,7 +53,7 @@ public class ThreadHandler implements Closeable {
     private final ParticipantIntermediaryApi intermediaryApi;
     private final CacheProvider cacheProvider;
 
-    private final Map<UUID, Future> executionMap = new ConcurrentHashMap<>();
+    private final Map<UUID, Future<?>> executionMap = new ConcurrentHashMap<>();
 
     private final ExecutorService executor =
             Context.taskWrapping(Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()));
@@ -295,21 +295,24 @@ public class ThreadHandler implements Closeable {
      * @param compositionElementTarget the information of the Automation Composition Definition Element Target
      * @param instanceElement the information of the Automation Composition Instance Element
      * @param instanceElementMigrate the information of the Automation Composition Instance Element updated
+     * @param stage the stage
      */
     public void migrate(UUID messageId, CompositionElementDto compositionElement,
         CompositionElementDto compositionElementTarget, InstanceElementDto instanceElement,
-        InstanceElementDto instanceElementMigrate) {
+        InstanceElementDto instanceElementMigrate, int stage) {
         cleanExecution(instanceElement.elementId(), messageId);
         var result = executor.submit(() ->
-            this.migrateProcess(compositionElement, compositionElementTarget, instanceElement, instanceElementMigrate));
+            this.migrateProcess(compositionElement, compositionElementTarget,
+                instanceElement, instanceElementMigrate, stage));
         executionMap.put(instanceElement.elementId(), result);
     }
 
     private void migrateProcess(CompositionElementDto compositionElement,
         CompositionElementDto compositionElementTarget, InstanceElementDto instanceElement,
-        InstanceElementDto instanceElementMigrate) {
+        InstanceElementDto instanceElementMigrate, int stage) {
         try {
-            listener.migrate(compositionElement, compositionElementTarget, instanceElement, instanceElementMigrate, 0);
+            listener.migrate(compositionElement, compositionElementTarget,
+                instanceElement, instanceElementMigrate, stage);
         } catch (PfModelException e) {
             LOGGER.error("Automation composition element migrate failed {} {}",
                 instanceElement.elementId(), e.getMessage());