diff --git a/engine/api/src/org/apache/cloudstack/engine/subsystem/api/storage/EndPointSelector.java b/engine/api/src/org/apache/cloudstack/engine/subsystem/api/storage/EndPointSelector.java
index 4657316dd8a..4d6465b8067 100644
--- a/engine/api/src/org/apache/cloudstack/engine/subsystem/api/storage/EndPointSelector.java
+++ b/engine/api/src/org/apache/cloudstack/engine/subsystem/api/storage/EndPointSelector.java
@@ -36,4 +36,6 @@ public interface EndPointSelector {
     EndPoint select(Scope scope, Long storeId);
 
     EndPoint selectHypervisorHost(Scope scope);
+
+    EndPoint select(DataStore store, String downloadUrl);
 }
diff --git a/engine/schema/src/com/cloud/host/dao/HostDao.java b/engine/schema/src/com/cloud/host/dao/HostDao.java
index 4d57bde0ea9..bd484825c81 100755
--- a/engine/schema/src/com/cloud/host/dao/HostDao.java
+++ b/engine/schema/src/com/cloud/host/dao/HostDao.java
@@ -88,4 +88,6 @@ public interface HostDao extends GenericDao<HostVO, Long>, StateDao<Status, Stat
     List<HostVO> listByDataCenterId(long id);
 
     List<Long> listAllHosts(long zoneId);
+
+    HostVO findByPublicIp(String publicIp);
 }
diff --git a/engine/schema/src/com/cloud/host/dao/HostDaoImpl.java b/engine/schema/src/com/cloud/host/dao/HostDaoImpl.java
index 58d472e9848..f1567ee51ea 100755
--- a/engine/schema/src/com/cloud/host/dao/HostDaoImpl.java
+++ b/engine/schema/src/com/cloud/host/dao/HostDaoImpl.java
@@ -84,6 +84,7 @@ public class HostDaoImpl extends GenericDaoBase<HostVO, Long> implements HostDao
     protected SearchBuilder<HostVO> MsStatusSearch;
     protected SearchBuilder<HostVO> DcPrivateIpAddressSearch;
     protected SearchBuilder<HostVO> DcStorageIpAddressSearch;
+    protected SearchBuilder<HostVO> PublicIpAddressSearch;
 
     protected SearchBuilder<HostVO> GuidSearch;
     protected SearchBuilder<HostVO> DcSearch;
@@ -207,6 +208,10 @@ public class HostDaoImpl extends GenericDaoBase<HostVO, Long> implements HostDao
         DcStorageIpAddressSearch.and("dc", DcStorageIpAddressSearch.entity().getDataCenterId(), SearchCriteria.Op.EQ);
         DcStorageIpAddressSearch.done();
 
+        PublicIpAddressSearch = createSearchBuilder();
+        PublicIpAddressSearch.and("publicIpAddress", PublicIpAddressSearch.entity().getPublicIpAddress(), SearchCriteria.Op.EQ);
+        PublicIpAddressSearch.done();
+
         GuidSearch = createSearchBuilder();
         GuidSearch.and("guid", GuidSearch.entity().getGuid(), SearchCriteria.Op.EQ);
         GuidSearch.done();
@@ -1068,6 +1073,14 @@ public class HostDaoImpl extends GenericDaoBase<HostVO, Long> implements HostDao
         return listBy(sc);
     }
 
+    @Override
+    public HostVO findByPublicIp(String publicIp) {
+        SearchCriteria<HostVO> sc = PublicIpAddressSearch.create();
+        sc.setParameters("publicIpAddress", publicIp);
+        return findOneBy(sc);
+    }
+
+
     @Override
     public List<HostVO> findHypervisorHostInCluster(long clusterId) {
         SearchCriteria<HostVO> sc = TypeClusterStatusSearch.create();
diff --git a/engine/schema/src/org/apache/cloudstack/storage/datastore/db/TemplateDataStoreDao.java b/engine/schema/src/org/apache/cloudstack/storage/datastore/db/TemplateDataStoreDao.java
index 13c4d315da8..9f8bccd83bf 100644
--- a/engine/schema/src/org/apache/cloudstack/storage/datastore/db/TemplateDataStoreDao.java
+++ b/engine/schema/src/org/apache/cloudstack/storage/datastore/db/TemplateDataStoreDao.java
@@ -78,4 +78,6 @@ public interface TemplateDataStoreDao extends GenericDao<TemplateDataStoreVO, Lo
     List<TemplateDataStoreVO> listTemplateDownloadUrls();
 
     void removeByTemplateStore(long templateId, long imageStoreId);
+
+    void expireDnldUrlsForZone(Long dcId);
 }
diff --git a/engine/schema/src/org/apache/cloudstack/storage/datastore/db/VolumeDataStoreDao.java b/engine/schema/src/org/apache/cloudstack/storage/datastore/db/VolumeDataStoreDao.java
index 0016185b9cb..625cc297274 100644
--- a/engine/schema/src/org/apache/cloudstack/storage/datastore/db/VolumeDataStoreDao.java
+++ b/engine/schema/src/org/apache/cloudstack/storage/datastore/db/VolumeDataStoreDao.java
@@ -44,4 +44,6 @@ public interface VolumeDataStoreDao extends GenericDao<VolumeDataStoreVO, Long>,
     void duplicateCacheRecordsOnRegionStore(long storeId);
 
     List<VolumeDataStoreVO> listVolumeDownloadUrls();
+
+    void expireDnldUrlsForZone(Long dcId);
 }
diff --git a/engine/storage/src/org/apache/cloudstack/storage/endpoint/DefaultEndPointSelector.java b/engine/storage/src/org/apache/cloudstack/storage/endpoint/DefaultEndPointSelector.java
index c620c40d981..18c8991e0c2 100644
--- a/engine/storage/src/org/apache/cloudstack/storage/endpoint/DefaultEndPointSelector.java
+++ b/engine/storage/src/org/apache/cloudstack/storage/endpoint/DefaultEndPointSelector.java
@@ -18,6 +18,8 @@
  */
 package org.apache.cloudstack.storage.endpoint;
 
+import java.net.URI;
+import java.net.URISyntaxException;
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
@@ -261,6 +263,33 @@ public class DefaultEndPointSelector implements EndPointSelector {
         return null;
     }
 
+    @Override
+    public EndPoint select(DataStore store, String downloadUrl){
+
+        HostVO host = null;
+        try {
+            URI uri = new URI(downloadUrl);
+            String scheme = uri.getScheme();
+            String publicIp = uri.getHost();
+            // If its https then public ip will be of the form xxx-xxx-xxx-xxx.mydomain.com
+            if(scheme.equalsIgnoreCase("https")){
+                publicIp = publicIp.split("\\.")[0]; // We want xxx-xxx-xxx-xxx
+                publicIp = publicIp.replace("-","."); // We not want the IP -  xxx.xxx.xxx.xxx
+            }
+            host = hostDao.findByPublicIp(publicIp);
+            if(host != null){
+                return RemoteHostEndPoint.getHypervisorHostEndPoint(host);
+            }
+
+        } catch (URISyntaxException e) {
+            s_logger.debug("Received URISyntaxException for url" +downloadUrl);
+        }
+
+        // If ssvm doesnt exist then find any ssvm in the zone.
+        s_logger.debug("Coudn't find ssvm for url" +downloadUrl);
+        return findEndpointForImageStorage(store);
+    }
+
     @Override
     public EndPoint select(DataStore store) {
         if (store.getRole() == DataStoreRole.Primary) {
diff --git a/engine/storage/src/org/apache/cloudstack/storage/image/db/TemplateDataStoreDaoImpl.java b/engine/storage/src/org/apache/cloudstack/storage/image/db/TemplateDataStoreDaoImpl.java
index a5cf6f8e8cb..dad0b6dc9f6 100644
--- a/engine/storage/src/org/apache/cloudstack/storage/image/db/TemplateDataStoreDaoImpl.java
+++ b/engine/storage/src/org/apache/cloudstack/storage/image/db/TemplateDataStoreDaoImpl.java
@@ -16,6 +16,7 @@
 // under the License.
 package org.apache.cloudstack.storage.image.db;
 
+import java.sql.PreparedStatement;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Date;
@@ -64,6 +65,7 @@ public class TemplateDataStoreDaoImpl extends GenericDaoBase<TemplateDataStoreVO
     private SearchBuilder<TemplateDataStoreVO> storeTemplateStateSearch;
     private SearchBuilder<TemplateDataStoreVO> storeTemplateDownloadStatusSearch;
     private SearchBuilder<TemplateDataStoreVO> downloadTemplateSearch;
+    private static final String EXPIRE_DOWNLOAD_URLS_FOR_ZONE = "update template_store_ref set download_url_created=? where store_id in (select id from image_store where data_center_id=?)";
 
     @Inject
     private DataStoreManager _storeMgr;
@@ -511,4 +513,22 @@ public class TemplateDataStoreDaoImpl extends GenericDaoBase<TemplateDataStoreVO
         return listBy(sc);
     }
 
+    @Override
+    public void expireDnldUrlsForZone(Long dcId){
+        TransactionLegacy txn = TransactionLegacy.currentTxn();
+        PreparedStatement pstmt = null;
+        try {
+            txn.start();
+            pstmt = txn.prepareAutoCloseStatement(EXPIRE_DOWNLOAD_URLS_FOR_ZONE);
+            pstmt.setDate(1, new java.sql.Date(-1l));// Set the time before the epoch time.
+            pstmt.setLong(2, dcId);
+            pstmt.executeUpdate();
+            txn.commit();
+        } catch (Exception e) {
+            txn.rollback();
+            s_logger.warn("Failed expiring download urls for dcId: " + dcId, e);
+        }
+
+    }
+
 }
diff --git a/engine/storage/src/org/apache/cloudstack/storage/image/db/VolumeDataStoreDaoImpl.java b/engine/storage/src/org/apache/cloudstack/storage/image/db/VolumeDataStoreDaoImpl.java
index 9309f6ef3c2..2e29ad79cf7 100644
--- a/engine/storage/src/org/apache/cloudstack/storage/image/db/VolumeDataStoreDaoImpl.java
+++ b/engine/storage/src/org/apache/cloudstack/storage/image/db/VolumeDataStoreDaoImpl.java
@@ -16,6 +16,7 @@
 // under the License.
 package org.apache.cloudstack.storage.image.db;
 
+import java.sql.PreparedStatement;
 import java.util.ArrayList;
 import java.util.Date;
 import java.util.List;
@@ -51,6 +52,8 @@ public class VolumeDataStoreDaoImpl extends GenericDaoBase<VolumeDataStoreVO, Lo
     private SearchBuilder<VolumeDataStoreVO> cacheSearch;
     private SearchBuilder<VolumeDataStoreVO> storeVolumeSearch;
     private SearchBuilder<VolumeDataStoreVO> downloadVolumeSearch;
+    private static final String EXPIRE_DOWNLOAD_URLS_FOR_ZONE = "update volume_store_ref set download_url_created=? where store_id in (select id from image_store where data_center_id=?)";
+
 
     @Inject
     DataStoreManager storeMgr;
@@ -267,4 +270,22 @@ public class VolumeDataStoreDaoImpl extends GenericDaoBase<VolumeDataStoreVO, Lo
         sc.setParameters("destroyed", false);
         return listBy(sc);
     }
+
+    @Override
+    public void expireDnldUrlsForZone(Long dcId){
+        TransactionLegacy txn = TransactionLegacy.currentTxn();
+        PreparedStatement pstmt = null;
+        try {
+            txn.start();
+            pstmt = txn.prepareAutoCloseStatement(EXPIRE_DOWNLOAD_URLS_FOR_ZONE);
+            pstmt.setDate(1, new java.sql.Date(-1l));// Set the time before the epoch time.
+            pstmt.setLong(2, dcId);
+            pstmt.executeUpdate();
+            txn.commit();
+        } catch (Exception e) {
+            txn.rollback();
+            s_logger.warn("Failed expiring download urls for dcId: " + dcId, e);
+        }
+
+    }
 }
diff --git a/plugins/storage/image/default/src/org/apache/cloudstack/storage/datastore/driver/CloudStackImageStoreDriverImpl.java b/plugins/storage/image/default/src/org/apache/cloudstack/storage/datastore/driver/CloudStackImageStoreDriverImpl.java
index c25e99f7271..05222c81f16 100644
--- a/plugins/storage/image/default/src/org/apache/cloudstack/storage/datastore/driver/CloudStackImageStoreDriverImpl.java
+++ b/plugins/storage/image/default/src/org/apache/cloudstack/storage/datastore/driver/CloudStackImageStoreDriverImpl.java
@@ -23,6 +23,7 @@ import java.util.UUID;
 import javax.inject.Inject;
 
 import com.cloud.agent.api.storage.DeleteEntityDownloadURLCommand;
+import com.cloud.host.dao.HostDao;
 import com.cloud.storage.Upload;
 import org.apache.log4j.Logger;
 
@@ -49,6 +50,8 @@ public class CloudStackImageStoreDriverImpl extends BaseImageStoreDriverImpl {
     @Inject
     ConfigurationDao _configDao;
     @Inject
+    HostDao _hostDao;
+    @Inject
     EndPointSelector _epSelector;
 
     @Override
@@ -113,8 +116,8 @@ public class CloudStackImageStoreDriverImpl extends BaseImageStoreDriverImpl {
 
     @Override
     public void deleteEntityExtractUrl(DataStore store, String installPath, String downloadUrl, Upload.Type entityType) {
-        // find an endpoint to send command
-        EndPoint ep = _epSelector.select(store);
+        // find an endpoint to send command based on the ssvm on which the url was created.
+        EndPoint ep = _epSelector.select(store, downloadUrl);
 
         // Delete Symlink at ssvm. In case of volume also delete the volume.
         DeleteEntityDownloadURLCommand cmd = new DeleteEntityDownloadURLCommand(installPath, entityType, downloadUrl, ((ImageStoreEntity) store).getMountPoint());
diff --git a/services/secondary-storage/controller/src/org/apache/cloudstack/secondarystorage/SecondaryStorageManagerImpl.java b/services/secondary-storage/controller/src/org/apache/cloudstack/secondarystorage/SecondaryStorageManagerImpl.java
index 9288be479a4..045c62283fc 100755
--- a/services/secondary-storage/controller/src/org/apache/cloudstack/secondarystorage/SecondaryStorageManagerImpl.java
+++ b/services/secondary-storage/controller/src/org/apache/cloudstack/secondarystorage/SecondaryStorageManagerImpl.java
@@ -43,6 +43,7 @@ import org.apache.cloudstack.framework.security.keystore.KeystoreManager;
 import org.apache.cloudstack.storage.datastore.db.ImageStoreDao;
 import org.apache.cloudstack.storage.datastore.db.ImageStoreVO;
 import org.apache.cloudstack.storage.datastore.db.TemplateDataStoreDao;
+import org.apache.cloudstack.storage.datastore.db.VolumeDataStoreDao;
 import org.apache.cloudstack.utils.identity.ManagementServerNode;
 import org.apache.log4j.Logger;
 
@@ -236,6 +237,8 @@ public class SecondaryStorageManagerImpl extends ManagerBase implements Secondar
     ImageStoreDao _imageStoreDao;
     @Inject
     TemplateDataStoreDao _tmplStoreDao;
+    @Inject
+    VolumeDataStoreDao _volumeStoreDao;
     private long _capacityScanInterval = DEFAULT_CAPACITY_SCAN_INTERVAL;
     private int _secStorageVmMtuSize;
 
@@ -987,7 +990,9 @@ public class SecondaryStorageManagerImpl extends ManagerBase implements Secondar
                 s_logger.debug("Removing host entry for ssvm id=" + vmId);
                 _hostDao.remove(host.getId());
             }
-
+            //Expire the download urls in the entire zone for templates and volumes.
+            _tmplStoreDao.expireDnldUrlsForZone(host.getDataCenterId());
+            _volumeStoreDao.expireDnldUrlsForZone(host.getDataCenterId());
             return true;
         } catch (ResourceUnavailableException e) {
             s_logger.warn("Unable to expunge " + ssvm, e);