YARN-8116. Nodemanager fails with NumberFormatException: For input string: . (Chandni Singh via wangda)

Change-Id: Idd30cfca59982d3fc6e47aa1b88f844a78fae94d

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/container/ContainerImpl.java

@@ -2191,7 +2191,8 @@ public class ContainerImpl implements Container {
   }
 
   private void storeRetryContext() {
-    if (windowRetryContext.getRestartTimes() != null) {
+    if (windowRetryContext.getRestartTimes() != null &&
+        !windowRetryContext.getRestartTimes().isEmpty()) {
       try {
         stateStore.storeContainerRestartTimes(containerId,
             windowRetryContext.getRestartTimes());

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/recovery/NMLeveldbStateStoreService.java

@@ -347,7 +347,9 @@ public class NMLeveldbStateStoreService extends NMStateStoreService {
             value.substring(1, value.length() - 1).split(", ");
         List<Long> restartTimes = new ArrayList<>();
         for (String restartTime : unparsedRestartTimes) {
-          restartTimes.add(Long.parseLong(restartTime));
+          if (!restartTime.isEmpty()) {
+            restartTimes.add(Long.parseLong(restartTime));
+          }
         }
         rcs.setRestartTimes(restartTimes);
       } else if (suffix.equals(CONTAINER_WORK_DIR_KEY_SUFFIX)) {

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/recovery/TestNMLeveldbStateStoreService.java

@@ -1216,6 +1216,22 @@ public class TestNMLeveldbStateStoreService {
     Assert.fail("Expected exception not thrown");
   }
 
+  @Test
+  public void testEmptyRestartTimes() throws IOException {
+    List<Long> restartTimes = new ArrayList<>();
+    ApplicationId appId = ApplicationId.newInstance(1234, 3);
+    ApplicationAttemptId appAttemptId = ApplicationAttemptId.newInstance(appId,
+        4);
+    ContainerId containerId = ContainerId.newContainerId(appAttemptId, 5);
+    storeMockContainer(containerId);
+    stateStore.storeContainerRestartTimes(containerId,
+        restartTimes);
+    restartStateStore();
+    RecoveredContainerState rcs = stateStore.loadContainersState().get(0);
+    List<Long> recoveredRestartTimes = rcs.getRestartTimes();
+    assertTrue(recoveredRestartTimes.isEmpty());
+  }
+
   private StartContainerRequest storeMockContainer(ContainerId containerId)
       throws IOException {
     // create a container request