flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-yarn/src/test/java/org/apache/flink/yarn/YarnFileStageTest.java
/**
	 * Verifies that nested directories are properly copied with the given filesystem and paths.
	 *
	 * @param targetFileSystem
	 * 		file system of the target path
	 * @param targetDir
	 * 		target path (URI like <tt>hdfs://...</tt>)
	 * @param tempFolder
	 * 		JUnit temporary folder rule to create the source directory with
	 * @param addSchemeToLocalPath
	 * 		whether add the <tt>file://</tt> scheme to the local path to copy from
	 */
Verifies that nested directories are properly copied with the given filesystem and paths.
@param targetFileSystem
file system of the target path
@param targetDir
target path (URI like hdfs://...)
@param tempFolder
JUnit temporary folder rule to create the source directory with
@param addSchemeToLocalPath
whether add the file:// scheme to the local path to copy from
Verifies that nested directories are properly copied with the given filesystem and paths . @param targetFileSystem file system of the target path @param targetDir target path ( URI like hdfs : // ... ) @param tempFolder JUnit temporary folder rule to create the source directory with @param addSchemeToLocalPath whether add the file : // scheme to the local path to copy from
static void testCopyFromLocalRecursive(
			FileSystem targetFileSystem,
			Path targetDir,
			TemporaryFolder tempFolder,
			boolean addSchemeToLocalPath) throws Exception {
		assertFalse(targetFileSystem.exists(targetDir));
		final File srcDir = tempFolder.newFolder();
		final Path srcPath;
		if (addSchemeToLocalPath) {
			srcPath = new Path("file://" + srcDir.getAbsolutePath());
		} else {
			srcPath = new Path(srcDir.getAbsolutePath());
		}
		HashMap<String /* (relative) path */, /* contents */ String> srcFiles = new HashMap<>(4);
		srcFiles.put("1", "Hello 1");
		srcFiles.put("2", "Hello 2");
		srcFiles.put("nested/3", "Hello nested/3");
		srcFiles.put("nested/4/5", "Hello nested/4/5");
		for (Map.Entry<String, String> src : srcFiles.entrySet()) {
			File file = new File(srcDir, src.getKey());
			file.getParentFile().mkdirs();
			try (DataOutputStream out = new DataOutputStream(new FileOutputStream(file))) {
				out.writeUTF(src.getValue());
			}
		}
		try {
			List<Path> remotePaths = new ArrayList<>();
			HashMap<String, LocalResource> localResources = new HashMap<>();
			AbstractYarnClusterDescriptor.uploadAndRegisterFiles(
				Collections.singletonList(new File(srcPath.toUri().getPath())),
				targetFileSystem,
				targetDir,
				ApplicationId.newInstance(0, 0),
				remotePaths,
				localResources,
				new StringBuilder());
			assertEquals(srcFiles.size(), localResources.size());
			Path workDir = ConverterUtils
				.getPathFromYarnURL(localResources.get(srcPath.getName() + "/1").getResource())
				.getParent();
			RemoteIterator<LocatedFileStatus> targetFilesIterator =
				targetFileSystem.listFiles(workDir, true);
			HashMap<String /* (relative) path */, /* contents */ String> targetFiles =
				new HashMap<>(4);
			final int workDirPrefixLength =
				workDir.toString().length() + 1; 
			while (targetFilesIterator.hasNext()) {
				LocatedFileStatus targetFile = targetFilesIterator.next();
				try (FSDataInputStream in = targetFileSystem.open(targetFile.getPath())) {
					String absolutePathString = targetFile.getPath().toString();
					String relativePath = absolutePathString.substring(workDirPrefixLength);
					targetFiles.put(relativePath, in.readUTF());
					assertEquals("extraneous data in file " + relativePath, -1, in.read());
				}
			}
			assertThat(targetFiles, equalTo(srcFiles));
		} finally {
			targetFileSystem.delete(targetDir, true);
		}
	}
static void testCopyFromLocalRecursive ( FileSystem targetFileSystem , Path targetDir , TemporaryFolder tempFolder , boolean addSchemeToLocalPath ) throws Exception { assertFalse ( targetFileSystem . exists ( targetDir ) ) ; final File srcDir = tempFolder . newFolder ( ) ; final Path srcPath ; if ( addSchemeToLocalPath ) { srcPath = new Path ( "file://" + srcDir . getAbsolutePath ( ) ) ; } else { srcPath = new Path ( srcDir . getAbsolutePath ( ) ) ; } HashMap < String /* (relative) path */ , /* contents */ String > srcFiles = new HashMap < > ( 4 ) ; srcFiles . put ( "1" , "Hello 1" ) ; srcFiles . put ( "2" , "Hello 2" ) ; srcFiles . put ( "nested/3" , "Hello nested/3" ) ; srcFiles . put ( "nested/4/5" , "Hello nested/4/5" ) ; for ( Map . Entry < String , String > src : srcFiles . entrySet ( ) ) { File file = new File ( srcDir , src . getKey ( ) ) ; file . getParentFile ( ) . mkdirs ( ) ; try ( DataOutputStream out = new DataOutputStream ( new FileOutputStream ( file ) ) ) { out . writeUTF ( src . getValue ( ) ) ; } } try { List < Path > remotePaths = new ArrayList < > ( ) ; HashMap < String , LocalResource > localResources = new HashMap < > ( ) ; AbstractYarnClusterDescriptor . uploadAndRegisterFiles ( Collections . singletonList ( new File ( srcPath . toUri ( ) . getPath ( ) ) ) , targetFileSystem , targetDir , ApplicationId . newInstance ( 0 , 0 ) , remotePaths , localResources , new StringBuilder ( ) ) ; assertEquals ( srcFiles . size ( ) , localResources . size ( ) ) ; Path workDir = ConverterUtils . getPathFromYarnURL ( localResources . get ( srcPath . getName ( ) + "/1" ) . getResource ( ) ) . getParent ( ) ; RemoteIterator < LocatedFileStatus > targetFilesIterator = targetFileSystem . listFiles ( workDir , true ) ; HashMap < String /* (relative) path */ , /* contents */ String > targetFiles = new HashMap < > ( 4 ) ; final int workDirPrefixLength = workDir . toString ( ) . length ( ) + 1 ; while ( targetFilesIterator . hasNext ( ) ) { LocatedFileStatus targetFile = targetFilesIterator . next ( ) ; try ( FSDataInputStream in = targetFileSystem . open ( targetFile . getPath ( ) ) ) { String absolutePathString = targetFile . getPath ( ) . toString ( ) ; String relativePath = absolutePathString . substring ( workDirPrefixLength ) ; targetFiles . put ( relativePath , in . readUTF ( ) ) ; assertEquals ( "extraneous data in file " + relativePath , - 1 , in . read ( ) ) ; } } assertThat ( targetFiles , equalTo ( srcFiles ) ) ; } finally { targetFileSystem . delete ( targetDir , true ) ; } }
Verifies that nested directories are properly copied with the given filesystem and paths.
Verifies that nested directories are properly copied with the given filesystem and paths .
// directory must not yet exist // create and fill source files //noinspection ResultOfMethodCallIgnored // copy the created directory recursively: // one more for the concluding "/" // clean up