hadoop-mapreduce-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ama...@apache.org
Subject svn commit: r1215141 [2/4] - in /hadoop/common/trunk/hadoop-mapreduce-project: ./ ivy/ src/contrib/gridmix/src/java/org/apache/hadoop/mapred/gridmix/ src/docs/src/documentation/content/xdocs/ src/test/mapred/org/apache/hadoop/tools/rumen/ src/tools/org...
Date Fri, 16 Dec 2011 14:21:00 GMT
Added: hadoop/common/trunk/hadoop-mapreduce-project/src/test/mapred/org/apache/hadoop/tools/rumen/TestRumenAnonymization.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-mapreduce-project/src/test/mapred/org/apache/hadoop/tools/rumen/TestRumenAnonymization.java?rev=1215141&view=auto
==============================================================================
--- hadoop/common/trunk/hadoop-mapreduce-project/src/test/mapred/org/apache/hadoop/tools/rumen/TestRumenAnonymization.java (added)
+++ hadoop/common/trunk/hadoop-mapreduce-project/src/test/mapred/org/apache/hadoop/tools/rumen/TestRumenAnonymization.java Fri Dec 16 14:20:58 2011
@@ -0,0 +1,1940 @@
+/**
+ * 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.tools.rumen;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.File;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
+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.io.WritableUtils;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.MiniMRCluster;
+import org.apache.hadoop.mapreduce.Cluster;
+import org.apache.hadoop.mapreduce.ID;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.JobID;
+import org.apache.hadoop.mapreduce.MRJobConfig;
+import org.apache.hadoop.mapreduce.MapReduceTestUtil;
+import org.apache.hadoop.mapreduce.TaskAttemptID;
+import org.apache.hadoop.mapreduce.TaskID;
+import org.apache.hadoop.mapreduce.TaskType;
+import org.apache.hadoop.mapreduce.jobhistory.JobHistory;
+import org.apache.hadoop.mapreduce.server.tasktracker.TTConfig;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.tools.rumen.anonymization.*;
+import org.apache.hadoop.tools.rumen.datatypes.*;
+import org.apache.hadoop.tools.rumen.datatypes.FileName.FileNameState;
+import org.apache.hadoop.tools.rumen.datatypes.NodeName.NodeNameState;
+import org.apache.hadoop.tools.rumen.datatypes.util.DefaultJobPropertiesParser;
+import org.apache.hadoop.tools.rumen.datatypes.util.MapReduceJobPropertiesParser;
+import org.apache.hadoop.tools.rumen.serializers.*;
+import org.apache.hadoop.tools.rumen.state.*;
+import org.apache.hadoop.util.Tool;
+import org.apache.hadoop.util.ToolRunner;
+import org.codehaus.jackson.JsonEncoding;
+import org.codehaus.jackson.JsonFactory;
+import org.codehaus.jackson.JsonGenerator;
+import org.codehaus.jackson.Version;
+import org.codehaus.jackson.annotate.JsonIgnore;
+import org.codehaus.jackson.map.JsonSerializer;
+import org.codehaus.jackson.map.ObjectMapper;
+import org.codehaus.jackson.map.module.SimpleModule;
+import org.junit.Test;
+import static org.junit.Assert.*;
+
+/**
+ * Tests Rumen output anonymization.
+ */
+@SuppressWarnings("deprecation")
+public class TestRumenAnonymization {
+  /**
+   * Test {@link UserName}, serialization and anonymization.
+   */
+  @Test
+  public void testUserNameSerialization() throws IOException {
+    JsonSerializer<?> defaultSerializer = new DefaultRumenSerializer();
+    
+    JsonSerializer<?> anonymizingSerializer = 
+      new DefaultAnonymizingRumenSerializer(new StatePool(), 
+                                            new Configuration());
+    // test username
+    UserName uname = new UserName("bob");
+    assertEquals("Username error!", "bob", uname.getValue());
+    
+    // test username serialization
+    //   test with no anonymization
+    //      test bob
+    testSerializer(new UserName("bob"), "bob", defaultSerializer);
+    //      test alice
+    testSerializer(new UserName("alice"), "alice", defaultSerializer);
+    
+    // test user-name serialization
+    //   test with anonymization
+    //      test bob
+    testSerializer(new UserName("bob"), "user0", anonymizingSerializer);
+    //      test alice
+    testSerializer(new UserName("alice"), "user1", anonymizingSerializer);
+  }
+  
+  /**
+   * Test {@link JobName}, serialization and anonymization.
+   */
+  @Test
+  public void testJobNameSerialization() throws IOException {
+    JsonSerializer<?> defaultSerializer = new DefaultRumenSerializer();
+    
+    JsonSerializer<?> anonymizingSerializer = 
+      new DefaultAnonymizingRumenSerializer(new StatePool(), 
+                                            new Configuration());
+    
+    // test jobname
+    JobName jname = new JobName("job-secret");
+    assertEquals("Jobname error!", "job-secret", jname.getValue());
+    
+    // test job-name serialization
+    //  test with no anonymization
+    //      test job1
+    testSerializer(new JobName("job-myjob"), "job-myjob", defaultSerializer);
+    //      test job2
+    testSerializer(new JobName("job-yourjob"), "job-yourjob", 
+                   defaultSerializer);
+    
+    // test job-name serialization
+    //   test with anonymization
+    //  test queue1
+    testSerializer(new JobName("secret-job-1"), "job0", anonymizingSerializer);
+    //      test queue2
+    testSerializer(new JobName("secret-job-2"), "job1", anonymizingSerializer);
+  }
+  
+  /**
+   * Test {@link QueueName}, serialization and anonymization.
+   */
+  @Test
+  public void testQueueNameSerialization() throws IOException {
+    JsonSerializer<?> defaultSerializer = new DefaultRumenSerializer();
+    
+    JsonSerializer<?> anonymizingSerializer = 
+      new DefaultAnonymizingRumenSerializer(new StatePool(), 
+                                            new Configuration());
+    
+    // test queuename
+    QueueName qname = new QueueName("queue-secret");
+    assertEquals("Queuename error!", "queue-secret", qname.getValue());
+    
+    // test queuename serialization
+    //  test with no anonymization
+    //      test queue1
+    testSerializer(new QueueName("project1-queue"), 
+                   "project1-queue", defaultSerializer);
+    //      test queue2
+    testSerializer(new QueueName("project2-queue"), 
+                   "project2-queue", defaultSerializer);
+    
+    // test queue-name serialization
+    //   test with anonymization
+    //  test queue1
+    testSerializer(new QueueName("project1-queue"), 
+                   "queue0", anonymizingSerializer);
+    //      test queue2
+    testSerializer(new QueueName("project2-queue"), 
+                   "queue1", anonymizingSerializer);
+  }
+  
+  /**
+   * Test {@link NodeName}.
+   */
+  @Test
+  public void testNodeNameDataType() throws IOException {
+    // test hostname
+    //   test only hostname
+    NodeName hname = new NodeName("host1.myorg.com");
+    assertNull("Expected missing rack name", hname.getRackName());
+    assertEquals("Hostname's test#1 hostname error!", 
+                 "host1.myorg.com", hname.getHostName());
+    assertEquals("Hostname test#1 error!", "host1.myorg.com", hname.getValue());
+    
+    //   test rack/hostname
+    hname = new NodeName("/rack1.myorg.com/host1.myorg.com");
+    assertEquals("Hostname's test#2 rackname error!", 
+                 "rack1.myorg.com", hname.getRackName());
+    assertEquals("Hostname test#2 hostname error!", 
+                 "host1.myorg.com", hname.getHostName());
+    assertEquals("Hostname test#2 error!", 
+                 "/rack1.myorg.com/host1.myorg.com", hname.getValue());
+    
+    //   test hostname and rackname separately
+    hname = new NodeName("rack1.myorg.com", "host1.myorg.com");
+    assertEquals("Hostname's test#3 rackname error!", 
+                 "rack1.myorg.com", hname.getRackName());
+    assertEquals("Hostname test#3 hostname error!", 
+                 "host1.myorg.com", hname.getHostName());
+    assertEquals("Hostname test#3 error!", 
+                 "/rack1.myorg.com/host1.myorg.com", hname.getValue());
+    
+    //   test hostname with no rackname
+    hname = new NodeName(null, "host1.myorg.com");
+    assertNull("Hostname's test#4 rackname error!", hname.getRackName());
+    assertEquals("Hostname test#4 hostname error!", 
+                 "host1.myorg.com", hname.getHostName());
+    assertEquals("Hostname test#4 error!", 
+                 "host1.myorg.com", hname.getValue());
+    
+    //  test rackname with no hostname
+    hname = new NodeName("rack1.myorg.com", null);
+    assertEquals("Hostname test#4 rackname error!", 
+                 "rack1.myorg.com", hname.getRackName());
+    assertNull("Hostname's test#5 hostname error!", hname.getHostName());
+    assertEquals("Hostname test#5 error!", 
+                 "rack1.myorg.com", hname.getValue());
+  }
+  
+  /**
+   * Test {@link NodeName} serialization.
+   */
+  @Test
+  public void testNodeNameDefaultSerialization() throws IOException {
+    JsonSerializer<?> defaultSerializer = new DefaultRumenSerializer();
+    
+    // test hostname serialization
+    //  test with no anonymization
+    //      test hostname
+    testSerializer(new NodeName("hostname.myorg.com"), "hostname.myorg.com",
+                   defaultSerializer);
+    //      test rack/hostname
+    testSerializer(new NodeName("/rackname.myorg.com/hostname.myorg.com"), 
+                   "/rackname.myorg.com/hostname.myorg.com",
+                   defaultSerializer);
+    //      test rack,hostname
+    testSerializer(new NodeName("rackname.myorg.com", "hostname.myorg.com"), 
+                   "/rackname.myorg.com/hostname.myorg.com",
+                   defaultSerializer);
+    //      test -,hostname
+    testSerializer(new NodeName(null, "hostname.myorg.com"), 
+                   "hostname.myorg.com", defaultSerializer);
+    //      test rack,-
+    testSerializer(new NodeName("rackname.myorg.com", null), 
+                   "rackname.myorg.com", defaultSerializer);
+  }
+  
+  /**
+   * Test {@link NodeName} anonymization.
+   */
+  @Test
+  public void testNodeNameAnonymization() throws IOException {
+    JsonSerializer<?> anonymizingSerializer = 
+      new DefaultAnonymizingRumenSerializer(new StatePool(), 
+                                            new Configuration());
+    
+    // test hostname serializer
+    //   test with anonymization
+    //      test hostname
+    testSerializer(new NodeName("hostname.myorg.com"), "host0",
+                   anonymizingSerializer);
+    //      test hostname reuse
+    testSerializer(new NodeName("hostname213.myorg.com"), "host1",
+                   anonymizingSerializer);
+    //      test rack/hostname
+    testSerializer(new NodeName("/rackname.myorg.com/hostname.myorg.com"), 
+                   "/rack0/host0", anonymizingSerializer);
+    //  test rack/hostname (hostname reuse)
+    testSerializer(new NodeName("/rackname654.myorg.com/hostname.myorg.com"), 
+                   "/rack1/host0", anonymizingSerializer);
+    //  test rack/hostname (rack reuse)
+    testSerializer(new NodeName("/rackname654.myorg.com/hostname765.myorg.com"), 
+                   "/rack1/host2", anonymizingSerializer);
+    //  test rack,hostname (rack & hostname reuse)
+    testSerializer(new NodeName("rackname.myorg.com", "hostname.myorg.com"), 
+                   "/rack0/host0", anonymizingSerializer);
+    //      test rack,hostname (rack reuse)
+    testSerializer(new NodeName("rackname.myorg.com", "hostname543.myorg.com"), 
+                   "/rack0/host3", anonymizingSerializer);
+    //      test rack,hostname (hostname reuse)
+    testSerializer(new NodeName("rackname987.myorg.com", "hostname.myorg.com"), 
+                   "/rack2/host0", anonymizingSerializer);
+    //      test rack,hostname (rack reuse)
+    testSerializer(new NodeName("rackname.myorg.com", "hostname654.myorg.com"), 
+                   "/rack0/host4", anonymizingSerializer);
+    //      test rack,hostname (host reuse)
+    testSerializer(new NodeName("rackname876.myorg.com", "hostname.myorg.com"), 
+                   "/rack3/host0", anonymizingSerializer);
+    //      test rack,hostname (rack & hostname reuse)
+    testSerializer(new NodeName("rackname987.myorg.com", 
+                                "hostname543.myorg.com"), 
+                   "/rack2/host3", anonymizingSerializer);
+    //      test -,hostname (hostname reuse)
+    testSerializer(new NodeName(null, "hostname.myorg.com"), 
+                   "host0", anonymizingSerializer);
+    //      test -,hostname 
+    testSerializer(new NodeName(null, "hostname15.myorg.com"), 
+                   "host5", anonymizingSerializer);
+    //      test rack,- (rack reuse)
+    testSerializer(new NodeName("rackname987.myorg.com", null), 
+                   "rack2", anonymizingSerializer);
+    //      test rack,- 
+    testSerializer(new NodeName("rackname15.myorg.com", null), 
+                   "rack4", anonymizingSerializer);
+  }
+  
+  /**
+   * Test {@link JobProperties}.
+   */
+  @Test
+  public void testJobPropertiesDataType() throws IOException {
+    // test job properties
+    Properties properties = new Properties();
+    JobProperties jp = new JobProperties(properties);
+    
+    // test empty job properties
+    assertEquals("Job Properties (default) store error", 
+                 0, jp.getValue().size());
+    // test by adding some data
+    properties.put("test-key", "test-value"); // user config
+    properties.put(MRJobConfig.USER_NAME, "bob"); // job config
+    properties.put(JobConf.MAPRED_TASK_JAVA_OPTS, "-Xmx1G"); // deprecated
+    jp = new JobProperties(properties);
+    assertEquals("Job Properties (default) store error", 
+                 3, jp.getValue().size());
+    assertEquals("Job Properties (default) key#1 error", 
+                 "test-value", jp.getValue().get("test-key"));
+    assertEquals("Job Properties (default) key#2 error", 
+                 "bob", jp.getValue().get(MRJobConfig.USER_NAME));
+    assertEquals("Job Properties (default) key#3 error", 
+                 "-Xmx1G", jp.getValue().get(JobConf.MAPRED_TASK_JAVA_OPTS));
+  }
+  
+  /**
+   * Test {@link JobProperties} serialization.
+   */
+  @Test
+  public void testJobPropertiesSerialization() throws IOException {
+    JsonSerializer<?> defaultSerializer = new DefaultRumenSerializer();
+    
+    // test job properties
+    Properties properties = new Properties();
+    properties.put("test-key", "test-value"); // user config
+    properties.put(MRJobConfig.USER_NAME, "bob"); // job config
+    properties.put(JobConf.MAPRED_TASK_JAVA_OPTS, "-Xmx1G"); // deprecated
+    JobProperties jp = new JobProperties(properties);
+    
+    testSerializer(jp, "{test-key:test-value," 
+                       + "mapreduce.job.user.name:bob," 
+                       + "mapred.child.java.opts:-Xmx1G}", defaultSerializer);
+  }
+  
+  /**
+   * Test {@link JobProperties} anonymization.
+   */
+  @Test
+  public void testJobPropertiesAnonymization() throws IOException {
+    // test job properties
+    Properties properties = new Properties();
+    Configuration conf = new Configuration();
+    
+    properties.put("test-key", "test-value"); // user config
+    properties.put(MRJobConfig.USER_NAME, "bob"); // job config
+    // deprecated
+    properties.put("mapred.map.child.java.opts", 
+                   "-Xmx2G -Xms500m -Dsecret=secret");
+    // deprecated and not supported
+    properties.put(JobConf.MAPRED_TASK_JAVA_OPTS, 
+                   "-Xmx1G -Xms200m -Dsecret=secret");
+    JobProperties jp = new JobProperties(properties);
+    
+    // define a module
+    SimpleModule module = new SimpleModule("Test Anonymization Serializer",  
+                                           new Version(0, 0, 0, "TEST"));
+    // add various serializers to the module
+    module.addSerializer(DataType.class, new DefaultRumenSerializer());
+    module.addSerializer(AnonymizableDataType.class, 
+                         new DefaultAnonymizingRumenSerializer(new StatePool(),
+                                                               conf));
+    
+    //TODO Support deprecated and un-supported keys
+    testSerializer(jp, "{mapreduce.job.user.name:user0," 
+                       + "mapred.map.child.java.opts:-Xmx2G -Xms500m}", module);
+  }
+  
+  /**
+   * Test {@link ClassName}, serialization and anonymization.
+   */
+  @Test
+  public void testClassNameSerialization() throws IOException {
+    JsonSerializer<?> defaultSerializer = new DefaultRumenSerializer();
+    
+    JsonSerializer<?> anonymizingSerializer = 
+      new DefaultAnonymizingRumenSerializer(new StatePool(), 
+                                            new Configuration());
+    
+    // test classname
+    ClassName cName = new ClassName(TestRumenAnonymization.class.getName());
+    assertEquals("Classname error!", TestRumenAnonymization.class.getName(), 
+                 cName.getValue());
+    
+    // test classname serialization
+    //  test with no anonymization
+    //      test class1
+    testSerializer(new ClassName("org.apache.hadoop.Test"), 
+                   "org.apache.hadoop.Test", defaultSerializer);
+    //      test class2
+    testSerializer(new ClassName("org.apache.hadoop.Test2"), 
+                   "org.apache.hadoop.Test2", defaultSerializer);
+    
+    // test class-name serialization
+    //  test with anonymization
+    //      test class1
+    testSerializer(new ClassName("org.apache.hadoop.Test1"), 
+                   "class0", anonymizingSerializer);
+    //      test class2
+    testSerializer(new ClassName("org.apache.hadoop.Test2"), 
+                   "class1", anonymizingSerializer);
+    
+    // test classnames with preserves
+    Configuration conf = new Configuration();
+    conf.set(ClassName.CLASSNAME_PRESERVE_CONFIG, "org.apache.hadoop.");
+    anonymizingSerializer = 
+      new DefaultAnonymizingRumenSerializer(new StatePool(), conf);
+    // test word with prefix
+    testSerializer(new ClassName("org.apache.hadoop.Test3"), 
+                   "org.apache.hadoop.Test3", anonymizingSerializer);
+    // test word without prefix
+    testSerializer(new ClassName("org.apache.hadoop2.Test4"), 
+                   "class0", anonymizingSerializer);
+  }
+  
+  /**
+   * Test {@link FileName}.
+   */
+  @Test
+  public void testFileName() throws IOException {
+    // test file on hdfs
+    FileName hFile = new FileName("hdfs://testnn:123/user/test.json");
+    assertEquals("Filename error!", "hdfs://testnn:123/user/test.json", 
+                 hFile.getValue());
+    // test file on local-fs
+    hFile = new FileName("file:///user/test.json");
+    assertEquals("Filename error!", "file:///user/test.json", 
+                 hFile.getValue());
+    // test dir on hdfs
+    hFile = new FileName("hdfs://testnn:123/user/");
+    assertEquals("Filename error!", "hdfs://testnn:123/user/",
+                 hFile.getValue());
+    // test dir on local-fs
+    hFile = new FileName("file:///user/");
+    assertEquals("Filename error!", "file:///user/", hFile.getValue());
+    // test absolute file
+    hFile = new FileName("/user/test/test.json");
+    assertEquals("Filename error!", "/user/test/test.json", hFile.getValue());
+    // test absolute directory
+    hFile = new FileName("/user/test/");
+    assertEquals("Filename error!", "/user/test/", hFile.getValue());
+    // test relative file
+    hFile = new FileName("user/test/test2.json");
+    assertEquals("Filename error!", "user/test/test2.json", hFile.getValue());
+    // test relative directory
+    hFile = new FileName("user/test/");
+    assertEquals("Filename error!", "user/test/", hFile.getValue());
+    // test absolute file
+    hFile = new FileName("user");
+    assertEquals("Filename error!", "user", hFile.getValue());
+    // test absolute directory
+    hFile = new FileName("user/");
+    assertEquals("Filename error!", "user/", hFile.getValue());
+    hFile = new FileName("./tmp");
+    assertEquals("Filename error!","./tmp", hFile.getValue());
+    hFile = new FileName("./tmp/");
+    assertEquals("Filename error!","./tmp/", hFile.getValue());
+    hFile = new FileName("../tmp");
+    assertEquals("Filename error!","../tmp", hFile.getValue());
+    hFile = new FileName("../tmp/");
+    assertEquals("Filename error!","../tmp/", hFile.getValue());
+    
+    // test comma separated filenames
+    //  test hdfs filenames, absolute and local-fs filenames
+    hFile = new FileName("hdfs://testnn:123/user/test1," 
+                         + "file:///user/test2,/user/test3");
+    assertEquals("Filename error!", 
+                 "hdfs://testnn:123/user/test1,file:///user/test2,/user/test3", 
+                 hFile.getValue());
+  }
+  
+  /**
+   * Test {@link FileName} serialization.
+   */
+  @Test
+  public void testFileNameSerialization() throws IOException {
+    JsonSerializer<?> defaultSerializer = new DefaultRumenSerializer();
+    
+    // test filename serialization
+    //  test with no anonymization
+    //      test a file on hdfs
+    testSerializer(new FileName("hdfs://mynn:123/home/user/test.json"), 
+                   "hdfs://mynn:123/home/user/test.json", defaultSerializer);
+    // test a file on local-fs
+    testSerializer(new FileName("file:///home/user/test.json"), 
+                   "file:///home/user/test.json", defaultSerializer);
+    // test directory on hdfs
+    testSerializer(new FileName("hdfs://mynn:123/home/user/"), 
+                   "hdfs://mynn:123/home/user/", defaultSerializer);
+    // test directory on local fs
+    testSerializer(new FileName("file:///home/user/"), 
+                   "file:///home/user/", defaultSerializer);
+    // test absolute file
+    testSerializer(new FileName("/home/user/test.json"), 
+                   "/home/user/test.json", defaultSerializer);
+    // test relative file
+    testSerializer(new FileName("home/user/test.json"), 
+                   "home/user/test.json", defaultSerializer);
+    // test absolute folder
+    testSerializer(new FileName("/home/user/"), "/home/user/", 
+                   defaultSerializer);
+    // test relative folder
+    testSerializer(new FileName("home/user/"), "home/user/", 
+                   defaultSerializer);
+    // relative file
+    testSerializer(new FileName("home"), "home", defaultSerializer);
+    // relative folder
+    testSerializer(new FileName("home/"), "home/", defaultSerializer);
+    // absolute file
+    testSerializer(new FileName("/home"), "/home", defaultSerializer);
+    // absolute folder
+    testSerializer(new FileName("/home/"), "/home/", defaultSerializer);
+    // relative folder
+    testSerializer(new FileName("./tmp"), "./tmp", defaultSerializer);
+    testSerializer(new FileName("./tmp/"), "./tmp/", defaultSerializer);
+    testSerializer(new FileName("../tmp"), "../tmp", defaultSerializer);
+    
+    // test comma separated filenames
+    //  test hdfs filenames, absolute and local-fs filenames
+    FileName fileName = 
+      new FileName("hdfs://testnn:123/user/test1,file:///user/test2,"
+                   + "/user/test3");
+    testSerializer(fileName, 
+        "hdfs://testnn:123/user/test1,file:///user/test2,/user/test3",
+        defaultSerializer);
+  }
+  
+  /**
+   * Test {@link FileName} anonymization.
+   */
+  @Test
+  public void testFileNameAnonymization() throws IOException {
+    JsonSerializer<?> anonymizingSerializer = 
+      new DefaultAnonymizingRumenSerializer(new StatePool(), 
+                                            new Configuration());
+    
+    // test filename serialization
+    //  test with no anonymization
+    //      test hdfs file
+    testSerializer(new FileName("hdfs://mynn:123/home/user/bob/test.json"),
+        "hdfs://host0/home/user/dir0/test.json", anonymizingSerializer);
+    //      test local-fs file
+    testSerializer(new FileName("file:///home/user/alice/test.jar"), 
+        "file:///home/user/dir1/test.jar", anonymizingSerializer);
+    //      test hdfs dir
+    testSerializer(new FileName("hdfs://mynn:123/home/user/"),
+                   "hdfs://host0/home/user/", anonymizingSerializer);
+    //      test local-fs dir
+    testSerializer(new FileName("file:///home/user/secret/more-secret/"), 
+                   "file:///home/user/dir2/dir3/", anonymizingSerializer);
+    //  test absolute filenames
+    testSerializer(new FileName("/home/user/top-secret.txt"),
+                   "/home/user/file0.txt", anonymizingSerializer);
+    //      test relative filenames
+    testSerializer(new FileName("home/user/top-top-secret.zip"), 
+                   "home/user/file1.zip", anonymizingSerializer);
+    //  test absolute dirnames
+    testSerializer(new FileName("/home/user/project1/"),
+                   "/home/user/dir4/", anonymizingSerializer);
+    //      test relative filenames
+    testSerializer(new FileName("home/user/project1"), 
+                   "home/user/file2", anonymizingSerializer);
+    //  test absolute dirnames (re-use)
+    testSerializer(new FileName("more-secret/"),
+                   "dir3/", anonymizingSerializer);
+    //      test relative filenames (re-use)
+    testSerializer(new FileName("secret/project1"), 
+                   "dir2/file2", anonymizingSerializer);
+    //  test absolute filenames (re-use)
+    testSerializer(new FileName("/top-secret.txt"),
+                   "/file0.txt", anonymizingSerializer);
+    //  test relative filenames (re-use)
+    testSerializer(new FileName("top-top-secret.tar"), 
+                   "file1.tar", anonymizingSerializer);
+    //  test absolute dirname
+    testSerializer(new FileName("sensitive-projectname/"),
+                   "dir5/", anonymizingSerializer);
+    //  test relative filenames 
+    testSerializer(new FileName("/real-sensitive-projectname/"), 
+                   "/dir6/", anonymizingSerializer);
+    //  test absolute filenames 
+    testSerializer(new FileName("/usernames.xml"),
+                   "/file3.xml", anonymizingSerializer);
+    //  test relative filenames 
+    testSerializer(new FileName("passwords.zip"), 
+                   "file4.zip", anonymizingSerializer);
+    //  test relative filenames 
+    testSerializer(new FileName("./tmp"), 
+                   "./tmp", anonymizingSerializer);
+    testSerializer(new FileName("./tmp/"), 
+                   "./tmp/", anonymizingSerializer);
+    testSerializer(new FileName("../tmp"), 
+                   "../tmp", anonymizingSerializer);
+    testSerializer(new FileName("../tmp/"), 
+                   "../tmp/", anonymizingSerializer);
+    
+    // test comma separated filenames
+    //  test hdfs filenames, absolute and local-fs filenames
+    FileName fileName = 
+      new FileName("hdfs://mynn:123/home/user/bob/test.json," 
+                   + "file:///home/user/bob/test.json,/user/alice/test.json");
+    testSerializer(fileName, 
+        "hdfs://host0/home/user/dir0/test.json,file:///home/user/dir0/test.json"
+        + ",/user/dir1/test.json",
+        anonymizingSerializer);
+  }
+  
+  
+  /**
+   * Test {@link DefaultDataType} serialization.
+   */
+  @Test
+  public void testDefaultDataTypeSerialization() throws IOException {
+    JsonSerializer<?> defaultSerializer = new DefaultRumenSerializer();
+    
+    // test default data-type
+    DefaultDataType dt = new DefaultDataType("test");
+    assertEquals("DefaultDataType error!", "test", dt.getValue());
+    
+    // test default data-type
+    //  test with no anonymization
+    //      test data
+    testSerializer(new DefaultDataType("test"), "test", defaultSerializer);
+  }
+  
+  // A faked OutputStream which stores the stream content into a StringBuffer.
+  private static class MyOutputStream extends OutputStream {
+    private StringBuffer data = new StringBuffer();
+    
+    @Override
+    public void write(int b) throws IOException {
+      data.append((char)b);
+    }
+    
+    @Override
+    public void write(byte[] b) throws IOException {
+      data.append(b);
+    }
+    
+    @Override
+    public String toString() {
+      // remove all the '"' for ease of testing
+      return data.toString().trim().replaceAll("\"", "");
+    }
+  }
+  
+  // tests the object serializing using the class of the specified object
+  @SuppressWarnings("unchecked")
+  private static void testSerializer(Object toBeSerialized, String expData, 
+                                     JsonSerializer serializer) 
+  throws IOException {
+    // define a module
+    SimpleModule module = new SimpleModule("Test Anonymization Serializer",  
+                                           new Version(0, 0, 0, "TEST"));
+    // add various serializers to the module
+    module.addSerializer(toBeSerialized.getClass(), serializer);
+    testSerializer(toBeSerialized, expData, module);
+  }
+  
+  // tests the object serializing using the specified class
+  private static void testSerializer(Object toBeSerialized, String expData, 
+                                     SimpleModule module) 
+  throws IOException {
+    // define a custom generator
+    ObjectMapper outMapper = new ObjectMapper();
+    
+    // register the module
+    outMapper.registerModule(module);
+    
+    // get the json factory
+    JsonFactory outFactory = outMapper.getJsonFactory();
+    // define a fake output stream which will cache the data
+    MyOutputStream output = new MyOutputStream();
+    // define the json output generator
+    JsonGenerator outGen = 
+      outFactory.createJsonGenerator(output, JsonEncoding.UTF8);
+    
+    // serialize the object
+    outGen.writeObject(toBeSerialized);
+    //serializer.serialize(toBeSerialized, outGen, null);
+    
+    // close the json generator so that it flushes out the data to the output
+    // stream
+    outGen.close();
+    
+    assertEquals("Serialization failed!", expData, output.toString());
+  }
+  
+  /**
+   * Test {@link DefaultRumenSerializer}.
+   */
+  @Test
+  public void testDefaultDataSerializers() throws Exception {
+    JsonSerializer<?> defaultSer = new DefaultRumenSerializer();
+    // test default data-type
+    //  test with no anonymization
+    //      test data
+    testSerializer(new DefaultDataType("test"), "test", defaultSer);
+  }
+  
+  @Test
+  public void testBlockingDataSerializers() throws Exception {
+    JsonSerializer<?> blockingSerializer = new BlockingSerializer();
+    
+    // test string serializer
+    testSerializer("username:password", "null", blockingSerializer);
+  }
+  
+  @Test
+  public void testObjectStringDataSerializers() throws Exception {
+    JsonSerializer<?> objectStringSerializer = new ObjectStringSerializer<ID>();
+    // test job/task/attempt id serializer
+    //   test job-id 
+    JobID jid = JobID.forName("job_1_1");
+    testSerializer(jid, jid.toString(), objectStringSerializer);
+    //   test task-id
+    TaskID tid = new TaskID(jid, TaskType.MAP, 1);
+    testSerializer(tid, tid.toString(), objectStringSerializer);
+    //   test attempt-id
+    TaskAttemptID aid = new TaskAttemptID(tid, 0);
+    testSerializer(aid, aid.toString(), objectStringSerializer);
+  }
+  
+  // test anonymizer
+  @Test
+  public void testRumenAnonymization() throws Exception {
+    Configuration conf = new Configuration();
+
+    // Run a MR job
+    // create a MR cluster
+    conf.setInt(TTConfig.TT_MAP_SLOTS, 1);
+    conf.setInt(TTConfig.TT_REDUCE_SLOTS, 1);
+    
+    MiniDFSCluster dfsCluster = null;
+    MiniMRCluster mrCluster =  null;
+    
+    // local filesystem for running TraceBuilder
+    FileSystem lfs = FileSystem.getLocal(conf);
+    Path rootTempDir =
+      new Path(System.getProperty("test.build.data", "/tmp"));
+
+    Path tempDir = new Path(rootTempDir, "testRumenAnonymization");
+    tempDir = lfs.makeQualified(tempDir);
+    lfs.delete(tempDir, true);
+    
+    try {
+      dfsCluster = new MiniDFSCluster(conf, 1, true, null);
+      String[] racks = new String[] {"/rack123.myorg.com", 
+                                     "/rack456.myorg.com"};
+      String[] hosts = new String[] {"host1230.myorg.com", 
+                                     "host4560.myorg.com"};
+      mrCluster = 
+        new MiniMRCluster(2, dfsCluster.getFileSystem().getUri().toString(), 
+                          1, racks, hosts, new JobConf(conf));
+
+      // run a job
+      Path inDir = new Path("secret-input");
+      Path outDir = new Path("secret-output");
+
+      JobConf jConf = mrCluster.createJobConf();
+      // add some usr sensitive data in the job conf
+      jConf.set("user-secret-code", "abracadabra");
+      
+      jConf.setJobName("top-secret");
+      // construct a job with 1 map and 1 reduce task.
+      Job job = MapReduceTestUtil.createJob(jConf, inDir, outDir, 2, 2);
+      // wait for the job to complete
+      job.waitForCompletion(false);
+
+      assertTrue("Job failed", job.isSuccessful());
+
+      JobID id = job.getJobID();
+      Cluster cluster = new Cluster(jConf);
+      String user = cluster.getAllJobStatuses()[0].getUsername();
+
+      // get the jobhistory filepath
+      Path jhPath = 
+        new Path(mrCluster.getJobTrackerRunner().getJobTracker()
+                          .getJobHistoryDir());
+      Path inputLogPath = JobHistory.getJobHistoryFile(jhPath, id, user);
+      Path inputConfPath = JobHistory.getConfFile(jhPath, id);
+      // wait for 10 secs for the jobhistory file to move into the done folder
+      FileSystem fs = inputLogPath.getFileSystem(jConf);
+      for (int i = 0; i < 100; ++i) {
+        if (fs.exists(inputLogPath)) {
+          break;
+        }
+        TimeUnit.MILLISECONDS.wait(100);
+      }
+
+      assertTrue("Missing job history file", fs.exists(inputLogPath));
+
+      // run trace builder on the job history logs
+      Path goldTraceFilename = new Path(tempDir, "trace.json");
+      Path goldTopologyFilename = new Path(tempDir, "topology.json");
+
+      // build the trace-builder command line args
+      String[] args = new String[] {goldTraceFilename.toString(), 
+                                    goldTopologyFilename.toString(),
+                                    inputLogPath.toString(),
+                                    inputConfPath.toString()};
+      Tool analyzer = new TraceBuilder();
+      int result = ToolRunner.run(analyzer, args);
+      assertEquals("Non-zero exit", 0, result);
+
+      // anonymize the job trace
+      Path anonymizedTraceFilename = new Path(tempDir, "trace-anonymized.json");
+      Path anonymizedClusterTopologyFilename = 
+        new Path(tempDir, "topology-anonymized.json");
+      args = new String[] {"-trace", goldTraceFilename.toString(), 
+                           anonymizedTraceFilename.toString(),
+                           "-topology", goldTopologyFilename.toString(), 
+                           anonymizedClusterTopologyFilename.toString()};
+      Tool anonymizer = new Anonymizer();
+      result = ToolRunner.run(anonymizer, args);
+      assertEquals("Non-zero exit", 0, result);
+
+      JobTraceReader reader = new JobTraceReader(anonymizedTraceFilename, conf);
+      LoggedJob anonymizedJob = reader.getNext();
+      reader.close(); // close the reader as we need only 1 job
+      // test
+      //   user-name
+      String currentUser = UserGroupInformation.getCurrentUser().getUserName();
+      assertFalse("Username not anonymized!", 
+                  currentUser.equals(anonymizedJob.getUser().getValue()));
+      //   jobid
+      assertEquals("JobID mismatch!", 
+                   id.toString(), anonymizedJob.getJobID().toString());
+      //   queue-name
+      assertFalse("Queuename mismatch!", 
+                  "default".equals(anonymizedJob.getQueue().getValue()));
+      //   job-name
+      assertFalse("Jobname mismatch!", 
+                  "top-secret".equals(anonymizedJob.getJobName().getValue()));
+      
+      //   job properties
+      for (Map.Entry<Object, Object> entry : 
+           anonymizedJob.getJobProperties().getValue().entrySet()) {
+        assertFalse("User sensitive configuration key not anonymized", 
+                    entry.getKey().toString().equals("user-secret-code"));
+        assertFalse("User sensitive data not anonymized", 
+                    entry.getValue().toString().contains(currentUser));
+        assertFalse("User sensitive data not anonymized", 
+                    entry.getValue().toString().contains("secret"));
+      }
+      
+      // test map tasks
+      testTasks(anonymizedJob.getMapTasks(), id, TaskType.MAP);
+      
+      // test reduce tasks
+      testTasks(anonymizedJob.getReduceTasks(), id, TaskType.REDUCE);
+      
+      // test other tasks
+      testTasks(anonymizedJob.getOtherTasks(), id, null);
+
+      // test the anonymized cluster topology file
+      ClusterTopologyReader cReader = 
+        new ClusterTopologyReader(anonymizedClusterTopologyFilename, conf);
+      LoggedNetworkTopology loggedNetworkTopology = cReader.get();
+      // test the cluster topology
+      testClusterTopology(loggedNetworkTopology, 0, "myorg");
+    } finally {
+      // shutdown and cleanup
+      if (mrCluster != null) {
+        mrCluster.shutdown();
+      }
+      
+      if (dfsCluster != null) {
+        dfsCluster.formatDataNodeDirs();
+        dfsCluster.shutdown();
+      }
+      lfs.delete(tempDir, true);
+    }
+  }
+  
+  // test task level details lije
+  //   - taskid
+  //   - locality info
+  //   - attempt details
+  //     - attempt execution hostname
+  private static void testTasks(List<LoggedTask> tasks, JobID id, 
+                                TaskType type) {
+    int index = 0;
+    for (LoggedTask task : tasks) {
+      // generate the expected task id for this task
+      if (type != null) {
+        TaskID tid = new TaskID(id, type, index++);
+        assertEquals("TaskID mismatch!", 
+                     tid.toString(), task.getTaskID().toString());
+      }
+
+      // check locality information
+      if (task.getPreferredLocations() != null) {
+        for (LoggedLocation loc : task.getPreferredLocations()) {
+          for (NodeName name : loc.getLayers()) {
+            assertFalse("Hostname mismatch!", 
+                        name.getValue().contains("myorg"));
+          }
+        }
+      }
+      
+      // check execution host
+      for (LoggedTaskAttempt attempt : task.getAttempts()) {
+        // generate the expected task id for this task
+        TaskAttemptID aid = new TaskAttemptID(task.getTaskID(), 0);
+        assertEquals("TaskAttemptID mismatch!", 
+                     aid.toString(), attempt.getAttemptID().toString());
+
+        assertNotNull("Hostname null!", attempt.getHostName());
+        assertFalse("Hostname mismatch!", 
+                    attempt.getHostName().getValue().contains("myorg"));
+      }
+    }
+  }
+  
+  // tests the logged network topology
+  private static void testClusterTopology(LoggedNetworkTopology topology, 
+                                          int level, String bannedString) {
+    assertFalse("Cluster topology test failed!", 
+                topology.getName().getValue().contains(bannedString));
+    if (level == 0) {
+      assertEquals("Level-1 data mismatch!", 
+                   "<root>", topology.getName().getValue());
+    } else if (level == 1) {
+      assertTrue("Level-2 data mismatch!", 
+                 topology.getName().getValue().contains("rack"));
+      assertFalse("Level-2 data mismatch!", 
+                 topology.getName().getValue().contains("host"));
+    } else {
+      assertTrue("Level-2 data mismatch!", 
+                 topology.getName().getValue().contains("host"));
+      assertFalse("Level-2 data mismatch!", 
+                  topology.getName().getValue().contains("rack"));
+    }
+    
+    // if the current node is a rack, then test the nodes under it
+    if (topology.getChildren() != null) {
+      for (LoggedNetworkTopology child : topology.getChildren()) {
+        testClusterTopology(child, level + 1, bannedString);
+      }
+    }
+  }
+  
+  @Test
+  public void testCLI() throws Exception {
+    Configuration conf = new Configuration();
+    FileSystem lfs = FileSystem.getLocal(conf);
+    Path rootTempDir =
+      new Path(System.getProperty("test.build.data", "/tmp"));
+
+    Path tempDir = new Path(rootTempDir, "testCLI");
+    tempDir = lfs.makeQualified(tempDir);
+    lfs.delete(tempDir, true);
+    
+    // test no args
+    String[] args = new String[] {};
+    testAnonymizerCLI(args, -1);
+    
+    // test with wrong args
+    args = new String[] {"-test"};
+    testAnonymizerCLI(args, -1);
+    
+    args = new String[] {"-trace"};
+    testAnonymizerCLI(args, -1);
+    
+    args = new String[] {"-topology"};
+    testAnonymizerCLI(args, -1);
+    
+    args = new String[] {"-trace -topology"};
+    testAnonymizerCLI(args, -1);
+    
+    Path testTraceInputFilename = new Path(tempDir, "trace-in.json");
+    args = new String[] {"-trace", testTraceInputFilename.toString()};
+    testAnonymizerCLI(args, -1);
+    
+    Path testTraceOutputFilename = new Path(tempDir, "trace-out.json");
+    args = new String[] {"-trace", testTraceInputFilename.toString(), 
+                         testTraceOutputFilename.toString()};
+    testAnonymizerCLI(args, -1);
+    
+    OutputStream out = lfs.create(testTraceInputFilename);
+    out.write("{\n}".getBytes());
+    out.close();
+    args = new String[] {"-trace", testTraceInputFilename.toString(), 
+                         testTraceOutputFilename.toString()};
+    testAnonymizerCLI(args, 0);
+    
+    Path testToplogyInputFilename = new Path(tempDir, "topology-in.json");
+    args = new String[] {"-topology", testToplogyInputFilename.toString()};
+    testAnonymizerCLI(args, -1);
+    
+    Path testTopologyOutputFilename = new Path(tempDir, "topology-out.json");
+    args = new String[] {"-topology", testToplogyInputFilename.toString(), 
+                         testTopologyOutputFilename.toString()};
+    testAnonymizerCLI(args, -1);
+    
+    out = lfs.create(testToplogyInputFilename);
+    out.write("{\n}".getBytes());
+    out.close();
+    args = new String[] {"-topology", testToplogyInputFilename.toString(), 
+                         testTopologyOutputFilename.toString()};
+    testAnonymizerCLI(args, 0);
+    
+    args = new String[] {"-trace", testTraceInputFilename.toString(), 
+                         "-topology", testToplogyInputFilename.toString()};
+    testAnonymizerCLI(args, -1);
+
+    args = new String[] {"-trace", testTraceInputFilename.toString(), 
+                         testTraceOutputFilename.toString(),
+                         "-topology", testToplogyInputFilename.toString(), 
+                         testTopologyOutputFilename.toString()};
+    testAnonymizerCLI(args, 0);
+  }
+  
+  // tests the Anonymizer CLI via the Tools interface
+  private static void testAnonymizerCLI(String[] args, int eExitCode) 
+  throws Exception {
+    Anonymizer anonymizer = new Anonymizer();
+    
+    int exitCode = ToolRunner.run(anonymizer, args);
+    assertEquals("Exit code mismatch", eExitCode, exitCode);
+  }
+  
+  /**
+   * Test {@link StatePool}'s reload and persistence feature.
+   */
+  @Test
+  public void testStatePool() throws Exception {
+    Configuration conf = new Configuration();
+    FileSystem lfs = FileSystem.getLocal(conf);
+    Path rootTempDir =
+      new Path(System.getProperty("test.build.data", "/tmp"));
+
+    Path tempDir = new Path(rootTempDir, "testStatePool");
+    tempDir = lfs.makeQualified(tempDir);
+    lfs.delete(tempDir, true);
+    
+    // set the state dir
+    conf.set(StatePool.DIR_CONFIG, tempDir.toString());
+    
+    StatePool pool = new StatePool();
+    // test reload, persist and dir config
+    //   test with no reload and persist
+    pool.initialize(conf);
+    
+    //  test with reload and/or persist enabled with no dir
+    assertNull("Default state pool error", 
+               pool.getState(MyState.class));
+    
+    // try persisting 
+    pool.persist();
+    assertFalse("State pool persisted when disabled", lfs.exists(tempDir));
+    
+    // test wrongly configured state-pool
+    conf.setBoolean(StatePool.RELOAD_CONFIG, true);
+    conf.unset(StatePool.DIR_CONFIG);
+    pool = new StatePool();
+    boolean success = true;
+    try {
+      pool.initialize(conf);
+    } catch (Exception e) {
+      success = false;
+    }
+    assertFalse("State pool bad configuration succeeded", success);
+    
+    // test wrongly configured state-pool
+    conf.setBoolean(StatePool.RELOAD_CONFIG, false);
+    conf.setBoolean(StatePool.PERSIST_CONFIG, true);
+    pool = new StatePool();
+    success = true;
+    try {
+      pool.initialize(conf);
+    } catch (Exception e) {
+      success = false;
+    }
+    assertFalse("State manager bad configuration succeeded", success);
+    
+    
+    // test persistence
+    conf.setBoolean(StatePool.RELOAD_CONFIG, false);
+    conf.set(StatePool.DIR_CONFIG, tempDir.toString());
+    conf.setBoolean(StatePool.PERSIST_CONFIG, true);
+    pool = new StatePool();
+    pool.initialize(conf);
+    
+    // add states to the state pool
+    MyState myState = new MyState();
+    pool.addState(MyState.class, myState);
+    myState.setState("test-1");
+    // try persisting 
+    pool.persist();
+    assertTrue("State pool persisted when enabled", lfs.exists(tempDir));
+    assertEquals("State pool persisted when enabled", 
+                 1, lfs.listStatus(tempDir).length);
+    
+    // reload
+    conf.setBoolean(StatePool.RELOAD_CONFIG, true);
+    conf.set(StatePool.DIR_CONFIG, tempDir.toString());
+    conf.setBoolean(StatePool.PERSIST_CONFIG, true);
+    pool = new StatePool();
+    pool.initialize(conf);
+    MyState pState = 
+      (MyState) pool.getState(MyState.class);
+    assertEquals("State pool persistence/reload failed", "test-1", 
+                 pState.getState());
+    
+    // try persisting with no state change
+    pool.persist();
+    assertEquals("State pool persisted when disabled", 
+                 1, lfs.listStatus(tempDir).length);
+    
+    // modify the state of the pool and check persistence
+    pState.setUpdated(true);
+    pool.persist();
+    assertEquals("State pool persisted when disabled", 
+                 2, lfs.listStatus(tempDir).length);
+    
+    // delete the temp directory if everything goes fine
+    lfs.delete(tempDir, true);
+  }
+  
+  /**
+   * Test state.
+   */
+  static class MyState implements State {
+    private boolean updated = false;
+    private String state = "null";
+    
+    @Override
+    @JsonIgnore
+    public String getName() {
+      return "test";
+    }
+    
+    @Override
+    public void setName(String name) {
+      // for now, simply assert since this class has a hardcoded name
+      if (!getName().equals(name)) {
+        throw new RuntimeException("State name mismatch! Expected '" 
+                                   + getName() + "' but found '" + name + "'.");
+      }
+    }
+    
+    public void setState(String state) {
+      this.state = state;
+    }
+    
+    public String getState() {
+      return state;
+    }
+    
+    void setUpdated(boolean up) {
+      this.updated = up;
+    }
+    
+    @Override
+    @JsonIgnore
+    public boolean isUpdated() {
+      return updated;
+    }
+  }
+  
+  @SuppressWarnings("unchecked")
+  private static String getValueFromDataType(Object object) {
+    DataType<String> dt = (DataType<String>) object;
+    return dt.getValue();
+  }
+  
+  @Test
+  public void testJobPropertiesParser() {
+    // test default parser
+    Properties properties = new Properties();
+    Configuration conf = new Configuration();
+    JobProperties jp = new JobProperties(properties);
+    assertEquals("Job Properties (default filter) store error", 
+                 0, jp.getAnonymizedValue(null, conf).size());
+    
+    // define key-value pairs for job configuration
+    String key1 = "test-key";
+    String value1 = "test-value";
+    properties.put(key1, value1); // user config
+    String key2 = MRJobConfig.USER_NAME;
+    String value2 = "bob";
+    properties.put(key2, value2); // job config
+    String key3 = JobConf.MAPRED_MAP_TASK_JAVA_OPTS;
+    String value3 = "-Xmx1G";
+    properties.put(key3, value3); // deprecated
+    String key4 = MRJobConfig.REDUCE_JAVA_OPTS;
+    String value4 = "-Xms100m";
+    properties.put(key4, value4);
+    
+    jp = new JobProperties(properties);
+    
+    // Configure the default parser
+    conf.set(JobProperties.PARSERS_CONFIG_KEY, 
+             DefaultJobPropertiesParser.class.getName());
+    // anonymize
+    Properties defaultProp = jp.getAnonymizedValue(null, conf);
+    assertEquals("Job Properties (all-pass filter) store error", 
+                 4, defaultProp.size());
+    assertEquals("Job Properties (default filter) key#1 error", value1, 
+                 getValueFromDataType(defaultProp.get(key1)));
+    assertEquals("Job Properties (default filter) key#2 error", value2, 
+                 getValueFromDataType(defaultProp.get(key2)));
+    assertEquals("Job Properties (default filter) key#3 error", value3, 
+                 getValueFromDataType(defaultProp.get(key3)));
+    assertEquals("Job Properties (default filter) key#4 error", value4, 
+                 getValueFromDataType(defaultProp.get(key4)));
+    
+    // test MR parser
+    conf.set(JobProperties.PARSERS_CONFIG_KEY, 
+             MapReduceJobPropertiesParser.class.getName());
+    // anonymize
+    Properties filteredProp = jp.getAnonymizedValue(null, conf);
+    assertEquals("Job Properties (MR filter) store error", 
+                 3, filteredProp.size());
+    assertNull("Job Properties (MR filter) key#1 error", 
+               filteredProp.get(key1));
+    assertEquals("Job Properties (MR filter) key#2 error", value2, 
+                 getValueFromDataType(filteredProp.get(key2)));
+    assertEquals("Job Properties (MR filter) key#3 error", value3, 
+                 getValueFromDataType(filteredProp.get(key3)));
+    assertEquals("Job Properties (MR filter) key#4 error", value4, 
+                 getValueFromDataType(filteredProp.get(key4)));
+  }
+  
+  /**
+   * Test {@link WordListAnonymizerUtility}. Test various features like
+   *   - test known words
+   *   - test known suffix
+   */
+  @Test
+  public void testWordListBasedAnonymizer() {
+    String[] knownSuffixes = new String[] {".1", ".2", ".3", ".4"};
+    
+    // test with valid suffix
+    assertTrue("suffix test#0 failed!", 
+               WordListAnonymizerUtility.hasSuffix("a.1", knownSuffixes));
+    String split[] = 
+      WordListAnonymizerUtility.extractSuffix("a.1", knownSuffixes);
+    assertEquals("suffix test#1 failed!", 2, split.length);
+    assertEquals("suffix test#2 failed!", "a", split[0]);
+    assertEquals("suffix test#3 failed!", ".1", split[1]);
+    
+    // test with valid suffix
+    assertTrue("suffix test#0 failed!",
+               WordListAnonymizerUtility.hasSuffix("a.1", knownSuffixes));
+    split = 
+      WordListAnonymizerUtility.extractSuffix("/a/b.2", knownSuffixes);
+    assertEquals("suffix test#0 failed!", 2, split.length);
+    assertEquals("suffix test#1 failed!", "/a/b", split[0]);
+    assertEquals("suffix test#2 failed!", ".2", split[1]);
+    
+    // test with invalid suffix
+    assertFalse("suffix test#0 failed!", 
+                WordListAnonymizerUtility.hasSuffix("a.b", knownSuffixes));
+    
+    boolean failed = false;
+    try {
+      split = WordListAnonymizerUtility.extractSuffix("a.b", knownSuffixes);
+    } catch (Exception e) {
+      failed = true;
+    }
+    assertTrue("Exception expected!", failed);
+    
+    String[] knownWords = new String[] {"a", "b"};
+    
+    // test with valid data
+    assertTrue("data test#0 failed!", 
+               WordListAnonymizerUtility.isKnownData("a", knownWords));
+    // test with valid data
+    assertTrue("data test#1 failed!", 
+               WordListAnonymizerUtility.isKnownData("b", knownWords));
+    // test with invalid data
+    assertFalse("data test#2 failed!", 
+                WordListAnonymizerUtility.isKnownData("c", knownWords));
+    
+    // test with valid known word
+    assertTrue("data test#3 failed!", 
+               WordListAnonymizerUtility.isKnownData("job"));
+    // test with invalid known word
+    assertFalse("data test#4 failed!", 
+                WordListAnonymizerUtility.isKnownData("bob"));
+    
+    // test numeric data
+    assertFalse("Numeric test failed!", 
+                 WordListAnonymizerUtility.needsAnonymization("123"));
+    // test numeric data (unsupported)
+    assertTrue("Numeric test failed!", 
+               WordListAnonymizerUtility.needsAnonymization("123.456"));
+    // test text data
+    assertTrue("Text test failed!", 
+               WordListAnonymizerUtility.needsAnonymization("123abc"));
+  }
+  
+  /**
+   * Test {@link WordList} features like
+   *   - add words
+   *   - index 
+   *   - contains
+   */
+  @Test
+  public void testWordList() throws Exception {
+    // test features with fresh state
+    WordList wordList = new WordList();
+    assertFalse("Word list state incorrect", wordList.isUpdated());
+    
+    // add some special word
+    String test = "abbracadabra";
+    wordList.add(test);
+    assertTrue("Word list failed to store", wordList.contains(test));
+    assertEquals("Word list index failed", 0, wordList.indexOf(test));
+    assertEquals("Word list size failed", 1, wordList.getSize());
+    assertTrue("Word list state incorrect", wordList.isUpdated());
+    
+    // add already added word
+    wordList.add(test);
+    assertEquals("Word list index failed", 0, wordList.indexOf(test));
+    assertEquals("Word list size failed", 1, wordList.getSize());
+    assertTrue("Word list state incorrect", wordList.isUpdated());
+    
+    String test2 = "hakuna-matata";
+    wordList.add(test2);
+    assertTrue("Word list failed to store", wordList.contains(test2));
+    assertEquals("Word list index failed", 1, wordList.indexOf(test2));
+    assertEquals("Word list size failed", 2, wordList.getSize());
+    assertTrue("Word list state incorrect", wordList.isUpdated());
+
+    // test persistence
+    Configuration conf = new Configuration();
+    FileSystem lfs = FileSystem.getLocal(conf);
+    Path rootTempDir =
+      new Path(System.getProperty("test.build.data", "/tmp"));
+
+    Path tempDir = new Path(rootTempDir, "testWordList");
+    tempDir = lfs.makeQualified(tempDir);
+    lfs.delete(tempDir, true);
+    
+    // define a state pool to help persist the wordlist
+    StatePool pool = new StatePool();
+    
+    try {
+      // set the persistence directory
+      conf.set(StatePool.DIR_CONFIG, tempDir.toString());
+      conf.setBoolean(StatePool.PERSIST_CONFIG, true);
+
+      // initialize the state-pool
+      pool.initialize(conf);
+
+      // add the wordlist to the pool
+      pool.addState(getClass(), wordList);
+
+      pool.persist();
+
+      // now clear the pool state
+      pool = new StatePool();
+      
+      // set reload to true
+      conf.setBoolean(StatePool.RELOAD_CONFIG, true);
+      // initialize the state-pool
+      pool.initialize(conf);
+
+      State state = pool.getState(getClass());
+      assertNotNull("Missing state!", state);
+      assertEquals("Incorrect state class!", WordList.class, state.getClass());
+      WordList pList = (WordList) state;
+
+      // check size
+      assertEquals("Word list size on reload failed", 2, pList.getSize());
+      assertFalse("Word list state incorrect", pList.isUpdated());
+
+      // add already added word
+      pList.add(test);
+      assertEquals("Word list index on reload failed", 0, pList.indexOf(test));
+      assertEquals("Word list size on reload failed", 2, pList.getSize());
+      assertFalse("Word list state on reload incorrect", pList.isUpdated());
+
+      String test3 = "disco-dancer";
+      assertFalse("Word list failed to after reload", pList.contains(test3));
+      pList.add(test3);
+      assertTrue("Word list failed to store on reload", pList.contains(test3));
+      assertEquals("Word list index on reload failed", 2, pList.indexOf(test3));
+      assertEquals("Word list size on reload failed", 3, pList.getSize());
+      assertTrue("Word list state on reload incorrect", pList.isUpdated());
+      
+      // test previously added (persisted) word
+      assertTrue("Word list failed to store on reload", pList.contains(test2));
+      assertEquals("Word list index on reload failed", 1, pList.indexOf(test2));
+    } finally {
+      lfs.delete(tempDir, true);
+    }
+  }
+  
+  /**
+   * Test {@link FileName#FileNameState} persistence with directories only.
+   */
+  @Test
+  public void testFileNameStateWithDir() throws Exception {
+    // test persistence
+    Configuration conf = new Configuration();
+    FileSystem lfs = FileSystem.getLocal(conf);
+    Path rootTempDir =
+      new Path(System.getProperty("test.build.data", "/tmp"));
+
+    Path tempDir = new Path(rootTempDir, "testFileNameStateWithDir");
+    tempDir = lfs.makeQualified(tempDir);
+    lfs.delete(tempDir, true);
+    
+    // set the persistence directory
+    conf.set(StatePool.DIR_CONFIG, tempDir.toString());
+    conf.setBoolean(StatePool.PERSIST_CONFIG, true);
+    
+    // define a state pool to help persist the dirs
+    StatePool pool = new StatePool();
+    
+    FileNameState fState = new FileNameState();
+    
+    // define the directory names
+    String test1 = "test";
+    String test2 = "home";
+    
+    // test dir only
+    WordList dirState = new WordList("dir");
+    dirState.add(test1);
+    dirState.add(test2);
+    
+    // set the directory state
+    fState.setDirectoryState(dirState);
+    
+    try {
+      // initialize the state-pool
+      pool.initialize(conf);
+
+      // add the wordlist to the pool
+      pool.addState(getClass(), fState);
+
+      // persist the state
+      pool.persist();
+
+      // now clear the pool state
+      pool = new StatePool();
+      
+      // set reload to true
+      conf.setBoolean(StatePool.RELOAD_CONFIG, true);
+      
+      // initialize the state-pool
+      pool.initialize(conf);
+
+      State state = pool.getState(getClass());
+      assertNotNull("Missing state!", state);
+      assertEquals("Incorrect state class!", 
+                    FileNameState.class, state.getClass());
+      FileNameState newFState = (FileNameState) state;
+
+      // check the state contents
+      WordList newStateWordList = newFState.getDirectoryState();
+      assertTrue("File state failed to store on reload", 
+                 newStateWordList.contains(test1));
+      assertEquals("File state index on reload failed", 
+                   0, newStateWordList.indexOf(test1));
+      
+      assertTrue("File state failed to store on reload", 
+                 newStateWordList.contains(test2));
+      assertEquals("File state index on reload failed", 
+                   1, newStateWordList.indexOf(test2));
+    } finally {
+      lfs.delete(tempDir, true);
+    }
+  }
+  
+  /**
+   * Test {@link FileName#FileNameState} persistence with files only.
+   */
+  @Test
+  public void testFileNameStateWithFiles() throws Exception {
+    // test persistence
+    Configuration conf = new Configuration();
+    FileSystem lfs = FileSystem.getLocal(conf);
+    Path rootTempDir =
+      new Path(System.getProperty("test.build.data", "/tmp"));
+
+    Path tempDir = new Path(rootTempDir, "testFileNameStateWithFiles");
+    tempDir = lfs.makeQualified(tempDir);
+    lfs.delete(tempDir, true);
+    
+    // set the persistence directory
+    conf.set(StatePool.DIR_CONFIG, tempDir.toString());
+    conf.setBoolean(StatePool.PERSIST_CONFIG, true);
+    
+    // define a state pool to help persist the filename parts
+    StatePool pool = new StatePool();
+    
+    FileNameState fState = new FileNameState();
+    
+    // define the file names
+    String test1 = "part-00.bzip";
+    String test2 = "file1.txt";
+    
+    // test filenames only
+    WordList fileNameState = new WordList("files");
+    fileNameState.add(test1);
+    fileNameState.add(test2);
+    
+    // set the filename state
+    fState.setDirectoryState(fileNameState);
+    
+    try {
+      // initialize the state-pool
+      pool.initialize(conf);
+
+      // add the wordlist to the pool
+      pool.addState(getClass(), fState);
+
+      // persist the state
+      pool.persist();
+
+      // now clear the pool state
+      pool = new StatePool();
+      
+      // set reload to true
+      conf.setBoolean(StatePool.RELOAD_CONFIG, true);
+      
+      // initialize the state-pool
+      pool.initialize(conf);
+
+      State state = pool.getState(getClass());
+      assertNotNull("Missing state!", state);
+      assertEquals("Incorrect state class!", 
+                    FileNameState.class, state.getClass());
+      FileNameState newFState = (FileNameState) state;
+
+      // check the state contents
+      WordList newFileWordList = newFState.getDirectoryState();
+      assertTrue("File state failed on reload", 
+                 newFileWordList.contains(test1));
+      assertEquals("File state indexing on reload failed", 
+                   0, newFileWordList.indexOf(test1));
+      
+      assertTrue("File state failed on reload", 
+                 newFileWordList.contains(test2));
+      assertEquals("File state indexing on reload failed", 
+                   1, newFileWordList.indexOf(test2));
+    } finally {
+      lfs.delete(tempDir, true);
+    }
+  }
+  
+  /**
+   * Test {@link FileName#FileNameState} persistence with files and directories.
+   */
+  @Test
+  public void testFileNameState() throws Exception {
+    // test persistence
+    Configuration conf = new Configuration();
+    FileSystem lfs = FileSystem.getLocal(conf);
+    Path rootTempDir =
+      new Path(System.getProperty("test.build.data", "/tmp"));
+
+    Path tempDir = new Path(rootTempDir, "testFileNameState");
+    tempDir = lfs.makeQualified(tempDir);
+    lfs.delete(tempDir, true);
+    
+    // set the persistence directory
+    conf.set(StatePool.DIR_CONFIG, tempDir.toString());
+    conf.setBoolean(StatePool.PERSIST_CONFIG, true);
+    
+    // define a state pool to help persist the filename parts
+    StatePool pool = new StatePool();
+    
+    FileNameState fState = new FileNameState();
+    
+    // define the directory names
+    String testD1 = "test";
+    String testD2 = "home";
+    String testD3 = "tmp";
+    
+    // test dir only
+    WordList dirState = new WordList("dir");
+    dirState.add(testD1);
+    dirState.add(testD2);
+    dirState.add(testD3);
+    
+    // define the file names
+    String testF1 = "part-00.bzip";
+    String testF2 = "file1.txt";
+    String testF3 = "tmp";
+    
+    // test filenames only
+    WordList fileNameState = new WordList("files");
+    fileNameState.add(testF1);
+    fileNameState.add(testF2);
+    fileNameState.add(testF3);
+    
+    // set the filename state
+    fState.setFileNameState(fileNameState);
+    // set the directory state
+    fState.setDirectoryState(dirState);
+    
+    try {
+      // initialize the state-pool
+      pool.initialize(conf);
+
+      // add the wordlist to the pool
+      pool.addState(getClass(), fState);
+
+      // persist the state
+      pool.persist();
+
+      // now clear the pool state
+      pool = new StatePool();
+      
+      // set reload to true
+      conf.setBoolean(StatePool.RELOAD_CONFIG, true);
+      
+      // initialize the state-pool
+      pool.initialize(conf);
+
+      State state = pool.getState(getClass());
+      assertNotNull("Missing state!", state);
+      assertEquals("Incorrect state class!", 
+                    FileNameState.class, state.getClass());
+      FileNameState newFState = (FileNameState) state;
+
+      // test filenames
+      WordList newStateWordList = newFState.getFileNameState();
+      assertTrue("File state failed on reload", 
+                 newStateWordList.contains(testF1));
+      assertEquals("File state indexing on reload failed", 
+                   0, newStateWordList.indexOf(testF1));
+      
+      assertTrue("File state failed on reload", 
+                 newStateWordList.contains(testF2));
+      assertEquals("File state indexing on reload failed", 
+                   1, newStateWordList.indexOf(testF2));
+      
+      assertTrue("File state failed on reload", 
+                 newStateWordList.contains(testF3));
+      assertEquals("File state indexing on reload failed", 
+                   2, newStateWordList.indexOf(testF3));
+      
+      // test dirs
+      WordList newDirWordList = newFState.getDirectoryState();
+      assertTrue("File state failed on reload", 
+                 newDirWordList.contains(testD1));
+      assertEquals("File state indexing on reload failed", 
+                   0, newDirWordList.indexOf(testD1));
+      
+      assertTrue("File state failed on reload", 
+                 newDirWordList.contains(testD2));
+      assertEquals("File state indexing on reload failed", 
+                   1, newDirWordList.indexOf(testD2));
+      assertTrue("File state failed on reload", 
+                 newDirWordList.contains(testD3));
+      assertEquals("File state indexing on reload failed", 
+                   2, newDirWordList.indexOf(testD3));
+    } finally {
+      lfs.delete(tempDir, true);
+    }
+  }
+  
+  /**
+   * Test {@link NodeName#NodeName} persistence with hostnames only.
+   */
+  @Test
+  public void testNodeNameStateWithHostNameOnly() throws Exception {
+    // test persistence
+    Configuration conf = new Configuration();
+    FileSystem lfs = FileSystem.getLocal(conf);
+    Path rootTempDir =
+      new Path(System.getProperty("test.build.data", "/tmp"));
+
+    Path tempDir = new Path(rootTempDir, "testNodeNameStateWithHostNameOnly");
+    tempDir = lfs.makeQualified(tempDir);
+    lfs.delete(tempDir, true);
+    
+    // set the persistence directory
+    conf.set(StatePool.DIR_CONFIG, tempDir.toString());
+    conf.setBoolean(StatePool.PERSIST_CONFIG, true);
+    
+    // define a state pool to help persist the hostnames
+    StatePool pool = new StatePool();
+    
+    NodeNameState nState = new NodeNameState();
+    
+    // define the host names
+    String test1 = "abc123";
+    String test2 = "xyz789";
+    
+    // test hostname only
+    WordList hostNameState = new WordList("hostname");
+    hostNameState.add(test1);
+    hostNameState.add(test2);
+    
+    // set the directory state
+    nState.setHostNameState(hostNameState);
+    
+    try {
+      // initialize the state-pool
+      pool.initialize(conf);
+
+      // add the wordlist to the pool
+      pool.addState(getClass(), nState);
+
+      // persist the state
+      pool.persist();
+
+      // now clear the pool state
+      pool = new StatePool();
+      
+      // set reload to true
+      conf.setBoolean(StatePool.RELOAD_CONFIG, true);
+      
+      // initialize the state-pool
+      pool.initialize(conf);
+
+      State state = pool.getState(getClass());
+      assertNotNull("Missing state!", state);
+      assertEquals("Incorrect state class!", 
+                   NodeNameState.class, state.getClass());
+      NodeNameState newNState = (NodeNameState) state;
+
+      // check the state contents
+      WordList newStateWordList = newNState.getHostNameState();
+      assertTrue("Node state failed to store on reload", 
+                 newStateWordList.contains(test1));
+      assertEquals("Node state index on reload failed", 
+                   0, newStateWordList.indexOf(test1));
+      
+      assertTrue("Node state failed to store on reload", 
+                 newStateWordList.contains(test2));
+      assertEquals("Node state index on reload failed", 
+                   1, newStateWordList.indexOf(test2));
+    } finally {
+      lfs.delete(tempDir, true);
+    }
+  }
+  
+  /**
+   * Test {@link NodeName#NodeNameState} persistence with racknames only.
+   */
+  @Test
+  public void testNodeNameWithRackNamesOnly() throws Exception {
+    // test persistence
+    Configuration conf = new Configuration();
+    FileSystem lfs = FileSystem.getLocal(conf);
+    Path rootTempDir =
+      new Path(System.getProperty("test.build.data", "/tmp"));
+
+    Path tempDir = new Path(rootTempDir, "testNodeNameWithRackNamesOnly");
+    tempDir = lfs.makeQualified(tempDir);
+    lfs.delete(tempDir, true);
+    
+    // set the persistence directory
+    conf.set(StatePool.DIR_CONFIG, tempDir.toString());
+    conf.setBoolean(StatePool.PERSIST_CONFIG, true);
+    
+    // define a state pool to help persist the rack names
+    StatePool pool = new StatePool();
+    
+    NodeNameState nState = new NodeNameState();
+    
+    // define the rack names
+    String test1 = "rack1";
+    String test2 = "rack2";
+    
+    // test filenames only
+    WordList rackNameState = new WordList("racknames");
+    rackNameState.add(test1);
+    rackNameState.add(test2);
+    
+    // set the rackname state
+    nState.setRackNameState(rackNameState);
+    
+    try {
+      // initialize the state-pool
+      pool.initialize(conf);
+
+      // add the wordlist to the pool
+      pool.addState(getClass(), nState);
+
+      // persist the state
+      pool.persist();
+
+      // now clear the pool state
+      pool = new StatePool();
+      
+      // set reload to true
+      conf.setBoolean(StatePool.RELOAD_CONFIG, true);
+      
+      // initialize the state-pool
+      pool.initialize(conf);
+
+      State state = pool.getState(getClass());
+      assertNotNull("Missing state!", state);
+      assertEquals("Incorrect state class!", 
+                   NodeNameState.class, state.getClass());
+      NodeNameState newNState = (NodeNameState) state;
+
+      // check the state contents
+      WordList newFileWordList = newNState.getRackNameState();
+      assertTrue("File state failed on reload", 
+                 newFileWordList.contains(test1));
+      assertEquals("File state indexing on reload failed", 
+                   0, newFileWordList.indexOf(test1));
+      
+      assertTrue("File state failed on reload", 
+                 newFileWordList.contains(test2));
+      assertEquals("File state indexing on reload failed", 
+                   1, newFileWordList.indexOf(test2));
+    } finally {
+      lfs.delete(tempDir, true);
+    }
+  }
+  
+  /**
+   * Test {@link NodeName#NodeNameState} persistence with hosts and racks.
+   */
+  @Test
+  public void testNodeNameState() throws Exception {
+    // test persistence
+    Configuration conf = new Configuration();
+    FileSystem lfs = FileSystem.getLocal(conf);
+    Path rootTempDir =
+      new Path(System.getProperty("test.build.data", "/tmp"));
+
+    Path tempDir = new Path(rootTempDir, "testNodeNameState");
+    tempDir = lfs.makeQualified(tempDir);
+    lfs.delete(tempDir, true);
+    
+    // set the persistence directory
+    conf.set(StatePool.DIR_CONFIG, tempDir.toString());
+    conf.setBoolean(StatePool.PERSIST_CONFIG, true);
+    
+    // define a state pool to help persist the node names.
+    StatePool pool = new StatePool();
+    
+    NodeNameState nState = new NodeNameState();
+    
+    // define the rack names
+    String testR1 = "rack1";
+    String testR2 = "rack2";
+    String testR3 = "rack3";
+    
+    WordList rackState = new WordList("rack");
+    rackState.add(testR1);
+    rackState.add(testR2);
+    rackState.add(testR3);
+    
+    String testH1 = "host1";
+    String testH2 = "host2";
+    String testH3 = "host3";
+    
+    WordList hostNameState = new WordList("host");
+    hostNameState.add(testH1);
+    hostNameState.add(testH2);
+    hostNameState.add(testH3);
+    
+    // set the filename state
+    nState.setHostNameState(hostNameState);
+    nState.setRackNameState(rackState);
+    
+    try {
+      // initialize the state-pool
+      pool.initialize(conf);
+
+      // add the wordlist to the pool
+      pool.addState(getClass(), nState);
+
+      // persist the state
+      pool.persist();
+
+      // now clear the pool state
+      pool = new StatePool();
+      
+      // set reload to true
+      conf.setBoolean(StatePool.RELOAD_CONFIG, true);
+      
+      // initialize the state-pool
+      pool.initialize(conf);
+
+      State state = pool.getState(getClass());
+      assertNotNull("Missing state!", state);
+      assertEquals("Incorrect state class!", 
+                   NodeNameState.class, state.getClass());
+      NodeNameState newNState = (NodeNameState) state;
+
+      // test nodenames
+      WordList newHostWordList = newNState.getHostNameState();
+      assertTrue("File state failed on reload", 
+                 newHostWordList.contains(testH1));
+      assertEquals("File state indexing on reload failed", 
+                   0, newHostWordList.indexOf(testH1));
+      
+      assertTrue("File state failed on reload", 
+                 newHostWordList.contains(testH2));
+      assertEquals("File state indexing on reload failed", 
+                   1, newHostWordList.indexOf(testH2));
+      
+      assertTrue("File state failed on reload", 
+                 newHostWordList.contains(testH3));
+      assertEquals("File state indexing on reload failed", 
+                   2, newHostWordList.indexOf(testH3));
+      
+      // test racknames
+      WordList newRackWordList = newNState.getRackNameState();
+      assertTrue("File state failed on reload", 
+                 newRackWordList.contains(testR1));
+      assertEquals("File state indexing on reload failed", 
+                   0, newRackWordList.indexOf(testR1));
+      
+      assertTrue("File state failed on reload", 
+                 newRackWordList.contains(testR2));
+      assertEquals("File state indexing on reload failed", 
+                   1, newRackWordList.indexOf(testR2));
+      assertTrue("File state failed on reload", 
+                 newRackWordList.contains(testR3));
+      assertEquals("File state indexing on reload failed", 
+                   2, newRackWordList.indexOf(testR3));
+    } finally {
+      lfs.delete(tempDir, true);
+    }
+  }
+}
\ No newline at end of file

Modified: hadoop/common/trunk/hadoop-mapreduce-project/src/test/mapred/org/apache/hadoop/tools/rumen/TestRumenFolder.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-mapreduce-project/src/test/mapred/org/apache/hadoop/tools/rumen/TestRumenFolder.java?rev=1215141&r1=1215140&r2=1215141&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-mapreduce-project/src/test/mapred/org/apache/hadoop/tools/rumen/TestRumenFolder.java (original)
+++ hadoop/common/trunk/hadoop-mapreduce-project/src/test/mapred/org/apache/hadoop/tools/rumen/TestRumenFolder.java Fri Dec 16 14:20:58 2011
@@ -20,12 +20,8 @@ package org.apache.hadoop.tools.rumen;
 
 import java.io.IOException;
 import java.io.InputStream;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.LinkedHashMap;
 import java.util.LinkedList;
 import java.util.List;
-import java.util.Map;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;

Modified: hadoop/common/trunk/hadoop-mapreduce-project/src/test/mapred/org/apache/hadoop/tools/rumen/TestRumenJobTraces.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-mapreduce-project/src/test/mapred/org/apache/hadoop/tools/rumen/TestRumenJobTraces.java?rev=1215141&r1=1215140&r2=1215141&view=diff
==============================================================================
--- hadoop/common/trunk/hadoop-mapreduce-project/src/test/mapred/org/apache/hadoop/tools/rumen/TestRumenJobTraces.java (original)
+++ hadoop/common/trunk/hadoop-mapreduce-project/src/test/mapred/org/apache/hadoop/tools/rumen/TestRumenJobTraces.java Fri Dec 16 14:20:58 2011
@@ -960,11 +960,11 @@ public class TestRumenJobTraces {
 
     for (LoggedNetworkTopology rack : racks) {
       List<LoggedNetworkTopology> nodes = rack.getChildren();
-      if (rack.getName().endsWith(".64")) {
+      if (rack.getName().getValue().endsWith(".64")) {
         assertEquals("The singleton rack has the wrong number of elements", 1,
             nodes.size());
         sawSingleton = true;
-      } else if (rack.getName().endsWith(".80")) {
+      } else if (rack.getName().getValue().endsWith(".80")) {
         assertEquals("The doubleton rack has the wrong number of elements", 2,
             nodes.size());
         sawDoubleton = true;

Added: hadoop/common/trunk/hadoop-mapreduce-project/src/tools/org/apache/hadoop/tools/rumen/Anonymizer.java
URL: http://svn.apache.org/viewvc/hadoop/common/trunk/hadoop-mapreduce-project/src/tools/org/apache/hadoop/tools/rumen/Anonymizer.java?rev=1215141&view=auto
==============================================================================
--- hadoop/common/trunk/hadoop-mapreduce-project/src/tools/org/apache/hadoop/tools/rumen/Anonymizer.java (added)
+++ hadoop/common/trunk/hadoop-mapreduce-project/src/tools/org/apache/hadoop/tools/rumen/Anonymizer.java Fri Dec 16 14:20:58 2011
@@ -0,0 +1,273 @@
+/**
+ * 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.tools.rumen;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.conf.Configured;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.compress.CodecPool;
+import org.apache.hadoop.io.compress.CompressionCodec;
+import org.apache.hadoop.io.compress.CompressionCodecFactory;
+import org.apache.hadoop.io.compress.Compressor;
+import org.apache.hadoop.mapreduce.ID;
+import org.apache.hadoop.util.Tool;
+import org.apache.hadoop.util.ToolRunner;
+import org.apache.hadoop.tools.rumen.datatypes.*;
+import org.apache.hadoop.tools.rumen.serializers.*;
+import org.apache.hadoop.tools.rumen.state.*;
+
+import org.codehaus.jackson.JsonEncoding;
+import org.codehaus.jackson.JsonFactory;
+import org.codehaus.jackson.JsonGenerator;
+import org.codehaus.jackson.Version;
+import org.codehaus.jackson.map.ObjectMapper;
+import org.codehaus.jackson.map.module.SimpleModule;
+
+public class Anonymizer extends Configured implements Tool {
+  private boolean anonymizeTrace = false;
+  private Path inputTracePath = null;
+  private Path outputTracePath = null;
+  private boolean anonymizeTopology = false;
+  private Path inputTopologyPath = null;
+  private Path outputTopologyPath = null;
+  
+  //TODO Make this final if not using JSON
+  // private final StatePool statePool = new StatePool();
+  private StatePool statePool;
+  
+  private ObjectMapper outMapper = null;
+  private JsonFactory outFactory = null;
+  
+  private void initialize(String[] args) throws Exception {
+    try {
+      for (int i = 0; i < args.length; ++i) {
+        if ("-trace".equals(args[i])) {
+          anonymizeTrace = true;
+          inputTracePath = new Path(args[i+1]);
+          outputTracePath = new Path(args[i+2]);
+          i +=2;
+        }
+        if ("-topology".equals(args[i])) {
+          anonymizeTopology = true;
+          inputTopologyPath = new Path(args[i+1]);
+          outputTopologyPath = new Path(args[i+2]);
+          i +=2;
+        }
+      }
+    } catch (Exception e) {
+      throw new IllegalArgumentException("Illegal arguments list!", e);
+    }
+    
+    if (!anonymizeTopology && !anonymizeTrace) {
+      throw new IllegalArgumentException("Invalid arguments list!");
+    }
+    
+    statePool = new StatePool();
+    // initialize the state manager after the anonymizers are registered
+    statePool.initialize(getConf());
+     
+    outMapper = new ObjectMapper();
+    // define a module
+    SimpleModule module = new SimpleModule("Anonymization Serializer",  
+                                           new Version(0, 1, 1, "FINAL"));
+    // add various serializers to the module
+    // use the default (as-is) serializer for default data types
+    module.addSerializer(DataType.class, new DefaultRumenSerializer());
+    // use a blocking serializer for Strings as they can contain sensitive 
+    // information
+    module.addSerializer(String.class, new BlockingSerializer());
+    // use object.toString() for object of type ID
+    module.addSerializer(ID.class, new ObjectStringSerializer<ID>());
+    // use getAnonymizedValue() for data types that have the anonymizing 
+    // feature
+    module.addSerializer(AnonymizableDataType.class, 
+        new DefaultAnonymizingRumenSerializer(statePool, getConf()));
+    
+    // register the module with the object-mapper
+    outMapper.registerModule(module);
+    
+    outFactory = outMapper.getJsonFactory();
+  }
+  
+  // anonymize the job trace file
+  private void anonymizeTrace() throws Exception {
+    if (anonymizeTrace) {
+      System.out.println("Anonymizing trace file: " + inputTracePath);
+      JobTraceReader reader = null;
+      JsonGenerator outGen = null;
+      Configuration conf = getConf();
+      
+      try {
+        // create a generator
+        outGen = createJsonGenerator(conf, outputTracePath);
+
+        // define the input trace reader
+        reader = new JobTraceReader(inputTracePath, conf);
+        
+        // read the plain unanonymized logged job
+        LoggedJob job = reader.getNext();
+        
+        while (job != null) {
+          // write it via an anonymizing channel
+          outGen.writeObject(job);
+          // read the next job
+          job = reader.getNext();
+        }
+        
+        System.out.println("Anonymized trace file: " + outputTracePath);
+      } finally {
+        if (outGen != null) {
+          outGen.close();
+        }
+        if (reader != null) {
+          reader.close();
+        }
+      }
+    }
+  }
+
+  // anonymize the cluster topology file
+  private void anonymizeTopology() throws Exception {
+    if (anonymizeTopology) {
+      System.out.println("Anonymizing topology file: " + inputTopologyPath);
+      ClusterTopologyReader reader = null;
+      JsonGenerator outGen = null;
+      Configuration conf = getConf();
+
+      try {
+        // create a generator
+        outGen = createJsonGenerator(conf, outputTopologyPath);
+
+        // define the input cluster topology reader
+        reader = new ClusterTopologyReader(inputTopologyPath, conf);
+        
+        // read the plain unanonymized logged job
+        LoggedNetworkTopology job = reader.get();
+        
+        // write it via an anonymizing channel
+        outGen.writeObject(job);
+        
+        System.out.println("Anonymized topology file: " + outputTopologyPath);
+      } finally {
+        if (outGen != null) {
+          outGen.close();
+        }
+      }
+    }
+  }
+  
+  // Creates a JSON generator
+  private JsonGenerator createJsonGenerator(Configuration conf, Path path) 
+  throws IOException {
+    FileSystem outFS = path.getFileSystem(conf);
+    CompressionCodec codec =
+      new CompressionCodecFactory(conf).getCodec(path);
+    OutputStream output;
+    Compressor compressor = null;
+    if (codec != null) {
+      compressor = CodecPool.getCompressor(codec);
+      output = codec.createOutputStream(outFS.create(path), compressor);
+    } else {
+      output = outFS.create(path);
+    }
+
+    JsonGenerator outGen = outFactory.createJsonGenerator(output, 
+                                                          JsonEncoding.UTF8);
+    outGen.useDefaultPrettyPrinter();
+    
+    return outGen;
+  }
+  
+  @Override
+  public int run(String[] args) throws Exception {
+    try {
+      initialize(args);
+    } catch (Exception e) {
+      e.printStackTrace();
+      printUsage();
+      return -1;
+    }
+    
+    return run();
+  }
+
+  /**
+   * Runs the actual anonymization tool.
+   */
+  public int run() throws Exception {
+    try {
+      anonymizeTrace();
+    } catch (IOException ioe) {
+      System.err.println("Error running the trace anonymizer!");
+      ioe.printStackTrace();
+      System.out.println("\n\nAnonymization unsuccessful!");
+      return -1;
+    }
+    
+    try {
+      anonymizeTopology();
+    } catch (IOException ioe) {
+      System.err.println("Error running the cluster topology anonymizer!");
+      ioe.printStackTrace();
+      System.out.println("\n\nAnonymization unsuccessful!");
+      return -1;
+    }
+    
+    statePool.persist();
+    
+    System.out.println("Anonymization completed successfully!");
+    
+    return 0;
+  }
+
+  private static void printUsage() {
+    System.out.println("\nUsage:-");
+    System.out.print("  Anonymizer");
+    System.out.print(" [-trace <input-trace-path> <output-trace-path>]");
+    System.out.println(" [-topology <input-topology-path> " 
+                       + "<output-topology-path>] ");
+    System.out.print("\n");
+  }
+  
+  /**
+   * The main driver program to use the anonymization utility.
+   * @param args
+   */
+  public static void main(String[] args) {
+    Anonymizer instance = new Anonymizer();
+    int result = 0;
+    
+    try {
+      result = ToolRunner.run(instance, args);
+    } catch (Exception e) {
+      e.printStackTrace(System.err);
+      System.exit(-1);
+    }
+
+    if (result != 0) {
+      System.exit(result);
+    }
+
+    return;
+  }
+}
\ No newline at end of file



Mime
View raw message