public Set<LocallyCachedBlob> cleanup(ClientBlobStore store) {
        Set<LocallyCachedBlob> deleted = new HashSet<>();
        LOG.debug("cleanup target size: {} current size is: {}", targetSize, currentSize);
        long bytesOver = currentSize - targetSize;
        //First delete everything that no longer exists...
        for (Iterator<Map.Entry<LocallyCachedBlob, Map<String, ? extends LocallyCachedBlob>>> i = noReferences.entrySet().iterator();
             i.hasNext(); ) {
            Map.Entry<LocallyCachedBlob, Map<String, ? extends LocallyCachedBlob>> rsrc = i.next();
            LocallyCachedBlob resource = rsrc.getKey();
            try {
                if (!store.isRemoteBlobExists(resource.getKey())) {
                    //The key was removed so we should delete it too.
                    Map<String, ? extends LocallyCachedBlob> set = rsrc.getValue();
                    if (removeBlob(resource, set)) {
                        bytesOver -= resource.getSizeOnDisk();
                        LOG.info("Deleted blob: {} (REMOVED FROM CLUSTER).", resource.getKey());
                        deleted.add(resource);
                        i.remove();
                    }
                }
            } catch (AuthorizationException e) {
                //Ignored
            }
        }

        for (Iterator<Map.Entry<LocallyCachedBlob, Map<String, ? extends LocallyCachedBlob>>> i = noReferences.entrySet().iterator();
             bytesOver > 0 && i.hasNext(); ) {
            Map.Entry<LocallyCachedBlob, Map<String, ? extends LocallyCachedBlob>> rsrc = i.next();
            LocallyCachedBlob resource = rsrc.getKey();
            Map<String, ? extends LocallyCachedBlob> set = rsrc.getValue();
            if (removeBlob(resource, set)) {
                bytesOver -= resource.getSizeOnDisk();
                LOG.info("Deleted blob: {} (OVER SIZE LIMIT).", resource.getKey());
                deleted.add(resource);
                i.remove();
            }
        }
        return deleted;
    }
@Test
    public void testCleanup() throws Exception {
        ClientBlobStore mockBlobstore = mock(ClientBlobStore.class);
        when(mockBlobstore.getBlobMeta(any())).thenReturn(new ReadableBlobMeta(new SettableBlobMeta(), 1));
        PortAndAssignment pna1 = new PortAndAssignmentImpl(1, new LocalAssignment("topo1", Collections.emptyList()));
        String user = "user";
        Map<String, Object> conf = new HashMap<>();
        IAdvancedFSOps ops = mock(IAdvancedFSOps.class);
        LocalizedResourceRetentionSet lrretset = spy(new LocalizedResourceRetentionSet(10));
        ConcurrentMap<String, LocalizedResource> lrFiles = new ConcurrentHashMap<>();
        ConcurrentMap<String, LocalizedResource> lrArchives = new ConcurrentHashMap<>();
        StormMetricsRegistry metricsRegistry = new StormMetricsRegistry();
        // no reference to key1
        LocalizedResource localresource1 = new LocalizedResource("key1", Paths.get("./target/TESTING/testfile1"), false, ops, conf,
                                                                 user, metricsRegistry);
        localresource1.setSize(10);
        // no reference to archive1
        LocalizedResource archiveresource1 = new LocalizedResource("archive1", Paths.get("./target/TESTING/testarchive1"), true, ops,
                                                                   conf, user, metricsRegistry);
        archiveresource1.setSize(20);
        // reference to key2
        LocalizedResource localresource2 = new LocalizedResource("key2", Paths.get("./target/TESTING/testfile2"), false, ops, conf,
                                                                 user, metricsRegistry);
        localresource2.addReference(pna1, null);
        // check adding reference to local resource with topology of same name
        localresource2.addReference(pna1, null);
        localresource2.setSize(10);
        lrFiles.put("key1", localresource1);
        lrFiles.put("key2", localresource2);
        lrArchives.put("archive1", archiveresource1);

        lrretset.addResources(lrFiles);
        lrretset.addResources(lrArchives);
        assertEquals(2, lrretset.getSizeWithNoReferences(), "number to clean is not 2");

        lrretset.cleanup(mockBlobstore);
        assertEquals(0, lrretset.getSizeWithNoReferences(), "resource not cleaned up");
    }