YARN-11126. ZKConfigurationStore Java deserialisation vulnerability. Contributed by Tamas Domok

(cherry picked from commit 45801fba8b00257ab32c02a7d1a05948ba687a49)

Conflicts:
	hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/conf/ZKConfigurationStore.java
	hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/conf/TestZKConfigurationStore.java

(cherry picked from commit b2be007db5bc0b731596943dbced1263a9594cde)

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/conf/ZKConfigurationStore.java

@@ -19,8 +19,12 @@
 package org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.conf;
 
 import com.google.common.annotations.VisibleForTesting;
+import org.apache.commons.io.serialization.ValidatingObjectInputStream;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.apache.zookeeper.KeeperException.NodeExistsException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.util.curator.ZKCuratorManager;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
@@ -33,7 +37,6 @@ import org.apache.zookeeper.data.ACL;
 
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
-import java.io.ObjectInputStream;
 import java.io.ObjectOutputStream;
 import java.util.HashMap;
 import java.util.LinkedList;
@@ -229,7 +232,8 @@ public class ZKConfigurationStore extends YarnConfigurationStore {
 
   private static Object deserializeObject(byte[] bytes) throws Exception {
     try (ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
-        ObjectInputStream ois = new ObjectInputStream(bais);) {
+         ValidatingObjectInputStream ois = new ValidatingObjectInputStream(bais);) {
+      ois.accept(LinkedList.class, LogMutation.class, HashMap.class, String.class);
       return ois.readObject();
     }
   }