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 6a903e4235a4..7f5e4996f55b 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
@@ -30,15 +30,12 @@
 
 import javax.inject.Inject;
 
-import com.cloud.dc.DedicatedResourceVO;
-import com.cloud.dc.dao.DedicatedResourceDao;
-import com.cloud.user.Account;
-import com.cloud.utils.Pair;
 import org.apache.cloudstack.context.CallContext;
 import org.apache.cloudstack.engine.subsystem.api.storage.DataObject;
 import org.apache.cloudstack.engine.subsystem.api.storage.DataStore;
 import org.apache.cloudstack.engine.subsystem.api.storage.EndPoint;
 import org.apache.cloudstack.engine.subsystem.api.storage.EndPointSelector;
+import org.apache.cloudstack.engine.subsystem.api.storage.PrimaryDataStore;
 import org.apache.cloudstack.engine.subsystem.api.storage.Scope;
 import org.apache.cloudstack.engine.subsystem.api.storage.SnapshotInfo;
 import org.apache.cloudstack.engine.subsystem.api.storage.StorageAction;
@@ -50,14 +47,19 @@
 import org.springframework.stereotype.Component;
 
 import com.cloud.capacity.CapacityManager;
+import com.cloud.dc.DedicatedResourceVO;
+import com.cloud.dc.dao.DedicatedResourceDao;
 import com.cloud.host.Host;
 import com.cloud.host.HostVO;
 import com.cloud.host.Status;
 import com.cloud.host.dao.HostDao;
 import com.cloud.hypervisor.Hypervisor;
+import com.cloud.resource.ResourceState;
 import com.cloud.storage.DataStoreRole;
 import com.cloud.storage.ScopeType;
 import com.cloud.storage.Storage.TemplateType;
+import com.cloud.user.Account;
+import com.cloud.utils.Pair;
 import com.cloud.utils.db.DB;
 import com.cloud.utils.db.QueryBuilder;
 import com.cloud.utils.db.SearchCriteria.Op;
@@ -318,6 +320,35 @@ protected EndPoint findEndpointForImageStorage(DataStore store) {
         return RemoteHostEndPoint.getHypervisorHostEndPoint(host);
     }
 
+    protected Host getVmwareHostFromVolumeToDelete(VolumeInfo volume) {
+        VirtualMachine vm = volume.getAttachedVM();
+        if (vm == null) {
+            return null;
+        }
+        Long hostId = vm.getHostId() != null ? vm.getHostId() : vm.getLastHostId();
+        if (hostId == null) {
+            return null;
+        }
+        HostVO host = hostDao.findById(hostId);
+        if (host == null || !ResourceState.Enabled.equals(host.getResourceState()) || !Status.Up.equals(host.getState())) {
+            return null;
+        }
+        if (volume.getDataStore() instanceof PrimaryDataStore) {
+            PrimaryDataStore dataStore = (PrimaryDataStore)volume.getDataStore();
+            if (dataStore.getScope() == null) {
+                return null;
+            }
+            if (ScopeType.HOST.equals(dataStore.getScope().getScopeType()) && !host.getStorageIpAddress().equals(dataStore.getHostAddress())) {
+                return null;
+            }
+            if (ScopeType.CLUSTER.equals(dataStore.getScope().getScopeType()) && !host.getClusterId().equals(dataStore.getClusterId())) {
+                return null;
+            }
+        }
+        s_logger.debug(String.format("Selected host %s for deleting volume ID: %d", host, volume.getId()));
+        return host;
+    }
+
     @Override
     public List<EndPoint> findAllEndpointsForScope(DataStore store) {
         Long dcId = null;
@@ -436,13 +467,10 @@ public EndPoint select(DataObject object, StorageAction action) {
             }
         } else if (action == StorageAction.DELETEVOLUME) {
             VolumeInfo volume = (VolumeInfo)object;
-            if (volume.getHypervisorType() == Hypervisor.HypervisorType.VMware) {
-                VirtualMachine vm = volume.getAttachedVM();
-                if (vm != null) {
-                    Long hostId = vm.getHostId() != null ? vm.getHostId() : vm.getLastHostId();
-                    if (hostId != null) {
-                        return getEndPointFromHostId(hostId);
-                    }
+            if (Hypervisor.HypervisorType.VMware.equals(volume.getHypervisorType()))  {
+                Host host = getVmwareHostFromVolumeToDelete(volume);
+                if (host != null) {
+                    return RemoteHostEndPoint.getHypervisorHostEndPoint(host);
                 }
             }
         }
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
new file mode 100644
index 000000000000..ab7cced831bd
--- /dev/null
+++ b/engine/storage/src/test/java/org/apache/cloudstack/storage/endpoint/DefaultEndPointSelectorTest.java
@@ -0,0 +1,179 @@
+// 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 org.apache.cloudstack.storage.endpoint;
+
+import org.apache.cloudstack.engine.subsystem.api.storage.PrimaryDataStore;
+import org.apache.cloudstack.engine.subsystem.api.storage.Scope;
+import org.apache.cloudstack.engine.subsystem.api.storage.VolumeInfo;
+import org.apache.cloudstack.storage.image.datastore.ImageStoreInfo;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.MockitoAnnotations;
+
+import com.cloud.host.HostVO;
+import com.cloud.host.Status;
+import com.cloud.host.dao.HostDao;
+import com.cloud.resource.ResourceState;
+import com.cloud.storage.ScopeType;
+import com.cloud.vm.VirtualMachine;
+
+public class DefaultEndPointSelectorTest {
+
+    @Mock
+    HostDao hostDao;
+
+    @Before
+    public void initMocks() {
+        MockitoAnnotations.initMocks(this);
+    }
+    @InjectMocks
+    private DefaultEndPointSelector defaultEndPointSelector = new DefaultEndPointSelector();
+
+    @Test
+    public void testGetVmwareHostFromVolumeToDeleteBasic() {
+        VolumeInfo volumeInfo = Mockito.mock(VolumeInfo.class);
+        Mockito.when(volumeInfo.getAttachedVM()).thenReturn(null);
+        Assert.assertNull(defaultEndPointSelector.getVmwareHostFromVolumeToDelete(volumeInfo));
+
+        VirtualMachine virtualMachine = Mockito.mock(VirtualMachine.class);
+        Mockito.when(virtualMachine.getHostId()).thenReturn(null);
+        Mockito.when(virtualMachine.getLastHostId()).thenReturn(null);
+        Mockito.when(volumeInfo.getAttachedVM()).thenReturn(virtualMachine);
+        Assert.assertNull(defaultEndPointSelector.getVmwareHostFromVolumeToDelete(volumeInfo));
+
+        Long hostId = 1L;
+        Mockito.when(virtualMachine.getHostId()).thenReturn(hostId);
+        Mockito.when(volumeInfo.getAttachedVM()).thenReturn(virtualMachine);
+        Mockito.when(hostDao.findById(hostId)).thenReturn(null);
+        Assert.assertNull(defaultEndPointSelector.getVmwareHostFromVolumeToDelete(volumeInfo));
+
+        HostVO hostVO = Mockito.mock(HostVO.class);
+        Mockito.when(hostVO.getResourceState()).thenReturn(ResourceState.Disabled);
+        Mockito.when(hostDao.findById(hostId)).thenReturn(hostVO);
+        Assert.assertNull(defaultEndPointSelector.getVmwareHostFromVolumeToDelete(volumeInfo));
+
+        Mockito.when(hostVO.getResourceState()).thenReturn(ResourceState.Enabled);
+        Mockito.when(hostVO.getState()).thenReturn(Status.Down);
+        Assert.assertNull(defaultEndPointSelector.getVmwareHostFromVolumeToDelete(volumeInfo));
+    }
+
+    @Test
+    public void testGetVmwareHostFromVolumeToDeleteNotPrimary() {
+        VolumeInfo volumeInfo = Mockito.mock(VolumeInfo.class);
+        Long hostId = 1L;
+        VirtualMachine virtualMachine = Mockito.mock(VirtualMachine.class);
+        Mockito.when(virtualMachine.getHostId()).thenReturn(hostId);
+        Mockito.when(volumeInfo.getAttachedVM()).thenReturn(virtualMachine);
+        HostVO hostVO = Mockito.mock(HostVO.class);
+        Mockito.when(hostVO.getResourceState()).thenReturn(ResourceState.Enabled);
+        Mockito.when(hostVO.getState()).thenReturn(Status.Up);
+        Mockito.when(hostDao.findById(hostId)).thenReturn(hostVO);
+        Mockito.when(volumeInfo.getDataStore()).thenReturn(Mockito.mock(ImageStoreInfo.class));
+        Assert.assertNotNull(defaultEndPointSelector.getVmwareHostFromVolumeToDelete(volumeInfo));
+    }
+
+    @Test
+    public void testGetVmwareHostFromVolumeToDeletePrimaryNoScope() {
+        VolumeInfo volumeInfo = Mockito.mock(VolumeInfo.class);
+        Long hostId = 1L;
+        VirtualMachine virtualMachine = Mockito.mock(VirtualMachine.class);
+        Mockito.when(virtualMachine.getHostId()).thenReturn(hostId);
+        Mockito.when(volumeInfo.getAttachedVM()).thenReturn(virtualMachine);
+        HostVO hostVO = Mockito.mock(HostVO.class);
+        Mockito.when(hostVO.getResourceState()).thenReturn(ResourceState.Enabled);
+        Mockito.when(hostVO.getState()).thenReturn(Status.Up);
+        Mockito.when(hostDao.findById(hostId)).thenReturn(hostVO);
+        PrimaryDataStore store = Mockito.mock(PrimaryDataStore.class);
+        Mockito.when(store.getScope()).thenReturn(null);
+        Mockito.when(volumeInfo.getDataStore()).thenReturn(store);
+        Assert.assertNull(defaultEndPointSelector.getVmwareHostFromVolumeToDelete(volumeInfo));
+    }
+
+    @Test
+    public void testGetVmwareHostFromVolumeToDeletePrimaryNotHostOrClusterScope() {
+        VolumeInfo volumeInfo = Mockito.mock(VolumeInfo.class);
+        Long hostId = 1L;
+        VirtualMachine virtualMachine = Mockito.mock(VirtualMachine.class);
+        Mockito.when(virtualMachine.getHostId()).thenReturn(hostId);
+        Mockito.when(volumeInfo.getAttachedVM()).thenReturn(virtualMachine);
+        HostVO hostVO = Mockito.mock(HostVO.class);
+        Mockito.when(hostVO.getResourceState()).thenReturn(ResourceState.Enabled);
+        Mockito.when(hostVO.getState()).thenReturn(Status.Up);
+        Mockito.when(hostDao.findById(hostId)).thenReturn(hostVO);
+        PrimaryDataStore store = Mockito.mock(PrimaryDataStore.class);
+        Scope scope = Mockito.mock(Scope.class);
+        Mockito.when(scope.getScopeType()).thenReturn(ScopeType.ZONE);
+        Mockito.when(store.getScope()).thenReturn(scope);
+        Mockito.when(volumeInfo.getDataStore()).thenReturn(store);
+        Assert.assertNotNull(defaultEndPointSelector.getVmwareHostFromVolumeToDelete(volumeInfo));
+    }
+
+    @Test
+    public void testGetVmwareHostFromVolumeToDeletePrimaryClusterScope() {
+        VolumeInfo volumeInfo = Mockito.mock(VolumeInfo.class);
+        Long hostId = 1L;
+        Long clusterId = 1L;
+        VirtualMachine virtualMachine = Mockito.mock(VirtualMachine.class);
+        Mockito.when(virtualMachine.getHostId()).thenReturn(hostId);
+        Mockito.when(volumeInfo.getAttachedVM()).thenReturn(virtualMachine);
+        HostVO hostVO = Mockito.mock(HostVO.class);
+        Mockito.when(hostVO.getResourceState()).thenReturn(ResourceState.Enabled);
+        Mockito.when(hostVO.getState()).thenReturn(Status.Up);
+        Mockito.when(hostVO.getClusterId()).thenReturn(clusterId);
+        Mockito.when(hostDao.findById(hostId)).thenReturn(hostVO);
+        PrimaryDataStore store = Mockito.mock(PrimaryDataStore.class);
+        Scope scope = Mockito.mock(Scope.class);
+        Mockito.when(scope.getScopeType()).thenReturn(ScopeType.CLUSTER);
+        Mockito.when(store.getScope()).thenReturn(scope);
+        Mockito.when(store.getClusterId()).thenReturn(clusterId);
+        Mockito.when(volumeInfo.getDataStore()).thenReturn(store);
+        Assert.assertNotNull(defaultEndPointSelector.getVmwareHostFromVolumeToDelete(volumeInfo));
+
+        Mockito.when(store.getClusterId()).thenReturn(2L);
+        Assert.assertNull(defaultEndPointSelector.getVmwareHostFromVolumeToDelete(volumeInfo));
+    }
+
+    @Test
+    public void testGetVmwareHostFromVolumeToDeletePrimaryHostScope() {
+        VolumeInfo volumeInfo = Mockito.mock(VolumeInfo.class);
+        Long hostId = 1L;
+        String storageIp = "something";
+        VirtualMachine virtualMachine = Mockito.mock(VirtualMachine.class);
+        Mockito.when(virtualMachine.getHostId()).thenReturn(hostId);
+        Mockito.when(volumeInfo.getAttachedVM()).thenReturn(virtualMachine);
+        HostVO hostVO = Mockito.mock(HostVO.class);
+        Mockito.when(hostVO.getResourceState()).thenReturn(ResourceState.Enabled);
+        Mockito.when(hostVO.getState()).thenReturn(Status.Up);
+        Mockito.when(hostVO.getStorageIpAddress()).thenReturn(storageIp);
+        Mockito.when(hostDao.findById(hostId)).thenReturn(hostVO);
+        PrimaryDataStore store = Mockito.mock(PrimaryDataStore.class);
+        Scope scope = Mockito.mock(Scope.class);
+        Mockito.when(scope.getScopeType()).thenReturn(ScopeType.HOST);
+        Mockito.when(store.getScope()).thenReturn(scope);
+        Mockito.when(store.getHostAddress()).thenReturn(storageIp);
+        Mockito.when(volumeInfo.getDataStore()).thenReturn(store);
+        Assert.assertNotNull(defaultEndPointSelector.getVmwareHostFromVolumeToDelete(volumeInfo));
+
+        Mockito.when(store.getHostAddress()).thenReturn("something-else");
+        Assert.assertNull(defaultEndPointSelector.getVmwareHostFromVolumeToDelete(volumeInfo));
+    }
+}