@Test public void testRequeuingFetchRequestOnFailure() throws IOException { 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<>(); bookKeeper = new CoordinatorBookKeeper(conf, new BookKeeperMetrics(conf, new MetricRegistry())); RemoteFetchProcessor remoteFetchProcessor = bookKeeper.getRemoteFetchProcessorInstance(); downloader = remoteFetchProcessor.getFileDownloaderInstance(); 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); assertTrue(requestChains.size() == 0, "Should not have created FileDownloadRequestChain as bookkeeper is not initialized"); assertTrue(remoteFetchProcessor.getProcessQueue().size() == context.getRanges().asRanges().size(), String.format("unexpected process queue size: %d, expected: %d", remoteFetchProcessor.getProcessQueue().size(), context.getRanges().asRanges().size())); for (Range<Long> range: context.getRanges().asRanges()) { FetchRequest actualFetchRequest = remoteFetchProcessor.getProcessQueue().poll(); FetchRequest expectedFetchRequest = new FetchRequest( backendPath.toString(), range.lowerEndpoint(), Math.toIntExact(range.upperEndpoint() - range.lowerEndpoint()), file.length(), 1000, 0 ); assertTrue(actualFetchRequest.equals(expectedFetchRequest), String.format("Actual: %s, Expected: %s", actualFetchRequest, expectedFetchRequest)); } }
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);   }