diff --git a/engine/storage/src/main/java/org/apache/cloudstack/storage/endpoint/DefaultEndPointSelector.java b/engine/storage/src/main/java/org/apache/cloudstack/storage/endpoint/DefaultEndPointSelector.java
index 0ce2cedb944c..96a5aef5140a 100644
--- a/engine/storage/src/main/java/org/apache/cloudstack/storage/endpoint/DefaultEndPointSelector.java
+++ b/engine/storage/src/main/java/org/apache/cloudstack/storage/endpoint/DefaultEndPointSelector.java
@@ -651,7 +651,12 @@ protected EndPoint getEndPointForBitmapRemoval(DataObject object, boolean encryp
             VirtualMachine attachedVM = volumeInfo.getAttachedVM();
             if (attachedVM.getHostId() != null) {
                 return getEndPointFromHostId(attachedVM.getHostId());
-            } else if (attachedVM.getLastHostId() != null) {
+            }
+            EndPoint clvmEp = tryRouteToClvmLockHolder(volumeInfo, _storagePoolDao.findById(volumeInfo.getPoolId()), "bitmap removal");
+            if (clvmEp != null) {
+                return clvmEp;
+            }
+            if (attachedVM.getLastHostId() != null) {
                 return getEndPointFromHostId(attachedVM.getLastHostId());
             }
         }
diff --git a/engine/storage/src/test/java/org/apache/cloudstack/storage/endpoint/DefaultEndPointSelectorTest.java b/engine/storage/src/test/java/org/apache/cloudstack/storage/endpoint/DefaultEndPointSelectorTest.java
index ca124ec30f04..e9f759b05436 100644
--- a/engine/storage/src/test/java/org/apache/cloudstack/storage/endpoint/DefaultEndPointSelectorTest.java
+++ b/engine/storage/src/test/java/org/apache/cloudstack/storage/endpoint/DefaultEndPointSelectorTest.java
@@ -173,6 +173,51 @@ public void getEndPointForBitmapRemovalTestVolumeIsAttachedNoHostIsSet() {
         Mockito.verify(defaultEndPointSelectorSpy, Mockito.times(1)).select(volumeInfoMock, false);
     }
 
+    private void mockClvmVolumeOfStoppedVm(Long lockHostId) {
+        Mockito.doReturn(STORE_ID).when(volumeInfoMock).getPoolId();
+        Mockito.when(_storagePoolDao.findById(STORE_ID)).thenReturn(storagePoolVOMock);
+        Mockito.when(storagePoolVOMock.getPoolType()).thenReturn(StoragePoolType.CLVM_NG);
+        Mockito.doReturn(lockHostId).when(defaultEndPointSelectorSpy).getClvmLockHostId(volumeInfoMock);
+        Mockito.doReturn(true).when(volumeInfoMock).isAttachedVM();
+        Mockito.doReturn(virtualMachineMock).when(volumeInfoMock).getAttachedVM();
+        Mockito.doReturn(null).when(virtualMachineMock).getHostId();
+        Mockito.doReturn(HOST_ID).when(virtualMachineMock).getLastHostId();
+    }
+
+    @Test
+    public void getEndPointForBitmapRemovalTestClvmVolumeOfStoppedVmRoutesToLockHolderInsteadOfLastHost() {
+        mockClvmVolumeOfStoppedVm(DEST_HOST_ID);
+        Mockito.doReturn(endPointMock).when(defaultEndPointSelectorSpy).getEndPointFromHostId(DEST_HOST_ID);
+
+        EndPoint result = defaultEndPointSelectorSpy.getEndPointForBitmapRemoval(snapshotInfoMock, false);
+
+        assertEquals(endPointMock, result);
+        Mockito.verify(defaultEndPointSelectorSpy, Mockito.never()).getEndPointFromHostId(HOST_ID);
+    }
+
+    @Test
+    public void getEndPointForBitmapRemovalTestClvmVolumeOfStoppedVmWithoutLockHolderFallsBackToLastHost() {
+        mockClvmVolumeOfStoppedVm(null);
+        Mockito.doReturn(endPointMock).when(defaultEndPointSelectorSpy).getEndPointFromHostId(HOST_ID);
+
+        EndPoint result = defaultEndPointSelectorSpy.getEndPointForBitmapRemoval(snapshotInfoMock, false);
+
+        assertEquals(endPointMock, result);
+    }
+
+    @Test
+    public void getEndPointForBitmapRemovalTestClvmVolumeOfRunningVmUsesVmHostWithoutLockLookup() {
+        Mockito.doReturn(true).when(volumeInfoMock).isAttachedVM();
+        Mockito.doReturn(virtualMachineMock).when(volumeInfoMock).getAttachedVM();
+        Mockito.doReturn(HOST_ID).when(virtualMachineMock).getHostId();
+        Mockito.doReturn(endPointMock).when(defaultEndPointSelectorSpy).getEndPointFromHostId(HOST_ID);
+
+        EndPoint result = defaultEndPointSelectorSpy.getEndPointForBitmapRemoval(snapshotInfoMock, false);
+
+        assertEquals(endPointMock, result);
+        Mockito.verify(defaultEndPointSelectorSpy, Mockito.never()).getClvmLockHostId(volumeInfoMock);
+    }
+
     @Test
     public void getEndPointForSnapshotOperationsInKvmTestVolumeIsNotAttachedToVMAndSnapshotOnPrimary() {
         Mockito.doReturn(null).when(volumeInfoMock).getAttachedVM();
diff --git a/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/storage/KVMStorageProcessor.java b/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/storage/KVMStorageProcessor.java
index 29b23e6c6451..d55e645b36b7 100644
--- a/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/storage/KVMStorageProcessor.java
+++ b/plugins/hypervisors/kvm/src/main/java/com/cloud/hypervisor/kvm/storage/KVMStorageProcessor.java
@@ -2097,7 +2097,15 @@ public Answer createSnapshot(final CreateObjectCommand cmd) {
             String diskPath = disk.getPath();
             String snapshotPath = diskPath + File.separator + snapshotName;
             SnapshotObjectTO newSnapshot = new SnapshotObjectTO();
-            if (DomainInfo.DomainState.VIR_DOMAIN_RUNNING.equals(state) && !primaryPool.isExternalSnapshot()) {
+            boolean isClvmNgIncremental = primaryPool.getType() == StoragePoolType.CLVM_NG && snapshotTO.isKvmIncrementalSnapshot();
+            if (isClvmNgIncremental && secondaryPool == null) {
+                String errorMsg = String.format("Incremental snapshots for CLVM_NG require secondary storage. " +
+                        "Please configure secondary storage or disable incremental snapshots for volume [%s].", volume.getName());
+                logger.error(errorMsg);
+                return new CreateObjectAnswer(errorMsg);
+            }
+            boolean isLiveClvmNgIncremental = isClvmNgIncremental && (DomainInfo.DomainState.VIR_DOMAIN_RUNNING.equals(state) || DomainInfo.DomainState.VIR_DOMAIN_PAUSED.equals(state));
+            if ((DomainInfo.DomainState.VIR_DOMAIN_RUNNING.equals(state) && !primaryPool.isExternalSnapshot()) || isLiveClvmNgIncremental) {
                 if (snapshotTO.isKvmIncrementalSnapshot()) {
                     newSnapshot = takeIncrementalVolumeSnapshotOfRunningVm(snapshotTO, primaryPool, secondaryPool, imageStoreTo != null ? imageStoreTo.getUrl() : null, snapshotName, volume, vm, conn, cmd.getWait());
                 } else {
@@ -2111,13 +2119,7 @@ public Answer createSnapshot(final CreateObjectCommand cmd) {
                         newSnapshot.setPhysicalSize(snapshotSize);
                     }
                 } else if (primaryPool.getType() == StoragePoolType.CLVM || primaryPool.getType() == StoragePoolType.CLVM_NG) {
-                    if (primaryPool.getType() == StoragePoolType.CLVM_NG && snapshotTO.isKvmIncrementalSnapshot()) {
-                        if (secondaryPool == null) {
-                            String errorMsg = String.format("Incremental snapshots for CLVM_NG require secondary storage. " +
-                                    "Please configure secondary storage or disable incremental snapshots for volume [%s].", volume.getName());
-                            logger.error(errorMsg);
-                            return new CreateObjectAnswer(errorMsg);
-                        }
+                    if (isClvmNgIncremental) {
                         logger.info("Taking incremental snapshot of CLVM_NG volume [{}] using QCOW2 backup to secondary storage.", volume.getName());
                         newSnapshot = takeIncrementalVolumeSnapshotOfStoppedVm(snapshotTO, primaryPool, secondaryPool,
                                 imageStoreTo.getUrl(), snapshotName, volume, conn, cmd.getWait());
diff --git a/server/src/main/java/com/cloud/storage/snapshot/SnapshotManagerImpl.java b/server/src/main/java/com/cloud/storage/snapshot/SnapshotManagerImpl.java
index 6ce7a0f92bfb..ba679457b161 100755
--- a/server/src/main/java/com/cloud/storage/snapshot/SnapshotManagerImpl.java
+++ b/server/src/main/java/com/cloud/storage/snapshot/SnapshotManagerImpl.java
@@ -1649,6 +1649,7 @@ public SnapshotInfo takeSnapshot(VolumeInfo volume) throws ResourceAllocationExc
         boolean backupSnapToSecondary = isBackupSnapshotToSecondaryForZone(volume.getDataCenterId());
 
         StoragePoolType poolType = volume.getStoragePoolType();
+        boolean isClvmNgIncrementalCandidate = StoragePoolType.CLVM_NG == poolType && kvmIncrementalSnapshot.valueIn(clusterId);
 
         updateSnapshotPayload(volume.getPoolId(), payload, isKvmAndFileBasedStorage, poolType, clusterId);
 
@@ -1658,7 +1659,7 @@ public SnapshotInfo takeSnapshot(VolumeInfo volume) throws ResourceAllocationExc
         boolean archiveSnapshotToSecondary = backupSnapToSecondary
                 && !isOntapManagedPrimaryLocationSnapshot(storagePool, payload);
 
-        if (isKvmAndFileBasedStorage && archiveSnapshotToSecondary) {
+        if ((isKvmAndFileBasedStorage || isClvmNgIncrementalCandidate) && archiveSnapshotToSecondary) {
             DataStore imageStore = snapshotSrv.findSnapshotImageStore(snapshot);
             if (imageStore == null) {
                 throw new CloudRuntimeException(String.format("Could not find any secondary storage to allocate snapshot [%s].", snapshot));
@@ -1681,7 +1682,7 @@ public SnapshotInfo takeSnapshot(VolumeInfo volume) throws ResourceAllocationExc
             SnapshotInfo snapshotOnPrimary = snapshotStrategy.takeSnapshot(snapshot);
 
             if (archiveSnapshotToSecondary) {
-                if (!isKvmAndFileBasedStorage) {
+                if (!isKvmAndFileBasedStorage && !isClvmNgIncrementalCandidate) {
                     backupSnapshotToSecondary(payload.getAsyncBackup(), snapshotStrategy, snapshotOnPrimary, payload.getZoneIds(), payload.getStoragePoolIds());
                     if (!payload.getAsyncBackup() && ClvmPoolManager.isClvmPoolType(storagePool.getPoolType())) {
                         _snapshotStoreDao.removeBySnapshotStore(snapshotId, snapshotOnPrimary.getDataStore().getId(), snapshotOnPrimary.getDataStore().getRole());
@@ -1926,7 +1927,7 @@ private void updateSnapshotPayload(long storagePoolId, CreateSnapshotPayload pay
             payload.setLocationType(null);
         }
 
-        if (isKvmAndFileBasedStorage && kvmIncrementalSnapshot.valueIn(clusterId)) {
+        if ((isKvmAndFileBasedStorage || StoragePoolType.CLVM_NG == poolType) && kvmIncrementalSnapshot.valueIn(clusterId)) {
             payload.setKvmIncrementalSnapshot(true);
         }
     }