diff --git a/api/src/main/java/com/cloud/vm/UserVmService.java b/api/src/main/java/com/cloud/vm/UserVmService.java
index 0747e193ab8..01f11b73cd4 100644
--- a/api/src/main/java/com/cloud/vm/UserVmService.java
+++ b/api/src/main/java/com/cloud/vm/UserVmService.java
@@ -64,6 +64,7 @@ import com.cloud.storage.StoragePool;
 import com.cloud.template.VirtualMachineTemplate;
 import com.cloud.user.Account;
 import com.cloud.uservm.UserVm;
+import com.cloud.utils.Pair;
 import com.cloud.utils.exception.ExecutionException;
 
 public interface UserVmService {
@@ -538,9 +539,10 @@ public interface UserVmService {
 
     /**
      * Unmanage a guest VM from CloudStack
-     * @return true if the VM is successfully unmanaged, false if not.
+     *
+     * @return (true if successful, false if not, hostUuid) if the VM is successfully unmanaged.
      */
-    boolean unmanageUserVM(Long vmId);
+    Pair<Boolean, String> unmanageUserVM(Long vmId, Long targetHostId);
 
     UserVm allocateVMFromBackup(CreateVMFromBackupCmd cmd) throws InsufficientCapacityException, ResourceAllocationException, ResourceUnavailableException;
 
diff --git a/api/src/main/java/org/apache/cloudstack/api/command/admin/vm/UnmanageVMInstanceCmd.java b/api/src/main/java/org/apache/cloudstack/api/command/admin/vm/UnmanageVMInstanceCmd.java
index bbcb8840f66..2c60b574126 100644
--- a/api/src/main/java/org/apache/cloudstack/api/command/admin/vm/UnmanageVMInstanceCmd.java
+++ b/api/src/main/java/org/apache/cloudstack/api/command/admin/vm/UnmanageVMInstanceCmd.java
@@ -27,6 +27,7 @@ import com.cloud.exception.ResourceAllocationException;
 import com.cloud.exception.ResourceUnavailableException;
 import com.cloud.user.Account;
 import com.cloud.uservm.UserVm;
+import com.cloud.utils.Pair;
 import com.cloud.vm.VirtualMachine;
 import org.apache.cloudstack.acl.RoleType;
 import org.apache.cloudstack.api.APICommand;
@@ -36,10 +37,12 @@ import org.apache.cloudstack.api.ApiErrorCode;
 import org.apache.cloudstack.api.BaseAsyncCmd;
 import org.apache.cloudstack.api.Parameter;
 import org.apache.cloudstack.api.ServerApiException;
+import org.apache.cloudstack.api.response.HostResponse;
 import org.apache.cloudstack.api.response.UnmanageVMInstanceResponse;
 import org.apache.cloudstack.api.response.UserVmResponse;
 import org.apache.cloudstack.context.CallContext;
 import org.apache.cloudstack.vm.UnmanagedVMsManager;
+import org.apache.commons.lang3.BooleanUtils;
 
 import javax.inject.Inject;
 
@@ -65,6 +68,20 @@ public class UnmanageVMInstanceCmd extends BaseAsyncCmd {
             description = "The ID of the virtual machine to unmanage")
     private Long vmId;
 
+    @Parameter(name = ApiConstants.HOST_ID, type = CommandType.UUID,
+            entityType = HostResponse.class, required = false,
+            description = "ID of the host which will be used for unmanaging the Instance. " +
+                    "Applicable only for KVM hypervisor and stopped Instances. Domain XML will be stored on this host.",
+            since = "4.22.0")
+    private Long hostId;
+
+    @Parameter(name = ApiConstants.FORCED,
+            type = CommandType.BOOLEAN,
+            required = false,
+            description = "Force unmanaging Instance with config drive. Applicable only for KVM Hypervisor.",
+            since = "4.22.0")
+    private Boolean forced;
+
     /////////////////////////////////////////////////////
     /////////////////// Accessors ///////////////////////
     /////////////////////////////////////////////////////
@@ -83,6 +100,18 @@ public class UnmanageVMInstanceCmd extends BaseAsyncCmd {
         return "unmanaging VM. VM ID = " + vmId;
     }
 
+    public Long getHostId() {
+        return hostId;
+    }
+
+    public void setHostId(Long hostId) {
+        this.hostId = hostId;
+    }
+
+    public Boolean isForced() {
+        return BooleanUtils.isTrue(forced);
+    }
+
     /////////////////////////////////////////////////////
     /////////////// API Implementation///////////////////
     /////////////////////////////////////////////////////
@@ -93,9 +122,10 @@ public class UnmanageVMInstanceCmd extends BaseAsyncCmd {
         UnmanageVMInstanceResponse response = new UnmanageVMInstanceResponse();
         try {
             CallContext.current().setEventDetails("VM ID = " + vmId);
-            boolean result = unmanagedVMsManager.unmanageVMInstance(vmId);
-            response.setSuccess(result);
-            if (result) {
+            Pair<Boolean, String> result = unmanagedVMsManager.unmanageVMInstance(vmId, hostId, isForced());
+            if (result.first()) {
+                response.setSuccess(true);
+                response.setHostId(result.second());
                 response.setDetails("VM unmanaged successfully");
             }
         } catch (Exception e) {
@@ -124,5 +154,4 @@ public class UnmanageVMInstanceCmd extends BaseAsyncCmd {
     public Long getApiResourceId() {
         return vmId;
     }
-
 }
diff --git a/api/src/main/java/org/apache/cloudstack/api/response/UnmanageVMInstanceResponse.java b/api/src/main/java/org/apache/cloudstack/api/response/UnmanageVMInstanceResponse.java
index e9d45cb506a..bd95ecee5ae 100644
--- a/api/src/main/java/org/apache/cloudstack/api/response/UnmanageVMInstanceResponse.java
+++ b/api/src/main/java/org/apache/cloudstack/api/response/UnmanageVMInstanceResponse.java
@@ -32,6 +32,10 @@ public class UnmanageVMInstanceResponse extends BaseResponse {
     @Param(description = "details of the unmanage VM operation")
     private String details;
 
+    @SerializedName(ApiConstants.HOST_ID)
+    @Param(description = "The ID of the host used for unmanaged Instance")
+    private String hostId;
+
     public UnmanageVMInstanceResponse() {
     }
 
@@ -55,4 +59,12 @@ public class UnmanageVMInstanceResponse extends BaseResponse {
     public void setDetails(String details) {
         this.details = details;
     }
+
+    public String getHostId() {
+        return hostId;
+    }
+
+    public void setHostId(String hostId) {
+        this.hostId = hostId;
+    }
 }
diff --git a/api/src/main/java/org/apache/cloudstack/vm/UnmanageVMService.java b/api/src/main/java/org/apache/cloudstack/vm/UnmanageVMService.java
index 2315e5f2e95..70d8795f305 100644
--- a/api/src/main/java/org/apache/cloudstack/vm/UnmanageVMService.java
+++ b/api/src/main/java/org/apache/cloudstack/vm/UnmanageVMService.java
@@ -17,11 +17,14 @@
 
 package org.apache.cloudstack.vm;
 
+import com.cloud.utils.Pair;
+
 public interface UnmanageVMService {
 
     /**
      * Unmanage a guest VM from CloudStack
-     * @return true if the VM is successfully unmanaged, false if not.
+     *
+     * @return (true if successful, false if not, hostUuid) if the VM is successfully unmanaged.
      */
-    boolean unmanageVMInstance(long vmId);
+    Pair<Boolean, String> unmanageVMInstance(long vmId, Long paramHostId, boolean isForced);
 }
diff --git a/core/src/main/java/com/cloud/agent/api/UnmanageInstanceAnswer.java b/core/src/main/java/com/cloud/agent/api/UnmanageInstanceAnswer.java
new file mode 100644
index 00000000000..39a35d49990
--- /dev/null
+++ b/core/src/main/java/com/cloud/agent/api/UnmanageInstanceAnswer.java
@@ -0,0 +1,27 @@
+//
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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.
+//
+
+package com.cloud.agent.api;
+
+public class UnmanageInstanceAnswer extends Answer {
+
+    public UnmanageInstanceAnswer(UnmanageInstanceCommand cmd, boolean success, String details) {
+        super(cmd, success, details);
+    }
+}
diff --git a/core/src/main/java/com/cloud/agent/api/UnmanageInstanceCommand.java b/core/src/main/java/com/cloud/agent/api/UnmanageInstanceCommand.java
new file mode 100644
index 00000000000..dd504b9ea26
--- /dev/null
+++ b/core/src/main/java/com/cloud/agent/api/UnmanageInstanceCommand.java
@@ -0,0 +1,61 @@
+//
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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.
+//
+
+package com.cloud.agent.api;
+
+import com.cloud.agent.api.to.VirtualMachineTO;
+
+/**
+ */
+public class UnmanageInstanceCommand extends Command {
+    String instanceName;
+    boolean executeInSequence = false;
+    VirtualMachineTO vm;
+    boolean isConfigDriveAttached;
+
+    @Override
+    public boolean executeInSequence() {
+        return executeInSequence;
+    }
+
+    public UnmanageInstanceCommand(VirtualMachineTO vm) {
+        this.vm = vm;
+        this.instanceName = vm.getName();
+    }
+
+    public UnmanageInstanceCommand(String instanceName) {
+        this.instanceName = instanceName;
+    }
+
+    public String getInstanceName() {
+        return instanceName;
+    }
+
+    public VirtualMachineTO getVm() {
+        return vm;
+    }
+
+    public boolean isConfigDriveAttached() {
+        return isConfigDriveAttached;
+    }
+
+    public void setConfigDriveAttached(boolean configDriveAttached) {
+        isConfigDriveAttached = configDriveAttached;
+    }
+}
diff --git a/engine/api/src/main/java/com/cloud/vm/VirtualMachineManager.java b/engine/api/src/main/java/com/cloud/vm/VirtualMachineManager.java
index c05c29add55..cffba3d9a13 100644
--- a/engine/api/src/main/java/com/cloud/vm/VirtualMachineManager.java
+++ b/engine/api/src/main/java/com/cloud/vm/VirtualMachineManager.java
@@ -274,7 +274,7 @@ public interface VirtualMachineManager extends Manager {
      * - Remove the references of the VM and its volumes, nics, IPs from database
      * - Keep the VM as it is on the hypervisor
      */
-    boolean unmanage(String vmUuid);
+    Pair<Boolean, String> unmanage(String vmUuid, Long paramHostId);
 
     UserVm restoreVirtualMachine(long vmId, Long newTemplateId, Long rootDiskOfferingId, boolean expunge, Map<String, String> details) throws ResourceUnavailableException, InsufficientCapacityException, ResourceAllocationException;
 
diff --git a/engine/orchestration/src/main/java/com/cloud/vm/VirtualMachineManagerImpl.java b/engine/orchestration/src/main/java/com/cloud/vm/VirtualMachineManagerImpl.java
index b5597280364..7c5d43fea97 100755
--- a/engine/orchestration/src/main/java/com/cloud/vm/VirtualMachineManagerImpl.java
+++ b/engine/orchestration/src/main/java/com/cloud/vm/VirtualMachineManagerImpl.java
@@ -71,6 +71,7 @@ import org.apache.cloudstack.engine.subsystem.api.storage.DataStoreProviderManag
 import org.apache.cloudstack.engine.subsystem.api.storage.PrimaryDataStoreDriver;
 import org.apache.cloudstack.engine.subsystem.api.storage.StoragePoolAllocator;
 import org.apache.cloudstack.engine.subsystem.api.storage.VolumeDataFactory;
+import org.apache.cloudstack.engine.subsystem.api.storage.VolumeInfo;
 import org.apache.cloudstack.framework.ca.Certificate;
 import org.apache.cloudstack.framework.config.ConfigKey;
 import org.apache.cloudstack.framework.config.Configurable;
@@ -150,11 +151,13 @@ import com.cloud.agent.api.StopAnswer;
 import com.cloud.agent.api.StopCommand;
 import com.cloud.agent.api.UnPlugNicAnswer;
 import com.cloud.agent.api.UnPlugNicCommand;
+import com.cloud.agent.api.UnmanageInstanceCommand;
 import com.cloud.agent.api.UnregisterVMCommand;
 import com.cloud.agent.api.VmDiskStatsEntry;
 import com.cloud.agent.api.VmNetworkStatsEntry;
 import com.cloud.agent.api.VmStatsEntry;
 import com.cloud.agent.api.routing.NetworkElementCommand;
+import com.cloud.agent.api.to.DataTO;
 import com.cloud.agent.api.to.DiskTO;
 import com.cloud.agent.api.to.DpdkTO;
 import com.cloud.agent.api.to.GPUDeviceTO;
@@ -2016,7 +2019,7 @@ public class VirtualMachineManagerImpl extends ManagerBase implements VirtualMac
     }
 
     @Override
-    public boolean unmanage(String vmUuid) {
+    public Pair<Boolean, String> unmanage(String vmUuid, Long paramHostId) {
         VMInstanceVO vm = _vmDao.findByUuid(vmUuid);
         if (vm == null || vm.getRemoved() != null) {
             throw new CloudRuntimeException("Could not find VM with id = " + vmUuid);
@@ -2029,6 +2032,10 @@ public class VirtualMachineManagerImpl extends ManagerBase implements VirtualMac
             throw new ConcurrentOperationException(msg);
         }
 
+        Long agentHostId = vm.getHostId();
+        if (HypervisorType.KVM.equals(vm.getHypervisorType())) {
+            agentHostId = persistDomainForKVM(vm, paramHostId);
+        }
         Boolean result = Transaction.execute(new TransactionCallback<Boolean>() {
             @Override
             public Boolean doInTransaction(TransactionStatus status) {
@@ -2052,21 +2059,66 @@ public class VirtualMachineManagerImpl extends ManagerBase implements VirtualMac
                 return true;
             }
         });
+        HostVO host = ApiDBUtils.findHostById(agentHostId);
+        if (host == null) {
+            return new Pair<>(result, null);
+        }
+        logger.debug("Selected host UUID: {} to unmanage Instance: {}.", host.getUuid(), vm.getName());
+        ActionEventUtils.onActionEvent(User.UID_SYSTEM, Account.ACCOUNT_ID_SYSTEM, Domain.ROOT_DOMAIN, EventTypes.EVENT_VM_UNMANAGE,
+                String.format("Successfully unmanaged Instance: %s (ID: %s) on host ID: %s", vm.getName(), vm.getUuid(), host.getUuid()),
+                vm.getId(), ApiCommandResourceType.VirtualMachine.toString());
+        return new Pair<>(result, host.getUuid());
+    }
 
-        return BooleanUtils.isTrue(result);
+    Long persistDomainForKVM(VMInstanceVO vm, Long paramHostId) {
+        Long agentHostId = vm.getHostId();
+        String vmName = vm.getName();
+        UnmanageInstanceCommand unmanageInstanceCommand;
+        if (State.Stopped.equals(vm.getState())) {
+            if (paramHostId == null) {
+                Pair<Long, Long> clusterAndHostId = findClusterAndHostIdForVm(vm, false);
+                agentHostId = clusterAndHostId.second();
+                if (agentHostId == null) {
+                    String errorMsg = "No available host to persist domain XML for Instance: " + vmName;
+                    logger.debug(errorMsg);
+                    throw new CloudRuntimeException(errorMsg);
+                }
+            } else {
+                agentHostId = paramHostId;
+            }
+            unmanageInstanceCommand = new UnmanageInstanceCommand(prepVmSpecForUnmanageCmd(vm.getId(), agentHostId)); // reconstruct vmSpec for stopped instance
+        } else {
+            unmanageInstanceCommand = new UnmanageInstanceCommand(vmName);
+            unmanageInstanceCommand.setConfigDriveAttached(vmInstanceDetailsDao.findDetail(vm.getId(), VmDetailConstants.CONFIG_DRIVE_LOCATION) != null);
+        }
+
+        logger.debug("Selected host ID: {} to persist domain XML for Instance: {}.", agentHostId, vmName);
+        try {
+            Answer answer = _agentMgr.send(agentHostId, unmanageInstanceCommand);
+            if (!answer.getResult()) {
+                String errorMsg = "Failed to persist domain XML for Instance: " + vmName + " on host ID: " + agentHostId;
+                logger.debug(errorMsg);
+                throw new CloudRuntimeException(errorMsg);
+            }
+        } catch (AgentUnavailableException | OperationTimedoutException e) {
+            String errorMsg = "Failed to send command to persist domain XML for Instance: " + vmName + " on host ID: " + agentHostId;
+            logger.error(errorMsg, e);
+            throw new CloudRuntimeException(errorMsg);
+        }
+        return agentHostId;
     }
 
     /**
      * Clean up VM snapshots (if any) from DB
      */
-    private void unmanageVMSnapshots(VMInstanceVO vm) {
+    void unmanageVMSnapshots(VMInstanceVO vm) {
         _vmSnapshotMgr.deleteVMSnapshotsFromDB(vm.getId(), true);
     }
 
     /**
      * Clean up volumes for a VM to be unmanaged from CloudStack
      */
-    private void unmanageVMVolumes(VMInstanceVO vm) {
+    void unmanageVMVolumes(VMInstanceVO vm) {
         final Long hostId = vm.getHostId() != null ? vm.getHostId() : vm.getLastHostId();
         if (hostId != null) {
             volumeMgr.revokeAccess(vm.getId(), hostId);
@@ -2084,7 +2136,7 @@ public class VirtualMachineManagerImpl extends ManagerBase implements VirtualMac
      * - If 'unmanage.vm.preserve.nics' = true: then the NICs are not removed but still Allocated, to preserve MAC addresses
      * - If 'unmanage.vm.preserve.nics' = false: then the NICs are removed while unmanaging
      */
-    private void unmanageVMNics(VirtualMachineProfile profile, VMInstanceVO vm) {
+    void unmanageVMNics(VirtualMachineProfile profile, VMInstanceVO vm) {
         logger.debug("Cleaning up NICs of {}.", vm.toString());
         Boolean preserveNics = UnmanagedVMsManager.UnmanageVMPreserveNic.valueIn(vm.getDataCenterId());
         if (BooleanUtils.isTrue(preserveNics)) {
@@ -4019,6 +4071,62 @@ public class VirtualMachineManagerImpl extends ManagerBase implements VirtualMac
         vmTo.setEnterHardwareSetup(enterSetup == null ? false : enterSetup);
     }
 
+    /**
+     * This method helps constructing vmSpec for Unmanage operation for Stopped Instance
+     * @param vmId
+     * @param hostId
+     * @return VirtualMachineTO
+     */
+    protected VirtualMachineTO prepVmSpecForUnmanageCmd(Long vmId, Long hostId) {
+        final VMInstanceVO vm = _vmDao.findById(vmId);
+        final Account owner = _entityMgr.findById(Account.class, vm.getAccountId());
+        final ServiceOfferingVO offering = _offeringDao.findById(vm.getId(), vm.getServiceOfferingId());
+        final VirtualMachineTemplate template = _entityMgr.findByIdIncludingRemoved(VirtualMachineTemplate.class, vm.getTemplateId());
+        Host host = _hostDao.findById(hostId);
+        VirtualMachineProfileImpl vmProfile = new VirtualMachineProfileImpl(vm, template, offering, owner, null);
+        updateOverCommitRatioForVmProfile(vmProfile, host.getClusterId());
+        final List<NicVO> nics = _nicsDao.listByVmId(vmProfile.getId());
+        Collections.sort(nics, (nic1, nic2) -> {
+            Long nicId1 = Long.valueOf(nic1.getDeviceId());
+            Long nicId2 = Long.valueOf(nic2.getDeviceId());
+            return nicId1.compareTo(nicId2);
+        });
+
+        for (final NicVO nic : nics) {
+            final Network network = _networkModel.getNetwork(nic.getNetworkId());
+            final NicProfile nicProfile =
+                    new NicProfile(nic, network, nic.getBroadcastUri(), nic.getIsolationUri(), null, _networkModel.isSecurityGroupSupportedInNetwork(network),
+                            _networkModel.getNetworkTag(vmProfile.getHypervisorType(), network));
+            vmProfile.addNic(nicProfile);
+        }
+
+        List<VolumeVO> volumes = _volsDao.findUsableVolumesForInstance(vmId);
+        for (VolumeVO vol: volumes) {
+            VolumeInfo volumeInfo = volumeDataFactory.getVolume(vol.getId());
+            DataTO dataTO = volumeInfo.getTO();
+            DiskTO disk = storageMgr.getDiskWithThrottling(dataTO, vol.getVolumeType(), vol.getDeviceId(), vol.getPath(), vm.getServiceOfferingId(), vol.getDiskOfferingId());
+            vmProfile.addDisk(disk);
+        }
+
+        Map<String, String> details = vmInstanceDetailsDao.listDetailsKeyPairs(vmId,
+                List.of(VirtualMachineProfile.Param.BootType.getName(), VirtualMachineProfile.Param.BootMode.getName(),
+                        VirtualMachineProfile.Param.UefiFlag.getName()));
+
+        if (details.containsKey(VirtualMachineProfile.Param.BootType.getName())) {
+            vmProfile.getParameters().put(VirtualMachineProfile.Param.BootType, details.get(VirtualMachineProfile.Param.BootType.getName()));
+        }
+
+        if (details.containsKey(VirtualMachineProfile.Param.BootMode.getName())) {
+            vmProfile.getParameters().put(VirtualMachineProfile.Param.BootMode, details.get(VirtualMachineProfile.Param.BootMode.getName()));
+        }
+
+        if (details.containsKey(VirtualMachineProfile.Param.UefiFlag.getName())) {
+            vmProfile.getParameters().put(VirtualMachineProfile.Param.UefiFlag, details.get(VirtualMachineProfile.Param.UefiFlag.getName()));
+        }
+
+        return toVmTO(vmProfile);
+    }
+
     protected VirtualMachineTO getVmTO(Long vmId) {
         final VMInstanceVO vm = _vmDao.findById(vmId);
         final VirtualMachineProfile profile = new VirtualMachineProfileImpl(vm);
@@ -6185,8 +6293,9 @@ public class VirtualMachineManagerImpl extends ManagerBase implements VirtualMac
         host = host == null ? _hostDao.findById(hostId) : host;
         if (host != null) {
             clusterId = host.getClusterId();
+            return new Pair<>(clusterId, hostId);
         }
-        return new Pair<>(clusterId, hostId);
+        return findClusterAndHostIdForVmFromVolumes(vm.getId());
     }
 
     private Pair<Long, Long> findClusterAndHostIdForVm(VirtualMachine vm) {
diff --git a/engine/orchestration/src/test/java/com/cloud/vm/VirtualMachineManagerImplTest.java b/engine/orchestration/src/test/java/com/cloud/vm/VirtualMachineManagerImplTest.java
index 4f6329f81cb..a07870d09af 100644
--- a/engine/orchestration/src/test/java/com/cloud/vm/VirtualMachineManagerImplTest.java
+++ b/engine/orchestration/src/test/java/com/cloud/vm/VirtualMachineManagerImplTest.java
@@ -19,14 +19,18 @@ package com.cloud.vm;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyList;
 import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.doNothing;
 import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
@@ -36,6 +40,8 @@ import static org.mockito.Mockito.when;
 import java.lang.reflect.Field;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
+import java.util.Date;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -43,15 +49,29 @@ import java.util.Random;
 import java.util.UUID;
 import java.util.stream.Collectors;
 
+import com.cloud.agent.api.UnmanageInstanceAnswer;
+import com.cloud.agent.api.UnmanageInstanceCommand;
+import com.cloud.agent.api.to.DataTO;
+import com.cloud.agent.api.to.DiskTO;
+import com.cloud.api.ApiDBUtils;
+import com.cloud.event.ActionEventUtils;
+import com.cloud.exception.ConcurrentOperationException;
+import com.cloud.ha.HighAvailabilityManager;
+import com.cloud.network.Network;
+import com.cloud.network.NetworkModel;
 import com.cloud.resource.ResourceManager;
 import org.apache.cloudstack.api.ApiConstants;
 import org.apache.cloudstack.context.CallContext;
 import org.apache.cloudstack.engine.subsystem.api.storage.StoragePoolAllocator;
+import org.apache.cloudstack.engine.subsystem.api.storage.VolumeDataFactory;
+import org.apache.cloudstack.engine.subsystem.api.storage.VolumeInfo;
 import org.apache.cloudstack.framework.config.ConfigKey;
 import org.apache.cloudstack.framework.config.impl.ConfigDepotImpl;
 import org.apache.cloudstack.framework.extensions.dao.ExtensionDetailsDao;
 import org.apache.cloudstack.framework.extensions.manager.ExtensionsManager;
 import org.apache.cloudstack.framework.extensions.vo.ExtensionDetailsVO;
+import org.apache.cloudstack.framework.jobs.dao.VmWorkJobDao;
+import org.apache.cloudstack.framework.jobs.impl.VmWorkJobVO;
 import org.apache.cloudstack.storage.datastore.db.PrimaryDataStoreDao;
 import org.apache.cloudstack.storage.datastore.db.StoragePoolVO;
 import org.apache.cloudstack.storage.to.VolumeObjectTO;
@@ -61,6 +81,7 @@ import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
 import org.mockito.InOrder;
 import org.mockito.InjectMocks;
 import org.mockito.Mock;
@@ -167,6 +188,9 @@ public class VirtualMachineManagerImplTest {
     private PrimaryDataStoreDao storagePoolDaoMock;
     @Mock
     private VMInstanceVO vmInstanceMock;
+    @Mock
+    private VmWorkJobDao _workJobDao;
+
     private long vmInstanceVoMockId = 1L;
 
     @Mock
@@ -181,6 +205,9 @@ public class VirtualMachineManagerImplTest {
     private long hostMockId = 1L;
     private long clusterMockId = 2L;
     private long zoneMockId = 3L;
+    private final String vmMockUuid = UUID.randomUUID().toString();
+    private final String hostUuid = UUID.randomUUID().toString();
+
     @Mock
     private HostVO hostMock;
     @Mock
@@ -192,6 +219,7 @@ public class VirtualMachineManagerImplTest {
     private StoragePoolVO storagePoolVoMock;
     private long storagePoolVoMockId = 11L;
     private long storagePoolVoMockClusterId = 234L;
+    private String vmName = "vm1";
 
     @Mock
     private VolumeVO volumeVoMock;
@@ -254,6 +282,16 @@ public class VirtualMachineManagerImplTest {
     NicDao _nicsDao;
     @Mock
     NetworkService networkService;
+    @Mock
+    NetworkModel networkModel;
+    @Mock
+    VolumeDataFactory volumeDataFactoryMock;
+    @Mock
+    StorageManager storageManager;
+    @Mock
+    private HighAvailabilityManager _haMgr;
+    @Mock
+    VirtualMachineGuru guru;
 
     private ConfigDepotImpl configDepotImpl;
     private boolean updatedConfigKeyDepot = false;
@@ -263,6 +301,7 @@ public class VirtualMachineManagerImplTest {
         ReflectionTestUtils.getField(VirtualMachineManager.VmMetadataManufacturer, "s_depot");
         virtualMachineManagerImpl.setHostAllocators(new ArrayList<>());
 
+        when(vmInstanceMock.getName()).thenReturn(vmName);
         when(vmInstanceMock.getId()).thenReturn(vmInstanceVoMockId);
         when(vmInstanceMock.getServiceOfferingId()).thenReturn(2L);
         when(hostMock.getId()).thenReturn(hostMockId);
@@ -1361,7 +1400,7 @@ public class VirtualMachineManagerImplTest {
         Mockito.doReturn(HypervisorType.KVM).when(vmInstanceMock).getHypervisorType();
         Mockito.doReturn(List.of(new VolumeObjectTO())).when(virtualMachineManagerImpl).getVmVolumesWithCheckpointsToRecreate(Mockito.any());
 
-        Mockito.doThrow(new AgentUnavailableException(0)).when(agentManagerMock).send(Mockito.anyLong(), (Command) any());
+        doThrow(new AgentUnavailableException(0)).when(agentManagerMock).send(Mockito.anyLong(), (Command) any());
         Mockito.doNothing().when(snapshotManagerMock).endSnapshotChainForVolume(Mockito.anyLong(), Mockito.any());
 
         virtualMachineManagerImpl.recreateCheckpointsKvmOnVmAfterMigration(vmInstanceMock, 0);
@@ -1374,7 +1413,7 @@ public class VirtualMachineManagerImplTest {
         Mockito.doReturn(HypervisorType.KVM).when(vmInstanceMock).getHypervisorType();
         Mockito.doReturn(List.of(new VolumeObjectTO())).when(virtualMachineManagerImpl).getVmVolumesWithCheckpointsToRecreate(Mockito.any());
 
-        Mockito.doThrow(new OperationTimedoutException(null, 0, 0, 0, false)).when(agentManagerMock).send(Mockito.anyLong(), (Command) any());
+        doThrow(new OperationTimedoutException(null, 0, 0, 0, false)).when(agentManagerMock).send(Mockito.anyLong(), (Command) any());
         Mockito.doNothing().when(snapshotManagerMock).endSnapshotChainForVolume(Mockito.anyLong(), Mockito.any());
 
         virtualMachineManagerImpl.recreateCheckpointsKvmOnVmAfterMigration(vmInstanceMock, 0);
@@ -1641,4 +1680,278 @@ public class VirtualMachineManagerImplTest {
         virtualMachineManagerImpl.processPrepareExternalProvisioning(true, host, vmProfile, mock(DataCenter.class));
         verify(agentManagerMock).send(anyLong(), any(Command.class));
     }
+
+    @Test
+    public void testPrepVMSpecForUnmanageInstance() {
+        // Arrange
+        final Long accountId = 1L;
+        final Long offeringId = 1L;
+        final Long templateId = 1L;
+
+        // Mock vm
+        VMInstanceVO vm = Mockito.mock(VMInstanceVO.class);
+        when(vm.getId()).thenReturn(vmInstanceVoMockId);
+        when(vm.getAccountId()).thenReturn(accountId);
+        when(vm.getServiceOfferingId()).thenReturn(offeringId);
+        when(vm.getTemplateId()).thenReturn(templateId);
+        when(vm.getHypervisorType()).thenReturn(HypervisorType.KVM);
+        when(vmInstanceDaoMock.findById(vmInstanceVoMockId)).thenReturn(vm);
+
+        // Mock owner
+        AccountVO owner = Mockito.mock(AccountVO.class);
+        when(_entityMgr.findById(Account.class, accountId)).thenReturn(owner);
+
+        ServiceOfferingVO offering = Mockito.mock(ServiceOfferingVO.class);
+        when(serviceOfferingDaoMock.findById(vmInstanceVoMockId, offeringId)).thenReturn(offering);
+
+        VMTemplateVO template = Mockito.mock(VMTemplateVO.class);
+        when(_entityMgr.findByIdIncludingRemoved(VirtualMachineTemplate.class, templateId)).thenReturn(template);
+
+        when(hostMock.getClusterId()).thenReturn(clusterMockId);
+
+        // Mock cpuOvercommitRatio and ramOvercommitRatio
+        ClusterDetailsVO cpuOvercommitRatio = Mockito.mock(ClusterDetailsVO.class);
+        when(cpuOvercommitRatio.getValue()).thenReturn("1.0");
+        when(_clusterDetailsDao.findDetail(clusterMockId, VmDetailConstants.CPU_OVER_COMMIT_RATIO)).thenReturn(cpuOvercommitRatio);
+        ClusterDetailsVO ramOvercommitRatio = Mockito.mock(ClusterDetailsVO.class);
+        when(ramOvercommitRatio.getValue()).thenReturn("1.0");
+        when(_clusterDetailsDao.findDetail(clusterMockId, VmDetailConstants.MEMORY_OVER_COMMIT_RATIO)).thenReturn(ramOvercommitRatio);
+
+        // Mock NICs
+        List<NicVO> nics = new ArrayList<>();
+        NicVO nic1 = Mockito.mock(NicVO.class);
+        when(nic1.getDeviceId()).thenReturn(1);
+        nics.add(nic1);
+        NicVO nic2 = Mockito.mock(NicVO.class);
+        when(nic2.getDeviceId()).thenReturn(0);
+        nics.add(nic2);
+        when(_nicsDao.listByVmId(vmInstanceVoMockId)).thenReturn(nics);
+
+        Network networkMock = Mockito.mock(Network.class);
+        when(networkModel.getNetwork(anyLong())).thenReturn(networkMock);
+
+        when(volumeVoMock.getVolumeType()).thenReturn(Volume.Type.ROOT);
+        when(volumeVoMock.getDeviceId()).thenReturn(0L);
+        when(volumeVoMock.getPath()).thenReturn("/");
+        when(volumeVoMock.getDiskOfferingId()).thenReturn(1L);
+        when(volumeDaoMock.findUsableVolumesForInstance(vmInstanceVoMockId)).thenReturn(List.of(volumeVoMock));
+
+        VolumeInfo volumeInfo = mock(VolumeInfo.class);
+        DataTO dataTO = mock(DataTO.class);
+        when(volumeInfo.getTO()).thenReturn(dataTO);
+        when(volumeDataFactoryMock.getVolume(anyLong())).thenReturn(volumeInfo);
+        when(storageManager.getDiskWithThrottling(any(), any(), anyLong(), anyString(), anyLong(), anyLong())).thenReturn(Mockito.mock(DiskTO.class));
+
+        Map<String, String> details = new HashMap<>();
+        details.put(VirtualMachineProfile.Param.BootType.getName(), "BIOS");
+        details.put(VirtualMachineProfile.Param.BootMode.getName(), "LEGACY");
+        details.put(VirtualMachineProfile.Param.UefiFlag.getName(), "Yes");
+        when(vmInstanceDetailsDao.listDetailsKeyPairs(anyLong(), anyList())).thenReturn(details);
+
+        com.cloud.hypervisor.HypervisorGuru guru = Mockito.mock(com.cloud.hypervisor.HypervisorGuru.class);
+        when(_hvGuruMgr.getGuru(HypervisorType.KVM)).thenReturn(guru);
+        VirtualMachineTO vmTO = new VirtualMachineTO() {};
+        when(guru.implement(any(VirtualMachineProfile.class))).thenAnswer((Answer<VirtualMachineTO>) invocation -> {
+            VirtualMachineProfile profile = invocation.getArgument(0);
+            assertEquals("BIOS", profile.getParameter(VirtualMachineProfile.Param.BootType));
+            return vmTO;
+        });
+
+        // Act
+        VirtualMachineTO result = virtualMachineManagerImpl.prepVmSpecForUnmanageCmd(vmInstanceVoMockId, hostMockId);
+
+        // Assert
+        assertNotNull(result);
+        assertEquals(vmTO, result);
+        verify(_clusterDetailsDao, times(2)).findDetail(eq(clusterMockId), anyString());
+        verify(vmInstanceDetailsDao).listDetailsKeyPairs(anyLong(), anyList());
+    }
+
+    @Test
+    public void testPersistDomainForKvmForRunningVmSuccess() throws AgentUnavailableException, OperationTimedoutException {
+        when(vmInstanceMock.getState()).thenReturn(VirtualMachine.State.Running);
+        when(vmInstanceMock.getHostId()).thenReturn(hostMockId);
+        UnmanageInstanceAnswer successAnswer = new UnmanageInstanceAnswer(null, true, "success");
+        when(agentManagerMock.send(anyLong(), any(Command.class))).thenReturn(successAnswer);
+        virtualMachineManagerImpl.persistDomainForKVM(vmInstanceMock, null);
+        ArgumentCaptor<Long> hostIdCaptor = ArgumentCaptor.forClass(Long.class);
+        ArgumentCaptor<UnmanageInstanceCommand> commandCaptor = ArgumentCaptor.forClass(UnmanageInstanceCommand.class);
+        verify(agentManagerMock).send(hostIdCaptor.capture(), commandCaptor.capture());
+        assertEquals(hostMockId, hostIdCaptor.getValue().longValue());
+    }
+
+    @Test
+    public void testPersistDomainForKvmForStoppedVmSuccess() throws AgentUnavailableException, OperationTimedoutException {
+        when(vmInstanceMock.getState()).thenReturn(VirtualMachine.State.Stopped);
+        VirtualMachineTO vmTO = new VirtualMachineTO() {};
+        vmTO.setName(vmName);
+        doReturn(vmTO).when(virtualMachineManagerImpl).prepVmSpecForUnmanageCmd(vmInstanceVoMockId, 1L);
+        UnmanageInstanceAnswer successAnswer = new UnmanageInstanceAnswer(null, true, "success");
+        when(agentManagerMock.send(anyLong(), any(UnmanageInstanceCommand.class))).thenReturn(successAnswer);
+        when(virtualMachineManagerImpl.findClusterAndHostIdForVm(vmInstanceMock, false)).thenReturn(new Pair<>(clusterMockId, hostMockId));
+        virtualMachineManagerImpl.persistDomainForKVM(vmInstanceMock, null);
+        ArgumentCaptor<Long> hostIdCaptor = ArgumentCaptor.forClass(Long.class);
+        ArgumentCaptor<UnmanageInstanceCommand> commandCaptor = ArgumentCaptor.forClass(UnmanageInstanceCommand.class);
+        verify(agentManagerMock).send(hostIdCaptor.capture(), commandCaptor.capture());
+        assertEquals(1L, hostIdCaptor.getValue().longValue());
+        UnmanageInstanceCommand sentCommand = commandCaptor.getValue();
+        assertNotNull(sentCommand.getVm());
+        assertEquals(vmTO, sentCommand.getVm());
+        assertEquals(vmName, sentCommand.getInstanceName());
+        verify(virtualMachineManagerImpl).prepVmSpecForUnmanageCmd(vmInstanceVoMockId, 1L);
+    }
+
+
+    @Test
+    public void testPersistDomainForKvmForStoppedVmNoHost() {
+        when(vmInstanceMock.getState()).thenReturn(VirtualMachine.State.Stopped);
+        VirtualMachineTO vmTO = new VirtualMachineTO() {};
+        vmTO.setName(vmName);
+        when(virtualMachineManagerImpl.findClusterAndHostIdForVm(vmInstanceMock, false)).thenReturn(new Pair<>(clusterMockId, null));
+        CloudRuntimeException exception = assertThrows(CloudRuntimeException.class, () -> virtualMachineManagerImpl.persistDomainForKVM(vmInstanceMock, null));
+        assertEquals("No available host to persist domain XML for Instance: " + vmName, exception.getMessage());
+    }
+
+    @Test
+    public void testPersistDomainForKvmForRunningVmAgentFailure() throws AgentUnavailableException, OperationTimedoutException {
+        when(vmInstanceMock.getState()).thenReturn(VirtualMachine.State.Running);
+        when(vmInstanceMock.getHostId()).thenReturn(hostMockId);
+        UnmanageInstanceAnswer failureAnswer = new UnmanageInstanceAnswer(null, false, "failure");
+        when(agentManagerMock.send(anyLong(), any(UnmanageInstanceCommand.class))).thenReturn(failureAnswer);
+        CloudRuntimeException exception = assertThrows(CloudRuntimeException.class, () -> virtualMachineManagerImpl.persistDomainForKVM(vmInstanceMock, null));
+        assertEquals("Failed to persist domain XML for Instance: " + vmName + " on host ID: " + hostMockId, exception.getMessage());
+    }
+
+    @Test
+    public void testPersistDomainForKvmAgentUnavailable() throws AgentUnavailableException, OperationTimedoutException {
+        when(vmInstanceMock.getState()).thenReturn(VirtualMachine.State.Running);
+        when(vmInstanceMock.getHostId()).thenReturn(hostMockId);
+        doThrow(new AgentUnavailableException("Agent down", hostMockId)).when(agentManagerMock).send(anyLong(), any(UnmanageInstanceCommand.class));
+        CloudRuntimeException exception = assertThrows(CloudRuntimeException.class, () -> virtualMachineManagerImpl.persistDomainForKVM(vmInstanceMock, null));
+        assertEquals("Failed to send command to persist domain XML for Instance: " + vmName + " on host ID: " + hostMockId, exception.getMessage());
+    }
+
+    @Test(expected = ConcurrentOperationException.class)
+    public void testUnmanagePendingHaWork() {
+        when(vmInstanceDaoMock.findByUuid(vmMockUuid)).thenReturn(vmInstanceMock);
+        when(_workJobDao.listPendingWorkJobs(VirtualMachine.Type.Instance, vmInstanceVoMockId)).thenReturn(Collections.emptyList());
+        when(_haMgr.hasPendingHaWork(vmInstanceVoMockId)).thenReturn(true);
+        virtualMachineManagerImpl.unmanage(vmMockUuid, null);
+    }
+
+    @Test
+    public void testPersistDomainForKvmOperationTimedOut() throws AgentUnavailableException, OperationTimedoutException {
+        when(vmInstanceMock.getState()).thenReturn(VirtualMachine.State.Running);
+        when(vmInstanceMock.getHostId()).thenReturn(hostMockId);
+        doThrow(new OperationTimedoutException(null, hostMockId, 123L, 60, false)).when(agentManagerMock).send(anyLong(), any(UnmanageInstanceCommand.class));
+        CloudRuntimeException exception = assertThrows(CloudRuntimeException.class, () -> virtualMachineManagerImpl.persistDomainForKVM(vmInstanceMock, null));
+        assertEquals("Failed to send command to persist domain XML for Instance: " + vmName + " on host ID: " + hostMockId, exception.getMessage());
+    }
+
+    @Test(expected = CloudRuntimeException.class)
+    public void testUnmanageVmRemoved() {
+        when(vmInstanceMock.getRemoved()).thenReturn(new Date());
+        when(vmInstanceDaoMock.findByUuid(vmMockUuid)).thenReturn(vmInstanceMock);
+        virtualMachineManagerImpl.unmanage(vmMockUuid, null);
+    }
+
+    @Test(expected = ConcurrentOperationException.class)
+    public void testUnmanagePendingWorkJobs() {
+        when(vmInstanceDaoMock.findByUuid(vmMockUuid)).thenReturn(vmInstanceMock);
+        List<VmWorkJobVO> pendingJobs = new ArrayList<>();
+        VmWorkJobVO vmWorkJobVO = mock(VmWorkJobVO.class);
+        pendingJobs.add(vmWorkJobVO);
+        when(_workJobDao.listPendingWorkJobs(VirtualMachine.Type.Instance, vmInstanceVoMockId)).thenReturn(pendingJobs);
+        virtualMachineManagerImpl.unmanage(vmMockUuid, null);
+    }
+
+    @Test
+    public void testUnmanageHostNotFoundAfterTransaction() {
+        when(vmInstanceMock.getHostId()).thenReturn(hostMockId);
+        when(vmInstanceDaoMock.findByUuid(vmMockUuid)).thenReturn(vmInstanceMock);
+        when(_workJobDao.listPendingWorkJobs(any(), anyLong())).thenReturn(Collections.emptyList());
+        when(_haMgr.hasPendingHaWork(anyLong())).thenReturn(false);
+        doReturn(guru).when(virtualMachineManagerImpl).getVmGuru(vmInstanceMock);
+        doNothing().when(virtualMachineManagerImpl).unmanageVMSnapshots(vmInstanceMock);
+        doNothing().when(virtualMachineManagerImpl).unmanageVMNics(any(VirtualMachineProfile.class), any(VMInstanceVO.class));
+        doNothing().when(virtualMachineManagerImpl).unmanageVMVolumes(vmInstanceMock);
+        doNothing().when(guru).finalizeUnmanage(vmInstanceMock);
+        try (MockedStatic<ApiDBUtils> ignored = Mockito.mockStatic(ApiDBUtils.class)) {
+            when(ApiDBUtils.findHostById(hostMockId)).thenReturn(null);
+            Pair<Boolean, String> result = virtualMachineManagerImpl.unmanage(vmMockUuid, null);
+            assertNull(result.second());
+        }
+    }
+
+    @Test
+    public void testUnmanageSuccessNonKvm() {
+        when(vmInstanceMock.getHostId()).thenReturn(hostMockId);
+        when(hostMock.getUuid()).thenReturn(hostUuid);
+        when(vmInstanceDaoMock.findByUuid(vmMockUuid)).thenReturn(vmInstanceMock);
+        when(_workJobDao.listPendingWorkJobs(any(), anyLong())).thenReturn(Collections.emptyList());
+        when(_haMgr.hasPendingHaWork(anyLong())).thenReturn(false);
+        doReturn(guru).when(virtualMachineManagerImpl).getVmGuru(vmInstanceMock);
+        doNothing().when(virtualMachineManagerImpl).unmanageVMSnapshots(vmInstanceMock);
+        doNothing().when(virtualMachineManagerImpl).unmanageVMNics(any(VirtualMachineProfile.class), any(VMInstanceVO.class));
+        doNothing().when(virtualMachineManagerImpl).unmanageVMVolumes(vmInstanceMock);
+        doNothing().when(guru).finalizeUnmanage(vmInstanceMock);
+        try (MockedStatic<ApiDBUtils> ignored = Mockito.mockStatic(ApiDBUtils.class)) {
+            when(ApiDBUtils.findHostById(hostMockId)).thenReturn(hostMock);
+            try (MockedStatic<ActionEventUtils> actionUtil = Mockito.mockStatic(ActionEventUtils.class)) {
+                actionUtil.when(() -> ActionEventUtils.onActionEvent(
+                        anyLong(), anyLong(), anyLong(),
+                        anyString(), anyString(),
+                        anyLong(), anyString()
+                )).thenReturn(1L);
+
+                Pair<Boolean, String> result = virtualMachineManagerImpl.unmanage(vmMockUuid, null);
+                assertNotNull(result);
+                assertTrue(result.first());
+                assertEquals(hostUuid, result.second());
+
+                verify(virtualMachineManagerImpl, never()).persistDomainForKVM(any(VMInstanceVO.class), anyLong());
+                verify(virtualMachineManagerImpl, times(1)).unmanageVMSnapshots(vmInstanceMock);
+                verify(virtualMachineManagerImpl, times(1)).unmanageVMNics(any(VirtualMachineProfile.class), any(VMInstanceVO.class));
+                verify(virtualMachineManagerImpl, times(1)).unmanageVMVolumes(vmInstanceMock);
+                verify(guru, times(1)).finalizeUnmanage(vmInstanceMock);
+            }
+        }
+    }
+
+    @Test
+    public void testUnmanageSuccessKvm() throws Exception {
+        when(vmInstanceMock.getHostId()).thenReturn(hostMockId);
+        when(hostMock.getUuid()).thenReturn(hostUuid);
+        when(vmInstanceMock.getHypervisorType()).thenReturn(HypervisorType.KVM);
+        when(vmInstanceDaoMock.findByUuid(vmMockUuid)).thenReturn(vmInstanceMock);
+        when(_workJobDao.listPendingWorkJobs(any(), anyLong())).thenReturn(Collections.emptyList());
+        when(_haMgr.hasPendingHaWork(anyLong())).thenReturn(false);
+        doReturn(guru).when(virtualMachineManagerImpl).getVmGuru(vmInstanceMock);
+        doNothing().when(virtualMachineManagerImpl).unmanageVMSnapshots(vmInstanceMock);
+        doNothing().when(virtualMachineManagerImpl).unmanageVMNics(any(VirtualMachineProfile.class), any(VMInstanceVO.class));
+        doNothing().when(virtualMachineManagerImpl).unmanageVMVolumes(vmInstanceMock);
+        doNothing().when(guru).finalizeUnmanage(vmInstanceMock);
+        try (MockedStatic<ApiDBUtils> ignored = Mockito.mockStatic(ApiDBUtils.class)) {
+            when(ApiDBUtils.findHostById(hostMockId)).thenReturn(hostMock);
+            try (MockedStatic<ActionEventUtils> actionUtil = Mockito.mockStatic(ActionEventUtils.class)) {
+                actionUtil.when(() -> ActionEventUtils.onActionEvent(
+                        anyLong(), anyLong(), anyLong(),
+                        anyString(), anyString(),
+                        anyLong(), anyString()
+                )).thenReturn(1L);
+                UnmanageInstanceAnswer successAnswer = new UnmanageInstanceAnswer(null, true, "success");
+                when(agentManagerMock.send(anyLong(), any(UnmanageInstanceCommand.class))).thenReturn(successAnswer);
+                Pair<Boolean, String> result = virtualMachineManagerImpl.unmanage(vmMockUuid, null);
+                assertNotNull(result);
+                assertTrue(result.first());
+                assertEquals(hostUuid, result.second());
+                verify(virtualMachineManagerImpl, times(1)).persistDomainForKVM(vmInstanceMock, null);
+                verify(virtualMachineManagerImpl, times(1)).unmanageVMSnapshots(vmInstanceMock);
+                verify(virtualMachineManagerImpl, times(1)).unmanageVMNics(any(VirtualMachineProfile.class), any(VMInstanceVO.class));
+                verify(virtualMachineManagerImpl, times(1)).unmanageVMVolumes(vmInstanceMock);
+                verify(guru, times(1)).finalizeUnmanage(vmInstanceMock);
+            }
+        }
+    }
+
 }
diff --git a/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtUnmanageInstanceCommandWrapper.java b/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtUnmanageInstanceCommandWrapper.java
new file mode 100644
index 00000000000..33ad274b8d1
--- /dev/null
+++ b/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtUnmanageInstanceCommandWrapper.java
@@ -0,0 +1,174 @@
+//
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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.
+//
+
+package com.cloud.hypervisor.kvm.resource.wrapper;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.StringWriter;
+import java.net.URISyntaxException;
+import java.nio.charset.StandardCharsets;
+
+import javax.xml.parsers.DocumentBuilder;
+import javax.xml.parsers.DocumentBuilderFactory;
+import javax.xml.parsers.ParserConfigurationException;
+import javax.xml.transform.Transformer;
+import javax.xml.transform.TransformerException;
+import javax.xml.transform.TransformerFactory;
+import javax.xml.transform.dom.DOMSource;
+import javax.xml.transform.stream.StreamResult;
+import javax.xml.xpath.XPath;
+import javax.xml.xpath.XPathConstants;
+import javax.xml.xpath.XPathExpressionException;
+import javax.xml.xpath.XPathFactory;
+
+import org.apache.cloudstack.utils.security.ParserUtils;
+import org.apache.commons.io.IOUtils;
+import org.libvirt.Connect;
+import org.libvirt.Domain;
+import org.libvirt.LibvirtException;
+import org.w3c.dom.Document;
+import org.w3c.dom.Node;
+import org.w3c.dom.NodeList;
+import org.xml.sax.SAXException;
+
+import com.cloud.agent.api.Answer;
+import com.cloud.agent.api.UnmanageInstanceAnswer;
+import com.cloud.agent.api.UnmanageInstanceCommand;
+import com.cloud.agent.api.to.VirtualMachineTO;
+import com.cloud.exception.InternalErrorException;
+import com.cloud.hypervisor.kvm.resource.LibvirtComputingResource;
+import com.cloud.hypervisor.kvm.resource.LibvirtKvmAgentHook;
+import com.cloud.hypervisor.kvm.resource.LibvirtVMDef;
+import com.cloud.resource.CommandWrapper;
+import com.cloud.resource.ResourceWrapper;
+
+@ResourceWrapper(handles =  UnmanageInstanceCommand.class)
+public final class LibvirtUnmanageInstanceCommandWrapper extends CommandWrapper<UnmanageInstanceCommand, Answer, LibvirtComputingResource> {
+
+
+    @Override
+    public Answer execute(final UnmanageInstanceCommand command, final LibvirtComputingResource libvirtComputingResource) {
+        String instanceName = command.getInstanceName();
+        VirtualMachineTO vmSpec = command.getVm();
+        final LibvirtUtilitiesHelper libvirtUtilitiesHelper = libvirtComputingResource.getLibvirtUtilitiesHelper();
+        logger.debug("Attempting to unmanage KVM instance: {}", instanceName);
+        Domain dom = null;
+        Connect conn = null;
+        String vmFinalSpecification;
+        try {
+            if (vmSpec == null) {
+                conn = libvirtUtilitiesHelper.getConnectionByVmName(instanceName);
+                dom = conn.domainLookupByName(instanceName);
+                vmFinalSpecification = dom.getXMLDesc(1);
+                if (command.isConfigDriveAttached()) {
+                    vmFinalSpecification = cleanupConfigDrive(vmFinalSpecification, instanceName);
+                }
+            } else {
+                // define domain using reconstructed vmSpec
+                logger.debug("Unmanaging Stopped KVM instance: {}", instanceName);
+                LibvirtVMDef vm = libvirtComputingResource.createVMFromSpec(vmSpec);
+                libvirtComputingResource.createVbd(conn, vmSpec, instanceName, vm);
+                conn = libvirtUtilitiesHelper.getConnectionByType(vm.getHvsType());
+                String vmInitialSpecification = vm.toString();
+                vmFinalSpecification = performXmlTransformHook(vmInitialSpecification, libvirtComputingResource);
+            }
+            conn.domainDefineXML(vmFinalSpecification).free();
+            logger.debug("Successfully unmanaged KVM instance: {} with domain XML: {}", instanceName, vmFinalSpecification);
+            return new UnmanageInstanceAnswer(command, true, "Successfully unmanaged");
+        } catch (final LibvirtException e) {
+            logger.error("LibvirtException occurred during unmanaging instance: {} ", instanceName, e);
+            return new UnmanageInstanceAnswer(command, false, e.getMessage());
+        } catch (final IOException
+                       | ParserConfigurationException
+                       | SAXException
+                       | TransformerException
+                       | XPathExpressionException
+                       | InternalErrorException
+                       | URISyntaxException e) {
+
+            logger.error("Failed to unmanage Instance: {}.", instanceName, e);
+            return new UnmanageInstanceAnswer(command, false, e.getMessage());
+        } finally {
+            if (dom != null) {
+                try {
+                    dom.free();
+                } catch (LibvirtException e) {
+                    logger.error("Ignore libvirt error on free.", e);
+                }
+            }
+        }
+    }
+
+    String cleanupConfigDrive(String domainXML, String instanceName) throws ParserConfigurationException, IOException, SAXException, XPathExpressionException, TransformerException {
+        String isoName = "/" + instanceName + ".iso";
+        DocumentBuilderFactory docFactory = ParserUtils.getSaferDocumentBuilderFactory();
+        DocumentBuilder docBuilder = docFactory.newDocumentBuilder();
+        Document document;
+        try (InputStream inputStream = IOUtils.toInputStream(domainXML, StandardCharsets.UTF_8)) {
+            document = docBuilder.parse(inputStream);
+        }
+        XPathFactory xPathFactory = XPathFactory.newInstance();
+        XPath xpath = xPathFactory.newXPath();
+
+        // Find all <disk device='cdrom'> elements with source file containing instanceName.iso
+        String expression = String.format("//disk[@device='cdrom'][source/@file[contains(., '%s')]]", isoName);
+        NodeList cdromDisks = (NodeList) xpath.evaluate(expression, document, XPathConstants.NODESET);
+
+        // If nothing matched, return original XML
+        if (cdromDisks == null || cdromDisks.getLength() == 0) {
+            logger.debug("No config drive found in domain XML for Instance: {}", instanceName);
+            return domainXML;
+        }
+
+        // Remove all matched config drive disks
+        for (int i = 0; i < cdromDisks.getLength(); i++) {
+            Node diskNode = cdromDisks.item(i);
+            if (diskNode != null && diskNode.getParentNode() != null) {
+                diskNode.getParentNode().removeChild(diskNode);
+            }
+        }
+        logger.debug("Removed {} config drive ISO CD-ROM entries for instance: {}", cdromDisks.getLength(), instanceName);
+
+        TransformerFactory transformerFactory = ParserUtils.getSaferTransformerFactory();
+        Transformer transformer = transformerFactory.newTransformer();
+        DOMSource source = new DOMSource(document);
+        StringWriter output = new StringWriter();
+        StreamResult result = new StreamResult(output);
+        transformer.transform(source, result);
+        return output.toString();
+    }
+
+    private String performXmlTransformHook(String vmInitialSpecification, final LibvirtComputingResource libvirtComputingResource) {
+        String vmFinalSpecification;
+        try {
+            // if transformer fails, everything must go as it's just skipped.
+            LibvirtKvmAgentHook t = libvirtComputingResource.getTransformer();
+            vmFinalSpecification = (String) t.handle(vmInitialSpecification);
+            if (null == vmFinalSpecification) {
+                logger.warn("Libvirt XML transformer returned NULL, will use XML specification unchanged.");
+                vmFinalSpecification = vmInitialSpecification;
+            }
+        } catch(Exception e) {
+            logger.warn("Exception occurred when handling LibVirt XML transformer hook: {}", e);
+            vmFinalSpecification = vmInitialSpecification;
+        }
+        return vmFinalSpecification;
+    }
+}
diff --git a/plugins/hypervisors/kvm/src/test/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtUnmanageInstanceCommandWrapperTest.java b/plugins/hypervisors/kvm/src/test/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtUnmanageInstanceCommandWrapperTest.java
new file mode 100644
index 00000000000..13ddc26610c
--- /dev/null
+++ b/plugins/hypervisors/kvm/src/test/java/com/cloud/hypervisor/kvm/resource/wrapper/LibvirtUnmanageInstanceCommandWrapperTest.java
@@ -0,0 +1,357 @@
+//
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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.
+//
+package com.cloud.hypervisor.kvm.resource.wrapper;
+
+import java.io.IOException;
+
+import javax.xml.parsers.ParserConfigurationException;
+import javax.xml.transform.TransformerException;
+import javax.xml.xpath.XPathExpressionException;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Spy;
+import org.mockito.junit.MockitoJUnitRunner;
+import org.xml.sax.SAXException;
+
+
+@RunWith(MockitoJUnitRunner.class)
+public class LibvirtUnmanageInstanceCommandWrapperTest {
+    @Spy
+    LibvirtUnmanageInstanceCommandWrapper unmanageInstanceCommandWrapper = new LibvirtUnmanageInstanceCommandWrapper();
+
+    @Test
+    public void testCleanupConfigDriveFromDomain() throws XPathExpressionException, ParserConfigurationException, IOException, TransformerException, SAXException {
+        String domainXML = "<domain type='kvm' id='6'>\n" +
+                "  <name>i-2-6-VM</name>\n" +
+                "  <uuid>071628d0-84f1-421e-a9cf-d18bca2283bc</uuid>\n" +
+                "  <description>CentOS 5.5 (64-bit)</description>\n" +
+                "  <memory unit='KiB'>524288</memory>\n" +
+                "  <currentMemory unit='KiB'>524288</currentMemory>\n" +
+                "  <vcpu placement='static'>1</vcpu>\n" +
+                "  <cputune>\n" +
+                "    <shares>250</shares>\n" +
+                "  </cputune>\n" +
+                "  <resource>\n" +
+                "    <partition>/machine</partition>\n" +
+                "  </resource>\n" +
+                "  <sysinfo type='smbios'>\n" +
+                "    <system>\n" +
+                "      <entry name='manufacturer'>Apache Software Foundation</entry>\n" +
+                "      <entry name='product'>CloudStack KVM Hypervisor</entry>\n" +
+                "      <entry name='serial'>071628d0-84f1-421e-a9cf-d18bca2283bc</entry>\n" +
+                "      <entry name='uuid'>071628d0-84f1-421e-a9cf-d18bca2283bc</entry>\n" +
+                "    </system>\n" +
+                "  </sysinfo>\n" +
+                "  <os>\n" +
+                "    <type arch='x86_64' machine='pc-i440fx-rhel7.6.0'>hvm</type>\n" +
+                "    <boot dev='cdrom'/>\n" +
+                "    <boot dev='hd'/>\n" +
+                "    <smbios mode='sysinfo'/>\n" +
+                "  </os>\n" +
+                "  <features>\n" +
+                "    <acpi/>\n" +
+                "    <apic/>\n" +
+                "    <pae/>\n" +
+                "  </features>\n" +
+                "  <cpu mode='custom' match='exact' check='full'>\n" +
+                "    <model fallback='forbid'>qemu64</model>\n" +
+                "    <topology sockets='1' dies='1' cores='1' threads='1'/>\n" +
+                "    <feature policy='require' name='x2apic'/>\n" +
+                "    <feature policy='require' name='hypervisor'/>\n" +
+                "    <feature policy='require' name='lahf_lm'/>\n" +
+                "    <feature policy='disable' name='svm'/>\n" +
+                "  </cpu>\n" +
+                "  <clock offset='utc'>\n" +
+                "    <timer name='kvmclock'/>\n" +
+                "  </clock>\n" +
+                "  <on_poweroff>destroy</on_poweroff>\n" +
+                "  <on_reboot>restart</on_reboot>\n" +
+                "  <on_crash>destroy</on_crash>\n" +
+                "  <devices>\n" +
+                "    <emulator>/usr/libexec/qemu-kvm</emulator>\n" +
+                "    <disk type='file' device='disk'>\n" +
+                "      <driver name='qemu' type='qcow2' cache='none'/>\n" +
+                "      <source file='/mnt/c88eb8f7-4516-3e35-a226-8737b9470a92/7617e6b5-570c-40d6-99a9-c74bb174c988' index='3'/>\n" +
+                "      <backingStore type='file' index='4'>\n" +
+                "        <format type='qcow2'/>\n" +
+                "        <source file='/mnt/c88eb8f7-4516-3e35-a226-8737b9470a92/7408cffa-9860-11f0-a923-1e0097000303'/>\n" +
+                "        <backingStore/>\n" +
+                "      </backingStore>\n" +
+                "      <target dev='vda' bus='virtio'/>\n" +
+                "      <serial>9329e4fa9db546c78b1a</serial>\n" +
+                "      <alias name='virtio-disk0'/>\n" +
+                "      <address type='pci' domain='0x0000' bus='0x00' slot='0x05' function='0x0'/>\n" +
+                "    </disk>\n" +
+                "    <disk type='file' device='cdrom'>\n" +
+                "      <driver name='qemu'/>\n" +
+                "      <target dev='hdc' bus='ide'/>\n" +
+                "      <readonly/>\n" +
+                "      <alias name='ide0-1-0'/>\n" +
+                "      <address type='drive' controller='0' bus='1' target='0' unit='0'/>\n" +
+                "    </disk>\n" +
+                "    <disk type='file' device='cdrom'>\n" +
+                "      <driver name='qemu' type='raw'/>\n" +
+                "      <source file='/mnt/fec63e8a-d390-31e0-a02d-8e634aabded7/i-2-6-VM.iso' index='1'/>\n" +
+                "      <backingStore/>\n" +
+                "      <target dev='hdd' bus='ide'/>\n" +
+                "      <readonly/>\n" +
+                "      <alias name='ide0-1-1'/>\n" +
+                "      <address type='drive' controller='0' bus='1' target='0' unit='1'/>\n" +
+                "    </disk>\n" +
+                "    <controller type='usb' index='0' model='piix3-uhci'>\n" +
+                "      <alias name='usb'/>\n" +
+                "      <address type='pci' domain='0x0000' bus='0x00' slot='0x01' function='0x2'/>\n" +
+                "    </controller>\n" +
+                "    <controller type='pci' index='0' model='pci-root'>\n" +
+                "      <alias name='pci.0'/>\n" +
+                "    </controller>\n" +
+                "    <controller type='ide' index='0'>\n" +
+                "      <alias name='ide'/>\n" +
+                "      <address type='pci' domain='0x0000' bus='0x00' slot='0x01' function='0x1'/>\n" +
+                "    </controller>\n" +
+                "    <controller type='virtio-serial' index='0'>\n" +
+                "      <alias name='virtio-serial0'/>\n" +
+                "      <address type='pci' domain='0x0000' bus='0x00' slot='0x04' function='0x0'/>\n" +
+                "    </controller>\n" +
+                "    <interface type='bridge'>\n" +
+                "      <mac address='02:01:00:cc:00:02'/>\n" +
+                "      <source bridge='breth1-2765'/>\n" +
+                "      <bandwidth>\n" +
+                "        <inbound average='25000' peak='25000'/>\n" +
+                "        <outbound average='25000' peak='25000'/>\n" +
+                "      </bandwidth>\n" +
+                "      <target dev='vnet6'/>\n" +
+                "      <model type='virtio'/>\n" +
+                "      <link state='up'/>\n" +
+                "      <alias name='net0'/>\n" +
+                "      <address type='pci' domain='0x0000' bus='0x00' slot='0x03' function='0x0'/>\n" +
+                "    </interface>\n" +
+                "    <serial type='pty'>\n" +
+                "      <source path='/dev/pts/1'/>\n" +
+                "      <target type='isa-serial' port='0'>\n" +
+                "        <model name='isa-serial'/>\n" +
+                "      </target>\n" +
+                "      <alias name='serial0'/>\n" +
+                "    </serial>\n" +
+                "    <console type='pty' tty='/dev/pts/1'>\n" +
+                "      <source path='/dev/pts/1'/>\n" +
+                "      <target type='serial' port='0'/>\n" +
+                "      <alias name='serial0'/>\n" +
+                "    </console>\n" +
+                "    <channel type='unix'>\n" +
+                "      <source mode='bind' path='/var/lib/libvirt/qemu/i-2-6-VM.org.qemu.guest_agent.0'/>\n" +
+                "      <target type='virtio' name='org.qemu.guest_agent.0' state='disconnected'/>\n" +
+                "      <alias name='channel0'/>\n" +
+                "      <address type='virtio-serial' controller='0' bus='0' port='1'/>\n" +
+                "    </channel>\n" +
+                "    <input type='tablet' bus='usb'>\n" +
+                "      <alias name='input0'/>\n" +
+                "      <address type='usb' bus='0' port='1'/>\n" +
+                "    </input>\n" +
+                "    <input type='mouse' bus='ps2'>\n" +
+                "      <alias name='input1'/>\n" +
+                "    </input>\n" +
+                "    <input type='keyboard' bus='ps2'>\n" +
+                "      <alias name='input2'/>\n" +
+                "    </input>\n" +
+                "    <graphics type='vnc' port='5901' autoport='yes' listen='10.0.35.235' passwd='l1wi9CFu'>\n" +
+                "      <listen type='address' address='10.0.35.235'/>\n" +
+                "    </graphics>\n" +
+                "    <audio id='1' type='none'/>\n" +
+                "    <video>\n" +
+                "      <model type='cirrus' vram='16384' heads='1' primary='yes'/>\n" +
+                "      <alias name='video0'/>\n" +
+                "      <address type='pci' domain='0x0000' bus='0x00' slot='0x02' function='0x0'/>\n" +
+                "    </video>\n" +
+                "    <watchdog model='i6300esb' action='none'>\n" +
+                "      <alias name='watchdog0'/>\n" +
+                "      <address type='pci' domain='0x0000' bus='0x00' slot='0x07' function='0x0'/>\n" +
+                "    </watchdog>\n" +
+                "    <memballoon model='virtio'>\n" +
+                "      <alias name='balloon0'/>\n" +
+                "      <address type='pci' domain='0x0000' bus='0x00' slot='0x06' function='0x0'/>\n" +
+                "    </memballoon>\n" +
+                "  </devices>\n" +
+                "  <seclabel type='dynamic' model='dac' relabel='yes'>\n" +
+                "    <label>+0:+0</label>\n" +
+                "    <imagelabel>+0:+0</imagelabel>\n" +
+                "  </seclabel>\n" +
+                "</domain>\n";
+
+        String cleanupXML = unmanageInstanceCommandWrapper.cleanupConfigDrive(domainXML, "i-2-6-VM");
+        Assert.assertNotEquals(domainXML, cleanupXML);
+    }
+
+
+    @Test
+    public void testCleanupConfigDriveFromDomainNoConfigDrive() throws XPathExpressionException, ParserConfigurationException, IOException, TransformerException, SAXException {
+        String domainXML = "<domain type='kvm' id='6'>\n" +
+                "  <name>i-2-6-VM</name>\n" +
+                "  <uuid>071628d0-84f1-421e-a9cf-d18bca2283bc</uuid>\n" +
+                "  <description>CentOS 5.5 (64-bit)</description>\n" +
+                "  <memory unit='KiB'>524288</memory>\n" +
+                "  <currentMemory unit='KiB'>524288</currentMemory>\n" +
+                "  <vcpu placement='static'>1</vcpu>\n" +
+                "  <cputune>\n" +
+                "    <shares>250</shares>\n" +
+                "  </cputune>\n" +
+                "  <resource>\n" +
+                "    <partition>/machine</partition>\n" +
+                "  </resource>\n" +
+                "  <sysinfo type='smbios'>\n" +
+                "    <system>\n" +
+                "      <entry name='manufacturer'>Apache Software Foundation</entry>\n" +
+                "      <entry name='product'>CloudStack KVM Hypervisor</entry>\n" +
+                "      <entry name='serial'>071628d0-84f1-421e-a9cf-d18bca2283bc</entry>\n" +
+                "      <entry name='uuid'>071628d0-84f1-421e-a9cf-d18bca2283bc</entry>\n" +
+                "    </system>\n" +
+                "  </sysinfo>\n" +
+                "  <os>\n" +
+                "    <type arch='x86_64' machine='pc-i440fx-rhel7.6.0'>hvm</type>\n" +
+                "    <boot dev='cdrom'/>\n" +
+                "    <boot dev='hd'/>\n" +
+                "    <smbios mode='sysinfo'/>\n" +
+                "  </os>\n" +
+                "  <features>\n" +
+                "    <acpi/>\n" +
+                "    <apic/>\n" +
+                "    <pae/>\n" +
+                "  </features>\n" +
+                "  <cpu mode='custom' match='exact' check='full'>\n" +
+                "    <model fallback='forbid'>qemu64</model>\n" +
+                "    <topology sockets='1' dies='1' cores='1' threads='1'/>\n" +
+                "    <feature policy='require' name='x2apic'/>\n" +
+                "    <feature policy='require' name='hypervisor'/>\n" +
+                "    <feature policy='require' name='lahf_lm'/>\n" +
+                "    <feature policy='disable' name='svm'/>\n" +
+                "  </cpu>\n" +
+                "  <clock offset='utc'>\n" +
+                "    <timer name='kvmclock'/>\n" +
+                "  </clock>\n" +
+                "  <on_poweroff>destroy</on_poweroff>\n" +
+                "  <on_reboot>restart</on_reboot>\n" +
+                "  <on_crash>destroy</on_crash>\n" +
+                "  <devices>\n" +
+                "    <emulator>/usr/libexec/qemu-kvm</emulator>\n" +
+                "    <disk type='file' device='disk'>\n" +
+                "      <driver name='qemu' type='qcow2' cache='none'/>\n" +
+                "      <source file='/mnt/c88eb8f7-4516-3e35-a226-8737b9470a92/7617e6b5-570c-40d6-99a9-c74bb174c988' index='3'/>\n" +
+                "      <backingStore type='file' index='4'>\n" +
+                "        <format type='qcow2'/>\n" +
+                "        <source file='/mnt/c88eb8f7-4516-3e35-a226-8737b9470a92/7408cffa-9860-11f0-a923-1e0097000303'/>\n" +
+                "        <backingStore/>\n" +
+                "      </backingStore>\n" +
+                "      <target dev='vda' bus='virtio'/>\n" +
+                "      <serial>9329e4fa9db546c78b1a</serial>\n" +
+                "      <alias name='virtio-disk0'/>\n" +
+                "      <address type='pci' domain='0x0000' bus='0x00' slot='0x05' function='0x0'/>\n" +
+                "    </disk>\n" +
+                "    <disk type='file' device='cdrom'>\n" +
+                "      <driver name='qemu'/>\n" +
+                "      <target dev='hdc' bus='ide'/>\n" +
+                "      <readonly/>\n" +
+                "      <alias name='ide0-1-0'/>\n" +
+                "      <address type='drive' controller='0' bus='1' target='0' unit='0'/>\n" +
+                "    </disk>\n" +
+                "    <controller type='usb' index='0' model='piix3-uhci'>\n" +
+                "      <alias name='usb'/>\n" +
+                "      <address type='pci' domain='0x0000' bus='0x00' slot='0x01' function='0x2'/>\n" +
+                "    </controller>\n" +
+                "    <controller type='pci' index='0' model='pci-root'>\n" +
+                "      <alias name='pci.0'/>\n" +
+                "    </controller>\n" +
+                "    <controller type='ide' index='0'>\n" +
+                "      <alias name='ide'/>\n" +
+                "      <address type='pci' domain='0x0000' bus='0x00' slot='0x01' function='0x1'/>\n" +
+                "    </controller>\n" +
+                "    <controller type='virtio-serial' index='0'>\n" +
+                "      <alias name='virtio-serial0'/>\n" +
+                "      <address type='pci' domain='0x0000' bus='0x00' slot='0x04' function='0x0'/>\n" +
+                "    </controller>\n" +
+                "    <interface type='bridge'>\n" +
+                "      <mac address='02:01:00:cc:00:02'/>\n" +
+                "      <source bridge='breth1-2765'/>\n" +
+                "      <bandwidth>\n" +
+                "        <inbound average='25000' peak='25000'/>\n" +
+                "        <outbound average='25000' peak='25000'/>\n" +
+                "      </bandwidth>\n" +
+                "      <target dev='vnet6'/>\n" +
+                "      <model type='virtio'/>\n" +
+                "      <link state='up'/>\n" +
+                "      <alias name='net0'/>\n" +
+                "      <address type='pci' domain='0x0000' bus='0x00' slot='0x03' function='0x0'/>\n" +
+                "    </interface>\n" +
+                "    <serial type='pty'>\n" +
+                "      <source path='/dev/pts/1'/>\n" +
+                "      <target type='isa-serial' port='0'>\n" +
+                "        <model name='isa-serial'/>\n" +
+                "      </target>\n" +
+                "      <alias name='serial0'/>\n" +
+                "    </serial>\n" +
+                "    <console type='pty' tty='/dev/pts/1'>\n" +
+                "      <source path='/dev/pts/1'/>\n" +
+                "      <target type='serial' port='0'/>\n" +
+                "      <alias name='serial0'/>\n" +
+                "    </console>\n" +
+                "    <channel type='unix'>\n" +
+                "      <source mode='bind' path='/var/lib/libvirt/qemu/i-2-6-VM.org.qemu.guest_agent.0'/>\n" +
+                "      <target type='virtio' name='org.qemu.guest_agent.0' state='disconnected'/>\n" +
+                "      <alias name='channel0'/>\n" +
+                "      <address type='virtio-serial' controller='0' bus='0' port='1'/>\n" +
+                "    </channel>\n" +
+                "    <input type='tablet' bus='usb'>\n" +
+                "      <alias name='input0'/>\n" +
+                "      <address type='usb' bus='0' port='1'/>\n" +
+                "    </input>\n" +
+                "    <input type='mouse' bus='ps2'>\n" +
+                "      <alias name='input1'/>\n" +
+                "    </input>\n" +
+                "    <input type='keyboard' bus='ps2'>\n" +
+                "      <alias name='input2'/>\n" +
+                "    </input>\n" +
+                "    <graphics type='vnc' port='5901' autoport='yes' listen='10.0.35.235' passwd='l1wi9CFu'>\n" +
+                "      <listen type='address' address='10.0.35.235'/>\n" +
+                "    </graphics>\n" +
+                "    <audio id='1' type='none'/>\n" +
+                "    <video>\n" +
+                "      <model type='cirrus' vram='16384' heads='1' primary='yes'/>\n" +
+                "      <alias name='video0'/>\n" +
+                "      <address type='pci' domain='0x0000' bus='0x00' slot='0x02' function='0x0'/>\n" +
+                "    </video>\n" +
+                "    <watchdog model='i6300esb' action='none'>\n" +
+                "      <alias name='watchdog0'/>\n" +
+                "      <address type='pci' domain='0x0000' bus='0x00' slot='0x07' function='0x0'/>\n" +
+                "    </watchdog>\n" +
+                "    <memballoon model='virtio'>\n" +
+                "      <alias name='balloon0'/>\n" +
+                "      <address type='pci' domain='0x0000' bus='0x00' slot='0x06' function='0x0'/>\n" +
+                "    </memballoon>\n" +
+                "  </devices>\n" +
+                "  <seclabel type='dynamic' model='dac' relabel='yes'>\n" +
+                "    <label>+0:+0</label>\n" +
+                "    <imagelabel>+0:+0</imagelabel>\n" +
+                "  </seclabel>\n" +
+                "</domain>\n";
+
+        String cleanupXML = unmanageInstanceCommandWrapper.cleanupConfigDrive(domainXML, "i-2-6-VM");
+        Assert.assertEquals(domainXML, cleanupXML);
+    }
+}
diff --git a/server/src/main/java/com/cloud/vm/UserVmManagerImpl.java b/server/src/main/java/com/cloud/vm/UserVmManagerImpl.java
index 854886d233e..a9877b97f84 100644
--- a/server/src/main/java/com/cloud/vm/UserVmManagerImpl.java
+++ b/server/src/main/java/com/cloud/vm/UserVmManagerImpl.java
@@ -9416,18 +9416,19 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
     }
 
     @Override
-    public boolean unmanageUserVM(Long vmId) {
+    public Pair<Boolean, String> unmanageUserVM(Long vmId, Long paramHostId) {
         UserVmVO vm = _vmDao.findById(vmId);
         if (vm == null || vm.getRemoved() != null) {
             throw new InvalidParameterValueException("Unable to find a VM with ID = " + vmId);
         }
 
         vm = _vmDao.acquireInLockTable(vm.getId());
-        boolean result;
+
         try {
             if (vm.getState() != State.Running && vm.getState() != State.Stopped) {
-                logger.debug("VM {} is not running or stopped, cannot be unmanaged", vm);
-                return false;
+                String errorMsg = "Instance: " + vm.getName() + " is not running or stopped, cannot be unmanaged";
+                logger.debug(errorMsg);
+                throw new CloudRuntimeException(errorMsg);
             }
 
             if (!UnmanagedVMsManager.isSupported(vm.getHypervisorType())) {
@@ -9439,22 +9440,21 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
             checkUnmanagingVMOngoingVolumeSnapshots(vm);
             checkUnmanagingVMVolumes(vm, volumes);
 
-            result = _itMgr.unmanage(vm.getUuid());
-            if (result) {
+            Pair<Boolean, String> result = _itMgr.unmanage(vm.getUuid(), paramHostId);
+            if (result.first()) {
                 cleanupUnmanageVMResources(vm);
                 unmanageVMFromDB(vm.getId());
                 publishUnmanageVMUsageEvents(vm, volumes);
             } else {
                 throw new CloudRuntimeException("Error while unmanaging VM: " + vm.getUuid());
             }
+            return result;
         } catch (Exception e) {
             logger.error("Could not unmanage VM {}", vm, e);
             throw new CloudRuntimeException(e);
         } finally {
             _vmDao.releaseFromLockTable(vm.getId());
         }
-
-        return true;
     }
 
     private void updateDetailsWithRootDiskAttributes(Map<String, String> details, VmDiskInfo rootVmDiskInfo) {
@@ -9687,7 +9687,7 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
     /*
         Generate usage events related to unmanaging a VM
      */
-    private void publishUnmanageVMUsageEvents(UserVmVO vm, List<VolumeVO> volumes) {
+    void publishUnmanageVMUsageEvents(UserVmVO vm, List<VolumeVO> volumes) {
         postProcessingUnmanageVMVolumes(volumes, vm);
         postProcessingUnmanageVM(vm);
     }
@@ -9695,12 +9695,12 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
     /*
         Cleanup the VM from resources and groups
      */
-    private void cleanupUnmanageVMResources(UserVmVO vm) {
+    void cleanupUnmanageVMResources(UserVmVO vm) {
         cleanupVmResources(vm);
         removeVMFromAffinityGroups(vm.getId());
     }
 
-    private void unmanageVMFromDB(long vmId) {
+    void unmanageVMFromDB(long vmId) {
         VMInstanceVO vm = _vmInstanceDao.findById(vmId);
         vmInstanceDetailsDao.removeDetails(vmId);
         vm.setState(State.Expunging);
@@ -9762,7 +9762,7 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
         }
     }
 
-    private void checkUnmanagingVMOngoingVolumeSnapshots(UserVmVO vm) {
+    void checkUnmanagingVMOngoingVolumeSnapshots(UserVmVO vm) {
         logger.debug("Checking if there are any ongoing snapshots on the ROOT volumes associated with VM {}", vm);
         if (checkStatusOfVolumeSnapshots(vm, Volume.Type.ROOT)) {
             throw new CloudRuntimeException("There is/are unbacked up snapshot(s) on ROOT volume, vm unmanage is not permitted, please try again later.");
@@ -9770,7 +9770,7 @@ public class UserVmManagerImpl extends ManagerBase implements UserVmManager, Vir
         logger.debug("Found no ongoing snapshots on volume of type ROOT, for the vm {}", vm);
     }
 
-    private void checkUnmanagingVMVolumes(UserVmVO vm, List<VolumeVO> volumes) {
+    void checkUnmanagingVMVolumes(UserVmVO vm, List<VolumeVO> volumes) {
         for (VolumeVO volume : volumes) {
             if (volume.getInstanceId() == null || !volume.getInstanceId().equals(vm.getId())) {
                 throw new CloudRuntimeException(String.format("Invalid state for volume %s of VM %s: it is not attached to VM", volume, vm));
diff --git a/server/src/main/java/org/apache/cloudstack/vm/UnmanagedVMsManagerImpl.java b/server/src/main/java/org/apache/cloudstack/vm/UnmanagedVMsManagerImpl.java
index 285760f811b..9f38182e4d5 100644
--- a/server/src/main/java/org/apache/cloudstack/vm/UnmanagedVMsManagerImpl.java
+++ b/server/src/main/java/org/apache/cloudstack/vm/UnmanagedVMsManagerImpl.java
@@ -1128,7 +1128,7 @@ public class UnmanagedVMsManagerImpl implements UnmanagedVMsManager {
         }
 
         String internalCSName = unmanagedInstance.getInternalCSName();
-        if (StringUtils.isEmpty(instanceNameInternal)) {
+        if (StringUtils.isEmpty(internalCSName)) {
             internalCSName = instanceNameInternal;
         }
         Map<String, String> allDetails = new HashMap<>(details);
@@ -2194,7 +2194,7 @@ public class UnmanagedVMsManagerImpl implements UnmanagedVMsManager {
      * - VM must not have any associated volume snapshot
      * - VM must not have an attached ISO
      */
-    private void performUnmanageVMInstancePrechecks(VMInstanceVO vmVO) {
+    void performUnmanageVMInstancePrechecks(VMInstanceVO vmVO) {
         if (hasVolumeSnapshotsPriorToUnmanageVM(vmVO)) {
             throw new UnsupportedServiceException("Cannot unmanage VM with id = " + vmVO.getUuid() +
                     " as there are volume snapshots for its volume(s). Please remove snapshots before unmanaging.");
@@ -2254,7 +2254,7 @@ public class UnmanagedVMsManagerImpl implements UnmanagedVMsManager {
 
     @Override
     @ActionEvent(eventType = EventTypes.EVENT_VM_UNMANAGE, eventDescription = "unmanaging VM", async = true)
-    public boolean unmanageVMInstance(long vmId) {
+    public Pair<Boolean, String> unmanageVMInstance(long vmId, Long paramHostId, boolean isForced) {
         VMInstanceVO vmVO = vmDao.findById(vmId);
         if (vmVO == null || vmVO.getRemoved() != null) {
             throw new InvalidParameterValueException("Could not find VM to unmanage, it is either removed or not existing VM");
@@ -2265,6 +2265,12 @@ public class UnmanagedVMsManagerImpl implements UnmanagedVMsManager {
                     vmVO.getHypervisorType().toString());
         } else if (vmVO.getType() != VirtualMachine.Type.User) {
             throw new UnsupportedServiceException("Unmanage VM is currently allowed for guest VMs only");
+        } else if (paramHostId != null &&
+                (vmVO.getHypervisorType() != Hypervisor.HypervisorType.KVM || vmVO.getState() != VirtualMachine.State.Stopped)) {
+            throw new UnsupportedServiceException("Param hostid is only supported for KVM hypervisor for stopped Instances.");
+        } else if (!isForced && vmVO.getHypervisorType() == Hypervisor.HypervisorType.KVM
+                && vmInstanceDetailsDao.findDetail(vmId, VmDetailConstants.CONFIG_DRIVE_LOCATION) != null) {
+            throw new UnsupportedServiceException("Config drive is attached to Instance, use forced param true from API to unmanage it.");
         }
 
         if (vmVO.getType().equals(VirtualMachine.Type.User)) {
@@ -2276,14 +2282,15 @@ public class UnmanagedVMsManagerImpl implements UnmanagedVMsManager {
 
         performUnmanageVMInstancePrechecks(vmVO);
 
-        Long hostId = findSuitableHostId(vmVO);
+        boolean isKvmVmStopped = VirtualMachine.State.Stopped.equals(vmVO.getState()) && vmVO.getHypervisorType() == Hypervisor.HypervisorType.KVM;
+        Long hostId = isKvmVmStopped ? vmVO.getLastHostId() : findSuitableHostId(vmVO);
         String instanceName = vmVO.getInstanceName();
 
-        if (!existsVMToUnmanage(instanceName, hostId)) {
+        if (!isKvmVmStopped && !existsVMToUnmanage(instanceName, hostId)) {
             throw new CloudRuntimeException(String.format("VM %s is not found in the hypervisor", vmVO));
         }
 
-        return userVmManager.unmanageUserVM(vmId);
+        return userVmManager.unmanageUserVM(vmId, paramHostId);
     }
 
     /**
diff --git a/server/src/test/java/com/cloud/vm/UserVmManagerImplTest.java b/server/src/test/java/com/cloud/vm/UserVmManagerImplTest.java
index 66e8783e77b..360800fac34 100644
--- a/server/src/test/java/com/cloud/vm/UserVmManagerImplTest.java
+++ b/server/src/test/java/com/cloud/vm/UserVmManagerImplTest.java
@@ -33,8 +33,10 @@ import static org.mockito.ArgumentMatchers.isNull;
 import static org.mockito.ArgumentMatchers.nullable;
 import static org.mockito.Mockito.doNothing;
 import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.lenient;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
@@ -45,6 +47,7 @@ import java.time.LocalDateTime;
 import java.time.ZoneOffset;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.Date;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -86,6 +89,7 @@ import org.apache.cloudstack.storage.datastore.db.PrimaryDataStoreDao;
 import org.apache.cloudstack.storage.datastore.db.StoragePoolVO;
 import org.apache.cloudstack.storage.template.VnfTemplateManager;
 import org.apache.cloudstack.userdata.UserDataManager;
+import org.apache.cloudstack.vm.UnmanagedVMsManager;
 import org.apache.cloudstack.vm.lease.VMLeaseManager;
 import org.junit.After;
 import org.junit.Assert;
@@ -431,6 +435,8 @@ public class UserVmManagerImplTest {
     @Mock
     private UUIDManager uuidMgr;
 
+    MockedStatic<UnmanagedVMsManager> unmanagedVMsManagerMockedStatic;
+
     private static final long vmId = 1l;
     private static final long zoneId = 2L;
     private static final long accountId = 3L;
@@ -438,6 +444,7 @@ public class UserVmManagerImplTest {
     private static final long templateId = 11L;
     private static final long volumeId = 1L;
     private static final long snashotId = 1L;
+    private static final String vmUuid = UUID.randomUUID().toString();
 
     private static final long GiB_TO_BYTES = 1024 * 1024 * 1024;
 
@@ -472,11 +479,13 @@ public class UserVmManagerImplTest {
         lenient().doNothing().when(resourceLimitMgr).decrementResourceCount(anyLong(), any(Resource.ResourceType.class), anyLong());
 
         Mockito.when(virtualMachineProfile.getId()).thenReturn(vmId);
+        unmanagedVMsManagerMockedStatic = mockStatic(UnmanagedVMsManager.class);
     }
 
     @After
     public void afterTest() {
         CallContext.unregister();
+        unmanagedVMsManagerMockedStatic.close();
     }
 
     @Test
@@ -1409,7 +1418,7 @@ public class UserVmManagerImplTest {
         CloudRuntimeException cre = new CloudRuntimeException("Error and CloudRuntimeException is thrown");
         cre.addProxyObject(vmId, "vmId");
 
-        Mockito.doThrow(cre).when(userVmManagerImpl).createBasicSecurityGroupVirtualMachine(any(), any(), any(), any(), any(), any(), any(),
+        doThrow(cre).when(userVmManagerImpl).createBasicSecurityGroupVirtualMachine(any(), any(), any(), any(), any(), any(), any(),
                 any(), any(), any(), any(), any(), any(), any(), any(), any(), any(), any(), any(), nullable(Boolean.class), any(), any(), any(),
                 any(), any(), any(), any(), eq(true), any(), any(), any());
 
@@ -1784,7 +1793,7 @@ public class UserVmManagerImplTest {
     public void checkExpungeVmPermissionTestAccountIsNotAdminConfigTrueNoApiAccessThrowsPermissionDeniedException () {
         Mockito.doReturn(false).when(accountManager).isAdmin(Mockito.anyLong());
         Mockito.doReturn(true).when(userVmManagerImpl).getConfigAllowUserExpungeRecoverVm(Mockito.anyLong());
-        Mockito.doThrow(PermissionDeniedException.class).when(accountManager).checkApiAccess(accountMock, "expungeVirtualMachine");
+        doThrow(PermissionDeniedException.class).when(accountManager).checkApiAccess(accountMock, "expungeVirtualMachine");
 
         Assert.assertThrows(PermissionDeniedException.class, () -> userVmManagerImpl.checkExpungeVmPermission(accountMock));
     }
@@ -1798,7 +1807,7 @@ public class UserVmManagerImplTest {
     @Test
     public void checkExpungeVmPermissionTestAccountIsAdminNoApiAccessThrowsPermissionDeniedException () {
         Mockito.doReturn(true).when(accountManager).isAdmin(Mockito.anyLong());
-        Mockito.doThrow(PermissionDeniedException.class).when(accountManager).checkApiAccess(accountMock, "expungeVirtualMachine");
+        doThrow(PermissionDeniedException.class).when(accountManager).checkApiAccess(accountMock, "expungeVirtualMachine");
 
         Assert.assertThrows(PermissionDeniedException.class, () -> userVmManagerImpl.checkExpungeVmPermission(accountMock));
     }
@@ -2604,7 +2613,7 @@ public class UserVmManagerImplTest {
         Mockito.doReturn(ControlledEntity.ACLType.Domain).when(networkMock).getAclType();
         Mockito.doReturn(Network.GuestType.L2).when(networkMock).getGuestType();
 
-        Mockito.doThrow(PermissionDeniedException.class).when(networkModel).checkNetworkPermissions(accountMock, networkMock);
+        doThrow(PermissionDeniedException.class).when(networkModel).checkNetworkPermissions(accountMock, networkMock);
 
         boolean canAccountUseNetwork = userVmManagerImpl.canAccountUseNetwork(accountMock, networkMock);
 
@@ -2616,7 +2625,7 @@ public class UserVmManagerImplTest {
         CallContext callContextMock = Mockito.mock(CallContext.class);
         NetworkVO currentNetwork = Mockito.mock(NetworkVO.class);
 
-        try (MockedStatic<CallContext> ignored = Mockito.mockStatic(CallContext.class)) {
+        try (MockedStatic<CallContext> ignored = mockStatic(CallContext.class)) {
             Mockito.when(CallContext.current()).thenReturn(callContextMock);
 
             Mockito.doReturn(1l).when(callContextMock).getCallingUserId();
@@ -2635,7 +2644,7 @@ public class UserVmManagerImplTest {
         CallContext callContextMock = Mockito.mock(CallContext.class);
         NetworkVO currentNetwork = Mockito.mock(NetworkVO.class);
 
-        try (MockedStatic<CallContext> ignored = Mockito.mockStatic(CallContext.class)) {
+        try (MockedStatic<CallContext> ignored = mockStatic(CallContext.class)) {
             Mockito.when(CallContext.current()).thenReturn(callContextMock);
 
             Mockito.doReturn(1l).when(callContextMock).getCallingUserId();
@@ -2657,7 +2666,7 @@ public class UserVmManagerImplTest {
         CallContext callContextMock = Mockito.mock(CallContext.class);
         NetworkVO currentNetwork = Mockito.mock(NetworkVO.class);
 
-        try (MockedStatic<CallContext> ignored = Mockito.mockStatic(CallContext.class)) {
+        try (MockedStatic<CallContext> ignored = mockStatic(CallContext.class)) {
             Mockito.when(CallContext.current()).thenReturn(callContextMock);
 
             Mockito.doReturn(1l).when(callContextMock).getCallingUserId();
@@ -2680,7 +2689,7 @@ public class UserVmManagerImplTest {
         CallContext callContextMock = Mockito.mock(CallContext.class);
         NetworkVO currentNetwork = Mockito.mock(NetworkVO.class);
 
-        try (MockedStatic<CallContext> ignored = Mockito.mockStatic(CallContext.class)) {
+        try (MockedStatic<CallContext> ignored = mockStatic(CallContext.class)) {
             Mockito.when(CallContext.current()).thenReturn(callContextMock);
 
             Mockito.doReturn(1l).when(callContextMock).getCallingUserId();
@@ -2704,7 +2713,7 @@ public class UserVmManagerImplTest {
 
         CallContext callContextMock = Mockito.mock(CallContext.class);
 
-        try (MockedStatic<CallContext> ignored = Mockito.mockStatic(CallContext.class)) {
+        try (MockedStatic<CallContext> ignored = mockStatic(CallContext.class)) {
             Mockito.when(CallContext.current()).thenReturn(callContextMock);
 
             Mockito.doReturn(1l).when(callContextMock).getCallingUserId();
@@ -2712,7 +2721,7 @@ public class UserVmManagerImplTest {
             Pair<? extends NetworkGuru, ? extends Network> implementedNetwork = Mockito.mock(Pair.class);
 
             Mockito.doReturn(callerUser).when(userDao).findById(Mockito.anyLong());
-            Mockito.doThrow(InvalidParameterValueException.class).when(_networkMgr).implementNetwork(Mockito.anyLong(), Mockito.any(), Mockito.any());
+            doThrow(InvalidParameterValueException.class).when(_networkMgr).implementNetwork(Mockito.anyLong(), Mockito.any(), Mockito.any());
 
             CloudRuntimeException assertThrows = Assert.assertThrows(expectedCloudRuntimeException, () -> {
                 userVmManagerImpl.implementNetwork(accountMock, _dcMock, networkMock);
@@ -2968,7 +2977,7 @@ public class UserVmManagerImplTest {
     public void moveVmToUserTestValidateVmExistsAndIsNotRunningThrowsInvalidParameterValueException() {
         Mockito.doReturn(true).when(accountManager).isRootAdmin(Mockito.anyLong());
 
-        Mockito.doThrow(InvalidParameterValueException.class).when(userVmManagerImpl).validateIfVmSupportsMigration(Mockito.any(), Mockito.anyLong());
+        doThrow(InvalidParameterValueException.class).when(userVmManagerImpl).validateIfVmSupportsMigration(Mockito.any(), Mockito.anyLong());
 
         Assert.assertThrows(InvalidParameterValueException.class, () -> userVmManagerImpl.moveVmToUser(assignVmCmdMock));
     }
@@ -3011,7 +3020,7 @@ public class UserVmManagerImplTest {
 
         configureDoNothingForMethodsThatWeDoNotWantToTest();
 
-        Mockito.doThrow(InvalidParameterValueException.class).when(userVmManagerImpl).validateIfVmHasNoRules(Mockito.any(), Mockito.anyLong());
+        doThrow(InvalidParameterValueException.class).when(userVmManagerImpl).validateIfVmHasNoRules(Mockito.any(), Mockito.anyLong());
 
         Assert.assertThrows(InvalidParameterValueException.class, () -> userVmManagerImpl.moveVmToUser(assignVmCmdMock));
     }
@@ -3030,7 +3039,7 @@ public class UserVmManagerImplTest {
 
         configureDoNothingForMethodsThatWeDoNotWantToTest();
 
-        Mockito.doThrow(InvalidParameterValueException.class).when(userVmManagerImpl).validateIfVolumesHaveNoSnapshots(Mockito.any());
+        doThrow(InvalidParameterValueException.class).when(userVmManagerImpl).validateIfVolumesHaveNoSnapshots(Mockito.any());
 
         Assert.assertThrows(InvalidParameterValueException.class, () -> userVmManagerImpl.moveVmToUser(assignVmCmdMock));
     }
@@ -3048,7 +3057,7 @@ public class UserVmManagerImplTest {
 
         configureDoNothingForMethodsThatWeDoNotWantToTest();
 
-        Mockito.doThrow(ResourceAllocationException.class).when(userVmManagerImpl).verifyResourceLimitsForAccountAndStorage(Mockito.any(), Mockito.any(),
+        doThrow(ResourceAllocationException.class).when(userVmManagerImpl).verifyResourceLimitsForAccountAndStorage(Mockito.any(), Mockito.any(),
                 Mockito.any(), Mockito.any(), Mockito.any());
 
         Assert.assertThrows(ResourceAllocationException.class, () -> userVmManagerImpl.moveVmToUser(assignVmCmdMock));
@@ -3067,7 +3076,7 @@ public class UserVmManagerImplTest {
 
         configureDoNothingForMethodsThatWeDoNotWantToTest();
 
-        Mockito.doThrow(InvalidParameterValueException.class).when(userVmManagerImpl).validateIfNewOwnerHasAccessToTemplate(Mockito.any(), Mockito.any(), Mockito.any());
+        doThrow(InvalidParameterValueException.class).when(userVmManagerImpl).validateIfNewOwnerHasAccessToTemplate(Mockito.any(), Mockito.any(), Mockito.any());
 
         Assert.assertThrows(InvalidParameterValueException.class, () -> userVmManagerImpl.moveVmToUser(assignVmCmdMock));
     }
@@ -3087,7 +3096,7 @@ public class UserVmManagerImplTest {
 
         configureDoNothingForMethodsThatWeDoNotWantToTest();
 
-        Mockito.doThrow(PermissionDeniedException.class).when(accountManager).checkAccess(Mockito.any(Account.class), Mockito.any());
+        doThrow(PermissionDeniedException.class).when(accountManager).checkAccess(Mockito.any(Account.class), Mockito.any());
 
         Assert.assertThrows(PermissionDeniedException.class, () -> userVmManagerImpl.moveVmToUser(assignVmCmdMock));
     }
@@ -3098,12 +3107,12 @@ public class UserVmManagerImplTest {
 
         LinkedList<VolumeVO> volumes = new LinkedList<VolumeVO>();
 
-        try (MockedStatic<UsageEventUtils> ignored = Mockito.mockStatic(UsageEventUtils.class)) {
+        try (MockedStatic<UsageEventUtils> ignored = mockStatic(UsageEventUtils.class)) {
             Mockito.doReturn(Hypervisor.HypervisorType.KVM).when(userVmVoMock).getHypervisorType();
 
             configureDoNothingForMethodsThatWeDoNotWantToTest();
 
-            Mockito.doThrow(InsufficientAddressCapacityException.class).when(userVmManagerImpl).updateVmNetwork(Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any(),
+            doThrow(InsufficientAddressCapacityException.class).when(userVmManagerImpl).updateVmNetwork(Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any(),
                     Mockito.any());
 
             Assert.assertThrows(CloudRuntimeException.class, () -> userVmManagerImpl.executeStepsToChangeOwnershipOfVm(assignVmCmdMock, callerAccount, accountMock, accountMock,
@@ -3121,12 +3130,12 @@ public class UserVmManagerImplTest {
 
         LinkedList<VolumeVO> volumes = new LinkedList<VolumeVO>();
 
-        try (MockedStatic<UsageEventUtils> ignored = Mockito.mockStatic(UsageEventUtils.class)) {
+        try (MockedStatic<UsageEventUtils> ignored = mockStatic(UsageEventUtils.class)) {
             Mockito.doReturn(Hypervisor.HypervisorType.KVM).when(userVmVoMock).getHypervisorType();
 
             configureDoNothingForMethodsThatWeDoNotWantToTest();
 
-            Mockito.doThrow(ResourceAllocationException.class).when(userVmManagerImpl).updateVmNetwork(Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any(),
+            doThrow(ResourceAllocationException.class).when(userVmManagerImpl).updateVmNetwork(Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any(),
                     Mockito.any());
 
             Assert.assertThrows(CloudRuntimeException.class, () -> userVmManagerImpl.executeStepsToChangeOwnershipOfVm(assignVmCmdMock, callerAccount, accountMock, accountMock,
@@ -3145,7 +3154,7 @@ public class UserVmManagerImplTest {
         LinkedList<VolumeVO> volumes = new LinkedList<VolumeVO>();
 
 
-        try (MockedStatic<UsageEventUtils> ignored = Mockito.mockStatic(UsageEventUtils.class)) {
+        try (MockedStatic<UsageEventUtils> ignored = mockStatic(UsageEventUtils.class)) {
             Mockito.doReturn(Hypervisor.HypervisorType.KVM).when(userVmVoMock).getHypervisorType();
             Mockito.doReturn(false).when(userVmManagerImpl).isResourceCountRunningVmsOnlyEnabled();
 
@@ -3168,7 +3177,7 @@ public class UserVmManagerImplTest {
 
         LinkedList<VolumeVO> volumes = new LinkedList<VolumeVO>();
 
-        try (MockedStatic<UsageEventUtils> ignored = Mockito.mockStatic(UsageEventUtils.class)) {
+        try (MockedStatic<UsageEventUtils> ignored = mockStatic(UsageEventUtils.class)) {
             Mockito.doReturn(Hypervisor.HypervisorType.KVM).when(userVmVoMock).getHypervisorType();
             Mockito.doReturn(true).when(userVmManagerImpl).isResourceCountRunningVmsOnlyEnabled();
 
@@ -3617,7 +3626,7 @@ public class UserVmManagerImplTest {
         when(callContext.getCallingAccount()).thenReturn(callingAccount);
         when(accountManager.isAdmin(callingAccount.getId())).thenReturn(true);
         doNothing().when(accountManager).checkApiAccess(callingAccount, BaseCmd.getCommandNameByClass(ExpungeVMCmd.class));
-        try (MockedStatic<CallContext> mockedCallContext = Mockito.mockStatic(CallContext.class)) {
+        try (MockedStatic<CallContext> mockedCallContext = mockStatic(CallContext.class)) {
             mockedCallContext.when(CallContext::current).thenReturn(callContext);
             mockedCallContext.when(() -> CallContext.register(callContext, ApiCommandResourceType.Volume)).thenReturn(callContext);
 
@@ -3649,7 +3658,7 @@ public class UserVmManagerImplTest {
             doReturn(vm).when(userVmManagerImpl).destroyVm(vmId, expunge);
             doReturn(true).when(userVmManagerImpl).expunge(vm);
 
-            try (MockedStatic<UsageEventUtils> mockedUsageEventUtils = Mockito.mockStatic(UsageEventUtils.class)) {
+            try (MockedStatic<UsageEventUtils> mockedUsageEventUtils = mockStatic(UsageEventUtils.class)) {
 
                 UserVm result = userVmManagerImpl.destroyVm(cmd);
 
@@ -3794,7 +3803,7 @@ public class UserVmManagerImplTest {
         UserVmVO userVm = Mockito.mock(UserVmVO.class);
         when(userVm.getId()).thenReturn(vmId);;
         when(vmInstanceDetailsDao.listDetailsKeyPairs(anyLong(), anyList())).thenReturn(getLeaseDetails(2, VMLeaseManager.LeaseActionExecution.PENDING.name()));
-        try (MockedStatic<ActionEventUtils> ignored = Mockito.mockStatic(ActionEventUtils.class)) {
+        try (MockedStatic<ActionEventUtils> ignored = mockStatic(ActionEventUtils.class)) {
             Mockito.when(ActionEventUtils.onActionEvent(Mockito.anyLong(), Mockito.anyLong(),
                     Mockito.anyLong(),
                     Mockito.anyString(), Mockito.anyString(),
@@ -4032,4 +4041,129 @@ public class UserVmManagerImplTest {
         verify(overrideDiskOffering).isComputeOnly();
     }
 
+
+    @Test
+    public void testUnmanageUserVMVmNotFound() {
+        when(userVmDao.findById(vmId)).thenReturn(null);
+        InvalidParameterValueException exception = assertThrows(InvalidParameterValueException.class, () -> {
+            userVmManagerImpl.unmanageUserVM(vmId, null);
+        });
+        assertEquals("Unable to find a VM with ID = " + vmId, exception.getMessage());
+        verify(userVmDao, never()).acquireInLockTable(anyLong());
+        verify(userVmDao, never()).releaseFromLockTable(anyLong());
+    }
+
+    @Test
+    public void testUnmanageUserVMAlreadyRemoved() {
+        when(userVmVoMock.getRemoved()).thenReturn(new Date());
+        when(userVmDao.findById(vmId)).thenReturn(userVmVoMock);
+        assertThrows(InvalidParameterValueException.class, () -> {
+            userVmManagerImpl.unmanageUserVM(vmId, null);
+        });
+    }
+
+    @Test
+    public void testUnmanageUserVMInvalidState() {
+        when(userVmVoMock.getId()).thenReturn(vmId);
+        when(userVmVoMock.getName()).thenReturn("test-vm");
+        when(userVmDao.findById(vmId)).thenReturn(userVmVoMock);
+        when(userVmDao.acquireInLockTable(vmId)).thenReturn(userVmVoMock);
+        when(userVmVoMock.getState()).thenReturn(VirtualMachine.State.Starting);
+        CloudRuntimeException exception = assertThrows(CloudRuntimeException.class, () -> {
+            userVmManagerImpl.unmanageUserVM(vmId, null);
+        });
+        assertEquals("Instance: test-vm is not running or stopped, cannot be unmanaged", exception.getMessage());
+        verify(userVmDao, times(1)).releaseFromLockTable(vmId);
+    }
+
+    @Test
+    public void testUnmanageUserVMUnsupportedHypervisor() {
+        when(userVmVoMock.getId()).thenReturn(vmId);
+        when(userVmDao.findById(vmId)).thenReturn(userVmVoMock);
+        when(userVmDao.acquireInLockTable(vmId)).thenReturn(userVmVoMock);
+        when(userVmVoMock.getState()).thenReturn(VirtualMachine.State.Stopped);
+        when(userVmVoMock.getHypervisorType()).thenReturn(Hypervisor.HypervisorType.Hyperv);
+        unmanagedVMsManagerMockedStatic.when(() -> UnmanagedVMsManager.isSupported(Hypervisor.HypervisorType.Hyperv)).thenReturn(false);
+
+        CloudRuntimeException exception = assertThrows(CloudRuntimeException.class, () -> {
+            userVmManagerImpl.unmanageUserVM(vmId, null);
+        });
+
+        assertEquals("Unmanaging a VM is currently not supported on hypervisor Hyperv", exception.getMessage());
+        verify(userVmDao, times(1)).releaseFromLockTable(vmId);
+    }
+
+    @Test
+    public void testUnmanageUserVMItManagerReturnsFalse() {
+        when(userVmDao.findById(vmId)).thenReturn(userVmVoMock);
+        when(userVmVoMock.getId()).thenReturn(vmId);
+        when(userVmVoMock.getUuid()).thenReturn(vmUuid);
+        when(userVmDao.acquireInLockTable(vmId)).thenReturn(userVmVoMock);
+        when(userVmVoMock.getState()).thenReturn(VirtualMachine.State.Running);
+        when(userVmVoMock.getHypervisorType()).thenReturn(Hypervisor.HypervisorType.KVM);
+        unmanagedVMsManagerMockedStatic.when(() -> UnmanagedVMsManager.isSupported(Hypervisor.HypervisorType.KVM)).thenReturn(true);
+        when(volumeDaoMock.findByInstance(vmId)).thenReturn(Collections.emptyList());
+        when(virtualMachineManager.unmanage(vmUuid, null)).thenReturn(new Pair<>(false, "Backend failure"));
+
+        doNothing().when(userVmManagerImpl).checkUnmanagingVMOngoingVolumeSnapshots(any(UserVmVO.class));
+        doNothing().when(userVmManagerImpl).checkUnmanagingVMVolumes(any(UserVmVO.class), any(List.class));
+
+        CloudRuntimeException exception = assertThrows(CloudRuntimeException.class, () -> {
+            userVmManagerImpl.unmanageUserVM(vmId, null);
+        });
+
+        assertEquals("Error while unmanaging VM: " + vmUuid, exception.getMessage());
+        verify(userVmManagerImpl, never()).cleanupUnmanageVMResources(any(UserVmVO.class));
+        verify(userVmManagerImpl, never()).unmanageVMFromDB(anyLong());
+        verify(userVmDao, times(1)).releaseFromLockTable(vmId);
+    }
+
+    @Test
+    public void testUnmanageUserVMGenericException() {
+        RuntimeException testException = new RuntimeException("Something went wrong");
+        when(userVmDao.findById(vmId)).thenReturn(userVmVoMock);
+        when(userVmVoMock.getId()).thenReturn(vmId);
+        when(userVmDao.acquireInLockTable(vmId)).thenReturn(userVmVoMock);
+        when(userVmVoMock.getState()).thenReturn(VirtualMachine.State.Running);
+        when(userVmVoMock.getHypervisorType()).thenReturn(Hypervisor.HypervisorType.KVM);
+        unmanagedVMsManagerMockedStatic.when(() -> UnmanagedVMsManager.isSupported(Hypervisor.HypervisorType.KVM)).thenReturn(true);
+        doThrow(testException).when(userVmManagerImpl).checkUnmanagingVMOngoingVolumeSnapshots(any(UserVmVO.class));
+
+        CloudRuntimeException exception = assertThrows(CloudRuntimeException.class, () -> {
+            userVmManagerImpl.unmanageUserVM(vmId, null);
+        });
+
+        assertNotNull(exception.getCause());
+        assertEquals(testException, exception.getCause());
+        verify(userVmDao, times(1)).releaseFromLockTable(vmId);
+    }
+
+    @Test
+    public void testUnmanageUserVMSuccess() {
+        when(userVmDao.findById(vmId)).thenReturn(userVmVoMock);
+        when(userVmVoMock.getId()).thenReturn(vmId);
+        when(userVmVoMock.getUuid()).thenReturn(vmUuid);
+        when(userVmDao.acquireInLockTable(vmId)).thenReturn(userVmVoMock);
+        when(userVmVoMock.getState()).thenReturn(VirtualMachine.State.Running);
+        when(userVmVoMock.getHypervisorType()).thenReturn(Hypervisor.HypervisorType.KVM);
+        unmanagedVMsManagerMockedStatic.when(() -> UnmanagedVMsManager.isSupported(Hypervisor.HypervisorType.KVM)).thenReturn(true);
+        when(volumeDaoMock.findByInstance(vmId)).thenReturn(Collections.emptyList());
+        when(virtualMachineManager.unmanage(vmUuid, null)).thenReturn(new Pair<>(true, "Unmanaged successfully"));
+        doNothing().when(userVmManagerImpl).checkUnmanagingVMOngoingVolumeSnapshots(any(UserVmVO.class));
+        doNothing().when(userVmManagerImpl).checkUnmanagingVMVolumes(any(UserVmVO.class), any(List.class));
+        doNothing().when(userVmManagerImpl).cleanupUnmanageVMResources(any(UserVmVO.class));
+        doNothing().when(userVmManagerImpl).unmanageVMFromDB(anyLong());
+        doNothing().when(userVmManagerImpl).publishUnmanageVMUsageEvents(any(UserVmVO.class), any(List.class));
+        Pair<Boolean, String> result = userVmManagerImpl.unmanageUserVM(vmId, null);
+        assertTrue(result.first());
+        assertEquals("Unmanaged successfully", result.second());
+        verify(userVmDao, times(1)).acquireInLockTable(vmId);
+        verify(userVmManagerImpl, times(1)).checkUnmanagingVMOngoingVolumeSnapshots(userVmVoMock);
+        verify(userVmManagerImpl, times(1)).checkUnmanagingVMVolumes(userVmVoMock, Collections.emptyList());
+        verify(userVmManagerImpl, times(1)).cleanupUnmanageVMResources(userVmVoMock);
+        verify(userVmManagerImpl, times(1)).unmanageVMFromDB(vmId);
+        verify(userVmManagerImpl, times(1)).publishUnmanageVMUsageEvents(userVmVoMock, Collections.emptyList());
+        verify(userVmDao, times(1)).releaseFromLockTable(vmId);
+    }
+
 }
diff --git a/server/src/test/java/org/apache/cloudstack/vm/UnmanagedVMsManagerImplTest.java b/server/src/test/java/org/apache/cloudstack/vm/UnmanagedVMsManagerImplTest.java
index 09f62f7a049..99e461949be 100644
--- a/server/src/test/java/org/apache/cloudstack/vm/UnmanagedVMsManagerImplTest.java
+++ b/server/src/test/java/org/apache/cloudstack/vm/UnmanagedVMsManagerImplTest.java
@@ -31,6 +31,7 @@ import static org.mockito.Mockito.when;
 
 import java.net.URI;
 import java.util.ArrayList;
+import java.util.Date;
 import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.List;
@@ -70,6 +71,7 @@ import org.mockito.Mock;
 import org.mockito.MockedStatic;
 import org.mockito.Mockito;
 import org.mockito.MockitoAnnotations;
+import org.mockito.Spy;
 import org.mockito.junit.MockitoJUnitRunner;
 
 import com.cloud.agent.AgentManager;
@@ -158,15 +160,19 @@ import com.cloud.vm.DiskProfile;
 import com.cloud.vm.NicProfile;
 import com.cloud.vm.UserVmManager;
 import com.cloud.vm.UserVmVO;
+import com.cloud.vm.VMInstanceDetailVO;
 import com.cloud.vm.VMInstanceVO;
 import com.cloud.vm.VirtualMachine;
+import com.cloud.vm.VmDetailConstants;
 import com.cloud.vm.dao.NicDao;
 import com.cloud.vm.dao.UserVmDao;
 import com.cloud.vm.dao.VMInstanceDao;
+import com.cloud.vm.dao.VMInstanceDetailsDao;
 
 @RunWith(MockitoJUnitRunner.class)
 public class UnmanagedVMsManagerImplTest {
 
+    @Spy
     @InjectMocks
     private UnmanagedVMsManagerImpl unmanagedVMsManager = new UnmanagedVMsManagerImpl();
 
@@ -237,6 +243,8 @@ public class UnmanagedVMsManagerImplTest {
     EntityManager entityMgr;
     @Mock
     DeploymentPlanningManager deploymentPlanningManager;
+    @Mock
+    private VMInstanceDetailsDao vmInstanceDetailsDao;
 
     private static final long virtualMachineId = 1L;
 
@@ -445,38 +453,107 @@ public class UnmanagedVMsManagerImplTest {
 
     @Test(expected = InvalidParameterValueException.class)
     public void unmanageVMInstanceMissingInstanceTest() {
-        long notExistingId = 10L;
-        unmanagedVMsManager.unmanageVMInstance(notExistingId);
+        when(vmDao.findById(anyLong())).thenReturn(null);
+        unmanagedVMsManager.unmanageVMInstance(1L, null, false);
+    }
+
+    @Test(expected = InvalidParameterValueException.class)
+    public void unmanageVMInstanceRemovedInstanceTest() {
+        VMInstanceVO userVmVO = mock(VMInstanceVO.class);
+        when(vmDao.findById(anyLong())).thenReturn(userVmVO);
+        when(userVmVO.getRemoved()).thenReturn(new Date());
+        unmanagedVMsManager.unmanageVMInstance(1L, null, false);
     }
 
     @Test(expected = InvalidParameterValueException.class)
     public void unmanageVMInstanceDestroyedInstanceTest() {
         when(virtualMachine.getState()).thenReturn(VirtualMachine.State.Destroyed);
-        unmanagedVMsManager.unmanageVMInstance(virtualMachineId);
+        unmanagedVMsManager.unmanageVMInstance(virtualMachineId, null, false);
     }
 
     @Test(expected = InvalidParameterValueException.class)
     public void unmanageVMInstanceExpungedInstanceTest() {
         when(virtualMachine.getState()).thenReturn(VirtualMachine.State.Expunging);
-        unmanagedVMsManager.unmanageVMInstance(virtualMachineId);
+        unmanagedVMsManager.unmanageVMInstance(virtualMachineId, null, false);
     }
 
     @Test(expected = UnsupportedServiceException.class)
     public void unmanageVMInstanceExistingVMSnapshotsTest() {
         when(virtualMachine.getHypervisorType()).thenReturn(Hypervisor.HypervisorType.None);
-        unmanagedVMsManager.unmanageVMInstance(virtualMachineId);
+        unmanagedVMsManager.unmanageVMInstance(virtualMachineId, null, false);
+    }
+
+    @Test(expected = UnsupportedServiceException.class)
+    public void unmanageVMInstanceInvalidHyperVisor() {
+        when(virtualMachine.getHypervisorType()).thenReturn(Hypervisor.HypervisorType.None);
+        unmanagedVMsManager.unmanageVMInstance(virtualMachineId, null, false);
+    }
+
+    @Test(expected = UnsupportedServiceException.class)
+    public void unmanageVMInstanceInvalidVmType() {
+        when(virtualMachine.getHypervisorType()).thenReturn(Hypervisor.HypervisorType.VMware);
+        when(virtualMachine.getType()).thenReturn(VirtualMachine.Type.ConsoleProxy);
+        unmanagedVMsManager.unmanageVMInstance(virtualMachineId, null, false);
     }
 
     @Test(expected = UnsupportedServiceException.class)
     public void unmanageVMInstanceExistingVolumeSnapshotsTest() {
         when(virtualMachine.getHypervisorType()).thenReturn(Hypervisor.HypervisorType.None);
-        unmanagedVMsManager.unmanageVMInstance(virtualMachineId);
+        unmanagedVMsManager.unmanageVMInstance(virtualMachineId, null, false);
     }
 
     @Test(expected = UnsupportedServiceException.class)
     public void unmanageVMInstanceExistingISOAttachedTest() {
         when(virtualMachine.getHypervisorType()).thenReturn(Hypervisor.HypervisorType.None);
-        unmanagedVMsManager.unmanageVMInstance(virtualMachineId);
+        unmanagedVMsManager.unmanageVMInstance(virtualMachineId, null, false);
+    }
+
+    @Test(expected = UnsupportedServiceException.class)
+    public void unmanageVMInstanceVmwareHostIdParamTest() {
+        when(virtualMachine.getHypervisorType()).thenReturn(Hypervisor.HypervisorType.VMware);
+        when(virtualMachine.getType()).thenReturn(VirtualMachine.Type.User);
+        unmanagedVMsManager.unmanageVMInstance(virtualMachineId, 1L, false);
+    }
+
+    @Test(expected = UnsupportedServiceException.class)
+    public void unmanageVMInstanceRunningHostIdParamTest() {
+        when(virtualMachine.getType()).thenReturn(VirtualMachine.Type.User);
+        when(virtualMachine.getState()).thenReturn(VirtualMachine.State.Running);
+        when(virtualMachine.getHypervisorType()).thenReturn(Hypervisor.HypervisorType.KVM);
+        unmanagedVMsManager.unmanageVMInstance(virtualMachineId, 1L, false);
+    }
+
+    @Test(expected = UnsupportedServiceException.class)
+    public void unmanageVMInstanceRunningForceParamTest() {
+        when(virtualMachine.getType()).thenReturn(VirtualMachine.Type.User);
+        when(virtualMachine.getState()).thenReturn(VirtualMachine.State.Running);
+        when(virtualMachine.getHypervisorType()).thenReturn(Hypervisor.HypervisorType.KVM);
+        VMInstanceDetailVO vmInstanceDetailVO = mock(VMInstanceDetailVO.class);
+        when(vmInstanceDetailsDao.findDetail(virtualMachineId, VmDetailConstants.CONFIG_DRIVE_LOCATION)).thenReturn(vmInstanceDetailVO);
+        unmanagedVMsManager.unmanageVMInstance(virtualMachineId, null, false);
+    }
+
+    @Test
+    public void unmanageVMInstanceVMwareHostId() {
+        when(virtualMachine.getType()).thenReturn(VirtualMachine.Type.User);
+        when(virtualMachine.getHypervisorType()).thenReturn(Hypervisor.HypervisorType.KVM);
+        when(virtualMachine.getState()).thenReturn(VirtualMachine.State.Stopped);
+        UserVmVO userVmVO = mock(UserVmVO.class);
+        when(userVmVO.getIsoId()).thenReturn(null);
+        when(userVmDao.findById(anyLong())).thenReturn(userVmVO);
+        when(vmDao.findById(virtualMachineId)).thenReturn(virtualMachine);
+        unmanagedVMsManager.unmanageVMInstance(virtualMachineId, 1L, false);
+    }
+
+    @Test
+    public void unmanageVMInstanceStoppedInstanceTest() {
+        when(virtualMachine.getHypervisorType()).thenReturn(Hypervisor.HypervisorType.KVM);
+        when(virtualMachine.getType()).thenReturn(VirtualMachine.Type.User);
+        when(virtualMachine.getState()).thenReturn(VirtualMachine.State.Stopped);
+        UserVmVO userVmVO = mock(UserVmVO.class);
+        when(userVmDao.findById(anyLong())).thenReturn(userVmVO);
+        Mockito.doNothing().when(unmanagedVMsManager).performUnmanageVMInstancePrechecks(any());
+        unmanagedVMsManager.unmanageVMInstance(virtualMachineId, null, false);
     }
 
     @Test
diff --git a/test/integration/smoke/test_vm_lifecycle_unmanage_kvm_import.py b/test/integration/smoke/test_vm_lifecycle_unmanage_kvm_import.py
new file mode 100644
index 00000000000..79c7513f2e2
--- /dev/null
+++ b/test/integration/smoke/test_vm_lifecycle_unmanage_kvm_import.py
@@ -0,0 +1,473 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you 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.
+""" BVT tests for Virtual Machine Life Cycle - Unmanage - Import
+"""
+# Import Local Modules
+from marvin.cloudstackTestCase import cloudstackTestCase
+from marvin.lib.utils import *
+from marvin.lib.base import (Account,
+                             ServiceOffering,
+                             VirtualMachine,
+                             Host,
+                             Network,
+                             NetworkOffering,
+                             VirtualMachine)
+from marvin.lib.common import (get_domain,
+                               get_zone,
+                               get_suitable_test_template)
+from marvin.codes import FAILED
+from nose.plugins.attrib import attr
+from marvin.lib.decoratorGenerators import skipTestIf
+import unittest
+from marvin.sshClient import SshClient
+
+_multiprocess_shared_ = True
+
+class TestUnmanageKvmVM(cloudstackTestCase):
+
+    @classmethod
+    def setUpClass(cls):
+        testClient = super(TestUnmanageKvmVM, cls).getClsTestClient()
+        cls.apiclient = testClient.getApiClient()
+        cls.services = testClient.getParsedTestDataConfig()
+        cls.hypervisor = testClient.getHypervisorInfo()
+        cls._cleanup = []
+
+        # Get Zone, Domain and templates
+        cls.domain = get_domain(cls.apiclient)
+        cls.zone = get_zone(cls.apiclient, cls.testClient.getZoneForTests())
+        cls.services['mode'] = cls.zone.networktype
+        cls.template = get_suitable_test_template(
+            cls.apiclient,
+            cls.zone.id,
+            cls.services["ostype"],
+            cls.hypervisor
+        )
+        if cls.template == FAILED:
+            assert False, "get_suitable_test_template() failed to return template with description %s" % cls.services["ostype"]
+
+        cls.hypervisorNotSupported = cls.hypervisor.lower() != "kvm"
+        if cls.hypervisorNotSupported:
+            return
+
+        cls.services["small"]["zoneid"] = cls.zone.id
+        cls.services["small"]["template"] = cls.template.id
+
+        cls.account = Account.create(
+            cls.apiclient,
+            cls.services["account"],
+            domainid=cls.domain.id
+        )
+        cls._cleanup.append(cls.account)
+
+        cls.small_offering = ServiceOffering.create(
+            cls.apiclient,
+            cls.services["service_offerings"]["small"]
+        )
+        cls._cleanup.append(cls.small_offering)
+
+        cls.network_offering = NetworkOffering.create(
+            cls.apiclient,
+            cls.services["l2-network_offering"],
+        )
+        cls._cleanup.append(cls.network_offering)
+        cls.network_offering.update(cls.apiclient, state='Enabled')
+        cls.isolated_network_offering = NetworkOffering.create(
+            cls.apiclient,
+            cls.services["nw_off_isolated_persistent"],
+        )
+        cls._cleanup.append(cls.isolated_network_offering)
+        cls.isolated_network_offering.update(cls.apiclient, state='Enabled')
+
+    @classmethod
+    def tearDownClass(cls):
+        super(TestUnmanageKvmVM, cls).tearDownClass()
+
+    def setUp(self):
+        self.apiclient = self.testClient.getApiClient()
+        self.dbclient = self.testClient.getDbConnection()
+        self.cleanup = []
+        self.created_networks = []
+        self.virtual_machine = None
+        self.unmanaged_instance = None
+        self.imported_vm = None
+        self.hostConfig = self.config.__dict__["zones"][0].__dict__["pods"][0].__dict__["clusters"][0].__dict__["hosts"][0].__dict__
+
+        if self.hypervisorNotSupported:
+            return
+        self.services["network"]["networkoffering"] = self.network_offering.id
+        network_data = self.services["l2-network"]
+        self.network = Network.create(
+            self.apiclient,
+            network_data,
+            zoneid=self.zone.id,
+            networkofferingid=self.network_offering.id
+        )
+        self.cleanup.append(self.network)
+        self.created_networks.append(self.network)
+        network_data['name'] = "Test L2 Network1"
+        network_data['displaytext'] = "Test L2 Network1"
+        self.network1 = Network.create(
+            self.apiclient,
+            network_data,
+            zoneid=self.zone.id,
+            networkofferingid=self.network_offering.id
+        )
+        self.cleanup.append(self.network1)
+        self.created_networks.append(self.network1)
+        self.network2 = Network.create(
+            self.apiclient,
+            self.services["isolated_network"],
+            zoneid=self.zone.id,
+            networkofferingid=self.isolated_network_offering.id
+        )
+        self.cleanup.append(self.network2)
+        self.created_networks.append(self.network2)
+
+
+    def tearDown(self):
+        super(TestUnmanageKvmVM, self).tearDown()
+
+    def check_vm_state(self, vm_id):
+        list_vm = VirtualMachine.list(
+            self.apiclient,
+            id=vm_id
+        )
+        self.assertEqual(
+            isinstance(list_vm, list),
+            True,
+            "Check if virtual machine is present"
+        )
+        vm_response = list_vm[0]
+        self.assertEqual(
+            vm_response.state,
+            "Running",
+            "VM state should be running after deployment"
+        )
+        return vm_response
+
+
+    @attr(tags=["advanced", "advancedns", "smoke", "sg"], required_hardware="true")
+    @skipTestIf("hypervisorNotSupported")
+    def test_01_unmanage_vm_cycle_persistent_domain(self):
+        """
+        Test the following:
+        1. Deploy VM
+        2. Unmanage VM
+        3. Verify VM is not listed in CloudStack
+        4. Verify VM is listed as part of the unmanaged instances
+        5. Verify VM domain is persistent for KVM hypervisor
+        6. Stop VM using virsh, confirm VM is stopped
+        7. Start VM using virsh, confirm VM is running
+        8. Import VM
+        9. Verify details of imported VM
+        10. Destroy VM
+        """
+
+        # 1 - Deploy VM
+        self.virtual_machine = VirtualMachine.create(
+            self.apiclient,
+            self.services["virtual_machine"],
+            templateid=self.template.id,
+            serviceofferingid=self.small_offering.id,
+            networkids=[self.network.id, self.network1.id, self.network2.id],
+            zoneid=self.zone.id
+        )
+
+        vm_id = self.virtual_machine.id
+        vm_instance_name = self.virtual_machine.instancename
+
+        networks = []
+        for network in self.created_networks:
+            n = Network.list(
+                self.apiclient,
+                id=network.id
+            )[0]
+            networks.append(n)
+        hostid = self.virtual_machine.hostid
+        hosts = Host.list(
+            self.apiclient,
+            id=hostid
+        )
+        host = hosts[0]
+        clusterid = host.clusterid
+        self.check_vm_state(vm_id)
+
+        # 2 - Unmanage VM from CloudStack
+        self.virtual_machine.unmanage(self.apiclient)
+
+        # 3 - Verify VM is not listed in CloudStack
+        list_vm = VirtualMachine.list(
+            self.apiclient,
+            id=vm_id
+        )
+        self.assertEqual(
+            list_vm,
+            None,
+            "VM should not be listed"
+        )
+        # 4 - Verify VM is listed as part of the unmanaged instances
+        unmanaged_vms = VirtualMachine.listUnmanagedInstances(
+            self.apiclient,
+            clusterid=clusterid,
+            name=vm_instance_name
+        )
+
+        self.assertEqual(
+            len(unmanaged_vms),
+            1,
+            "Unmanaged VMs matching instance name list size is 1"
+        )
+        unmanaged_vm = unmanaged_vms[0]
+        self.assertEqual(
+            unmanaged_vm.powerstate,
+            "PowerOn",
+            "Unmanaged VM is still running"
+        )
+
+        # 5 - Verify VM domain is persistent for KVM
+        ssh_host = self.get_ssh_client(host.ipaddress,
+                                    self.hostConfig["username"],
+                                    self.hostConfig["password"], 10)
+
+        cmd = f"virsh dominfo {vm_instance_name}"
+        result = ssh_host.execute(cmd)
+        if result == None or result == "":
+            raise Exception(f"Failed to fetch domain info for VM: {vm_instance_name} on host: {host.name}. Error: {result}")
+        persistent_line = next((line for line in result if "Persistent:" in line), None)
+        if not persistent_line:
+            raise Exception(f"'Persistent' info not found in dominfo output for VM: {vm_instance_name} on host: {host.name}")
+        if "yes" not in persistent_line.lower():
+            raise Exception(f"VM: {vm_instance_name} is NOT persistent on host: {host.name}")
+
+
+        # 6 - Stop VM using virsh, confirm VM is stopped
+        host = Host.list(
+            self.apiclient,
+            id=unmanaged_vm.hostid
+        )[0]
+        cmd = "virsh destroy %s" % vm_instance_name
+        result = ssh_host.execute(cmd)
+        if result == None or result == "":
+            raise Exception("Failed to stop VM: %s on host: %s" % (vm_instance_name, host.name))
+
+        cmd = "virsh list --all | grep %s" % vm_instance_name
+        result = ssh_host.execute(cmd)
+        if result == None or result == "":
+            raise Exception("Failed to fetch VM: %s state on host: %s" % (vm_instance_name, host.name))
+        state_line = next((line for line in result if vm_instance_name in line), None)
+        if not state_line:
+            raise Exception(f"VM: {vm_instance_name} not found in 'virsh list --all' output on host: {host.name}")
+        if 'shut off' not in state_line.lower():
+            raise Exception(f"VM: {vm_instance_name} is NOT stopped on host: {host.name}, state: {state_line}")
+
+        # 7 - Start VM using virsh, confirm VM is running
+        cmd = "virsh start %s" % vm_instance_name
+        result = ssh_host.execute(cmd)
+        if result == None or result == "":
+            raise Exception("Failed to start VM: %s on host: %s" % (vm_instance_name, host.name))
+        time.sleep(30)
+        cmd = "virsh list | grep %s" % vm_instance_name
+        result = ssh_host.execute(cmd)
+        if result == None or result == "":
+            raise Exception("Failed to fetch VM: %s state on host: %s" % (vm_instance_name, host.name))
+        if 'running' not in str(result).lower():
+            raise Exception(f"VM: {vm_instance_name} is NOT running on host: {host.name}, state: {state_line}")
+
+        # 8 - Import VM
+
+        nicnetworklist = []
+        nicipaddresslist = []
+        for nic in unmanaged_vm.nic:
+            for network in networks:
+                if int(network.vlan) == int(nic.vlanid):
+                    nicnetworklist.append({
+                        "nic": nic.id,
+                        "network": network.id
+                    })
+                    if network.type == "Isolated":
+                        nicipaddresslist.append({
+                            "nic": nic.id,
+                            "ip4Address": "auto"
+                        })
+        import_vm_service = {
+            "nicnetworklist": nicnetworklist,
+            "nicipaddresslist": nicipaddresslist
+        }
+
+        self.imported_vm = VirtualMachine.importUnmanagedInstance(
+            self.apiclient,
+            clusterid=clusterid,
+            name=vm_instance_name,
+            serviceofferingid=self.small_offering.id,
+            services=import_vm_service,
+            templateid=self.template.id)
+        self.cleanup.append(self.imported_vm)
+        self.unmanaged_instance = None
+        # 9 - Verify details of the imported VM
+        self.assertEqual(
+            self.small_offering.id,
+            self.imported_vm.serviceofferingid,
+            "Imported VM service offering is different, expected: %s, actual: %s" % (self.small_offering.id, self.imported_vm.serviceofferingid)
+        )
+        self.assertEqual(
+            self.template.id,
+            self.imported_vm.templateid,
+            "Imported VM template is different, expected: %s, actual: %s" % (self.template.id, self.imported_vm.templateid)
+        )
+        self.check_vm_state(self.imported_vm.id)
+        # 10 - Destroy VM. This will be done during cleanup
+
+
+
+    @attr(tags=["advanced", "advancedns", "smoke", "sg"], required_hardware="true")
+    @skipTestIf("hypervisorNotSupported")
+    def test_02_unmanage_stopped_vm_cycle_persistent_domain(self):
+        """
+        Test the following:
+        1. Deploy VM
+        2. Stop VM
+        3. Unmanage VM
+        4. Verify VM is not listed in CloudStack
+        5. Verify VM is listed as part of the unmanaged instances
+        6. Start VM using virsh, confirm VM is running
+        7. Import VM
+        8. Verify details of imported VM
+        9. Destroy VM
+        """
+        # 1 - Deploy VM
+        self.virtual_machine = VirtualMachine.create(
+            self.apiclient,
+            self.services["virtual_machine"],
+            templateid=self.template.id,
+            serviceofferingid=self.small_offering.id,
+            networkids=[self.network.id, self.network1.id, self.network2.id],
+            zoneid=self.zone.id
+        )
+
+        vm_id = self.virtual_machine.id
+        vm_instance_name = self.virtual_machine.instancename
+
+        networks = []
+        for network in self.created_networks:
+            n = Network.list(
+                self.apiclient,
+                id=network.id
+            )[0]
+            networks.append(n)
+        hostid = self.virtual_machine.hostid
+        hosts = Host.list(
+            self.apiclient,
+            id=hostid
+        )
+        host = hosts[0]
+        clusterid = host.clusterid
+        self.check_vm_state(vm_id)
+
+        #2 - Stop VM
+        self.virtual_machine.stop(self.apiclient)
+        list_vm = VirtualMachine.list(
+            self.apiclient,
+            id=vm_id
+        )
+        self.assertEqual(
+            isinstance(list_vm, list),
+            True,
+            "Check if virtual machine is present"
+        )
+        vm_response = list_vm[0]
+        self.assertEqual(
+            vm_response.state,
+            "Stopped",
+            "VM state should be Stopped after stopping the VM"
+        )
+
+        # 3 - Unmanage VM from CloudStack
+        self.virtual_machine.unmanage(self.apiclient)
+
+        # 4 - Verify VM is not listed in CloudStack
+        list_vm = VirtualMachine.list(
+            self.apiclient,
+            id=vm_id
+        )
+        self.assertEqual(
+            list_vm,
+            None,
+            "VM should not be listed"
+        )
+
+        # 5 - Verify VM is listed as part of the unmanaged instances
+        ssh_host = self.get_ssh_client(host.ipaddress,
+                                    self.hostConfig["username"],
+                                    self.hostConfig["password"], 10)
+        cmd = "virsh list --all | grep %s" % vm_instance_name
+        result = ssh_host.execute(cmd)
+        if result == None or result == "":
+            raise Exception("Failed to fetch VM: %s state on host: %s" % (vm_instance_name, host.name))
+        if 'shut off' not in str(result).lower():
+            raise Exception(f"VM: {vm_instance_name} is NOT in stopped on host: {host.name}")
+
+        # 6 - Start VM using virsh, confirm VM is running
+        cmd = "virsh start %s" % vm_instance_name
+        result = ssh_host.execute(cmd)
+        if result == None or result == "":
+            raise Exception("Failed to start VM: %s on host: %s" % (vm_instance_name, host.name))
+        time.sleep(30)
+        cmd = "virsh list | grep %s" % vm_instance_name
+        result = ssh_host.execute(cmd)
+        if result == None or result == "":
+            raise Exception("Failed to fetch VM: %s state on host: %s" % (vm_instance_name, host.name))
+        if 'running' not in str(result).lower():
+            raise Exception(f"VM: {vm_instance_name} is NOT running on host: {host.name}")
+
+        # 8 - Import VM
+        self.imported_vm = VirtualMachine.importUnmanagedInstance(
+            self.apiclient,
+            clusterid=clusterid,
+            name=vm_instance_name,
+            serviceofferingid=self.small_offering.id,
+            services={},
+            templateid=self.template.id)
+        self.cleanup.append(self.imported_vm)
+        self.unmanaged_instance = None
+
+        # 9 - Verify details of the imported VM
+        self.assertEqual(
+            self.small_offering.id,
+            self.imported_vm.serviceofferingid,
+            "Imported VM service offering is different, expected: %s, actual: %s" % (self.small_offering.id, self.imported_vm.serviceofferingid)
+        )
+        self.assertEqual(
+            self.template.id,
+            self.imported_vm.templateid,
+            "Imported VM template is different, expected: %s, actual: %s" % (self.template.id, self.imported_vm.templateid)
+        )
+        self.check_vm_state(self.imported_vm.id)
+        # 10 - Destroy VM. This will be done during cleanup
+
+
+    def get_ssh_client(self, ip, username, password, retries=10):
+        """ Setup ssh client connection and return connection """
+        try:
+            ssh_client = SshClient(ip, 22, username, password, retries)
+        except Exception as e:
+            raise unittest.SkipTest("Unable to create ssh connection: " % e)
+
+        self.assertIsNotNone(
+            ssh_client, "Failed to setup ssh connection to ip=%s" % ip)
+
+        return ssh_client
diff --git a/tools/marvin/marvin/lib/base.py b/tools/marvin/marvin/lib/base.py
index d0da9a26e5c..bbec2027bdc 100755
--- a/tools/marvin/marvin/lib/base.py
+++ b/tools/marvin/marvin/lib/base.py
@@ -401,7 +401,8 @@ class VirtualMachine:
             self.ssh_port = 22
         self.ssh_client = None
         # extract out the ipaddress
-        self.ipaddress = self.nic[0].ipaddress
+        if self.nic:
+            self.ipaddress = self.nic[0].ipaddress
 
     @classmethod
     def ssh_access_group(cls, apiclient, cmd):
@@ -1083,7 +1084,7 @@ class VirtualMachine:
         return apiclient.scaleVirtualMachine(cmd)
 
     def unmanage(self, apiclient):
-        """Unmanage a VM from CloudStack (currently VMware only)"""
+        """Unmanage a VM from CloudStack"""
         cmd = unmanageVirtualMachine.unmanageVirtualMachineCmd()
         cmd.id = self.id
         return apiclient.unmanageVirtualMachine(cmd)
diff --git a/ui/public/locales/en.json b/ui/public/locales/en.json
index 1dd29d5a929..d3e75d895ad 100644
--- a/ui/public/locales/en.json
+++ b/ui/public/locales/en.json
@@ -4025,7 +4025,7 @@
 "message.vr.alert.upon.network.offering.creation.others": "As none of the obligatory services for creating a virtual router (VPN, DHCP, DNS, Firewall, LB, UserData, SourceNat, StaticNat, PortForwarding) are enabled, the virtual router will not be created and the compute offering will not be used.",
 "message.warn.change.primary.storage.scope": "This feature is tested and supported for the following configurations:<br>KVM - NFS/Ceph - DefaultPrimary<br>VMware - NFS - DefaultPrimary<br>*There might be extra steps involved to make it work for other configurations.",
 "message.warn.filetype": "jpg, jpeg, png, bmp and svg are the only supported image formats.",
-"message.warn.importing.instance.without.nic": "WARNING: This Instance is being imported without NICs and many Network resources will not be available. Consider creating a NIC via vCenter before importing or as soon as the Instance is imported.",
+"message.warn.importing.instance.without.nic": "WARNING: This Instance is being imported without NICs and many Network resources will not be available. Consider creating a NIC via vCenter before importing or as soon as the Instance is imported. For KVM host, allocate a NIC to Instance after import.",
 "message.warn.select.template": "Please select a Template for Registration.",
 "message.warn.zone.mtu.update": "Please note that this limit won't affect pre-existing Network's MTU settings",
 "message.webhook.deliveries.time.filter": "Webhook deliveries list can be filtered based on date-time. Select 'Custom' for specifying start and end date range.",