public CompletableFuture<Void> requestDownloadTopologyBlobs(final LocalAssignment assignment, final int port,
                                                                final BlobChangingCallback cb) throws IOException {
        final PortAndAssignment pna = new TimePortAndAssignment(new PortAndAssignmentImpl(port, assignment), blobLocalizationDuration);
        final String topologyId = pna.getToplogyId();

        LOG.info("requestDownloadTopologyBlobs for {}", pna);

        CompletableFuture<Void> baseBlobs = requestDownloadBaseTopologyBlobs(pna, cb);
        return baseBlobs.thenComposeAsync((v) ->
            blobPending.compute(topologyId, (tid, old) -> {
                CompletableFuture<Void> ret = old;
                if (ret == null) {
                    ret = CompletableFuture.supplyAsync(new DownloadBlobs(pna, cb), taskExecService);
                } else {
                    try {
                        addReferencesToBlobs(pna, cb);
                    } catch (Exception e) {
                        LOG.error("Failed adding references to blobs for " + pna, e);
                        throw new RuntimeException(e);
                    } finally {
                        pna.complete();
                    }
                }
                LOG.debug("Reserved blobs {} {}", topologyId, ret);
                return ret;
            }));
    }
@Test
    public void testRequestDownloadTopologyBlobs() throws Exception {
        ConfigUtils mockedConfigUtils = mock(ConfigUtils.class);
        ConfigUtils previousConfigUtils = ConfigUtils.setInstance(mockedConfigUtils);

        AsyncLocalizer victim = null;

        try (TmpPath stormLocal = new TmpPath(); TmpPath localizerRoot = new TmpPath()) {

            Map<String, Object> conf = new HashMap<>();
            conf.put(Config.STORM_LOCAL_DIR, stormLocal.getPath());

            AdvancedFSOps ops = AdvancedFSOps.make(conf);
            StormMetricsRegistry metricsRegistry = new StormMetricsRegistry();

            victim = spy(new AsyncLocalizer(conf, ops, localizerRoot.getPath(), metricsRegistry));

            final String topoId = "TOPO-12345";
            final String user = "user";

            final Path userDir = Paths.get(stormLocal.getPath(), user);
            final Path topologyDirRoot = Paths.get(stormLocal.getPath(), topoId);

            final String simpleLocalName = "simple.txt";
            final String simpleKey = "simple";
            Map<String, Map<String, Object>> topoBlobMap = new HashMap<>();
            Map<String, Object> simple = new HashMap<>();
            simple.put("localname", simpleLocalName);
            simple.put("uncompress", false);
            topoBlobMap.put(simpleKey, simple);

            final int port = 8080;

            Map<String, Object> topoConf = new HashMap<>(conf);
            topoConf.put(Config.TOPOLOGY_BLOBSTORE_MAP, topoBlobMap);
            topoConf.put(Config.TOPOLOGY_NAME, "TOPO");

            List<LocalizedResource> localizedList = new ArrayList<>();
            LocalizedResource simpleLocal = new LocalizedResource(simpleKey, localizerRoot.getFile().toPath(), false, ops, conf, user, metricsRegistry);
            localizedList.add(simpleLocal);

            when(mockedConfigUtils.supervisorStormDistRootImpl(conf, topoId)).thenReturn(topologyDirRoot.toString());
            when(mockedConfigUtils.readSupervisorStormConfImpl(conf, topoId)).thenReturn(topoConf);
            when(mockedConfigUtils.readSupervisorTopologyImpl(conf, topoId, ops)).thenReturn(constructEmptyStormTopology());

            //Write the mocking backwards so the actual method is not called on the spy object
            doReturn(CompletableFuture.supplyAsync(() -> null)).when(victim)
                    .requestDownloadBaseTopologyBlobs(any(), eq(null));

            Files.createDirectories(topologyDirRoot);

            doReturn(userDir.toFile()).when(victim).getLocalUserFileCacheDir(user);
            doReturn(localizedList).when(victim).getBlobs(any(List.class), any(), any());

            Future<Void> f = victim.requestDownloadTopologyBlobs(constructLocalAssignment(topoId, user), port, null);
            f.get(20, TimeUnit.SECONDS);
            // We should be done now...

            verify(victim).getLocalUserFileCacheDir(user);

            assertTrue(ops.fileExists(userDir));

            verify(victim).getBlobs(any(List.class), any(), any());

            // symlink was created
            assertTrue(Files.isSymbolicLink(topologyDirRoot.resolve(simpleLocalName)));

        } finally {
            ConfigUtils.setInstance(previousConfigUtils);
            if (victim != null) {
                victim.close();
            }
        }
    }