Localizer { public void handleLocalizations(Service service) throws IOException { Path stagingDir = remoteDirectoryManager.getJobStagingArea( parameters.getName(), true); List<Localization> localizations = parameters.getLocalizations(); checkFilesExist(localizations); for (Localization localization : localizations) { LocalizationState localizationState = new LocalizationState(localization, remoteDirectoryManager); Path resourceToLocalize = new Path(localizationState.remoteUri); String sourceFile = determineSourceFile(localizationState); if (localizationState.needUploadToHDFS) { resourceToLocalize = fsOperations.uploadToRemoteFile(stagingDir, sourceFile); } if (localizationState.needToDeleteTempFile) { fsOperations.deleteFiles(sourceFile); } if (isZippedArchive(sourceFile, localizationState.destFileType)) { fsOperations.deleteFiles(sourceFile); sourceFile = getNameUntilUnderscore(sourceFile); } String containerLocalPath = localizationState.containerLocalPath; if (!containerLocalPath.equals(".") && !containerLocalPath.equals("./")) { sourceFile = getLastNameFromPath(containerLocalPath); } String localizedName = getLastNameFromPath(sourceFile); LOG.info("The file or directory to be localized is {}. " + "Its localized filename will be {}", resourceToLocalize.toString(), localizedName); ConfigFile configFile = new ConfigFile() .srcFile(resourceToLocalize.toUri().toString()) .destFile(localizedName) .type(localizationState.destFileType); service.getConfiguration().getFiles().add(configFile); if (containerLocalPath.startsWith("/")) { addToMounts(service, localization, containerLocalPath, sourceFile); } } } Localizer(FileSystemOperations fsOperations,
      RemoteDirectoryManager remoteDirectoryManager,
      RunJobParameters parameters); void handleLocalizations(Service service);  }
@Test(expected = FileNotFoundException.class) public void testLocalizeNotExistingRemoteFile() throws IOException { when(remoteDirectoryManager.isRemote(anyString())).thenReturn(true); when(remoteDirectoryManager.existsRemoteFile(anyString())) .thenReturn(false); Localization localization = new Localization(); localization.setRemoteUri("hdfs: Localizer localizer = createLocalizerWithLocalizations(localization); localizer.handleLocalizations(service); }
@Test(expected = FileNotFoundException.class) public void testLocalizeNotExistingLocalFile() throws IOException { when(remoteDirectoryManager.isRemote(anyString())).thenReturn(false); Localization localization = new Localization(); localization.setRemoteUri("file: Localizer localizer = createLocalizerWithLocalizations(localization); localizer.handleLocalizations(service); }
@Test(expected = IllegalStateException.class) public void testLocalizeExistingRemoteDirectoryNoUnderscoreInName() throws IOException { when(remoteDirectoryManager.isDir(anyString())).thenReturn(true); when(remoteDirectoryManager.isRemote(anyString())).thenReturn(true); when(remoteDirectoryManager.existsRemoteFile(anyString())) .thenReturn(true); String remoteUri = "hdfs: when(fsOperations.uploadToRemoteFile(any(Path.class), anyString())) .thenReturn(new Path(remoteUri)); when(fsOperations.downloadAndZip(anyString(), anyString())) .thenReturn("remotedir2.zip"); Localization localization = new Localization(); localization.setLocalPath("remotedir2"); localization.setRemoteUri(remoteUri); Localizer localizer = createLocalizerWithLocalizations(localization); localizer.handleLocalizations(service); verify(fsOperations).validFileSize(anyString()); }
@Test public void testLocalizeExistingRemoteDirectory() throws IOException { when(remoteDirectoryManager.isDir(anyString())).thenReturn(true); when(remoteDirectoryManager.isRemote(anyString())).thenReturn(true); when(remoteDirectoryManager.existsRemoteFile(anyString())) .thenReturn(true); String remoteUri = "hdfs: when(fsOperations.uploadToRemoteFile(any(Path.class), anyString())) .thenReturn(new Path(remoteUri)); String zipFileName = "remotedir2_221424.zip"; when(fsOperations.downloadAndZip(anyString(), anyString())) .thenReturn(zipFileName); Localization localization = new Localization(); localization.setLocalPath("/remotedir2"); localization.setRemoteUri(remoteUri); Localizer localizer = createLocalizerWithLocalizations(localization); localizer.handleLocalizations(service); verify(fsOperations).validFileSize(anyString()); verify(fsOperations).deleteFiles(eq(zipFileName)); verify(fsOperations, never()).uploadToRemoteFile(any(Path.class), eq(remoteUri)); List<ConfigFile> files = service.getConfiguration().getFiles(); assertEquals(1, files.size()); ConfigFile configFile = files.get(0); assertEquals(TypeEnum.ARCHIVE, configFile.getType()); assertEquals(remoteUri, configFile.getSrcFile()); assertEquals("remotedir2", configFile.getDestFile()); assertEquals(1, service.getConfiguration().getEnv().size()); String dockerMounts = service.getConfiguration().getEnv() .get(EnvironmentUtilities.ENV_DOCKER_MOUNTS_FOR_CONTAINER_RUNTIME); assertEquals("remotedir2:/remotedir2:rw", dockerMounts); }
@Test public void testLocalizeNonHdfsRemoteUri() throws IOException { when(remoteDirectoryManager.isRemote(anyString())).thenReturn(true); when(remoteDirectoryManager.existsRemoteFile(anyString())) .thenReturn(true); String remoteUri = "remote: when(fsOperations.uploadToRemoteFile(any(Path.class), anyString())) .thenReturn(new Path(remoteUri)); String downloadedFileName = "remotedir2_221424"; when(fsOperations.download(anyString(), anyString())) .thenReturn(downloadedFileName); Localization localization = new Localization(); localization.setLocalPath("/remotedir2"); localization.setRemoteUri(remoteUri); Localizer localizer = createLocalizerWithLocalizations(localization); localizer.handleLocalizations(service); verify(fsOperations).validFileSize(anyString()); verify(fsOperations).deleteFiles(eq(downloadedFileName)); verify(fsOperations).uploadToRemoteFile(any(Path.class), eq(downloadedFileName)); List<ConfigFile> files = service.getConfiguration().getFiles(); assertEquals(1, files.size()); ConfigFile configFile = files.get(0); assertEquals(TypeEnum.STATIC, configFile.getType()); assertEquals(remoteUri, configFile.getSrcFile()); assertEquals("remotedir2", configFile.getDestFile()); assertEquals(1, service.getConfiguration().getEnv().size()); String dockerMounts = service.getConfiguration().getEnv() .get(EnvironmentUtilities.ENV_DOCKER_MOUNTS_FOR_CONTAINER_RUNTIME); assertEquals("remotedir2:/remotedir2:rw", dockerMounts); }