LIHADOOP-55654 Duplicate application init calls trigger NPE and wrong…

… local dirs update in shuffle service. Also fixing a memory leak.

RB=2281730
BUG=LIHADOOP-55654
G=spark-reviewers
R=vsowrira,chsingh,mshen
A=vsowrira,chsingh

common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java

@@ -207,13 +207,15 @@ private File getFile(String appId, String filename) {
     String relativePath = getRelativePath(appPathsInfo.user, appId);
     Path filePath = localDir.resolve(relativePath);
     File targetFile = new File(filePath.toFile(), filename);
-    logger.info("Get the file for " + targetFile.getAbsolutePath());
+    logger.debug("Get the file for {}", targetFile.getAbsolutePath());
     return targetFile;
   }
 
   private Path[] getActiveLocalDirs(String[] activeLocalDirs) {
-    return Arrays.stream(activeLocalDirs)
-        .map(localDir -> Paths.get(localDir)).toArray(Path[]::new);
+    Preconditions.checkNotNull(activeLocalDirs,
+        "Active local dirs list has not been updated by any executor registration");
+    return
+        Arrays.stream(activeLocalDirs).map(localDir -> Paths.get(localDir)).toArray(Path[]::new);
   }
 
   private String getRelativePath(String user, String appId) {
@@ -240,7 +242,11 @@ public String[] getMergedBlockDirs(String appId) {
     AppPathsInfo appPathsInfo = Preconditions.checkNotNull(
         appsPathInfo.get(appId),
         "application " + appId + " is not registered or NM was restarted.");
-    return Arrays.stream(appPathsInfo.activeLocalDirs)
+    String[] activeLocalDirs = Preconditions.checkNotNull(
+        appsPathInfo.get(appId).activeLocalDirs,
+        "application " + appId +
+            " active local dirs list has not been updated by any executor registration");
+    return Arrays.stream(activeLocalDirs)
         .map(dir -> dir + getRelativePath(appPathsInfo.user, appId))
         .toArray(String[]::new);
   }
@@ -250,7 +256,7 @@ public void applicationRemoved(String appId, boolean cleanupLocalDirs) {
     logger.info("Application {} removed, cleanupLocalDirs = {}", appId, cleanupLocalDirs);
     // TODO: Change the message when this service is able to handle NM restart
     AppPathsInfo appPathsInfo = Preconditions.checkNotNull(
-        appsPathInfo.get(appId),
+        appsPathInfo.remove(appId),
         "application " + appId + " is not registered or NM was restarted.");
     Iterator<Map.Entry<AppShufflePartitionId, AppShufflePartitionInfo>> iterator =
         partitions.entrySet().iterator();
@@ -619,7 +625,7 @@ public MergeStatuses finalizeShuffleMerge(FinalizeShuffleMerge msg) throws IOExc
   @Override
   public void registerApplication(String appId, String user) {
     logger.debug("register application with RemoteBlockPushResolver {} {}", appId, user);
-    appsPathInfo.put(appId, new AppPathsInfo(user));
+    appsPathInfo.putIfAbsent(appId, new AppPathsInfo(user));
   }
 
   @Override
@@ -840,7 +846,8 @@ private AppPathsInfo updateActiveLocalDirs(
         activeLocalDirs = Arrays.stream(localDirs)
             .map(localDir -> localDir.substring(0, localDir.indexOf(relativePath)))
             .toArray(String[]::new);
-        logger.info("Updated the active local dirs " + Arrays.toString(activeLocalDirs));
+        logger.info("Updated the active local dirs {} for application {}",
+            Arrays.toString(activeLocalDirs), appId);
       }
       return this;
     }

common/network-shuffle/src/test/java/org/apache/spark/network/shuffle/RemoteBlockPushResolverSuite.java

@@ -88,6 +88,7 @@ public void testNoIndexFile() {
     try {
       String appId = "app_NoIndexFile";
       registerApplication(appId, USER);
+      registerExecutor(appId, prepareBlockManagerLocalDirs(appId, USER, localDirs));
       pushResolver.getMergedBlockMeta(appId, 0, 0);
       removeApplication(appId);
     } catch (Throwable t) {
@@ -162,6 +163,37 @@ public void testAppUsingFewerLocalDirs() throws IOException {
     removeApplication(appId);
   }
 
+  @Test(expected = NullPointerException.class)
+  public void testUpdateLocalDirsOnlyOnce() throws IOException {
+    String appId = "app_App1";
+    // First app init and executor register will store the active local dirs list
+    registerApplication(appId, USER);
+    String[] activeLocalDirs = Arrays.stream(localDirs).skip(1).toArray(String[]::new);
+    registerExecutor(appId, prepareBlockManagerLocalDirs(appId, USER, activeLocalDirs));
+    assertEquals(pushResolver.getMergedBlockDirs(appId).length, 1);
+    assertTrue(pushResolver.getMergedBlockDirs(appId)[0].contains(
+        "l2/usercache/testuser/appcache/app_App1/merge_manager"));
+    // Any later app init or executor register from the same application
+    // won't change the active local dirs list
+    registerApplication(appId, USER);
+    assertEquals(pushResolver.getMergedBlockDirs(appId).length, 1);
+    assertTrue(pushResolver.getMergedBlockDirs(appId)[0].contains(
+        "l2/usercache/testuser/appcache/app_App1/merge_manager"));
+    activeLocalDirs = Arrays.stream(localDirs).toArray(String[]::new);
+    registerExecutor(appId, prepareBlockManagerLocalDirs(appId, USER, activeLocalDirs));
+    assertEquals(pushResolver.getMergedBlockDirs(appId).length, 1);
+    assertTrue(pushResolver.getMergedBlockDirs(appId)[0].contains(
+        "l2/usercache/testuser/appcache/app_App1/merge_manager"));
+    removeApplication(appId);
+    try {
+      pushResolver.getMergedBlockDirs(appId);
+    } catch (Throwable e) {
+      assertTrue(e.getMessage()
+          .startsWith("application app_App1 is not registered or NM was restarted."));
+      Throwables.propagate(e);
+    }
+  }
+
   /**
    * Registers the app with RemoteBlockPushResolver. Use a different appId for different tests.
    * This is because when the application gets removed, the directory cleaner removes the merged