Cleanup

Author: Will-Lo

gobblin-runtime/src/main/java/org/apache/gobblin/runtime/mapreduce/MRJobLauncher.java

@@ -572,7 +572,7 @@ private void addJars(Path jarFileDir, String jarFileList, Configuration conf) th
       Path srcJarFile = new Path(jarFile);
       FileStatus[] fileStatusList = lfs.globStatus(srcJarFile);
       for (FileStatus status : fileStatusList) {
-        Path destJarFile = HdfsJarUploadUtils.calculateDestJarFile(fs, status, this.unsharedJarsDir, jarFileDir);
+        Path destJarFile = HdfsJarUploadUtils.calculateDestJarFilePath(fs, status.getPath().getName(), this.unsharedJarsDir, jarFileDir);
         // For each FileStatus there are chances it could fail in copying at the first attempt, due to file-existence
         // or file-copy is ongoing by other job instance since all Gobblin jobs share the same jar file directory.
         // the retryCount is to avoid cases (if any) where retry is going too far and causes job hanging.

gobblin-utility/src/main/java/org/apache/gobblin/util/filesystem/HdfsJarUploadUtils.java

@@ -39,15 +39,15 @@ public class HdfsJarUploadUtils {
    * given the {@link FileStatus} of a jarFile and the path of directory that contains jar.
    * Snapshot dirs should not be shared, as different jobs may be using different versions of it.
    * @param fs
-   * @param localJar
+   * @param localJarPath
    * @param unsharedJarsDir
    * @param jarCacheDir
    * @return
    * @throws IOException
    */
-  public static Path calculateDestJarFile(FileSystem fs, FileStatus localJar, Path unsharedJarsDir, Path jarCacheDir) throws IOException {
-    Path uploadDir = localJar.getPath().getName().contains("SNAPSHOT") ? unsharedJarsDir : jarCacheDir;
-    Path destJarFile = new Path(fs.makeQualified(uploadDir), localJar.getPath().getName());
+  public static Path calculateDestJarFilePath(FileSystem fs, String localJarPath, Path unsharedJarsDir, Path jarCacheDir) throws IOException {
+    Path uploadDir = localJarPath.contains("SNAPSHOT") ? unsharedJarsDir : jarCacheDir;
+    Path destJarFile = new Path(fs.makeQualified(uploadDir), localJarPath);
     return destJarFile;
   }
   /**
@@ -67,10 +67,9 @@ public static boolean uploadJarToHdfs(FileSystem fs, FileStatus localJar, int ma
           Thread.sleep(WAITING_TIME_ON_INCOMPLETE_UPLOAD_MILLIS);
           throw new IOException("Waiting for file to complete on uploading ... ");
         }
-        // Set the first parameter as false for not deleting sourceFile
-        // Set the second parameter as false for not overwriting existing file on the target, by default it is true.
-        // If the file is preExisted but overwrite flag set to false, then an IOException if thrown.
-        fs.copyFromLocalFile(false, false, localJar.getPath(), destJarFile);
+        boolean deleteSourceFile = false;
+        boolean overwriteAnyExistingDestFile = false; // IOException will be thrown if does already exist
+        fs.copyFromLocalFile(deleteSourceFile, overwriteAnyExistingDestFile, localJar.getPath(), destJarFile);
       } catch (IOException | InterruptedException e) {
         log.warn("Path:" + destJarFile + " is not copied successfully. Will require retry.");
         retryCount += 1;

gobblin-yarn/src/main/java/org/apache/gobblin/yarn/GobblinYarnAppLauncher.java

@@ -740,7 +740,7 @@ private void addLibJars(Path srcLibJarDir, Optional<Map<String, LocalResource>>
     }
 
     for (FileStatus libJarFile : libJarFiles) {
-      Path destFilePath = HdfsJarUploadUtils.calculateDestJarFile(this.fs, libJarFile, unsharedDir, destCacheDir);
+      Path destFilePath = HdfsJarUploadUtils.calculateDestJarFilePath(this.fs, libJarFile.getPath().getName(), unsharedDir, destCacheDir);
       if (HdfsJarUploadUtils.uploadJarToHdfs(fs, libJarFile, MAXIMUM_JAR_COPY_RETRY_TIMES_DEFAULT, destFilePath) && resourceMap.isPresent()) {
         YarnHelixUtils.addFileAsLocalResource(this.fs, destFilePath, LocalResourceType.FILE, resourceMap.get());
       } else {
@@ -753,7 +753,7 @@ private void addAppJars(String jarFilePathList, Optional<Map<String, LocalResour
     for (String jarFilePath : SPLITTER.split(jarFilePathList)) {
       Path srcFilePath = new Path(jarFilePath);
       FileStatus localJar = localFs.getFileStatus(srcFilePath);
-      Path destFilePath = HdfsJarUploadUtils.calculateDestJarFile(this.fs, localJar, unsharedDir, destCacheDir);
+      Path destFilePath = HdfsJarUploadUtils.calculateDestJarFilePath(this.fs, localJar.getPath().getName(), unsharedDir, destCacheDir);
       if (HdfsJarUploadUtils.uploadJarToHdfs(fs, localJar, MAXIMUM_JAR_COPY_RETRY_TIMES_DEFAULT, destFilePath)) {
         if (resourceMap.isPresent()) {
           YarnHelixUtils.addFileAsLocalResource(this.fs, destFilePath, LocalResourceType.FILE, resourceMap.get());