YARN-11741 Follow-up on ResourceManager quit due to ApplicationStateD…

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/RMStateStore.java

@@ -247,7 +247,7 @@ public RMStateStoreState transition(RMStateStore store,
         LOG.error("Error storing app: " + appId, e);
         if (e instanceof StoreLimitException) {
           store.notifyApplication(
-              new RMAppEvent(appId, RMAppEventType.APP_SAVE_FAILED,
+              new RMAppEvent(appId, RMAppEventType.APP_REJECTED,
                   e.getMessage()));
         } else {
           isFenced = store.notifyStoreOperationFailedInternal(e);
@@ -294,7 +294,12 @@ public RMStateStoreState transition(RMStateStore store,
       } catch (Exception e) {
         String msg = "Error updating app: " + appId;
         LOG.error(msg, e);
-        isFenced = store.notifyStoreOperationFailedInternal(e);
+        if (e instanceof StoreLimitException) {
+          store.notifyApplication(new RMAppEvent(appId,
+                  RMAppEventType.APP_REJECTED));
+        } else {
+          isFenced = store.notifyStoreOperationFailedInternal(e);
+        }
         if (result != null) {
           result.setException(new YarnException(msg, e));
         }
@@ -385,7 +390,13 @@ public RMStateStoreState transition(RMStateStore store,
                RMAppAttemptEventType.ATTEMPT_NEW_SAVED));
       } catch (Exception e) {
         LOG.error("Error storing appAttempt: " + attemptState.getAttemptId(), e);
-        isFenced = store.notifyStoreOperationFailedInternal(e);
+        if (e instanceof StoreLimitException) {
+          store.notifyApplicationAttempt(new RMAppAttemptEvent
+                  (attemptState.getAttemptId(),
+                          RMAppAttemptEventType.ATTEMPT_UPDATE_SAVED));
+        } else {
+          isFenced = store.notifyStoreOperationFailedInternal(e);
+        }
       }
       return finalState(isFenced);
     };
@@ -415,7 +426,13 @@ public RMStateStoreState transition(RMStateStore store,
                RMAppAttemptEventType.ATTEMPT_UPDATE_SAVED));
       } catch (Exception e) {
         LOG.error("Error updating appAttempt: " + attemptState.getAttemptId(), e);
-        isFenced = store.notifyStoreOperationFailedInternal(e);
+        if (e instanceof StoreLimitException) {
+          store.notifyApplicationAttempt(new RMAppAttemptEvent
+                  (attemptState.getAttemptId(),
+                          RMAppAttemptEventType.ATTEMPT_UPDATE_SAVED));
+        } else {
+          isFenced = store.notifyStoreOperationFailedInternal(e);
+        }
       }
       return finalState(isFenced);
     };

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/ZKRMStateStore.java

@@ -857,9 +857,11 @@ public synchronized void storeApplicationStateInternal(ApplicationId appId,
       LOG.debug("Application state data size for {} is {}",
           appId, appStateData.length);
 
-      throw new StoreLimitException("Application " + appId
-          + " exceeds the maximum allowed size for application data. "
-          + "See yarn.resourcemanager.zk-max-znode-size.bytes.");
+      throw new StoreLimitException("Application " + appId.toString()
+              + " exceeds the maximum allowed size for application data. "
+              + "The length is " + appStateData.length
+              + ", The node update path is " + nodeCreatePath
+              + ". Current yarn.resourcemanager.zk-max-znode-size.bytes is " + zknodeLimit);
     }
     opDurations.addStoreApplicationStateCallDuration(clock.getTime() - start);
   }
@@ -896,16 +898,33 @@ protected synchronized void updateApplicationStateInternal(
 
     byte[] appStateData = appStateDataPB.getProto().toByteArray();
 
-    if (pathExists) {
-      zkManager.safeSetData(nodeUpdatePath, appStateData, -1, zkAcl,
-          fencingNodePath);
+    if (appStateData.length <= zknodeLimit) {
+      if (pathExists) {
+        zkManager.safeSetData(nodeUpdatePath, appStateData, -1, zkAcl,
+            fencingNodePath);
+      } else {
+        zkManager.safeCreate(nodeUpdatePath, appStateData, zkAcl,
+            CreateMode.PERSISTENT, zkAcl, fencingNodePath);
+        LOG.debug("Path {} for {} didn't exist. Creating a new znode to update"
+            + " the application state.", nodeUpdatePath, appId);
+      }
+      opDurations.addUpdateApplicationStateCallDuration(clock.getTime() - start);
     } else {
-      zkManager.safeCreate(nodeUpdatePath, appStateData, zkAcl,
-          CreateMode.PERSISTENT, zkAcl, fencingNodePath);
-      LOG.debug("Path {} for {} didn't exist. Creating a new znode to update"
-          + " the application state.", nodeUpdatePath, appId);
+      if (LOG.isDebugEnabled()) {
+        LOG.debug("Application state data size for " + appId + " is "
+                + appStateData.length);
+      }
+      LOG.error("Application " + appId.toString()
+              + " exceeds the maximum allowed size for application data. "
+              + "The length is " + appStateData.length
+              + ", The node update path is " + nodeUpdatePath
+              + ". Current yarn.resourcemanager.zk-max-znode-size.bytes is " + zknodeLimit);
+      throw new StoreLimitException("Application " + appId.toString()
+              + " exceeds the maximum allowed size for application data. "
+              + "The length is " + appStateData.length
+              + ", The node update path is " + nodeUpdatePath
+              + ". Current yarn.resourcemanager.zk-max-znode-size.bytes is " + zknodeLimit);
     }
-    opDurations.addUpdateApplicationStateCallDuration(clock.getTime() - start);
   }
 
   /*
@@ -940,20 +959,54 @@ private void handleApplicationAttemptStateOp(
 
     switch (operation) {
     case UPDATE:
-      if (exists(path)) {
-        zkManager.safeSetData(path, attemptStateData, -1, zkAcl,
-            fencingNodePath);
-      } else {
-        zkManager.safeCreate(path, attemptStateData, zkAcl,
-            CreateMode.PERSISTENT, zkAcl, fencingNodePath);
-        LOG.debug("Path {} for {} didn't exist. Created a new znode to update"
-            + " the application attempt state.", path, appAttemptId);
+      if (attemptStateData.length <= zknodeLimit) {
+        if (exists(path)) {
+          zkManager.safeSetData(path, attemptStateData, -1, zkAcl,
+                  fencingNodePath);
+        } else {
+          zkManager.safeCreate(path, attemptStateData, zkAcl,
+                  CreateMode.PERSISTENT, zkAcl, fencingNodePath);
+          LOG.debug("Path {} for {} didn't exist. Created a new znode to update"
+                  + " the application attempt state.", path, appAttemptId);
 
+        }
+      } else {
+        if (LOG.isDebugEnabled()) {
+          LOG.debug("Application state data size for " + appId + " is "
+                  + attemptStateData.length);
+        }
+        LOG.error("Application " + appId.toString()
+                + " exceeds the maximum allowed size for application data. "
+                + "The length is " + attemptStateData.length
+                + ", The node update path is " + path
+                + ". Current yarn.resourcemanager.zk-max-znode-size.bytes is " + zknodeLimit);
+        throw new StoreLimitException("Application " + appId.toString()
+                + " exceeds the maximum allowed size for application data. "
+                + "The length is " + attemptStateData.length
+                + ", The node update path is " + path
+                + ". Current yarn.resourcemanager.zk-max-znode-size.bytes is " + zknodeLimit);
       }
       break;
     case STORE:
-      zkManager.safeCreate(path, attemptStateData, zkAcl, CreateMode.PERSISTENT,
-          zkAcl, fencingNodePath);
+      if (attemptStateData.length <= zknodeLimit) {
+        zkManager.safeCreate(path, attemptStateData, zkAcl, CreateMode.PERSISTENT,
+                zkAcl, fencingNodePath);
+      } else {
+        if (LOG.isDebugEnabled()) {
+          LOG.debug("Application state data size for " + appId + " is "
+                  + attemptStateData.length);
+        }
+        LOG.error("Application " + appId.toString()
+                + " exceeds the maximum allowed size for application data. "
+                + "The length is " + attemptStateData.length
+                + ", The node update path is " + path
+                + ". Current yarn.resourcemanager.zk-max-znode-size.bytes is " + zknodeLimit);
+        throw new StoreLimitException("Application " + appId.toString()
+                + " exceeds the maximum allowed size for application data. "
+                + "The length is " + attemptStateData.length
+                + ", The node update path is " + path
+                + ". Current yarn.resourcemanager.zk-max-znode-size.bytes is " + zknodeLimit);
+      }
       break;
     case REMOVE:
       safeDeleteAndCheckNode(path, zkAcl, fencingNodePath);