@Test public void persistenceFileWithBlocksOnMultipleWorkers() throws Exception { DefaultAsyncPersistHandler handler = new DefaultAsyncPersistHandler(new FileSystemMasterView(mFileSystemMaster)); AlluxioURI path = new AlluxioURI("/test"); List<FileBlockInfo> blockInfoList = new ArrayList<>(); BlockLocation location1 = new BlockLocation().setWorkerId(1); blockInfoList.add(new FileBlockInfo() .setBlockInfo(new BlockInfo().setLocations(Lists.newArrayList(location1)))); BlockLocation location2 = new BlockLocation().setWorkerId(2); blockInfoList.add(new FileBlockInfo() .setBlockInfo(new BlockInfo().setLocations(Lists.newArrayList(location2)))); long fileId = 2; when(mFileSystemMaster.getFileId(path)).thenReturn(fileId); when(mFileSystemMaster.getFileInfo(fileId)) .thenReturn(new FileInfo().setLength(1).setCompleted(true)); when(mFileSystemMaster.getFileBlockInfoList(path)).thenReturn(blockInfoList); assertEquals(0, handler.pollFilesToPersist(1).size()); assertEquals(0, handler.pollFilesToPersist(2).size()); }
@Override public synchronized List<PersistFile> pollFilesToPersist(long workerId) throws FileDoesNotExistException, InvalidPathException, AccessControlException { List<PersistFile> filesToPersist = new ArrayList<>(); List<Long> fileIdsToPersist = new ArrayList<>(); if (!mWorkerToAsyncPersistFiles.containsKey(workerId)) { return filesToPersist; } Set<Long> scheduledFiles = mWorkerToAsyncPersistFiles.get(workerId); try { for (long fileId : scheduledFiles) { try { FileInfo fileInfo = mFileSystemMasterView.getFileInfo(fileId); if (fileInfo.isCompleted()) { fileIdsToPersist.add(fileId); List<Long> blockIds = new ArrayList<>(); for (FileBlockInfo fileBlockInfo : mFileSystemMasterView .getFileBlockInfoList(mFileSystemMasterView.getPath(fileId))) { blockIds.add(fileBlockInfo.getBlockInfo().getBlockId()); } filesToPersist.add(new PersistFile(fileId, blockIds)); } } catch (FileDoesNotExistException e) { LOG.warn("FileId {} does not exist, ignore persistence it", fileId); } } } catch (UnavailableException e) { return filesToPersist; } mWorkerToAsyncPersistFiles.get(workerId).removeAll(fileIdsToPersist); return filesToPersist; }
DefaultAsyncPersistHandler implements AsyncPersistHandler { @Override public synchronized List<PersistFile> pollFilesToPersist(long workerId) throws FileDoesNotExistException, InvalidPathException, AccessControlException { List<PersistFile> filesToPersist = new ArrayList<>(); List<Long> fileIdsToPersist = new ArrayList<>(); if (!mWorkerToAsyncPersistFiles.containsKey(workerId)) { return filesToPersist; } Set<Long> scheduledFiles = mWorkerToAsyncPersistFiles.get(workerId); try { for (long fileId : scheduledFiles) { try { FileInfo fileInfo = mFileSystemMasterView.getFileInfo(fileId); if (fileInfo.isCompleted()) { fileIdsToPersist.add(fileId); List<Long> blockIds = new ArrayList<>(); for (FileBlockInfo fileBlockInfo : mFileSystemMasterView .getFileBlockInfoList(mFileSystemMasterView.getPath(fileId))) { blockIds.add(fileBlockInfo.getBlockInfo().getBlockId()); } filesToPersist.add(new PersistFile(fileId, blockIds)); } } catch (FileDoesNotExistException e) { LOG.warn("FileId {} does not exist, ignore persistence it", fileId); } } } catch (UnavailableException e) { return filesToPersist; } mWorkerToAsyncPersistFiles.get(workerId).removeAll(fileIdsToPersist); return filesToPersist; } }
DefaultAsyncPersistHandler implements AsyncPersistHandler { @Override public synchronized List<PersistFile> pollFilesToPersist(long workerId) throws FileDoesNotExistException, InvalidPathException, AccessControlException { List<PersistFile> filesToPersist = new ArrayList<>(); List<Long> fileIdsToPersist = new ArrayList<>(); if (!mWorkerToAsyncPersistFiles.containsKey(workerId)) { return filesToPersist; } Set<Long> scheduledFiles = mWorkerToAsyncPersistFiles.get(workerId); try { for (long fileId : scheduledFiles) { try { FileInfo fileInfo = mFileSystemMasterView.getFileInfo(fileId); if (fileInfo.isCompleted()) { fileIdsToPersist.add(fileId); List<Long> blockIds = new ArrayList<>(); for (FileBlockInfo fileBlockInfo : mFileSystemMasterView .getFileBlockInfoList(mFileSystemMasterView.getPath(fileId))) { blockIds.add(fileBlockInfo.getBlockInfo().getBlockId()); } filesToPersist.add(new PersistFile(fileId, blockIds)); } } catch (FileDoesNotExistException e) { LOG.warn("FileId {} does not exist, ignore persistence it", fileId); } } } catch (UnavailableException e) { return filesToPersist; } mWorkerToAsyncPersistFiles.get(workerId).removeAll(fileIdsToPersist); return filesToPersist; } DefaultAsyncPersistHandler(FileSystemMasterView view); }
DefaultAsyncPersistHandler implements AsyncPersistHandler { @Override public synchronized List<PersistFile> pollFilesToPersist(long workerId) throws FileDoesNotExistException, InvalidPathException, AccessControlException { List<PersistFile> filesToPersist = new ArrayList<>(); List<Long> fileIdsToPersist = new ArrayList<>(); if (!mWorkerToAsyncPersistFiles.containsKey(workerId)) { return filesToPersist; } Set<Long> scheduledFiles = mWorkerToAsyncPersistFiles.get(workerId); try { for (long fileId : scheduledFiles) { try { FileInfo fileInfo = mFileSystemMasterView.getFileInfo(fileId); if (fileInfo.isCompleted()) { fileIdsToPersist.add(fileId); List<Long> blockIds = new ArrayList<>(); for (FileBlockInfo fileBlockInfo : mFileSystemMasterView .getFileBlockInfoList(mFileSystemMasterView.getPath(fileId))) { blockIds.add(fileBlockInfo.getBlockInfo().getBlockId()); } filesToPersist.add(new PersistFile(fileId, blockIds)); } } catch (FileDoesNotExistException e) { LOG.warn("FileId {} does not exist, ignore persistence it", fileId); } } } catch (UnavailableException e) { return filesToPersist; } mWorkerToAsyncPersistFiles.get(workerId).removeAll(fileIdsToPersist); return filesToPersist; } DefaultAsyncPersistHandler(FileSystemMasterView view); @Override synchronized void scheduleAsyncPersistence(AlluxioURI path); @Override synchronized List<PersistFile> pollFilesToPersist(long workerId); }
DefaultAsyncPersistHandler implements AsyncPersistHandler { @Override public synchronized List<PersistFile> pollFilesToPersist(long workerId) throws FileDoesNotExistException, InvalidPathException, AccessControlException { List<PersistFile> filesToPersist = new ArrayList<>(); List<Long> fileIdsToPersist = new ArrayList<>(); if (!mWorkerToAsyncPersistFiles.containsKey(workerId)) { return filesToPersist; } Set<Long> scheduledFiles = mWorkerToAsyncPersistFiles.get(workerId); try { for (long fileId : scheduledFiles) { try { FileInfo fileInfo = mFileSystemMasterView.getFileInfo(fileId); if (fileInfo.isCompleted()) { fileIdsToPersist.add(fileId); List<Long> blockIds = new ArrayList<>(); for (FileBlockInfo fileBlockInfo : mFileSystemMasterView .getFileBlockInfoList(mFileSystemMasterView.getPath(fileId))) { blockIds.add(fileBlockInfo.getBlockInfo().getBlockId()); } filesToPersist.add(new PersistFile(fileId, blockIds)); } } catch (FileDoesNotExistException e) { LOG.warn("FileId {} does not exist, ignore persistence it", fileId); } } } catch (UnavailableException e) { return filesToPersist; } mWorkerToAsyncPersistFiles.get(workerId).removeAll(fileIdsToPersist); return filesToPersist; } DefaultAsyncPersistHandler(FileSystemMasterView view); @Override synchronized void scheduleAsyncPersistence(AlluxioURI path); @Override synchronized List<PersistFile> pollFilesToPersist(long workerId);  }