commit efe1304df361e2e77c4eef09f0df8c37b1aac601

Author: Chris Douglas <cdouglas@apache.org>
Date:   Fri Jan 22 16:03:23 2010 -0800

    MAPREDUCE:1376 from https://issues.apache.org/jira/secure/attachment/12431174/M1376-4.patch
    
    +++ b/YAHOO-CHANGES.txt
    +    MAPREDUCE-1376. Add support for submitting jobs as configured users,
    +    pluggable mapping of trace users to target users in Gridmix. (cdouglas)
    +


git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-0.20-security-patches@1077192 13f79535-47bb-0310-9956-ffa450edef68

src/contrib/gridmix/src/java/org/apache/hadoop/mapred/gridmix/EchoUserResolver.java

@@ -0,0 +1,43 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.mapred.gridmix;
+
+import java.io.IOException;
+import java.net.URI;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.security.UserGroupInformation;
+
+/**
+ * Echos the UGI offered.
+ */
+public class EchoUserResolver extends UserResolver {
+
+  public EchoUserResolver() { }
+
+  public synchronized boolean setTargetUsers(URI userdesc, Configuration conf)
+      throws IOException {
+    return false;
+  }
+
+  public synchronized UserGroupInformation getTargetUgi(
+      UserGroupInformation ugi) {
+    return ugi;
+  }
+
+}

src/contrib/gridmix/src/java/org/apache/hadoop/mapred/gridmix/GenerateData.java

@@ -31,6 +31,7 @@ import java.util.regex.Pattern;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.io.BytesWritable;
 import org.apache.hadoop.io.LongWritable;
 import org.apache.hadoop.io.NullWritable;
@@ -73,6 +74,16 @@ class GenerateData extends GridmixJob {
    */
   public static final String GRIDMIX_GEN_INTERVAL = "gendata.interval.mb";
 
+  /**
+   * Blocksize of generated data.
+   */
+  public static final String GRIDMIX_GEN_BLOCKSIZE = "gridmix.gen.blocksize";
+
+  /**
+   * Replication of generated data.
+   */
+  public static final String GRIDMIX_GEN_REPLICATION = "gridmix.gen.replicas";
+
   public GenerateData(Configuration conf, Path outdir, long genbytes)
       throws IOException {
     super(conf, 0L, "GRIDMIX_GENDATA");
@@ -248,7 +259,10 @@ class GenerateData extends GridmixJob {
     static class ChunkWriter extends RecordWriter<NullWritable,BytesWritable> {
       private final Path outDir;
       private final FileSystem fs;
+      private final int blocksize;
+      private final short replicas;
       private final long maxFileBytes;
+      private final FsPermission genPerms = new FsPermission((short) 0755);
 
       private long accFileBytes = 0L;
       private long fileIdx = -1L;
@@ -257,6 +271,8 @@ class GenerateData extends GridmixJob {
       public ChunkWriter(Path outDir, Configuration conf) throws IOException {
         this.outDir = outDir;
         fs = outDir.getFileSystem(conf);
+        blocksize = conf.getInt(GRIDMIX_GEN_BLOCKSIZE, 1 << 28);
+        replicas = (short) conf.getInt(GRIDMIX_GEN_REPLICATION, 3);
         maxFileBytes = conf.getLong(GRIDMIX_GEN_CHUNK, 1L << 30);
         nextDestination();
       }
@@ -264,7 +280,8 @@ class GenerateData extends GridmixJob {
         if (fileOut != null) {
           fileOut.close();
         }
-        fileOut = fs.create(new Path(outDir, "segment-" + (++fileIdx)), false);
+        fileOut = fs.create(new Path(outDir, "segment-" + (++fileIdx)),
+            genPerms, false, 64 * 1024, replicas, blocksize, null);
         accFileBytes = 0L;
       }
       @Override
@@ -273,14 +290,14 @@ class GenerateData extends GridmixJob {
         int written = 0;
         final int total = value.getLength();
         while (written < total) {
+          if (accFileBytes >= maxFileBytes) {
+            nextDestination();
+          }
           final int write = (int)
             Math.min(total - written, maxFileBytes - accFileBytes);
           fileOut.write(value.getBytes(), written, write);
           written += write;
           accFileBytes += write;
-          if (accFileBytes >= maxFileBytes) {
-            nextDestination();
-          }
         }
       }
       @Override

src/contrib/gridmix/src/java/org/apache/hadoop/mapred/gridmix/Gridmix.java

@@ -20,15 +20,21 @@ package org.apache.hadoop.mapred.gridmix;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.PrintStream;
+import java.net.URI;
+import java.util.Collection;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.conf.Configured;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.io.IOUtils;
 import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.util.ReflectionUtils;
 import org.apache.hadoop.util.StringUtils;
 import org.apache.hadoop.util.Tool;
 import org.apache.hadoop.util.ToolRunner;
@@ -78,6 +84,12 @@ public class Gridmix extends Configured implements Tool {
    */
   public static final String GRIDMIX_SUB_MUL = "gridmix.submit.multiplier";
 
+  /**
+   * Class used to resolve users in the trace to the list of target users
+   * on the cluster.
+   */
+  public static final String GRIDMIX_USR_RSV = "gridmix.user.resolve.class";
+
   // Submit data structures
   private JobFactory factory;
   private JobSubmitter submitter;
@@ -128,14 +140,16 @@ public class Gridmix extends Configured implements Tool {
    * @param startFlag Semaphore for starting job trace pipeline
    */
   private void startThreads(Configuration conf, String traceIn, Path ioPath,
-      Path scratchDir, CountDownLatch startFlag) throws IOException {
+      Path scratchDir, CountDownLatch startFlag, UserResolver userResolver)
+      throws IOException {
     monitor = createJobMonitor();
     submitter = createJobSubmitter(monitor,
         conf.getInt(GRIDMIX_SUB_THR,
           Runtime.getRuntime().availableProcessors() + 1),
         conf.getInt(GRIDMIX_QUE_DEP, 5),
         new FilePool(conf, ioPath));
-    factory = createJobFactory(submitter, traceIn, scratchDir, conf, startFlag);
+    factory = createJobFactory(submitter, traceIn, scratchDir, conf, startFlag,
+        userResolver);
     monitor.start();
     submitter.start();
     factory.start();
@@ -151,45 +165,64 @@ public class Gridmix extends Configured implements Tool {
   }
 
   protected JobFactory createJobFactory(JobSubmitter submitter, String traceIn,
-      Path scratchDir, Configuration conf, CountDownLatch startFlag)
-      throws IOException {
+      Path scratchDir, Configuration conf, CountDownLatch startFlag,
+      UserResolver userResolver) throws IOException {
     return new JobFactory(submitter, createInputStream(traceIn), scratchDir,
-        conf, startFlag);
+        conf, startFlag, userResolver);
   }
 
-  public int run(String[] argv) throws IOException, InterruptedException {
+  public int run(final String[] argv) throws IOException, InterruptedException {
     if (argv.length < 2) {
       printUsage(System.err);
       return 1;
     }
-    long genbytes = 0;
+    final Configuration conf = getConf();
+    long genbytes = -1L;
     String traceIn = null;
     Path ioPath = null;
+    URI userRsrc = null;
+    final UserResolver userResolver = ReflectionUtils.newInstance(
+        conf.getClass(GRIDMIX_USR_RSV, SubmitterUserResolver.class,
+          UserResolver.class), conf);
     try {
-      int i = 0;
-      genbytes = "-generate".equals(argv[i++])
-        ? StringUtils.TraditionalBinaryPrefix.string2long(argv[i++])
-        : --i;
-      ioPath = new Path(argv[i++]);
-      traceIn = argv[i++];
-      if (i != argv.length) {
-        printUsage(System.err);
-        return 1;
+      for (int i = 0; i < argv.length - 2; ++i) {
+        if ("-generate".equals(argv[i])) {
+          genbytes = StringUtils.TraditionalBinaryPrefix.string2long(argv[++i]);
+        } else if ("-users".equals(argv[i])) {
+          userRsrc = new URI(argv[++i]);
+        } else {
+          printUsage(System.err);
+          return 1;
+        }
+      }
+      if (!userResolver.setTargetUsers(userRsrc, conf)) {
+        LOG.warn("Resource " + userRsrc + " ignored");
       }
+      ioPath = new Path(argv[argv.length - 2]);
+      traceIn = argv[argv.length - 1];
     } catch (Exception e) {
+      e.printStackTrace();
       printUsage(System.err);
       return 1;
     }
+    return start(conf, traceIn, ioPath, genbytes, userResolver);
+  }
+
+  int start(Configuration conf, String traceIn, Path ioPath, long genbytes,
+      UserResolver userResolver) throws IOException, InterruptedException {
     InputStream trace = null;
     try {
-      final Configuration conf = getConf();
       Path scratchDir = new Path(ioPath, conf.get(GRIDMIX_OUT_DIR, "gridmix"));
+      final FileSystem scratchFs = scratchDir.getFileSystem(conf);
+      scratchFs.mkdirs(scratchDir, new FsPermission((short) 0777));
+      scratchFs.setPermission(scratchDir, new FsPermission((short) 0777));
       // add shutdown hook for SIGINT, etc.
       Runtime.getRuntime().addShutdownHook(sdh);
       CountDownLatch startFlag = new CountDownLatch(1);
       try {
         // Create, start job submission threads
-        startThreads(conf, traceIn, ioPath, scratchDir, startFlag);
+        startThreads(conf, traceIn, ioPath, scratchDir, startFlag,
+            userResolver);
         // Write input data if specified
         if (genbytes > 0) {
           writeInputData(genbytes, ioPath);
@@ -301,7 +334,7 @@ public class Gridmix extends Configured implements Tool {
 
   protected void printUsage(PrintStream out) {
     ToolRunner.printGenericCommandUsage(out);
-    out.println("Usage: gridmix [-generate <MiB>] <iopath> <trace>");
+    out.println("Usage: gridmix [-generate <MiB>] [-users URI] <iopath> <trace>");
     out.println("  e.g. gridmix -generate 100m foo -");
     out.println("Configuration parameters:");
     out.printf("       %-40s : Output directory\n", GRIDMIX_OUT_DIR);
@@ -309,6 +342,7 @@ public class Gridmix extends Configured implements Tool {
     out.printf("       %-40s : Queued job desc\n", GRIDMIX_QUE_DEP);
     out.printf("       %-40s : Key fraction of rec\n",
         AvgRecordFactory.GRIDMIX_KEY_FRC);
+    out.printf("       %-40s : User resolution class\n", GRIDMIX_USR_RSV);
   }
 
   /**

src/contrib/gridmix/src/java/org/apache/hadoop/mapred/gridmix/GridmixJob.java

@@ -26,6 +26,7 @@ import java.util.concurrent.Callable;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Delayed;
 import java.util.concurrent.TimeUnit;
+import javax.security.auth.login.LoginException;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FSDataOutputStream;
@@ -50,6 +51,8 @@ import org.apache.hadoop.mapreduce.TaskAttemptContext;
 import org.apache.hadoop.mapreduce.TaskType;
 import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
 import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.UnixUserGroupInformation;
 import org.apache.hadoop.tools.rumen.JobStory;
 import org.apache.hadoop.tools.rumen.TaskInfo;
 
@@ -79,10 +82,15 @@ class GridmixJob implements Callable<Job>, Delayed {
   private final Path outdir;
   protected final Job job;
   private final JobStory jobdesc;
+  private final UserGroupInformation ugi;
   private final long submissionTimeNanos;
 
   public GridmixJob(Configuration conf, long submissionMillis,
-      JobStory jobdesc, Path outRoot, int seq) throws IOException {
+      JobStory jobdesc, Path outRoot, UserGroupInformation ugi, int seq)
+      throws IOException {
+    this.ugi = ugi;
+    UserGroupInformation.setCurrentUser(ugi);
+    conf.set(UnixUserGroupInformation.UGI_PROPERTY_NAME, ugi.toString());
     ((StringBuilder)nameFormat.get().out()).setLength(JOBNAME.length());
     job = new Job(conf, nameFormat.get().format("%05d", seq).toString());
     submissionTimeNanos = TimeUnit.NANOSECONDS.convert(
@@ -100,6 +108,15 @@ class GridmixJob implements Callable<Job>, Delayed {
     jobdesc = null;
     outdir = null;
     seq = -1;
+    try {
+      ugi = UnixUserGroupInformation.login(conf);
+    } catch (LoginException e) {
+      throw new IOException("Could not identify submitter", e);
+    }
+  }
+
+  public UserGroupInformation getUgi() {
+    return ugi;
   }
 
   public String toString() {

src/contrib/gridmix/src/java/org/apache/hadoop/mapred/gridmix/JobFactory.java

@@ -60,6 +60,7 @@ class JobFactory implements Gridmix.Component<Void> {
   private final AtomicInteger sequence;
   private final JobSubmitter submitter;
   private final CountDownLatch startFlag;
+  private final UserResolver userResolver;
   private volatile IOException error = null;
   protected final JobStoryProducer jobProducer;
 
@@ -73,10 +74,10 @@ class JobFactory implements Gridmix.Component<Void> {
    * @param startFlag Latch released from main to start pipeline
    */
   public JobFactory(JobSubmitter submitter, InputStream jobTrace,
-      Path scratch, Configuration conf, CountDownLatch startFlag)
-      throws IOException {
+      Path scratch, Configuration conf, CountDownLatch startFlag,
+      UserResolver userResolver) throws IOException {
     this(submitter, new ZombieJobProducer(jobTrace, null), scratch, conf,
-        startFlag);
+        startFlag, userResolver);
   }
 
   /**
@@ -88,7 +89,8 @@ class JobFactory implements Gridmix.Component<Void> {
    * @param startFlag Latch released from main to start pipeline
    */
   protected JobFactory(JobSubmitter submitter, JobStoryProducer jobProducer,
-      Path scratch, Configuration conf, CountDownLatch startFlag) {
+      Path scratch, Configuration conf, CountDownLatch startFlag,
+      UserResolver userResolver) {
     sequence = new AtomicInteger(0);
     this.scratch = scratch;
     this.rateFactor = conf.getFloat(Gridmix.GRIDMIX_SUB_MUL, 1.0f);
@@ -97,6 +99,7 @@ class JobFactory implements Gridmix.Component<Void> {
     this.submitter = submitter;
     this.startFlag = startFlag;
     this.rThread = new ReaderThread();
+    this.userResolver = userResolver;
   }
 
   static class MinTaskInfo extends TaskInfo {
@@ -206,9 +209,10 @@ class JobFactory implements Gridmix.Component<Void> {
               continue;
             }
             last = current;
-            submitter.add(new GridmixJob(conf, initTime +
-                  Math.round(rateFactor * (current - first)),
-                job, scratch, sequence.getAndIncrement()));
+            submitter.add(new GridmixJob(new Configuration(conf), initTime +
+                  Math.round(rateFactor * (current - first)), job, scratch,
+                  userResolver.getTargetUgi(job.getUser()),
+                sequence.getAndIncrement()));
           } catch (IOException e) {
             JobFactory.this.error = e;
             return;

src/contrib/gridmix/src/java/org/apache/hadoop/mapred/gridmix/JobSubmitter.java

@@ -29,6 +29,8 @@ import java.util.concurrent.TimeUnit;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
+import org.apache.hadoop.security.UserGroupInformation;
+
 /**
  * Component accepting deserialized job traces, computing split data, and
  * submitting to the cluster on deadline. Each job added from an upstream
@@ -79,9 +81,11 @@ class JobSubmitter implements Gridmix.Component<GridmixJob> {
       try {
         // pre-compute split information
         try {
+          UserGroupInformation.setCurrentUser(job.getUgi());
           job.buildSplits(inputDir);
         } catch (IOException e) {
-          LOG.warn("Failed to submit " + job.getJob().getJobName(), e);
+          LOG.warn("Failed to submit " + job.getJob().getJobName() + " as " +
+              job.getUgi(), e);
           return;
         }
         // Sleep until deadline
@@ -96,7 +100,8 @@ class JobSubmitter implements Gridmix.Component<GridmixJob> {
           LOG.debug("SUBMIT " + job + "@" + System.currentTimeMillis() +
               " (" + job.getJob().getJobID() + ")");
         } catch (IOException e) {
-          LOG.warn("Failed to submit " + job.getJob().getJobName(), e);
+          LOG.warn("Failed to submit " + job.getJob().getJobName() + " as " +
+              job.getUgi(), e);
           if (e.getCause() instanceof ClosedByInterruptException) {
             throw new InterruptedException("Failed to submit " +
                 job.getJob().getJobName());

src/contrib/gridmix/src/java/org/apache/hadoop/mapred/gridmix/RoundRobinUserResolver.java

@@ -0,0 +1,63 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.mapred.gridmix;
+
+import java.io.IOException;
+import java.net.URI;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.UnixUserGroupInformation;
+
+public class RoundRobinUserResolver extends UserResolver {
+
+  private int uidx = 0;
+  private List<UserGroupInformation> users = Collections.emptyList();
+  private final HashMap<UserGroupInformation,UserGroupInformation> usercache =
+    new HashMap<UserGroupInformation,UserGroupInformation>();
+
+  public RoundRobinUserResolver() { }
+
+  @Override
+  public synchronized boolean setTargetUsers(URI userloc, Configuration conf)
+      throws IOException {
+    users = parseUserList(userloc, conf);
+    if (users.size() == 0) {
+      throw new IOException("Empty user list");
+    }
+    usercache.keySet().retainAll(users);
+    return true;
+  }
+
+  @Override
+  public synchronized UserGroupInformation getTargetUgi(
+      UserGroupInformation ugi) {
+    UserGroupInformation ret = usercache.get(ugi);
+    if (null == ret) {
+      ret = users.get(uidx++ % users.size());
+      usercache.put(ugi, ret);
+    }
+    return ret;
+  }
+
+}

src/contrib/gridmix/src/java/org/apache/hadoop/mapred/gridmix/SubmitterUserResolver.java

@@ -0,0 +1,52 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.mapred.gridmix;
+
+import java.io.IOException;
+import java.net.URI;
+import javax.security.auth.login.LoginException;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.UnixUserGroupInformation;
+
+/**
+ * Resolves all UGIs to the submitting user.
+ */
+public class SubmitterUserResolver extends UserResolver {
+
+  private UserGroupInformation ugi = null;
+
+  public SubmitterUserResolver() { }
+
+  public synchronized boolean setTargetUsers(URI userdesc, Configuration conf)
+      throws IOException {
+    try {
+      ugi = UnixUserGroupInformation.login(conf, false);
+    } catch (LoginException e) {
+      throw new IOException("Failed to get submitter UGI", e);
+    }
+    return false;
+  }
+
+  public synchronized UserGroupInformation getTargetUgi(
+      UserGroupInformation ugi) {
+    return this.ugi;
+  }
+
+}

src/contrib/gridmix/src/java/org/apache/hadoop/mapred/gridmix/UserResolver.java

@@ -0,0 +1,112 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.mapred.gridmix;
+
+import java.io.IOException;
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.security.UnixUserGroupInformation;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.util.LineReader;
+
+/**
+ * Maps users in the trace to a set of valid target users on the test cluster.
+ */
+public abstract class UserResolver {
+
+  /**
+   * Userlist assumes one UGI per line, each UGI matching
+   * <username>,<group>[,group]*
+   */
+  protected List<UserGroupInformation> parseUserList(
+      URI userUri, Configuration conf) throws IOException {
+    if (null == userUri) {
+      return Collections.emptyList();
+    }
+    final Path userloc = new Path(userUri.toString());
+    final Text rawUgi = new Text();
+    final FileSystem fs = userloc.getFileSystem(conf);
+    final ArrayList<UserGroupInformation> ret = new ArrayList();
+
+    LineReader in = null;
+    try {
+      final ArrayList<String> groups = new ArrayList();
+      in = new LineReader(fs.open(userloc));
+      while (in.readLine(rawUgi) > 0) {
+        int e = rawUgi.find(",");
+        if (e <= 0) {
+          throw new IOException("Missing username: " + rawUgi);
+        }
+        final String username = Text.decode(rawUgi.getBytes(), 0, e);
+        int s = e;
+        while ((e = rawUgi.find(",", ++s)) != -1) {
+          groups.add(Text.decode(rawUgi.getBytes(), s, e - s));
+          s = e;
+        }
+        groups.add(Text.decode(rawUgi.getBytes(), s, rawUgi.getLength() - s));
+        if (groups.size() == 0) {
+          throw new IOException("Missing groups: " + rawUgi);
+        }
+        ret.add(new UnixUserGroupInformation(
+              username, groups.toArray(new String[groups.size()])));
+        groups.clear();
+      }
+    } finally {
+      if (in != null) {
+        in.close();
+      }
+    }
+    return ret;
+  }
+
+  /**
+   * Configure the user map given the URI and configuration. The resolver's
+   * contract will define how the resource will be interpreted, but the default
+   * will typically interpret the URI as a {@link org.apache.hadoop.fs.Path}
+   * listing target users. The format of this file is defined by {@link
+   * #parseUserList}.
+   * @param userdesc URI (possibly null) from which user information may be
+   * loaded per the subclass contract.
+   * @param conf The tool configuration.
+   * @return true if the resource provided was used in building the list of
+   * target users
+   */
+  public abstract boolean setTargetUsers(URI userdesc, Configuration conf)
+    throws IOException;
+
+  // tmp compatibility hack prior to UGI from Rumen
+  public UserGroupInformation getTargetUgi(String user)
+      throws IOException {
+    return getTargetUgi(new UnixUserGroupInformation(
+          user, new String[] { "users" }));
+  }
+
+  /**
+   * Map the given UGI to another per the subclass contract.
+   * @param ugi User information from the trace.
+   */
+  public abstract UserGroupInformation getTargetUgi(UserGroupInformation ugi);
+
+}

src/contrib/gridmix/src/test/org/apache/hadoop/mapred/gridmix/DebugJobFactory.java

@@ -44,9 +44,10 @@ import org.apache.hadoop.tools.rumen.Pre21JobHistoryConstants.Values;;
 class DebugJobFactory extends JobFactory {
 
   public DebugJobFactory(JobSubmitter submitter, Path scratch, int numJobs,
-      Configuration conf, CountDownLatch startFlag) throws IOException {
+      Configuration conf, CountDownLatch startFlag, UserResolver userResolver)
+      throws IOException {
     super(submitter, new DebugJobProducer(numJobs, conf), scratch, conf,
-        startFlag);
+        startFlag, userResolver);
   }
 
   ArrayList<JobStory> getSubmitted() {
@@ -210,7 +211,7 @@ class DebugJobFactory extends JobFactory {
 
     @Override
     public String getUser() {
-      return "FOOBAR";
+      return String.format("foobar%d", id);
     }
 
     @Override

src/contrib/gridmix/src/test/org/apache/hadoop/mapred/gridmix/TestGridmixSubmission.java

@@ -30,6 +30,7 @@ import org.apache.hadoop.fs.ContentSummary;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.hdfs.MiniDFSCluster;
 import org.apache.hadoop.mapred.Counters;
 import org.apache.hadoop.mapred.JobClient;
@@ -60,6 +61,8 @@ public class TestGridmixSubmission {
         ).getLogger().setLevel(Level.DEBUG);
   }
 
+  private static final Path DEST = new Path("/gridmix");
+
   private static FileSystem dfs = null;
   private static MiniDFSCluster dfsCluster = null;
   private static MiniMRCluster mrCluster = null;
@@ -124,6 +127,11 @@ public class TestGridmixSubmission {
           sub.get(job.getJobName().replace("GRIDMIX", "MOCKJOB"));
         assertNotNull("No spec for " + job.getJobName(), spec);
         assertNotNull("No counters for " + job.getJobName(), job.getCounters());
+        final String specname = spec.getName();
+        final FileStatus stat = dfs.getFileStatus(new Path(DEST, "" +
+              Integer.valueOf(specname.substring(specname.length() - 5))));
+        assertEquals("Wrong owner for " + job.getJobName(), spec.getUser(),
+            stat.getOwner());
 
         final int nMaps = spec.getNumberMaps();
         final int nReds = spec.getNumberReduces();
@@ -295,9 +303,11 @@ public class TestGridmixSubmission {
     @Override
     protected JobFactory createJobFactory(JobSubmitter submitter,
         String traceIn, Path scratchDir, Configuration conf,
-        CountDownLatch startFlag) throws IOException {
+        CountDownLatch startFlag, UserResolver userResolver)
+        throws IOException {
       factory =
-        new DebugJobFactory(submitter, scratchDir, NJOBS, conf, startFlag);
+        new DebugJobFactory(submitter, scratchDir, NJOBS, conf, startFlag,
+            userResolver);
       return factory;
     }
   }
@@ -305,17 +315,21 @@ public class TestGridmixSubmission {
   @Test
   public void testSubmit() throws Exception {
     final Path in = new Path("foo").makeQualified(dfs);
-    final Path out = new Path("/gridmix").makeQualified(dfs);
+    final Path out = DEST.makeQualified(dfs);
     final String[] argv = {
       "-D" + FilePool.GRIDMIX_MIN_FILE + "=0",
       "-D" + Gridmix.GRIDMIX_OUT_DIR + "=" + out,
+      "-D" + Gridmix.GRIDMIX_USR_RSV + "=" + EchoUserResolver.class.getName(),
       "-generate", String.valueOf(GENDATA) + "m",
       in.toString(),
       "-" // ignored by DebugGridmix
     };
     DebugGridmix client = new DebugGridmix();
     final Configuration conf = mrCluster.createJobConf();
-    //conf.setInt(Gridmix.GRIDMIX_KEY_LEN, 2);
+    // allow synthetic users to create home directories
+    final Path root = new Path("/user");
+    dfs.mkdirs(root, new FsPermission((short)0777));
+    dfs.setPermission(root, new FsPermission((short)0777));
     int res = ToolRunner.run(conf, client, argv);
     assertEquals("Client exited with nonzero status", 0, res);
     client.checkMonitor();

src/contrib/gridmix/src/test/org/apache/hadoop/mapred/gridmix/TestUserResolve.java

@@ -0,0 +1,97 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.mapred.gridmix;
+
+import java.io.IOException;
+import java.net.URI;
+
+import org.junit.BeforeClass;
+import org.junit.Test;
+import static org.junit.Assert.*;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.security.UnixUserGroupInformation;
+import org.apache.hadoop.security.UserGroupInformation;
+
+public class TestUserResolve {
+
+  static Path userlist;
+
+  @BeforeClass
+  public static void writeUserList() throws IOException {
+    final Configuration conf = new Configuration();
+    final FileSystem fs = FileSystem.getLocal(conf);
+    final Path wd = new Path(new Path(
+          System.getProperty("test.build.data", "/tmp")).makeQualified(fs),
+        "gridmixUserResolve");
+    userlist = new Path(wd, "users");
+    FSDataOutputStream out = null;
+    try {
+      out = fs.create(userlist, true);
+      out.writeBytes("user0,groupA,groupB,groupC\n");
+      out.writeBytes("user1,groupA,groupC\n");
+      out.writeBytes("user2,groupB\n");
+      out.writeBytes("user3,groupA,groupB,groupC\n");
+    } finally {
+      if (out != null) {
+        out.close();
+      }
+    }
+  }
+
+  @Test
+  public void testRoundRobinResolver() throws Exception {
+    final Configuration conf = new Configuration();
+    final UserResolver rslv = new RoundRobinUserResolver();
+
+    boolean fail = false;
+    try {
+      rslv.setTargetUsers(null, conf);
+    } catch (IOException e) {
+      fail = true;
+    }
+    assertTrue("User list required for RoundRobinUserResolver", fail);
+
+    rslv.setTargetUsers(new URI(userlist.toString()), conf);
+    assertEquals("user0", rslv.getTargetUgi("hfre0").getUserName());
+    assertEquals("user1", rslv.getTargetUgi("hfre1").getUserName());
+    assertEquals("user2", rslv.getTargetUgi("hfre2").getUserName());
+    assertEquals("user0", rslv.getTargetUgi("hfre0").getUserName());
+    assertEquals("user3", rslv.getTargetUgi("hfre3").getUserName());
+    assertEquals("user0", rslv.getTargetUgi("hfre0").getUserName());
+    assertEquals("user0", rslv.getTargetUgi("hfre4").getUserName());
+    assertArrayEquals(new String[] { "groupA", "groupB", "groupC" },
+        rslv.getTargetUgi("hfre0").getGroupNames());
+    assertArrayEquals(new String[] { "groupB" },
+        rslv.getTargetUgi("hfre2").getGroupNames());
+  }
+
+  @Test
+  public void testSubmitterResolver() throws Exception {
+    final Configuration conf = new Configuration();
+    final UserResolver rslv = new SubmitterUserResolver();
+    rslv.setTargetUsers(null, conf);
+    assertEquals(UnixUserGroupInformation.login(),
+        rslv.getTargetUgi((UserGroupInformation)null));
+  }
+
+}