azkaban-aplcache

diff --git a/azkaban-common/src/main/java/azkaban/utils/FileIOUtils.java b/azkaban-common/src/main/java/azkaban/utils/FileIOUtils.java
index 68d64f1..e9e6e9b 100644
--- a/azkaban-common/src/main/java/azkaban/utils/FileIOUtils.java
+++ b/azkaban-common/src/main/java/azkaban/utils/FileIOUtils.java
@@ -26,6 +26,8 @@ import java.io.InputStream;
 import java.io.InputStreamReader;
 import java.io.OutputStream;
 import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Paths;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
@@ -86,10 +88,10 @@ public class FileIOUtils {
     }
   }
 
+
   /**
-   * Run a unix command that will hard link files and recurse into directories.
+   * Hard link files and recurse into directories.
    */
-
   public static void createDeepHardlink(final File sourceDir, final File destDir)
       throws IOException {
     if (!sourceDir.exists()) {
@@ -105,51 +107,18 @@ public class FileIOUtils {
     final Set<String> paths = new HashSet<>();
     createDirsFindFiles(sourceDir, sourceDir, destDir, paths);
 
-    final StringBuffer buffer = new StringBuffer();
     for (String path : paths) {
       final File sourceLink = new File(sourceDir, path);
-      path = "." + path;
-
-      buffer.append("ln ").append(sourceLink.getAbsolutePath()).append("/*")
-          .append(" ").append(path).append(";");
-    }
-
-    runShellCommand(buffer.toString(), destDir);
-  }
-
-  private static void runShellCommand(final String command, final File workingDir)
-      throws IOException {
-    final ProcessBuilder builder = new ProcessBuilder().command("sh", "-c", command);
-    builder.directory(workingDir);
+      path = destDir + path;
 
-    // XXX what about stopping threads ??
-    final Process process = builder.start();
-    try {
-      final NullLogger errorLogger = new NullLogger(process.getErrorStream());
-      final NullLogger inputLogger = new NullLogger(process.getInputStream());
-      errorLogger.start();
-      inputLogger.start();
-
-      try {
-        if (process.waitFor() < 0) {
-          // Assume that the error will be in standard out. Otherwise it'll be
-          // in standard in.
-          String errorMessage = errorLogger.getLastMessages();
-          if (errorMessage.isEmpty()) {
-            errorMessage = inputLogger.getLastMessages();
-          }
-
-          throw new IOException(errorMessage);
+      final File[] targetFiles = sourceLink.listFiles();
+      for (final File targetFile : targetFiles) {
+        if (targetFile.isFile()) {
+          final File linkFile = new File(path, targetFile.getName());
+          Files.createLink(linkFile.toPath(), Paths.get(targetFile.getAbsolutePath()));
         }
-      } catch (final InterruptedException e) {
-        logger.error(e);
       }
-    } finally {
-      IOUtils.closeQuietly(process.getInputStream());
-      IOUtils.closeQuietly(process.getOutputStream());
-      IOUtils.closeQuietly(process.getErrorStream());
     }
-
   }
 
   private static void createDirsFindFiles(final File baseDir, final File sourceDir,

diff --git a/azkaban-common/src/test/java/azkaban/utils/FileIOUtilsTest.java b/azkaban-common/src/test/java/azkaban/utils/FileIOUtilsTest.java
index 0b9b6c3..679fe33 100644
--- a/azkaban-common/src/test/java/azkaban/utils/FileIOUtilsTest.java
+++ b/azkaban-common/src/test/java/azkaban/utils/FileIOUtilsTest.java
@@ -16,13 +16,17 @@
 
 package azkaban.utils;
 
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
 
 import java.io.File;
 import java.io.FileOutputStream;
 import java.io.IOException;
 import java.io.UnsupportedEncodingException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
 import java.util.Arrays;
 import org.apache.commons.io.FileUtils;
 import org.apache.commons.io.comparator.NameFileComparator;
@@ -38,16 +42,46 @@ public class FileIOUtilsTest {
   public TemporaryFolder temp = new TemporaryFolder();
   private File sourceDir, destDir, baseDir;
 
+  /**
+   * Create a very big dir which would cause linux shell command to hard link the dir to
+   * exceed the allowed limit.
+   */
+  private void createBigDir(final String path) throws IOException {
+    final String verylongprefix =
+        "123123123123123123123123123123123113123111111111111111111111111111111111111111111111111111"
+            + "111111111111111111111111111111111111111111111111111111111111111111111111111111111111"
+            + "11231312312312312312313121111111111111111111111111111111111111111111111111";
+
+    for (int i = 1; i <= 400; i++) {
+      final Path tmpDirPath = Paths.get(path, verylongprefix + "dir" + i);
+      Files.createDirectory(tmpDirPath);
+      for (int j = 1; j <= 100; j++) {
+        final Path tmp = Paths.get(tmpDirPath.toAbsolutePath().toString(), String.valueOf(j));
+        Files.createFile(tmp);
+      }
+    }
+  }
+
   @Before
   public void setUp() throws Exception {
     // setup base dir
+
     this.baseDir = this.temp.newFolder("base");
     final File file1 = new File(this.baseDir.getAbsolutePath() + "/a.out");
     final File file2 = new File(this.baseDir.getAbsolutePath() + "/testdir");
     final File file3 = new File(file2.getAbsolutePath() + "/b.out");
+    final File file4 = new File(file2.getAbsolutePath() + "/testdir");
+    final File file5 = new File(file2.getAbsolutePath() + "/c.out");
+    final File file6 = new File(file4.getAbsolutePath() + "/c1.out");
+    final File file7 = new File(file4.getAbsolutePath() + "/c2.out");
+
     file1.createNewFile();
     file2.mkdir();
     file3.createNewFile();
+    file4.mkdir();
+    file5.createNewFile();
+    file6.createNewFile();
+    file7.createNewFile();
 
     byte[] fileData = new byte[]{1, 2, 3};
     FileOutputStream out = new FileOutputStream(file1);
@@ -77,23 +111,28 @@ public class FileIOUtilsTest {
   @Test
   public void testHardlinkCopy() throws IOException {
     FileIOUtils.createDeepHardlink(this.sourceDir, this.destDir);
-    assertTrue(areDirsEqual(this.sourceDir, this.destDir, true));
+    assertThat(areDirsEqual(this.sourceDir, this.destDir, true)).isTrue();
     FileUtils.deleteDirectory(this.destDir);
-    assertTrue(areDirsEqual(this.baseDir, this.sourceDir, true));
+    assertThat(areDirsEqual(this.baseDir, this.sourceDir, true)).isTrue();
+  }
+
+  @Test
+  public void testHardlinkCopyOfBigDir() throws IOException {
+    final File bigDir = new File(this.baseDir.getAbsolutePath() + "/bigdir");
+    bigDir.mkdir();
+    createBigDir(bigDir.getAbsolutePath());
+
+    FileIOUtils.createDeepHardlink(bigDir, this.destDir);
+    assertThat(areDirsEqual(this.destDir, bigDir, true)).isTrue();
+    FileUtils.deleteDirectory(bigDir);
+
   }
 
   @Test
   public void testHardlinkCopyNonSource() {
-    boolean exception = false;
-    try {
+    assertThatThrownBy(() -> {
       FileIOUtils.createDeepHardlink(new File(this.sourceDir, "idonotexist"), this.destDir);
-    } catch (final IOException e) {
-      System.out.println(e.getMessage());
-      System.out.println("Handled this case nicely.");
-      exception = true;
-    }
-
-    assertTrue(exception);
+    }).isInstanceOf(IOException.class);
   }
 
   private boolean areDirsEqualUtil(final File file1, final File file2, final boolean isRoot,