HDFS-5314. Do not expose CachePool type in AddCachePoolOp (Colin Patrick McCabe)

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/HDFS-4949@1530073 13f79535-47bb-0310-9956-ffa450edef68

hadoop-hdfs-project/hadoop-hdfs/CHANGES-HDFS-4949.txt

@@ -74,3 +74,6 @@ HDFS-4949 (Unreleased)
     HDFS-5266. ElasticByteBufferPool#Key does not implement equals. (cnauroth)
 
     HDFS-5309. Fix failing caching unit tests. (Andrew Wang)
+
+    HDFS-5314. Do not expose CachePool type in AddCachePoolOp (Colin Patrick
+    McCabe)

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/CachePoolInfo.java

@@ -26,17 +26,31 @@ import javax.annotation.Nullable;
 
 import org.apache.commons.lang.builder.EqualsBuilder;
 import org.apache.commons.lang.builder.HashCodeBuilder;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.fs.permission.PermissionStatus;
+import org.apache.hadoop.hdfs.server.namenode.FSEditLogOp;
+import org.apache.hadoop.hdfs.util.XMLUtils;
+import org.apache.hadoop.hdfs.util.XMLUtils.InvalidXmlException;
+import org.apache.hadoop.hdfs.util.XMLUtils.Stanza;
 import org.apache.hadoop.io.Text;
+import org.xml.sax.ContentHandler;
+import org.xml.sax.SAXException;
 
 /**
- * Information about a cache pool.
+ * CachePoolInfo describes a cache pool.
+ *
+ * This class is used in RPCs to create and modify cache pools.
+ * It is serializable and can be stored in the edit log.
  */
 @InterfaceAudience.Private
 @InterfaceStability.Evolving
 public class CachePoolInfo {
+  public static final Log LOG = LogFactory.getLog(CachePoolInfo.class);
+
   final String poolName;
 
   @Nullable
@@ -191,4 +205,23 @@ public class CachePoolInfo {
       out.writeInt(weight);
     }
   }
-}
+
+  public void writeXmlTo(ContentHandler contentHandler) throws SAXException {
+    XMLUtils.addSaxString(contentHandler, "POOLNAME", poolName);
+    PermissionStatus perm = new PermissionStatus(ownerName,
+        groupName, mode);
+    FSEditLogOp.permissionStatusToXml(contentHandler, perm);
+    XMLUtils.addSaxString(contentHandler, "WEIGHT", Integer.toString(weight));
+  }
+
+  public static CachePoolInfo readXmlFrom(Stanza st) throws InvalidXmlException {
+    String poolName = st.getValue("POOLNAME");
+    PermissionStatus perm = FSEditLogOp.permissionStatusFromXml(st);
+    int weight = Integer.parseInt(st.getValue("WEIGHT"));
+    return new CachePoolInfo(poolName).
+        setOwnerName(perm.getUserName()).
+        setGroupName(perm.getGroupName()).
+        setMode(perm.getPermission()).
+        setWeight(weight);
+  }
+}

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/CacheManager.java

@@ -371,7 +371,7 @@ public final class CacheManager {
    * @param info The info for the cache pool to create.
    * @return the created CachePool
    */
-  public synchronized CachePool addCachePool(CachePoolInfo info)
+  public synchronized CachePoolInfo addCachePool(CachePoolInfo info)
       throws IOException {
     CachePoolInfo.validate(info);
     String poolName = info.getPoolName();
@@ -379,11 +379,9 @@ public final class CacheManager {
     if (pool != null) {
       throw new IOException("cache pool " + poolName + " already exists.");
     }
-    CachePool cachePool = new CachePool(poolName,
-      info.getOwnerName(), info.getGroupName(), info.getMode(),
-      info.getWeight());
-    unprotectedAddCachePool(cachePool);
-    return cachePool;
+    pool = CachePool.createFromInfoAndDefaults(info);
+    cachePools.put(pool.getPoolName(), pool);
+    return pool.getInfo(true);
   }
 
   /**
@@ -392,8 +390,9 @@ public final class CacheManager {
    * 
    * @param pool to be added
    */
-  void unprotectedAddCachePool(CachePool pool) {
+  void unprotectedAddCachePool(CachePoolInfo info) {
     assert namesystem.hasWriteLock();
+    CachePool pool = CachePool.createFromInfo(info);
     cachePools.put(pool.getPoolName(), pool);
     LOG.info("created new cache pool " + pool);
   }
@@ -538,7 +537,7 @@ public final class CacheManager {
     Counter counter = prog.getCounter(Phase.SAVING_CHECKPOINT, step);
     out.writeInt(cachePools.size());
     for (CachePool pool: cachePools.values()) {
-      pool.writeTo(out);
+      pool.getInfo(true).writeTo(out);
       counter.increment();
     }
     prog.endStep(Phase.SAVING_CHECKPOINT, step);
@@ -576,8 +575,8 @@ public final class CacheManager {
     prog.setTotal(Phase.LOADING_FSIMAGE, step, numberOfPools);
     Counter counter = prog.getCounter(Phase.LOADING_FSIMAGE, step);
     for (int i = 0; i < numberOfPools; i++) {
-      CachePool pool = CachePool.readFrom(in);
-      unprotectedAddCachePool(pool);
+      CachePoolInfo info = CachePoolInfo.readFrom(in);
+      unprotectedAddCachePool(info);
       counter.increment();
     }
     prog.endStep(Phase.LOADING_FSIMAGE, step);

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/CachePool.java

@@ -17,8 +17,6 @@
  */
 package org.apache.hadoop.hdfs.server.namenode;
 
-import java.io.DataInput;
-import java.io.DataOutput;
 import java.io.IOException;
 
 import javax.annotation.Nonnull;
@@ -28,15 +26,10 @@ import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.fs.permission.FsAction;
 import org.apache.hadoop.fs.permission.FsPermission;
-import org.apache.hadoop.fs.permission.PermissionStatus;
 import org.apache.hadoop.hdfs.protocol.CachePoolInfo;
-import org.apache.hadoop.hdfs.util.XMLUtils;
-import org.apache.hadoop.hdfs.util.XMLUtils.InvalidXmlException;
-import org.apache.hadoop.hdfs.util.XMLUtils.Stanza;
-import org.apache.hadoop.io.Text;
 import org.apache.hadoop.security.UserGroupInformation;
-import org.xml.sax.ContentHandler;
-import org.xml.sax.SAXException;
+
+import com.google.common.base.Preconditions;
 
 /**
  * A CachePool describes a set of cache resources being managed by the NameNode.
@@ -44,6 +37,8 @@ import org.xml.sax.SAXException;
  *
  * This is an internal class, only used on the NameNode.  For identifying or
  * describing a cache pool to clients, please use CachePoolInfo.
+ * 
+ * CachePools must be accessed under the FSNamesystem lock.
  */
 @InterfaceAudience.Private
 public final class CachePool {
@@ -73,29 +68,57 @@ public final class CachePool {
   
   private int weight;
 
-  public CachePool(String poolName, String ownerName, String groupName,
-      FsPermission mode, Integer weight) throws IOException {
-    this.poolName = poolName;
+  /**
+   * Create a new cache pool based on a CachePoolInfo object and the defaults.
+   * We will fill in information that was not supplied according to the
+   * defaults.
+   */
+  static CachePool createFromInfoAndDefaults(CachePoolInfo info)
+      throws IOException {
     UserGroupInformation ugi = null;
+    String ownerName = info.getOwnerName();
     if (ownerName == null) {
       if (ugi == null) {
         ugi = NameNode.getRemoteUser();
       }
-      this.ownerName = ugi.getShortUserName();
-    } else {
-      this.ownerName = ownerName;
+      ownerName = ugi.getShortUserName();
     }
+    String groupName = info.getGroupName();
     if (groupName == null) {
       if (ugi == null) {
         ugi = NameNode.getRemoteUser();
       }
-      this.groupName = ugi.getPrimaryGroupName();
-    } else {
-      this.groupName = groupName;
+      groupName = ugi.getPrimaryGroupName();
     }
-    this.mode = mode != null ? 
-        new FsPermission(mode): FsPermission.getCachePoolDefault();
-    this.weight = weight != null ? weight : DEFAULT_WEIGHT;
+    FsPermission mode = (info.getMode() == null) ? 
+        FsPermission.getCachePoolDefault() : info.getMode();
+    Integer weight = (info.getWeight() == null) ?
+        DEFAULT_WEIGHT : info.getWeight();
+    return new CachePool(info.getPoolName(),
+        ownerName, groupName, mode, weight);
+  }
+
+  /**
+   * Create a new cache pool based on a CachePoolInfo object.
+   * No fields in the CachePoolInfo can be blank.
+   */
+  static CachePool createFromInfo(CachePoolInfo info) {
+    return new CachePool(info.getPoolName(),
+        info.getOwnerName(), info.getGroupName(),
+        info.getMode(), info.getWeight());
+  }
+
+  CachePool(String poolName, String ownerName, String groupName,
+      FsPermission mode, int weight) {
+    Preconditions.checkNotNull(poolName);
+    Preconditions.checkNotNull(ownerName);
+    Preconditions.checkNotNull(groupName);
+    Preconditions.checkNotNull(mode);
+    this.poolName = poolName;
+    this.ownerName = ownerName;
+    this.groupName = groupName;
+    this.mode = new FsPermission(mode);
+    this.weight = weight;
   }
 
   public String getPoolName() {
@@ -171,42 +194,4 @@ public final class CachePool {
         append(", weight:").append(weight).
         append(" }").toString();
   }
-
-  public void writeTo(DataOutput out) throws IOException {
-    Text.writeString(out, poolName);
-    PermissionStatus perm = PermissionStatus.createImmutable(
-        ownerName, groupName, mode);
-    perm.write(out);
-    out.writeInt(weight);
-  }
-
-  public static CachePool readFrom(DataInput in) throws IOException {
-    String poolName = Text.readString(in);
-    PermissionStatus perm = PermissionStatus.read(in);
-    int weight = in.readInt();
-    return new CachePool(poolName, perm.getUserName(), perm.getGroupName(),
-        perm.getPermission(), weight);
-  }
-
-  public void writeXmlTo(ContentHandler contentHandler) throws SAXException {
-    XMLUtils.addSaxString(contentHandler, "POOLNAME", poolName);
-    PermissionStatus perm = new PermissionStatus(ownerName,
-        groupName, mode);
-    FSEditLogOp.permissionStatusToXml(contentHandler, perm);
-    XMLUtils.addSaxString(contentHandler, "WEIGHT", Integer.toString(weight));
-  }
-
-  public static CachePool readXmlFrom(Stanza st) throws InvalidXmlException {
-    String poolName = st.getValue("POOLNAME");
-    PermissionStatus perm = FSEditLogOp.permissionStatusFromXml(st);
-    int weight = Integer.parseInt(st.getValue("WEIGHT"));
-    try {
-      return new CachePool(poolName, perm.getUserName(), perm.getGroupName(),
-          perm.getPermission(), weight);
-    } catch (IOException e) {
-      String error = "Invalid cache pool XML, missing fields.";
-      LOG.warn(error);
-      throw new InvalidXmlException(error);
-    }
-  }
 }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSEditLog.java

@@ -971,7 +971,7 @@ public class FSEditLog implements LogsPurgeable {
     logEdit(op);
   }
 
-  void logAddCachePool(CachePool pool, boolean toLogRpcIds) {
+  void logAddCachePool(CachePoolInfo pool, boolean toLogRpcIds) {
     AddCachePoolOp op =
         AddCachePoolOp.getInstance(cache.get()).setPool(pool);
     logRpcIds(op, toLogRpcIds);

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSEditLogLoader.java

@@ -664,7 +664,7 @@ public class FSEditLogLoader {
     }
     case OP_ADD_CACHE_POOL: {
       AddCachePoolOp addOp = (AddCachePoolOp) op;
-      fsNamesys.getCacheManager().unprotectedAddCachePool(addOp.pool);
+      fsNamesys.getCacheManager().unprotectedAddCachePool(addOp.info);
 
       if (toAddRetryCache) {
         fsNamesys.addCacheEntry(op.rpcClientId, op.rpcCallId);

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSEditLogOp.java

@@ -2966,7 +2966,7 @@ public abstract class FSEditLogOp {
   }
 
   static class AddCachePoolOp extends FSEditLogOp {
-    CachePool pool;
+    CachePoolInfo info;
 
     public AddCachePoolOp() {
       super(OP_ADD_CACHE_POOL);
@@ -2976,40 +2976,40 @@ public abstract class FSEditLogOp {
       return (AddCachePoolOp) cache.get(OP_ADD_CACHE_POOL);
     }
 
-    public AddCachePoolOp setPool(CachePool pool) {
-      this.pool = pool;
+    public AddCachePoolOp setPool(CachePoolInfo info) {
+      this.info = info;
       return this;
     }
 
     @Override
     void readFields(DataInputStream in, int logVersion) throws IOException {
-      pool = CachePool.readFrom(in);
+      info = CachePoolInfo.readFrom(in);
     }
 
     @Override
     public void writeFields(DataOutputStream out) throws IOException {
-      pool.writeTo(out);
+      info .writeTo(out);
     }
 
     @Override
     protected void toXml(ContentHandler contentHandler) throws SAXException {
-      pool.writeXmlTo(contentHandler);
+      info .writeXmlTo(contentHandler);
     }
 
     @Override
     void fromXml(Stanza st) throws InvalidXmlException {
-      this.pool = CachePool.readXmlFrom(st);
+      this.info = CachePoolInfo.readXmlFrom(st);
     }
 
     @Override
     public String toString() {
       StringBuilder builder = new StringBuilder();
       builder.append("AddCachePoolOp [");
-      builder.append("poolName=" + pool.getPoolName() + ",");
-      builder.append("ownerName=" + pool.getOwnerName() + ",");
-      builder.append("groupName=" + pool.getGroupName() + ",");
-      builder.append("mode=" + Short.toString(pool.getMode().toShort()) + ",");
-      builder.append("weight=" + Integer.toString(pool.getWeight()) + "]");
+      builder.append("poolName=" + info.getPoolName() + ",");
+      builder.append("ownerName=" + info.getOwnerName() + ",");
+      builder.append("groupName=" + info.getGroupName() + ",");
+      builder.append("mode=" + Short.toString(info.getMode().toShort()) + ",");
+      builder.append("weight=" + Integer.toString(info.getWeight()) + "]");
       return builder.toString();
     }
   }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java

@@ -6989,8 +6989,8 @@ public class FSNamesystem implements Namesystem, FSClusterStats,
       if (pc != null) {
         pc.checkSuperuserPrivilege();
       }
-      CachePool pool = cacheManager.addCachePool(req);
-      getEditLog().logAddCachePool(pool, cacheEntry != null);
+      CachePoolInfo info = cacheManager.addCachePool(req);
+      getEditLog().logAddCachePool(info, cacheEntry != null);
       success = true;
     } finally {
       writeUnlock();