@@ -40,7 +40,9 @@ public Answer execute(final DestroyCommand command, final LibvirtComputingResour
         try {
             final KVMStoragePoolManager storagePoolMgr = libvirtComputingResource.getStoragePoolMgr();
             final KVMStoragePool pool = storagePoolMgr.getStoragePool(vol.getPoolType(), vol.getPoolUuid());
-            pool.deletePhysicalDisk(vol.getPath(), null);
+            if (!pool.deletePhysicalDisk(vol.getPath(), null)) {
+                return new Answer(command, false, "Failed to delete " + vol.getPath());
+            }
             return new Answer(command, true, "Success");
         } catch (final CloudRuntimeException e) {
             logger.debug("Failed to delete volume: " + e.toString());
 
@@ -962,8 +962,12 @@ private boolean cleanupCLVMVolume(String uuid, KVMStoragePool pool) {
             String checkResult = checkLvs.execute();
 
             if (checkResult != null) {
-                logger.info("CLVM volume {} does not exist in LVM (check returned: {}), considering it as already deleted", uuid, checkResult);
-                return true;
+                if (isLvNotFound(checkResult)) {
+                    logger.info("CLVM volume {} does not exist in LVM (check returned: {}), considering it as already deleted", uuid, checkResult);
+                    return true;
+                }
+                logger.warn("Cannot check CLVM volume {}: {}", uuid, checkResult);
+                return false;
             }
 
             logger.info("Volume {} exists, proceeding with cleanup", uuid);
@@ -990,18 +994,22 @@ private boolean cleanupCLVMVolume(String uuid, KVMStoragePool pool) {
                 return true;
             } else {
                 logger.warn("Command 'lvremove -f {}' returned error: {}", lvPath, removeResult);
-                if (removeResult.contains("not found") || removeResult.contains("Failed to find")) {
+                if (isLvNotFound(removeResult)) {
                     logger.info("CLVM volume {} not found during cleanup, considering it as already deleted", uuid);
                     return true;
                 }
                 return false;
             }
         } catch (Exception ex) {
             logger.error("Exception during CLVM volume cleanup for {}: {}", uuid, ex.getMessage(), ex);
-            return true;
+            return false;
         }
     }
 
+    private boolean isLvNotFound(String lvmError) {
+        return lvmError.contains("not found") || lvmError.contains("Failed to find");
+    }
+
     private boolean shouldSecureZeroFill(KVMStoragePool pool) {
         Map<String, String> details = pool.getDetails();
         String secureZeroFillStr = (details != null) ? details.get(KVMStoragePool.CLVM_SECURE_ZERO_FILL) : null;
 
@@ -2269,6 +2269,7 @@ public void testDestroyCommand() {
 
         when(libvirtComputingResourceMock.getStoragePoolMgr()).thenReturn(poolManager);
         when(poolManager.getStoragePool(vol.getPoolType(), vol.getPoolUuid())).thenReturn(primary);
+        when(primary.deletePhysicalDisk(vol.getPath(), null)).thenReturn(true);
 
         final LibvirtRequestWrapper wrapper = LibvirtRequestWrapper.getInstance();
         assertNotNull(wrapper);
@@ -2280,6 +2281,22 @@ public void testDestroyCommand() {
         verify(poolManager, times(1)).getStoragePool(vol.getPoolType(), vol.getPoolUuid());
     }
 
+    @Test
+    public void testDestroyCommandDeleteFailed() {
+        final DestroyCommand command = new DestroyCommand(Mockito.mock(StoragePool.class), Mockito.mock(Volume.class), "Test");
+        final KVMStoragePoolManager poolManager = Mockito.mock(KVMStoragePoolManager.class);
+        final KVMStoragePool primary = Mockito.mock(KVMStoragePool.class);
+        final VolumeTO vol = command.getVolume();
+
+        when(libvirtComputingResourceMock.getStoragePoolMgr()).thenReturn(poolManager);
+        when(poolManager.getStoragePool(vol.getPoolType(), vol.getPoolUuid())).thenReturn(primary);
+        when(primary.deletePhysicalDisk(vol.getPath(), null)).thenReturn(false);
+
+        final Answer answer = LibvirtRequestWrapper.getInstance().execute(command, libvirtComputingResourceMock);
+
+        assertFalse(answer.getResult());
+    }
+
     @SuppressWarnings("unchecked")
     @Test
     public void testDestroyCommandError() {
 
@@ -36,6 +36,7 @@
 import java.util.Map;
 import java.util.UUID;
 import org.junit.After;
+import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -821,6 +822,38 @@ public void testCleanupCLVMVolume_VolumeNotFound() throws Exception {
         assert result : "Should return true when volume not found (already deleted)";
     }
 
+    @Test
+    public void testCleanupCLVMVolume_LockedByOtherHost() throws Exception {
+        Mockito.when(mockPool.getLocalPath()).thenReturn("testvg");
+        Mockito.when(mockPool.getUuid()).thenReturn(UUID.randomUUID().toString());
+
+        mockScriptConstruction = Mockito.mockConstruction(Script.class, (mock, context) -> {
+            boolean lvremove = "lvremove".equals(context.arguments().get(0));
+            when(mock.execute()).thenReturn(lvremove ? "  LV locked by other host: testvg/template-1" : null);
+        });
+
+        Method method = ClvmStorageAdaptor.class.getDeclaredMethod(
+            "cleanupCLVMVolume", String.class, KVMStoragePool.class);
+        method.setAccessible(true);
+
+        Assert.assertFalse((boolean) method.invoke(clvmStorageAdaptor, "template-1", mockPool));
+    }
+
+    @Test
+    public void testCleanupCLVMVolume_CheckFailsForOtherReason() throws Exception {
+        Mockito.when(mockPool.getLocalPath()).thenReturn("testvg");
+        Mockito.when(mockPool.getUuid()).thenReturn(UUID.randomUUID().toString());
+
+        mockScriptConstruction = Mockito.mockConstruction(Script.class,
+            (mock, context) -> when(mock.execute()).thenReturn("  Lock manager lvmlockd is not running"));
+
+        Method method = ClvmStorageAdaptor.class.getDeclaredMethod(
+            "cleanupCLVMVolume", String.class, KVMStoragePool.class);
+        method.setAccessible(true);
+
+        Assert.assertFalse((boolean) method.invoke(clvmStorageAdaptor, "template-1", mockPool));
+    }
+
     @Test
     public void testCleanupCLVMVolume_NullSourceDir() throws Exception {
         String volumeUuid = UUID.randomUUID().toString();
 
@@ -23,8 +23,10 @@
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutionException;
@@ -33,6 +35,7 @@
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
 import java.util.stream.Collectors;
 import javax.inject.Inject;
 
@@ -58,6 +61,8 @@
 import org.apache.cloudstack.managed.context.ManagedContextRunnable;
 import org.apache.cloudstack.storage.clvm.command.ClvmLockTransferCommand;
 import org.apache.cloudstack.storage.clvm.command.ClvmLockTransferAnswer;
+import org.apache.cloudstack.storage.datastore.db.PrimaryDataStoreDao;
+import org.apache.cloudstack.storage.datastore.db.StoragePoolVO;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.springframework.stereotype.Component;
@@ -72,6 +77,8 @@ public class ClvmPoolManager implements Configurable {
     private HostDao _hostDao;
     @Inject
     private HostDetailsDao _hostDetailsDao;
+    @Inject
+    private PrimaryDataStoreDao _storagePoolDao;
 
     protected Logger logger = LogManager.getLogger(getClass());
 
@@ -93,10 +100,12 @@ public class ClvmPoolManager implements Configurable {
     public static final ConfigKey<Boolean> CLVMSecureZeroFill = new ConfigKey<>("Advanced", Boolean.class, "clvm.secure.zero.fill", "false",
             "When enabled, CLVM volumes to be zero-filled at the time of deletion to prevent data from being recovered by VMs reusing the space, as thick LVM volumes write data linearly. Note: This setting is propagated to hosts when they connect to the storage pool. Changing this setting requires disconnecting and reconnecting hosts or restarting the KVM agent for it to take effect.", false, ConfigKey.Scope.StoragePool);
 
-    private static final int LOCK_QUERY_THREADS = 16;
+    public static final ConfigKey<Integer> ClvmLockQueryThreads = new ConfigKey<>("Advanced", Integer.class, "clvm.lock.query.threads", "5",
+            "Number of threads a management server uses to send CLVM lock commands to several hosts in parallel: the fallback lock holder lookup that asks every host, and the deactivation of a template LV on all hosts before it is removed. Requires a management server restart.", false);
+
     private static final int MAX_LEASE_READERS = 2;
 
-    private final ExecutorService lockQueryExecutor = newLockQueryExecutor();
+    private ExecutorService lockQueryExecutor;
 
     /** State shared by the lock lookups of all volumes of one operation. */
     public static class LockLookup {
@@ -114,17 +123,86 @@ boolean isUnreachable(Long hostId) {
         }
     }
 
-    private static ExecutorService newLockQueryExecutor() {
-        ThreadPoolExecutor executor = new ThreadPoolExecutor(LOCK_QUERY_THREADS, LOCK_QUERY_THREADS, 60L, TimeUnit.SECONDS,
-                new LinkedBlockingQueue<>(), new NamedThreadFactory("ClvmLockQuery"));
-        executor.allowCoreThreadTimeOut(true);
-        return executor;
+    /** Created on first use, as the config value is not available when the bean is constructed. */
+    private synchronized ExecutorService getLockQueryExecutor() {
+        if (lockQueryExecutor == null) {
+            int threads = Math.max(1, ClvmLockQueryThreads.value());
+            ThreadPoolExecutor executor = new ThreadPoolExecutor(threads, threads, 60L, TimeUnit.SECONDS,
+                    new LinkedBlockingQueue<>(), new NamedThreadFactory("ClvmLockQuery"));
+            executor.allowCoreThreadTimeOut(true);
+            lockQueryExecutor = executor;
+        }
+        return lockQueryExecutor;
     }
 
     public static boolean isClvmPoolType(Storage.StoragePoolType poolType) {
         return Arrays.asList(Storage.StoragePoolType.CLVM, Storage.StoragePoolType.CLVM_NG).contains(poolType);
     }
 
+    private static String getVgName(StoragePool pool) {
+        String vgName = pool.getPath();
+        return vgName.startsWith("/") ? vgName.substring(1) : vgName;
+    }
+
+    private static String getLvPath(StoragePool pool, String lvName) {
+        return String.format("/dev/%s/%s", getVgName(pool), lvName);
+    }
+
+    /**
+     * Pools of the same zone backed by the same VG share its LVs, e.g. one template LV used as backing file by volumes of all of them.
+     *
+     * @return ids of the pools sharing the VG of {@code pool}, including {@code pool}
+     */
+    public Set<Long> findPoolIdsSharingVg(StoragePool pool) {
+        String vgName = getVgName(pool);
+        Set<Long> poolIds = _storagePoolDao.findPoolsByStorageTypeAndZone(pool.getPoolType(), pool.getDataCenterId()).stream()
+                .filter(p -> vgName.equals(getVgName(p)))
+                .map(StoragePoolVO::getId)
+                .collect(Collectors.toCollection(HashSet::new));
+        poolIds.add(pool.getId());
+        return poolIds;
+    }
+
+    /**
+     * Deactivates an LV on every Up host of the pools sharing its VG, releasing their shared locks so it can be removed.
+     * Best effort: a host still using the LV refuses, and removing it then fails on the lock.
+     */
+    public void deactivateOnAllHosts(StoragePool pool, String lvName) {
+        String lvPath = getLvPath(pool, lvName);
+        Map<Long, HostVO> hosts = new LinkedHashMap<>();
+        findPoolIdsSharingVg(pool).stream()
+                .map(_storagePoolDao::findById)
+                .filter(Objects::nonNull)
+                .flatMap(p -> findPoolHosts(p).stream())
+                .filter(h -> h.getHypervisorType() == Hypervisor.HypervisorType.KVM)
+                .forEach(h -> hosts.putIfAbsent(h.getId(), h));
+        List<Long> upHostIds = hosts.values().stream().filter(this::isUpKvmRoutingHost).map(HostVO::getId).collect(Collectors.toList());
+        Map<Long, Boolean> results = callOnHosts(upHostIds, lvName, hostId -> deactivateOnHost(hosts.get(hostId), lvPath, lvName));
+        List<String> notDeactivated = hosts.values().stream()
+                .filter(h -> !Boolean.TRUE.equals(results.get(h.getId())))
+                .map(h -> String.format("%s (%s, %s)", h.getName(), h.getUuid(), h.getStatus()))
+                .collect(Collectors.toList());
+        if (notDeactivated.isEmpty()) {
+            logger.info("Deactivated {} on all hosts {}", lvPath, hosts.values().stream().map(HostVO::getName).collect(Collectors.toList()));
+            return;
+        }
+        logger.warn("{} could not be deactivated on hosts {}. It cannot be removed while active there; deactivate it on those hosts with 'lvchange -an {}'.",
+                lvPath, notDeactivated, lvPath);
+    }
+
+    private Boolean deactivateOnHost(HostVO host, String lvPath, String lvName) {
+        try {
+            Answer answer = _agentMgr.send(host.getId(), new ClvmLockTransferCommand(ClvmLockTransferCommand.Operation.DEACTIVATE, lvPath, lvName));
+            if (answer != null && answer.getResult()) {
+                return true;
+            }
+            logger.warn("Failed to deactivate {} on host {}: {}", lvPath, host.getName(), answer != null ? answer.getDetails() : "null answer");
+        } catch (AgentUnavailableException | OperationTimedoutException e) {
+            logger.warn("Could not deactivate {} on host {}: {}", lvPath, host.getName(), e.getMessage());
+        }
+        return false;
+    }
+
     /**
      * Gets the CLVM lock host ID for a volume, optionally querying actual LVM state.
      *
@@ -234,11 +312,7 @@ public Long queryCurrentLockHolder(Long volumeId, String volumeUuid, String volu
             return null;
         }
 
-        String vgName = pool.getPath();
-        if (vgName.startsWith("/")) {
-            vgName = vgName.substring(1);
-        }
-        String lvPath = String.format("/dev/%s/%s", vgName, volumePath);
+        String lvPath = getLvPath(pool, volumePath);
         Long dbHostId = getClvmLockHostId(volumeId, volumeUuid);
         // Hosts that did not answer the lock-state query for this volume
         Set<Long> noAnswerHostIds = new HashSet<>();
@@ -533,21 +607,26 @@ private Long recoverInactiveVolume(Long volumeId, String volumeUuid, String lvPa
         return null;
     }
 
+    /** @return lock state per host that answered; hosts that did not answer are absent */
+    private Map<Long, Boolean> queryHostsLockState(List<Long> hostIds, String lvPath, String volumeUuid, LockLookup lookup) {
+        return callOnHosts(hostIds, volumeUuid, hostId -> querySingleHostLockState(hostId, lvPath, volumeUuid, lookup));
+    }
+
     /**
-     * Queries the hosts in parallel, so one slow host does not delay the others.
+     * Calls the hosts in parallel, so one slow host does not delay the others.
      *
-     * @return lock state per host that answered; hosts that did not answer are absent
+     * @return result per host; hosts whose call returned null are absent
      */
-    private Map<Long, Boolean> queryHostsLockState(List<Long> hostIds, String lvPath, String volumeUuid, LockLookup lookup) {
-        Map<Long, Boolean> states = new ConcurrentHashMap<>();
+    private <T> Map<Long, T> callOnHosts(List<Long> hostIds, String volumeUuid, Function<Long, T> call) {
+        Map<Long, T> results = new ConcurrentHashMap<>();
         List<Future<?>> futures = new ArrayList<>();
         for (Long hostId : hostIds) {
-            futures.add(lockQueryExecutor.submit(new ManagedContextRunnable() {
+            futures.add(getLockQueryExecutor().submit(new ManagedContextRunnable() {
                 @Override
                 protected void runInContext() {
-                    Boolean active = querySingleHostLockState(hostId, lvPath, volumeUuid, lookup);
-                    if (active != null) {
-                        states.put(hostId, active);
+                    T result = call.apply(hostId);
+                    if (result != null) {
+                        results.put(hostId, result);
                     }
                 }
             }));
@@ -559,10 +638,10 @@ protected void runInContext() {
                 Thread.currentThread().interrupt();
                 break;
             } catch (ExecutionException e) {
-                logger.warn("Lock state query for volume {} failed: {}", volumeUuid, e.getMessage());
+                logger.warn("Host call for volume {} failed: {}", volumeUuid, e.getMessage());
             }
         }
-        return new HashMap<>(states);
+        return new HashMap<>(results);
     }
 
     /**
@@ -769,12 +848,7 @@ public boolean transferClvmVolumeLock(String volumeUuid, Long volumeId, String v
             return false;
         }
 
-        String vgName = pool.getPath();
-        if (vgName.startsWith("/")) {
-            vgName = vgName.substring(1);
-        }
-
-        String lvPath = String.format("/dev/%s/%s", vgName, volumePath);
+        String lvPath = getLvPath(pool, volumePath);
 
         try {
             // sourceHostId is trusted as pre-validated by the caller
@@ -833,7 +907,8 @@ public ConfigKey<?>[] getConfigKeys() {
         return new ConfigKey<?>[] {
                 CLVMSecureZeroFill,
                 ClvmLockRecoveryRetryTimeout,
-                ClvmLockRecoveryRetryInterval
+                ClvmLockRecoveryRetryInterval,
+                ClvmLockQueryThreads
         };
     }
 }