@Test public void testProcessDownloadRequests() throws Exception { DataGen.populateFile(TEST_BACKEND_FILE_NAME); final File file = new File(TEST_BACKEND_FILE_NAME); final Path backendPath = new Path("file: final ConcurrentMap<String, DownloadRequestContext> contextMap = new ConcurrentHashMap<>(); CacheStatusRequest request = new CacheStatusRequest(backendPath.toString(), file.length(), 1000, 0, 5) .setClusterType(TEST_CLUSTER_MANAGER.ordinal()); List<BlockLocation> cacheStatus = bookKeeper.getCacheStatus(request).getBlocks(); DownloadRequestContext context = new DownloadRequestContext(backendPath.toString(), file.length(), 1000); contextMap.put(backendPath.toString(), context); context.addDownloadRange(100, 300); context.addDownloadRange(250, 400); context.addDownloadRange(500, 800); final List<FileDownloadRequestChain> requestChains = downloader.getFileDownloadRequestChains(contextMap); int dataDownloaded = (int) downloader.processDownloadRequests(requestChains); int expectedDownloadedDataSize = 800; assertTrue(expectedDownloadedDataSize == dataDownloaded, "Download size didn't match"); assertTrue(metrics.getCounters().get(BookKeeperMetrics.CacheMetric.ASYNC_DOWNLOADED_MB_COUNT.getMetricName()) .getCount() == BYTES.toMB(expectedDownloadedDataSize), "Total downloaded bytes didn't match"); cacheStatus = bookKeeper.getCacheStatus(request).getBlocks(); int i = 0; for (i = 0; i < 4; i++) { assertTrue(cacheStatus.get(i).getLocation() == Location.CACHED, "Data is not cached"); } assertTrue(cacheStatus.get(i).getLocation() == Location.LOCAL, "Data is cached"); }
protected long processDownloadRequests(List<FileDownloadRequestChain> readRequestChainList) { if (readRequestChainList.size() == 0) { return 0; } long sizeRead = 0; List<Future<Long>> futures = new ArrayList<>(); for (FileDownloadRequestChain requestChain : readRequestChainList) { requestChain.lock(); Future<Long> result = processService.submit(requestChain); futures.add(result); } for (Future<Long> future : futures) { FileDownloadRequestChain requestChain = readRequestChainList.get(futures.indexOf(future)); long totalBytesToBeDownloaded = 0; for (ReadRequest request : requestChain.getReadRequests()) { totalBytesToBeDownloaded += request.getBackendReadLength(); } try { long read = future.get(); if (read == totalBytesToBeDownloaded) { requestChain.updateCacheStatus(requestChain.getRemotePath(), requestChain.getFileSize(), requestChain.getLastModified(), CacheConfig.getBlockSize(conf), conf); stats.addReadRequestChainStats(requestChain.getStats()); sizeRead += read; this.totalTimeToDownload.inc(requestChain.getTimeSpentOnDownload()); } else { log.error("ReadData didn't match with requested value. RequestedData: " + totalBytesToBeDownloaded + " ReadData: " + read); } } catch (ExecutionException | InterruptedException ex) { if (ex.getCause() instanceof FileNotFoundException) { log.debug("Could not process download request: " + ex.getCause().getMessage()); } else { log.error("Could not process download request", ex); } requestChain.cancel(); } } long dataDownloadedInMB = BYTES.toMB(sizeRead); this.totalMBDownloaded.inc(dataDownloadedInMB); return sizeRead; }
FileDownloader { protected long processDownloadRequests(List<FileDownloadRequestChain> readRequestChainList) { if (readRequestChainList.size() == 0) { return 0; } long sizeRead = 0; List<Future<Long>> futures = new ArrayList<>(); for (FileDownloadRequestChain requestChain : readRequestChainList) { requestChain.lock(); Future<Long> result = processService.submit(requestChain); futures.add(result); } for (Future<Long> future : futures) { FileDownloadRequestChain requestChain = readRequestChainList.get(futures.indexOf(future)); long totalBytesToBeDownloaded = 0; for (ReadRequest request : requestChain.getReadRequests()) { totalBytesToBeDownloaded += request.getBackendReadLength(); } try { long read = future.get(); if (read == totalBytesToBeDownloaded) { requestChain.updateCacheStatus(requestChain.getRemotePath(), requestChain.getFileSize(), requestChain.getLastModified(), CacheConfig.getBlockSize(conf), conf); stats.addReadRequestChainStats(requestChain.getStats()); sizeRead += read; this.totalTimeToDownload.inc(requestChain.getTimeSpentOnDownload()); } else { log.error("ReadData didn't match with requested value. RequestedData: " + totalBytesToBeDownloaded + " ReadData: " + read); } } catch (ExecutionException | InterruptedException ex) { if (ex.getCause() instanceof FileNotFoundException) { log.debug("Could not process download request: " + ex.getCause().getMessage()); } else { log.error("Could not process download request", ex); } requestChain.cancel(); } } long dataDownloadedInMB = BYTES.toMB(sizeRead); this.totalMBDownloaded.inc(dataDownloadedInMB); return sizeRead; } }
FileDownloader { protected long processDownloadRequests(List<FileDownloadRequestChain> readRequestChainList) { if (readRequestChainList.size() == 0) { return 0; } long sizeRead = 0; List<Future<Long>> futures = new ArrayList<>(); for (FileDownloadRequestChain requestChain : readRequestChainList) { requestChain.lock(); Future<Long> result = processService.submit(requestChain); futures.add(result); } for (Future<Long> future : futures) { FileDownloadRequestChain requestChain = readRequestChainList.get(futures.indexOf(future)); long totalBytesToBeDownloaded = 0; for (ReadRequest request : requestChain.getReadRequests()) { totalBytesToBeDownloaded += request.getBackendReadLength(); } try { long read = future.get(); if (read == totalBytesToBeDownloaded) { requestChain.updateCacheStatus(requestChain.getRemotePath(), requestChain.getFileSize(), requestChain.getLastModified(), CacheConfig.getBlockSize(conf), conf); stats.addReadRequestChainStats(requestChain.getStats()); sizeRead += read; this.totalTimeToDownload.inc(requestChain.getTimeSpentOnDownload()); } else { log.error("ReadData didn't match with requested value. RequestedData: " + totalBytesToBeDownloaded + " ReadData: " + read); } } catch (ExecutionException | InterruptedException ex) { if (ex.getCause() instanceof FileNotFoundException) { log.debug("Could not process download request: " + ex.getCause().getMessage()); } else { log.error("Could not process download request", ex); } requestChain.cancel(); } } long dataDownloadedInMB = BYTES.toMB(sizeRead); this.totalMBDownloaded.inc(dataDownloadedInMB); return sizeRead; } FileDownloader(BookKeeper bookKeeper, MetricRegistry metrics, Configuration conf, RemoteFetchProcessor remoteFetchProcessor, CachingFileSystemStatsProvider stats); }
FileDownloader { protected long processDownloadRequests(List<FileDownloadRequestChain> readRequestChainList) { if (readRequestChainList.size() == 0) { return 0; } long sizeRead = 0; List<Future<Long>> futures = new ArrayList<>(); for (FileDownloadRequestChain requestChain : readRequestChainList) { requestChain.lock(); Future<Long> result = processService.submit(requestChain); futures.add(result); } for (Future<Long> future : futures) { FileDownloadRequestChain requestChain = readRequestChainList.get(futures.indexOf(future)); long totalBytesToBeDownloaded = 0; for (ReadRequest request : requestChain.getReadRequests()) { totalBytesToBeDownloaded += request.getBackendReadLength(); } try { long read = future.get(); if (read == totalBytesToBeDownloaded) { requestChain.updateCacheStatus(requestChain.getRemotePath(), requestChain.getFileSize(), requestChain.getLastModified(), CacheConfig.getBlockSize(conf), conf); stats.addReadRequestChainStats(requestChain.getStats()); sizeRead += read; this.totalTimeToDownload.inc(requestChain.getTimeSpentOnDownload()); } else { log.error("ReadData didn't match with requested value. RequestedData: " + totalBytesToBeDownloaded + " ReadData: " + read); } } catch (ExecutionException | InterruptedException ex) { if (ex.getCause() instanceof FileNotFoundException) { log.debug("Could not process download request: " + ex.getCause().getMessage()); } else { log.error("Could not process download request", ex); } requestChain.cancel(); } } long dataDownloadedInMB = BYTES.toMB(sizeRead); this.totalMBDownloaded.inc(dataDownloadedInMB); return sizeRead; } FileDownloader(BookKeeper bookKeeper, MetricRegistry metrics, Configuration conf, RemoteFetchProcessor remoteFetchProcessor, CachingFileSystemStatsProvider stats);  }
FileDownloader { protected long processDownloadRequests(List<FileDownloadRequestChain> readRequestChainList) { if (readRequestChainList.size() == 0) { return 0; } long sizeRead = 0; List<Future<Long>> futures = new ArrayList<>(); for (FileDownloadRequestChain requestChain : readRequestChainList) { requestChain.lock(); Future<Long> result = processService.submit(requestChain); futures.add(result); } for (Future<Long> future : futures) { FileDownloadRequestChain requestChain = readRequestChainList.get(futures.indexOf(future)); long totalBytesToBeDownloaded = 0; for (ReadRequest request : requestChain.getReadRequests()) { totalBytesToBeDownloaded += request.getBackendReadLength(); } try { long read = future.get(); if (read == totalBytesToBeDownloaded) { requestChain.updateCacheStatus(requestChain.getRemotePath(), requestChain.getFileSize(), requestChain.getLastModified(), CacheConfig.getBlockSize(conf), conf); stats.addReadRequestChainStats(requestChain.getStats()); sizeRead += read; this.totalTimeToDownload.inc(requestChain.getTimeSpentOnDownload()); } else { log.error("ReadData didn't match with requested value. RequestedData: " + totalBytesToBeDownloaded + " ReadData: " + read); } } catch (ExecutionException | InterruptedException ex) { if (ex.getCause() instanceof FileNotFoundException) { log.debug("Could not process download request: " + ex.getCause().getMessage()); } else { log.error("Could not process download request", ex); } requestChain.cancel(); } } long dataDownloadedInMB = BYTES.toMB(sizeRead); this.totalMBDownloaded.inc(dataDownloadedInMB); return sizeRead; } FileDownloader(BookKeeper bookKeeper, MetricRegistry metrics, Configuration conf, RemoteFetchProcessor remoteFetchProcessor, CachingFileSystemStatsProvider stats);   }