@Test public void testGetFileDownloadRequestChains() throws Exception { final String remoteFilePath1 = "file: final String remoteFilePath2 = "file: final ConcurrentMap<String, DownloadRequestContext> contextMap = new ConcurrentHashMap<>(); DownloadRequestContext context = new DownloadRequestContext(remoteFilePath1, 1000, 1000); contextMap.put(remoteFilePath1, context); context.addDownloadRange(100, 300); context.addDownloadRange(250, 400); context.addDownloadRange(500, 800); context = new DownloadRequestContext(remoteFilePath2, 1000, 1000); contextMap.put(remoteFilePath2, context); context.addDownloadRange(100, 200); context.addDownloadRange(500, 800); final List<FileDownloadRequestChain> requestChains = downloader.getFileDownloadRequestChains(contextMap); assertTrue(requestChains.size() == 2, "Wrong Number of Request Chains. Expected = 2 Got = " + requestChains.size()); final FileDownloadRequestChain request1 = requestChains.get(0); assertTrue(request1.getReadRequests().size() == 1, "Wrong Number of Requests For File-1. Expected = 1 Got = " + request1.getReadRequests().size()); final FileDownloadRequestChain request2 = requestChains.get(1); assertTrue(request2.getReadRequests().size() == 2, "Wrong Number of Requests For File-2. Expected = 2 Got = " + request2.getReadRequests().size()); }
protected List<FileDownloadRequestChain> getFileDownloadRequestChains(ConcurrentMap<String, DownloadRequestContext> contextMap) throws IOException { List<FileDownloadRequestChain> readRequestChainList = new ArrayList<FileDownloadRequestChain>(); for (Map.Entry<String, DownloadRequestContext> entry : contextMap.entrySet()) { Path path = new Path(entry.getKey()); DownloadRequestContext context = entry.getValue(); int generationNumber = UNKONWN_GENERATION_NUMBER; FileSystem fs = FileSystem.get(path.toUri(), conf); fs.initialize(path.toUri(), conf); ByteBuffer directWriteBuffer = bufferPool.getBuffer(diskReadBufferSize); FileDownloadRequestChain requestChain = null; String remotePath = entry.getKey(); Range<Long> previousRange = null; for (Range<Long> range : context.getRanges().asRanges()) { long startBlock = toStartBlock(range.lowerEndpoint(), conf); long endBlock = toEndBlock(range.upperEndpoint(), conf); Range<Long> currentRange = Range.closedOpen(startBlock, endBlock); if (previousRange != null && previousRange.encloses(currentRange)) { continue; } previousRange = currentRange; List<BlockLocation> blockLocations = null; try { CacheStatusResponse response = bookKeeper.getCacheStatus( new CacheStatusRequest( context.getRemoteFilePath(), context.getFileSize(), context.getLastModifiedTime(), startBlock, endBlock)); blockLocations = response.getBlocks(); if(generationNumber != UNKONWN_GENERATION_NUMBER && response.getGenerationNumber() != generationNumber) { log.debug(String.format("Mismatch in generation-number in download requests for file %s, expected=%d but found=%d, skipping the file", remotePath, generationNumber, response.getGenerationNumber())); requestChain = null; break; } generationNumber = response.getGenerationNumber(); if (requestChain == null) { String localPath = CacheUtil.getLocalPath(remotePath, conf, generationNumber); log.debug("Processing Request for File : " + path.toString() + " LocalFile : " + localPath); requestChain = new FileDownloadRequestChain(bookKeeper, fs, localPath, directWriteBuffer, conf, context.getRemoteFilePath(), context.getFileSize(), context.getLastModifiedTime(), generationNumber); } } catch (Exception e) { log.warn("Error communicating with bookKeeper", e); remoteFetchProcessor.addToProcessQueueSafe(context.getRemoteFilePath(), context.getRanges().asRanges(), context.getFileSize(), context.getLastModifiedTime()); requestChain = null; break; } for (int i = 0; i < blockLocations.size(); i++) { if (!blockLocations.get(i).getLocation().equals(Location.LOCAL)) { continue; } long block = startBlock + i; long startPosition = toBlockStartPosition(block, conf); long endPosition = Math.min(toBlockStartPosition(block + 1, conf), context.getFileSize()); ReadRequest readRequest = new ReadRequest(startPosition, endPosition, startPosition, endPosition, null, 0, context.getFileSize()); requestChain.addReadRequest(readRequest); } } if (requestChain != null) { log.debug("Request added for file: " + requestChain.getRemotePath() + " Number of Requests : " + requestChain.getReadRequests().size()); readRequestChainList.add(requestChain); } } return readRequestChainList; }
FileDownloader { protected List<FileDownloadRequestChain> getFileDownloadRequestChains(ConcurrentMap<String, DownloadRequestContext> contextMap) throws IOException { List<FileDownloadRequestChain> readRequestChainList = new ArrayList<FileDownloadRequestChain>(); for (Map.Entry<String, DownloadRequestContext> entry : contextMap.entrySet()) { Path path = new Path(entry.getKey()); DownloadRequestContext context = entry.getValue(); int generationNumber = UNKONWN_GENERATION_NUMBER; FileSystem fs = FileSystem.get(path.toUri(), conf); fs.initialize(path.toUri(), conf); ByteBuffer directWriteBuffer = bufferPool.getBuffer(diskReadBufferSize); FileDownloadRequestChain requestChain = null; String remotePath = entry.getKey(); Range<Long> previousRange = null; for (Range<Long> range : context.getRanges().asRanges()) { long startBlock = toStartBlock(range.lowerEndpoint(), conf); long endBlock = toEndBlock(range.upperEndpoint(), conf); Range<Long> currentRange = Range.closedOpen(startBlock, endBlock); if (previousRange != null && previousRange.encloses(currentRange)) { continue; } previousRange = currentRange; List<BlockLocation> blockLocations = null; try { CacheStatusResponse response = bookKeeper.getCacheStatus( new CacheStatusRequest( context.getRemoteFilePath(), context.getFileSize(), context.getLastModifiedTime(), startBlock, endBlock)); blockLocations = response.getBlocks(); if(generationNumber != UNKONWN_GENERATION_NUMBER && response.getGenerationNumber() != generationNumber) { log.debug(String.format("Mismatch in generation-number in download requests for file %s, expected=%d but found=%d, skipping the file", remotePath, generationNumber, response.getGenerationNumber())); requestChain = null; break; } generationNumber = response.getGenerationNumber(); if (requestChain == null) { String localPath = CacheUtil.getLocalPath(remotePath, conf, generationNumber); log.debug("Processing Request for File : " + path.toString() + " LocalFile : " + localPath); requestChain = new FileDownloadRequestChain(bookKeeper, fs, localPath, directWriteBuffer, conf, context.getRemoteFilePath(), context.getFileSize(), context.getLastModifiedTime(), generationNumber); } } catch (Exception e) { log.warn("Error communicating with bookKeeper", e); remoteFetchProcessor.addToProcessQueueSafe(context.getRemoteFilePath(), context.getRanges().asRanges(), context.getFileSize(), context.getLastModifiedTime()); requestChain = null; break; } for (int i = 0; i < blockLocations.size(); i++) { if (!blockLocations.get(i).getLocation().equals(Location.LOCAL)) { continue; } long block = startBlock + i; long startPosition = toBlockStartPosition(block, conf); long endPosition = Math.min(toBlockStartPosition(block + 1, conf), context.getFileSize()); ReadRequest readRequest = new ReadRequest(startPosition, endPosition, startPosition, endPosition, null, 0, context.getFileSize()); requestChain.addReadRequest(readRequest); } } if (requestChain != null) { log.debug("Request added for file: " + requestChain.getRemotePath() + " Number of Requests : " + requestChain.getReadRequests().size()); readRequestChainList.add(requestChain); } } return readRequestChainList; } }
FileDownloader { protected List<FileDownloadRequestChain> getFileDownloadRequestChains(ConcurrentMap<String, DownloadRequestContext> contextMap) throws IOException { List<FileDownloadRequestChain> readRequestChainList = new ArrayList<FileDownloadRequestChain>(); for (Map.Entry<String, DownloadRequestContext> entry : contextMap.entrySet()) { Path path = new Path(entry.getKey()); DownloadRequestContext context = entry.getValue(); int generationNumber = UNKONWN_GENERATION_NUMBER; FileSystem fs = FileSystem.get(path.toUri(), conf); fs.initialize(path.toUri(), conf); ByteBuffer directWriteBuffer = bufferPool.getBuffer(diskReadBufferSize); FileDownloadRequestChain requestChain = null; String remotePath = entry.getKey(); Range<Long> previousRange = null; for (Range<Long> range : context.getRanges().asRanges()) { long startBlock = toStartBlock(range.lowerEndpoint(), conf); long endBlock = toEndBlock(range.upperEndpoint(), conf); Range<Long> currentRange = Range.closedOpen(startBlock, endBlock); if (previousRange != null && previousRange.encloses(currentRange)) { continue; } previousRange = currentRange; List<BlockLocation> blockLocations = null; try { CacheStatusResponse response = bookKeeper.getCacheStatus( new CacheStatusRequest( context.getRemoteFilePath(), context.getFileSize(), context.getLastModifiedTime(), startBlock, endBlock)); blockLocations = response.getBlocks(); if(generationNumber != UNKONWN_GENERATION_NUMBER && response.getGenerationNumber() != generationNumber) { log.debug(String.format("Mismatch in generation-number in download requests for file %s, expected=%d but found=%d, skipping the file", remotePath, generationNumber, response.getGenerationNumber())); requestChain = null; break; } generationNumber = response.getGenerationNumber(); if (requestChain == null) { String localPath = CacheUtil.getLocalPath(remotePath, conf, generationNumber); log.debug("Processing Request for File : " + path.toString() + " LocalFile : " + localPath); requestChain = new FileDownloadRequestChain(bookKeeper, fs, localPath, directWriteBuffer, conf, context.getRemoteFilePath(), context.getFileSize(), context.getLastModifiedTime(), generationNumber); } } catch (Exception e) { log.warn("Error communicating with bookKeeper", e); remoteFetchProcessor.addToProcessQueueSafe(context.getRemoteFilePath(), context.getRanges().asRanges(), context.getFileSize(), context.getLastModifiedTime()); requestChain = null; break; } for (int i = 0; i < blockLocations.size(); i++) { if (!blockLocations.get(i).getLocation().equals(Location.LOCAL)) { continue; } long block = startBlock + i; long startPosition = toBlockStartPosition(block, conf); long endPosition = Math.min(toBlockStartPosition(block + 1, conf), context.getFileSize()); ReadRequest readRequest = new ReadRequest(startPosition, endPosition, startPosition, endPosition, null, 0, context.getFileSize()); requestChain.addReadRequest(readRequest); } } if (requestChain != null) { log.debug("Request added for file: " + requestChain.getRemotePath() + " Number of Requests : " + requestChain.getReadRequests().size()); readRequestChainList.add(requestChain); } } return readRequestChainList; } FileDownloader(BookKeeper bookKeeper, MetricRegistry metrics, Configuration conf, RemoteFetchProcessor remoteFetchProcessor, CachingFileSystemStatsProvider stats); }
FileDownloader { protected List<FileDownloadRequestChain> getFileDownloadRequestChains(ConcurrentMap<String, DownloadRequestContext> contextMap) throws IOException { List<FileDownloadRequestChain> readRequestChainList = new ArrayList<FileDownloadRequestChain>(); for (Map.Entry<String, DownloadRequestContext> entry : contextMap.entrySet()) { Path path = new Path(entry.getKey()); DownloadRequestContext context = entry.getValue(); int generationNumber = UNKONWN_GENERATION_NUMBER; FileSystem fs = FileSystem.get(path.toUri(), conf); fs.initialize(path.toUri(), conf); ByteBuffer directWriteBuffer = bufferPool.getBuffer(diskReadBufferSize); FileDownloadRequestChain requestChain = null; String remotePath = entry.getKey(); Range<Long> previousRange = null; for (Range<Long> range : context.getRanges().asRanges()) { long startBlock = toStartBlock(range.lowerEndpoint(), conf); long endBlock = toEndBlock(range.upperEndpoint(), conf); Range<Long> currentRange = Range.closedOpen(startBlock, endBlock); if (previousRange != null && previousRange.encloses(currentRange)) { continue; } previousRange = currentRange; List<BlockLocation> blockLocations = null; try { CacheStatusResponse response = bookKeeper.getCacheStatus( new CacheStatusRequest( context.getRemoteFilePath(), context.getFileSize(), context.getLastModifiedTime(), startBlock, endBlock)); blockLocations = response.getBlocks(); if(generationNumber != UNKONWN_GENERATION_NUMBER && response.getGenerationNumber() != generationNumber) { log.debug(String.format("Mismatch in generation-number in download requests for file %s, expected=%d but found=%d, skipping the file", remotePath, generationNumber, response.getGenerationNumber())); requestChain = null; break; } generationNumber = response.getGenerationNumber(); if (requestChain == null) { String localPath = CacheUtil.getLocalPath(remotePath, conf, generationNumber); log.debug("Processing Request for File : " + path.toString() + " LocalFile : " + localPath); requestChain = new FileDownloadRequestChain(bookKeeper, fs, localPath, directWriteBuffer, conf, context.getRemoteFilePath(), context.getFileSize(), context.getLastModifiedTime(), generationNumber); } } catch (Exception e) { log.warn("Error communicating with bookKeeper", e); remoteFetchProcessor.addToProcessQueueSafe(context.getRemoteFilePath(), context.getRanges().asRanges(), context.getFileSize(), context.getLastModifiedTime()); requestChain = null; break; } for (int i = 0; i < blockLocations.size(); i++) { if (!blockLocations.get(i).getLocation().equals(Location.LOCAL)) { continue; } long block = startBlock + i; long startPosition = toBlockStartPosition(block, conf); long endPosition = Math.min(toBlockStartPosition(block + 1, conf), context.getFileSize()); ReadRequest readRequest = new ReadRequest(startPosition, endPosition, startPosition, endPosition, null, 0, context.getFileSize()); requestChain.addReadRequest(readRequest); } } if (requestChain != null) { log.debug("Request added for file: " + requestChain.getRemotePath() + " Number of Requests : " + requestChain.getReadRequests().size()); readRequestChainList.add(requestChain); } } return readRequestChainList; } FileDownloader(BookKeeper bookKeeper, MetricRegistry metrics, Configuration conf, RemoteFetchProcessor remoteFetchProcessor, CachingFileSystemStatsProvider stats);  }
FileDownloader { protected List<FileDownloadRequestChain> getFileDownloadRequestChains(ConcurrentMap<String, DownloadRequestContext> contextMap) throws IOException { List<FileDownloadRequestChain> readRequestChainList = new ArrayList<FileDownloadRequestChain>(); for (Map.Entry<String, DownloadRequestContext> entry : contextMap.entrySet()) { Path path = new Path(entry.getKey()); DownloadRequestContext context = entry.getValue(); int generationNumber = UNKONWN_GENERATION_NUMBER; FileSystem fs = FileSystem.get(path.toUri(), conf); fs.initialize(path.toUri(), conf); ByteBuffer directWriteBuffer = bufferPool.getBuffer(diskReadBufferSize); FileDownloadRequestChain requestChain = null; String remotePath = entry.getKey(); Range<Long> previousRange = null; for (Range<Long> range : context.getRanges().asRanges()) { long startBlock = toStartBlock(range.lowerEndpoint(), conf); long endBlock = toEndBlock(range.upperEndpoint(), conf); Range<Long> currentRange = Range.closedOpen(startBlock, endBlock); if (previousRange != null && previousRange.encloses(currentRange)) { continue; } previousRange = currentRange; List<BlockLocation> blockLocations = null; try { CacheStatusResponse response = bookKeeper.getCacheStatus( new CacheStatusRequest( context.getRemoteFilePath(), context.getFileSize(), context.getLastModifiedTime(), startBlock, endBlock)); blockLocations = response.getBlocks(); if(generationNumber != UNKONWN_GENERATION_NUMBER && response.getGenerationNumber() != generationNumber) { log.debug(String.format("Mismatch in generation-number in download requests for file %s, expected=%d but found=%d, skipping the file", remotePath, generationNumber, response.getGenerationNumber())); requestChain = null; break; } generationNumber = response.getGenerationNumber(); if (requestChain == null) { String localPath = CacheUtil.getLocalPath(remotePath, conf, generationNumber); log.debug("Processing Request for File : " + path.toString() + " LocalFile : " + localPath); requestChain = new FileDownloadRequestChain(bookKeeper, fs, localPath, directWriteBuffer, conf, context.getRemoteFilePath(), context.getFileSize(), context.getLastModifiedTime(), generationNumber); } } catch (Exception e) { log.warn("Error communicating with bookKeeper", e); remoteFetchProcessor.addToProcessQueueSafe(context.getRemoteFilePath(), context.getRanges().asRanges(), context.getFileSize(), context.getLastModifiedTime()); requestChain = null; break; } for (int i = 0; i < blockLocations.size(); i++) { if (!blockLocations.get(i).getLocation().equals(Location.LOCAL)) { continue; } long block = startBlock + i; long startPosition = toBlockStartPosition(block, conf); long endPosition = Math.min(toBlockStartPosition(block + 1, conf), context.getFileSize()); ReadRequest readRequest = new ReadRequest(startPosition, endPosition, startPosition, endPosition, null, 0, context.getFileSize()); requestChain.addReadRequest(readRequest); } } if (requestChain != null) { log.debug("Request added for file: " + requestChain.getRemotePath() + " Number of Requests : " + requestChain.getReadRequests().size()); readRequestChainList.add(requestChain); } } return readRequestChainList; } FileDownloader(BookKeeper bookKeeper, MetricRegistry metrics, Configuration conf, RemoteFetchProcessor remoteFetchProcessor, CachingFileSystemStatsProvider stats);   }