HDFS-4651. Merge changes 1462875 and 1462876 from branch-1

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

bin/hadoop

@@ -81,6 +81,7 @@ print_usage()
   echo "  fsck                 run a DFS filesystem checking utility"
   echo "  fs                   run a generic filesystem user client"
   echo "  balancer             run a cluster balancing utility"
+  echo "  oiv                  apply the offline fsimage viewer to an fsimage"
   echo "  fetchdt              fetch a delegation token from the NameNode"
   echo "  jobtracker           run the MapReduce job Tracker node" 
   echo "  pipes                run a Pipes job"
@@ -287,6 +288,8 @@ elif [ "$COMMAND" = "fsck" ] ; then
 elif [ "$COMMAND" = "balancer" ] ; then
   CLASS=org.apache.hadoop.hdfs.server.balancer.Balancer
   HADOOP_OPTS="$HADOOP_OPTS $HADOOP_BALANCER_OPTS"
+elif [ "$COMMAND" = "oiv" ] ; then
+  CLASS=org.apache.hadoop.hdfs.tools.offlineImageViewer.OfflineImageViewer
 elif [ "$COMMAND" = "fetchdt" ] ; then
   CLASS=org.apache.hadoop.hdfs.tools.DelegationTokenFetcher
 elif [ "$COMMAND" = "jobtracker" ] ; then

build.xml

@@ -936,6 +936,8 @@
     <copy file="${test.src.dir}/org/apache/hadoop/cli/clitest_data/data30bytes" todir="${test.cache.data}"/>
     <copy file="${test.src.dir}/org/apache/hadoop/cli/clitest_data/data60bytes" todir="${test.cache.data}"/>
     <copy file="${test.src.dir}/org/apache/hadoop/cli/clitest_data/data120bytes" todir="${test.cache.data}"/>
+    <copy file="${test.src.dir}/org/apache/hadoop/hdfs/tools/offlineImageViewer/fsimageV18" todir="${test.cache.data}"/>
+    <copy file="${test.src.dir}/org/apache/hadoop/hdfs/tools/offlineImageViewer/fsimageV19" todir="${test.cache.data}"/>
   </target>
 
   <!-- ================================================================== -->

src/core/org/apache/hadoop/io/compress/CompressionCodecFactory.java

@@ -39,10 +39,30 @@ public class CompressionCodecFactory {
    * automatically supports finding the longest matching suffix. 
    */
   private SortedMap<String, CompressionCodec> codecs = null;
+
+    /**
+     * A map from the reversed filename suffixes to the codecs.
+     * This is probably overkill, because the maps should be small, but it
+     * automatically supports finding the longest matching suffix.
+     */
+    private Map<String, CompressionCodec> codecsByName = null;
+
+  /**
+   * A map from class names to the codecs
+   */
+  private HashMap<String, CompressionCodec> codecsByClassName = null;
   
   private void addCodec(CompressionCodec codec) {
     String suffix = codec.getDefaultExtension();
     codecs.put(new StringBuffer(suffix).reverse().toString(), codec);
+    codecsByClassName.put(codec.getClass().getCanonicalName(), codec);
+
+    String codecName = codec.getClass().getSimpleName();
+    codecsByName.put(codecName.toLowerCase(), codec);
+    if (codecName.endsWith("Codec")) {
+      codecName = codecName.substring(0, codecName.length() - "Codec".length());
+      codecsByName.put(codecName.toLowerCase(), codec);
+    }
   }
   
   /**
@@ -131,6 +151,8 @@ public class CompressionCodecFactory {
    */
   public CompressionCodecFactory(Configuration conf) {
     codecs = new TreeMap<String, CompressionCodec>();
+    codecsByClassName = new HashMap<String, CompressionCodec>();
+    codecsByName = new HashMap<String, CompressionCodec>();
     List<Class<? extends CompressionCodec>> codecClasses = getCodecClasses(conf);
     if (codecClasses == null) {
       addCodec(new GzipCodec());
@@ -167,6 +189,68 @@ public class CompressionCodecFactory {
     return result;
   }
   
+  /**
+   * Find the relevant compression codec for the codec's canonical class name.
+   * @param classname the canonical class name of the codec
+   * @return the codec object
+   */
+  public CompressionCodec getCodecByClassName(String classname) {
+    if (codecsByClassName == null) {
+      return null;
+    }
+    return codecsByClassName.get(classname);
+  }
+
+    /**
+     * Find the relevant compression codec for the codec's canonical class name
+     * or by codec alias.
+     * <p/>
+     * Codec aliases are case insensitive.
+     * <p/>
+     * The code alias is the short class name (without the package name).
+     * If the short class name ends with 'Codec', then there are two aliases for
+     * the codec, the complete short class name and the short class name without
+     * the 'Codec' ending. For example for the 'GzipCodec' codec class name the
+     * alias are 'gzip' and 'gzipcodec'.
+     *
+     * @param codecName the canonical class name of the codec
+     * @return the codec object
+     */
+    public CompressionCodec getCodecByName(String codecName) {
+      if (codecsByClassName == null) {
+        return null;
+      }
+      CompressionCodec codec = getCodecByClassName(codecName);
+      if (codec == null) {
+        // trying to get the codec by name in case the name was specified instead a class
+        codec = codecsByName.get(codecName.toLowerCase());
+      }
+      return codec;
+    }
+
+    /**
+     * Find the relevant compression codec for the codec's canonical class name
+     * or by codec alias and returns its implemetation class.
+     * <p/>
+     * Codec aliases are case insensitive.
+     * <p/>
+     * The code alias is the short class name (without the package name).
+     * If the short class name ends with 'Codec', then there are two aliases for
+     * the codec, the complete short class name and the short class name without
+     * the 'Codec' ending. For example for the 'GzipCodec' codec class name the
+     * alias are 'gzip' and 'gzipcodec'.
+     *
+     * @param codecName the canonical class name of the codec
+     * @return the codec class
+     */
+    public Class<? extends CompressionCodec> getCodecClassByName(String codecName) {
+      CompressionCodec codec = getCodecByName(codecName);
+      if (codec == null) {
+        return null;
+      }
+      return codec.getClass();
+    }
+  
   /**
    * Removes a suffix from a filename, if it has it.
    * @param filename the filename to strip

src/docs/src/documentation/content/xdocs/hdfs_imageviewer.xml

@@ -0,0 +1,427 @@
+<?xml version="1.0"?>
+<!--
+  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.
+-->
+
+<!DOCTYPE document PUBLIC "-//APACHE//DTD Documentation V2.0//EN" "http://forrest.apache.org/dtd/document-v20.dtd">
+
+<document>
+
+  <header>
+    <title>Offline Image Viewer Guide</title>
+  </header>
+
+  <body>
+
+    <section>
+      <title>Overview</title>
+
+      <p>The Offline Image Viewer is a tool to dump the contents of hdfs
+      fsimage files to human-readable formats in order to allow offline analysis
+      and examination of an Hadoop cluster's namespace. The tool is able to
+      process very large image files relatively quickly, converting them to
+      one of several output formats. The tool handles the layout formats that
+      were included with Hadoop versions 16 and up. If the tool is not able to
+      process an image file, it will exit cleanly. The Offline Image Viewer does not require
+      an Hadoop cluster to be running; it is entirely offline in its operation.</p>
+
+      <p>The Offline Image Viewer provides several output processors:</p>
+        <ol>
+        <li><strong>Ls</strong> is the default output processor. It closely mimics the format of
+          the <code>lsr </code> command. It includes the same fields, in the same order, as
+          <code>lsr </code>: directory or file flag, permissions, replication, owner, group,
+          file size, modification date, and full path. Unlike the <code>lsr </code> command,
+          the root path is included. One important difference between the output
+          of the <code>lsr </code> command this processor, is that this output is not sorted
+          by directory name and contents. Rather, the files are listed in the
+          order in which they are stored in the fsimage file. Therefore, it is
+          not possible to directly compare the output of the <code>lsr </code> command this
+          this tool. The Ls processor uses information contained within the Inode blocks to
+          calculate file sizes and ignores the <code>-skipBlocks</code> option.</li>
+        <li><strong>Indented</strong> provides a more complete view of the fsimage's contents,
+          including all of the information included in the image, such as image
+          version, generation stamp and inode- and block-specific listings. This
+          processor uses indentation to organize the output into a hierarchal manner.
+          The <code>lsr </code> format is suitable for easy human comprehension.</li>
+        <li><strong>Delimited</strong> provides one file per line consisting of the path,
+        replication, modification time, access time, block size, number of blocks, file size,
+        namespace quota, diskspace quota, permissions, username and group name. If run against
+        an fsimage that does not contain any of these fields, the field's column will be included,
+        but no data recorded. The default record delimiter is a tab, but this may be changed
+        via the <code>-delimiter</code> command line argument. This processor is designed to
+        create output that is easily analyzed by other tools, such as <a href="http://hadoop.apache.org/pig/">Apache Pig</a>. 
+        See the <a href="#analysis">Analyzing Results</a> section
+        for further information on using this processor to analyze the contents of fsimage files.</li>
+        <li><strong>XML</strong> creates an XML document of the fsimage and includes all of the
+          information within the fsimage, similar to the <code>lsr </code> processor. The output
+          of this processor is amenable to automated processing and analysis with XML tools.
+          Due to the verbosity of the XML syntax, this processor will also generate
+          the largest amount of output.</li>
+        <li><strong>FileDistribution</strong> is the tool for analyzing file 
+          sizes in the namespace image. In order to run the tool one should 
+          define a range of integers <code>[0, maxSize]</code> by specifying
+          <code>maxSize</code> and a <code>step</code>.
+          The range of integers is divided into segments of size
+          <code>step</code>:
+          <code>[0, s</code><sub>1</sub><code>, ..., s</code><sub>n-1</sub><code>, maxSize]</code>, 
+          and the processor calculates how many files in the system fall into 
+          each segment <code>[s</code><sub>i-1</sub><code>, s</code><sub>i</sub><code>)</code>.
+          Note that files larger than <code>maxSize</code> always fall into 
+          the very last segment.
+          The output file is formatted as a tab separated two column table:
+          Size and NumFiles. Where Size represents the start of the segment,
+          and numFiles is the number of files form the image which size falls
+          in this segment.</li>
+        </ol>
+
+    </section> <!-- overview -->
+
+    <section>
+      <title>Usage</title>
+
+      <section>
+        <title>Basic</title>
+        <p>The simplest usage of the Offline Image Viewer is to provide just an input and output
+          file, via the <code>-i</code> and <code>-o</code> command-line switches:</p>
+
+        <p><code>bash$ bin/hadoop oiv -i fsimage -o fsimage.txt</code><br/></p>
+
+        <p>This will create a file named fsimage.txt in the current directory using
+        the Ls output processor.  For very large image files, this process may take
+        several minutes.</p>
+
+        <p>One can specify which output processor via the command-line switch <code>-p</code>.
+        For instance:</p>
+        <p><code>bash$ bin/hadoop oiv -i fsimage -o fsimage.xml -p XML</code><br/></p>
+
+        <p>or</p>
+
+        <p><code>bash$ bin/hadoop oiv -i fsimage -o fsimage.txt -p Indented</code><br/></p>
+
+        <p>This will run the tool using either the XML or Indented output processor,
+        respectively.</p>
+
+        <p>One command-line option worth considering is <code>-skipBlocks</code>, which
+        prevents the tool from explicitly enumerating all of the blocks that make up
+        a file in the namespace. This is useful for file systems that have very large
+        files. Enabling this option can significantly decrease the size of the resulting
+        output, as individual blocks are not included. Note, however, that the Ls processor
+        needs to enumerate the blocks and so overrides this option.</p>
+
+      </section> <!-- Basic -->
+      <section id="Example">
+        <title>Example</title>
+
+<p>Consider the following contrived namespace:</p>
+<source>
+drwxr-xr-x   - theuser supergroup          0 2009-03-16 21:17 /anotherDir 
+
+-rw-r--r--   3 theuser supergroup  286631664 2009-03-16 21:15 /anotherDir/biggerfile 
+
+-rw-r--r--   3 theuser supergroup       8754 2009-03-16 21:17 /anotherDir/smallFile 
+
+drwxr-xr-x   - theuser supergroup          0 2009-03-16 21:11 /mapredsystem 
+
+drwxr-xr-x   - theuser supergroup          0 2009-03-16 21:11 /mapredsystem/theuser 
+
+drwxr-xr-x   - theuser supergroup          0 2009-03-16 21:11 /mapredsystem/theuser/mapredsystem 
+
+drwx-wx-wx   - theuser supergroup          0 2009-03-16 21:11 /mapredsystem/theuser/mapredsystem/ip.redacted.com 
+
+drwxr-xr-x   - theuser supergroup          0 2009-03-16 21:12 /one 
+
+drwxr-xr-x   - theuser supergroup          0 2009-03-16 21:12 /one/two 
+
+drwxr-xr-x   - theuser supergroup          0 2009-03-16 21:16 /user 
+
+drwxr-xr-x   - theuser supergroup          0 2009-03-16 21:19 /user/theuser 
+</source>          
+
+<p>Applying the Offline Image Processor against this file with default options would result in the following output:</p>
+<source>
+machine:hadoop-0.21.0-dev theuser$ bin/hadoop oiv -i fsimagedemo -o fsimage.txt 
+
+drwxr-xr-x  -   theuser supergroup            0 2009-03-16 14:16 / 
+
+drwxr-xr-x  -   theuser supergroup            0 2009-03-16 14:17 /anotherDir 
+
+drwxr-xr-x  -   theuser supergroup            0 2009-03-16 14:11 /mapredsystem 
+
+drwxr-xr-x  -   theuser supergroup            0 2009-03-16 14:12 /one 
+
+drwxr-xr-x  -   theuser supergroup            0 2009-03-16 14:16 /user 
+
+-rw-r--r--  3   theuser supergroup    286631664 2009-03-16 14:15 /anotherDir/biggerfile 
+
+-rw-r--r--  3   theuser supergroup         8754 2009-03-16 14:17 /anotherDir/smallFile 
+
+drwxr-xr-x  -   theuser supergroup            0 2009-03-16 14:11 /mapredsystem/theuser 
+
+drwxr-xr-x  -   theuser supergroup            0 2009-03-16 14:11 /mapredsystem/theuser/mapredsystem 
+
+drwx-wx-wx  -   theuser supergroup            0 2009-03-16 14:11 /mapredsystem/theuser/mapredsystem/ip.redacted.com 
+
+drwxr-xr-x  -   theuser supergroup            0 2009-03-16 14:12 /one/two 
+
+drwxr-xr-x  -   theuser supergroup            0 2009-03-16 14:19 /user/theuser 
+</source>
+
+<p>Similarly, applying the Indented processor would generate output that begins with:</p>
+<source>
+machine:hadoop-0.21.0-dev theuser$ bin/hadoop oiv -i fsimagedemo -p Indented -o fsimage.txt 
+
+FSImage 
+
+  ImageVersion = -19 
+
+  NamespaceID = 2109123098 
+
+  GenerationStamp = 1003 
+
+  INodes [NumInodes = 12] 
+
+    Inode 
+
+      INodePath =  
+
+      Replication = 0 
+
+      ModificationTime = 2009-03-16 14:16 
+
+      AccessTime = 1969-12-31 16:00 
+
+      BlockSize = 0 
+
+      Blocks [NumBlocks = -1] 
+
+      NSQuota = 2147483647 
+
+      DSQuota = -1 
+
+      Permissions 
+
+        Username = theuser 
+
+        GroupName = supergroup 
+
+        PermString = rwxr-xr-x 
+
+   remaining output omitted
+</source>          
+          
+      </section> <!-- example-->
+
+    </section>
+
+    <section id="options">
+        <title>Options</title>
+
+        <section>
+        <title>Option Index</title>
+        <table>
+          <tr><th> Flag </th><th> Description </th></tr>
+          <tr><td><code>[-i|--inputFile] &lt;input file&gt;</code></td>
+              <td>Specify the input fsimage file to process. Required.</td></tr>
+          <tr><td><code>[-o|--outputFile] &lt;output file&gt;</code></td>
+              <td>Specify the output filename, if the specified output processor
+              generates one. If the specified file already exists, it is silently overwritten. Required.
+              </td></tr>
+          <tr><td><code>[-p|--processor] &lt;processor&gt;</code></td>
+                  <td>Specify the image processor to apply against the image file. Currently
+                    valid options are Ls (default), XML and Indented..
+                  </td></tr>
+          <tr><td><code>-skipBlocks</code></td>
+              <td>Do not enumerate individual blocks within files. This may save processing time
+              and outfile file space on namespaces with very large files. The <code>Ls</code> processor reads
+              the blocks to correctly determine file sizes and ignores this option.</td></tr>
+          <tr><td><code>-printToScreen</code></td>
+              <td>Pipe output of processor to console as well as specified file. On extremely 
+              large namespaces, this may increase processing time by an order of magnitude.</td></tr>
+           <tr><td><code>-delimiter &lt;arg&gt;</code></td>
+                  <td>When used in conjunction with the Delimited processor, replaces the default
+	                    tab delimiter with the string specified by <code>arg</code>.</td></tr>
+          <tr><td><code>[-h|--help]</code></td>
+              <td>Display the tool usage and help information and exit.</td></tr>
+            </table>
+          </section> <!-- options -->
+    </section>
+   
+    <section id="analysis">
+      <title>Analyzing Results</title>
+      <p>The Offline Image Viewer makes it easy to gather large amounts of data about the hdfs namespace.
+         This information can then be used to explore file system usage patterns or find
+        specific files that match arbitrary criteria, along with other types of namespace analysis. The Delimited 
+         image processor in particular creates
+        output that is amenable to further processing by tools such as <a href="http://hadoop.apache.org/pig/">Apache Pig</a>. Pig provides a particularly
+        good choice for analyzing these data as it is able to deal with the output generated from a small fsimage
+        but also scales up to consume data from extremely large file systems.</p>
+      <p>The Delimited image processor generates lines of text separated, by default, by tabs and includes
+        all of the fields that are common between constructed files and files that were still under constructed
+        when the fsimage was generated. Examples scripts are provided demonstrating how to use this output to 
+        accomplish three tasks: determine the number of files each user has created on the file system,
+        find files were created but have not accessed, and find probable duplicates of large files by comparing
+        the size of each file.</p>
+      <p>Each of the following scripts assumes you have generated an output file using the Delimited processor named
+        <code>foo</code> and will be storing the results of the Pig analysis in a file named <code>results</code>.</p>
+      <section>
+      <title>Total Number of Files for Each User</title>
+      <p>This script processes each path within the namespace, groups them by the file owner and determines the total
+      number of files each user owns.</p>
+      <p><strong>numFilesOfEachUser.pig:</strong></p>
+        <source>
+-- This script determines the total number of files each user has in
+-- the namespace. Its output is of the form:
+--   username, totalNumFiles
+
+-- Load all of the fields from the file
+A = LOAD '$inputFile' USING PigStorage('\t') AS (path:chararray,
+                                                 replication:int,
+                                                 modTime:chararray,
+                                                 accessTime:chararray,
+                                                 blockSize:long,
+                                                 numBlocks:int,
+                                                 fileSize:long,
+                                                 NamespaceQuota:int,
+                                                 DiskspaceQuota:int,
+                                                 perms:chararray,
+                                                 username:chararray,
+                                                 groupname:chararray);
+
+
+-- Grab just the path and username
+B = FOREACH A GENERATE path, username;
+
+-- Generate the sum of the number of paths for each user
+C = FOREACH (GROUP B BY username) GENERATE group, COUNT(B.path);
+
+-- Save results
+STORE C INTO '$outputFile';
+        </source>
+      <p>This script can be run against pig with the following command:</p>
+      <p><code>bin/pig -x local -param inputFile=../foo -param outputFile=../results ../numFilesOfEachUser.pig</code><br/></p>
+      <p>The output file's content will be similar to that below:</p>
+      <p>
+        <code>bart  1</code><br/>
+        <code>lisa  16</code><br/>
+        <code>homer 28</code><br/>
+        <code>marge 2456</code><br/>
+      </p>
+      </section>
+      
+      <section><title>Files That Have Never Been Accessed</title>
+      <p>This script finds files that were created but whose access times were never changed, meaning they were never opened or viewed.</p>
+            <p><strong>neverAccessed.pig:</strong></p>
+      <source>
+-- This script generates a list of files that were created but never
+-- accessed, based on their AccessTime
+
+-- Load all of the fields from the file
+A = LOAD '$inputFile' USING PigStorage('\t') AS (path:chararray,
+                                                 replication:int,
+                                                 modTime:chararray,
+                                                 accessTime:chararray,
+                                                 blockSize:long,
+                                                 numBlocks:int,
+                                                 fileSize:long,
+                                                 NamespaceQuota:int,
+                                                 DiskspaceQuota:int,
+                                                 perms:chararray,
+                                                 username:chararray,
+                                                 groupname:chararray);
+
+-- Grab just the path and last time the file was accessed
+B = FOREACH A GENERATE path, accessTime;
+
+-- Drop all the paths that don't have the default assigned last-access time
+C = FILTER B BY accessTime == '1969-12-31 16:00';
+
+-- Drop the accessTimes, since they're all the same
+D = FOREACH C GENERATE path;
+
+-- Save results
+STORE D INTO '$outputFile';
+      </source>
+      <p>This script can be run against pig with the following command and its output file's content will be a list of files that were created but never viewed afterwards.</p>
+      <p><code>bin/pig -x local -param inputFile=../foo -param outputFile=../results ../neverAccessed.pig</code><br/></p>
+      </section>
+      <section><title>Probable Duplicated Files Based on File Size</title>
+      <p>This script groups files together based on their size, drops any that are of less than 100mb and returns a list of the file size, number of files found and a tuple of the file paths.  This can be used to find likely duplicates within the filesystem namespace.</p>
+      
+            <p><strong>probableDuplicates.pig:</strong></p>
+      <source>
+-- This script finds probable duplicate files greater than 100 MB by
+-- grouping together files based on their byte size. Files of this size
+-- with exactly the same number of bytes can be considered probable
+-- duplicates, but should be checked further, either by comparing the
+-- contents directly or by another proxy, such as a hash of the contents.
+-- The scripts output is of the type:
+--    fileSize numProbableDuplicates {(probableDup1), (probableDup2)}
+
+-- Load all of the fields from the file
+A = LOAD '$inputFile' USING PigStorage('\t') AS (path:chararray,
+                                                 replication:int,
+                                                 modTime:chararray,
+                                                 accessTime:chararray,
+                                                 blockSize:long,
+                                                 numBlocks:int,
+                                                 fileSize:long,
+                                                 NamespaceQuota:int,
+                                                 DiskspaceQuota:int,
+                                                 perms:chararray,
+                                                 username:chararray,
+                                                 groupname:chararray);
+
+-- Grab the pathname and filesize
+B = FOREACH A generate path, fileSize;
+
+-- Drop files smaller than 100 MB
+C = FILTER B by fileSize > 100L  * 1024L * 1024L;
+
+-- Gather all the files of the same byte size
+D = GROUP C by fileSize;
+
+-- Generate path, num of duplicates, list of duplicates
+E = FOREACH D generate group AS fileSize, COUNT(C) as numDupes, C.path AS files;
+
+-- Drop all the files where there are only one of them
+F = FILTER E by numDupes > 1L;
+
+-- Sort by the size of the files
+G = ORDER F by fileSize;
+
+-- Save results
+STORE G INTO '$outputFile';
+      </source>
+      <p>This script can be run against pig with the following command:</p>
+      <p><code>bin/pig -x local -param inputFile=../foo -param outputFile=../results ../probableDuplicates.pig</code><br/></p>
+      <p> The output file's content will be similar to that below:</p>
+      
+<source>
+1077288632 2 {(/user/tennant/work1/part-00501),(/user/tennant/work1/part-00993)} 
+1077288664 4 {(/user/tennant/work0/part-00567),(/user/tennant/work0/part-03980),(/user/tennant/work1/part-00725),(/user/eccelston/output/part-03395)} 
+1077288668 3 {(/user/tennant/work0/part-03705),(/user/tennant/work0/part-04242),(/user/tennant/work1/part-03839)} 
+1077288698 2 {(/user/tennant/work0/part-00435),(/user/eccelston/output/part-01382)} 
+1077288702 2 {(/user/tennant/work0/part-03864),(/user/eccelston/output/part-03234)} 
+</source>      
+      <p>Each line includes the file size in bytes that was found to be duplicated, the number of duplicates found, and a list of the duplicated paths. 
+      Files less than 100MB are ignored, providing a reasonable likelihood that files of these exact sizes may be duplicates.</p>
+      </section>
+    </section>
+
+
+  </body>
+
+</document>

src/docs/src/documentation/content/xdocs/site.xml

@@ -63,6 +63,7 @@ See http://forrest.apache.org/docs/linking.html for more info.
     <hdfs_perm        label="Permissions" href="hdfs_permissions_guide.html" />
     <hdfs_quotas      label="Quotas" href="hdfs_quota_admin_guide.html" />
     <hdfs_SLG         label="Synthetic Load Generator"  href="SLG_user_guide.html" />
+    <hdfs_imageviewer	label="Offline Image Viewer"	href="hdfs_imageviewer.html" />
     <hftp label="HFTP" href="hftp.html"/>
     <webhdfs label="WebHDFS REST API" href="webhdfs.html" />
     <hdfs_libhdfs       label="C API libhdfs" href="libhdfs.html" />

src/hdfs/org/apache/hadoop/hdfs/DFSConfigKeys.java

@@ -130,6 +130,8 @@ public class DFSConfigKeys extends CommonConfigurationKeys {
   public static final long    DFS_NAMENODE_DELEGATION_TOKEN_RENEW_INTERVAL_DEFAULT = 24*60*60*1000;
   public static final String  DFS_NAMENODE_DELEGATION_TOKEN_MAX_LIFETIME_KEY = "dfs.namenode.delegation.token.max-lifetime";
   public static final long    DFS_NAMENODE_DELEGATION_TOKEN_MAX_LIFETIME_DEFAULT = 7*24*60*60*1000;
+  public static final String  DFS_NAMENODE_DELEGATION_TOKEN_ALWAYS_USE_KEY = "dfs.namenode.delegation.token.always-use"; // for tests
+  public static final boolean DFS_NAMENODE_DELEGATION_TOKEN_ALWAYS_USE_DEFAULT = false;
 
   //Following keys have no defaults
   public static final String  DFS_DATANODE_DATA_DIR_KEY = "dfs.datanode.data.dir";

src/hdfs/org/apache/hadoop/hdfs/protocol/LayoutVersion.java

@@ -0,0 +1,231 @@
+/**
+ * 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.hdfs.protocol;
+
+import java.util.EnumSet;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+
+/**
+ * This class tracks changes in the layout version of HDFS.
+ * 
+ * Layout version is changed for following reasons:
+ * <ol>
+ * <li>The layout of how namenode or datanode stores information 
+ * on disk changes.</li>
+ * <li>A new operation code is added to the editlog.</li>
+ * <li>Modification such as format of a record, content of a record 
+ * in editlog or fsimage.</li>
+ * </ol>
+ * <br>
+ * <b>How to update layout version:<br></b>
+ * When a change requires new layout version, please add an entry into
+ * {@link Feature} with a short enum name, new layout version and description
+ * of the change. Please see {@link Feature} for further details.
+ * <br>
+ */
+@InterfaceAudience.Private
+public class LayoutVersion {
+ 
+  /**
+   * Version in which HDFS-2991 was fixed. This bug caused OP_ADD to
+   * sometimes be skipped for append() calls. If we see such a case when
+   * loading the edits, but the version is known to have that bug, we
+   * workaround the issue. Otherwise we should consider it a corruption
+   * and bail.
+   */
+  public static final int BUGFIX_HDFS_2991_VERSION = -40;
+
+  /**
+   * Enums for features that change the layout version.
+   * <br><br>
+   * To add a new layout version:
+   * <ul>
+   * <li>Define a new enum constant with a short enum name, the new layout version 
+   * and description of the added feature.</li>
+   * <li>When adding a layout version with an ancestor that is not same as
+   * its immediate predecessor, use the constructor where a spacific ancestor
+   * can be passed.
+   * </li>
+   * </ul>
+   */
+  public static enum Feature {
+    NAMESPACE_QUOTA(-16, "Support for namespace quotas"),
+    FILE_ACCESS_TIME(-17, "Support for access time on files"),
+    DISKSPACE_QUOTA(-18, "Support for disk space quotas"),
+    STICKY_BIT(-19, "Support for sticky bits"),
+    APPEND_RBW_DIR(-20, "Datanode has \"rbw\" subdirectory for append"),
+    ATOMIC_RENAME(-21, "Support for atomic rename"),
+    CONCAT(-22, "Support for concat operation"),
+    SYMLINKS(-23, "Support for symbolic links"),
+    DELEGATION_TOKEN(-24, "Support for delegation tokens for security"),
+    FSIMAGE_COMPRESSION(-25, "Support for fsimage compression"),
+    FSIMAGE_CHECKSUM(-26, "Support checksum for fsimage"),
+    REMOVE_REL13_DISK_LAYOUT_SUPPORT(-27, "Remove support for 0.13 disk layout"),
+    EDITS_CHESKUM(-28, "Support checksum for editlog"),
+    UNUSED(-29, "Skipped version"),
+    FSIMAGE_NAME_OPTIMIZATION(-30, "Store only last part of path in fsimage"),
+    RESERVED_REL20_203(-31, -19, "Reserved for release 0.20.203", true,
+        DELEGATION_TOKEN),
+    RESERVED_REL20_204(-32, -31, "Reserved for release 0.20.204", true),
+    RESERVED_REL22(-33, -27, "Reserved for release 0.22", true),
+    RESERVED_REL23(-34, -30, "Reserved for release 0.23", true),
+    // layout versions -35 - -40 are features not present on this branch
+    RESERVED_REL1_2_0(-41, -32, "Reserved for release 1.2.0", true, CONCAT);
+    
+    final int lv;
+    final int ancestorLV;
+    final String description;
+    final boolean reserved;
+    final Feature[] specialFeatures;
+    
+    /**
+     * Feature that is added at layout version {@code lv} - 1. 
+     * @param lv new layout version with the addition of this feature
+     * @param description description of the feature
+     */
+    Feature(final int lv, final String description) {
+      this(lv, lv + 1, description, false);
+    }
+
+    /**
+     * Feature that is added at layout version {@code ancestoryLV}.
+     * @param lv new layout version with the addition of this feature
+     * @param ancestorLV layout version from which the new lv is derived from.
+     * @param description description of the feature
+     * @param reserved true when this is a layout version reserved for previous
+     *          verions
+     * @param features set of features that are to be enabled for this version
+     */
+    Feature(final int lv, final int ancestorLV, final String description,
+        boolean reserved, Feature... features) {
+      this.lv = lv;
+      this.ancestorLV = ancestorLV;
+      this.description = description;
+      this.reserved = reserved;
+      specialFeatures = features;
+    }
+    
+    /** 
+     * Accessor method for feature layout version 
+     * @return int lv value
+     */
+    public int getLayoutVersion() {
+      return lv;
+    }
+
+    /** 
+     * Accessor method for feature ancestor layout version 
+     * @return int ancestor LV value
+     */
+    public int getAncestorLayoutVersion() {
+      return ancestorLV;
+    }
+
+    /** 
+     * Accessor method for feature description 
+     * @return String feature description 
+     */
+    public String getDescription() {
+      return description;
+    }
+    
+    public boolean isReservedForOldRelease() {
+      return reserved;
+    }
+  }
+  
+  // Build layout version and corresponding feature matrix
+  static final Map<Integer, EnumSet<Feature>>map = 
+    new HashMap<Integer, EnumSet<Feature>>();
+  
+  // Static initialization 
+  static {
+    initMap();
+  }
+  
+  /**
+   * Initialize the map of a layout version and EnumSet of {@link Feature}s 
+   * supported.
+   */
+  private static void initMap() {
+    // Go through all the enum constants and build a map of
+    // LayoutVersion <-> EnumSet of all supported features in that LayoutVersion
+    for (Feature f : Feature.values()) {
+      EnumSet<Feature> ancestorSet = map.get(f.ancestorLV);
+      if (ancestorSet == null) {
+        ancestorSet = EnumSet.noneOf(Feature.class); // Empty enum set
+        map.put(f.ancestorLV, ancestorSet);
+      }
+      EnumSet<Feature> featureSet = EnumSet.copyOf(ancestorSet);
+      if (f.specialFeatures != null) {
+        for (Feature specialFeature : f.specialFeatures) {
+          featureSet.add(specialFeature);
+        }
+      }
+      featureSet.add(f);
+      map.put(f.lv, featureSet);
+    }
+  }
+  
+  /**
+   * Gets formatted string that describes {@link LayoutVersion} information.
+   */
+  public static String getString() {
+    final StringBuilder buf = new StringBuilder();
+    buf.append("Feature List:\n");
+    for (Feature f : Feature.values()) {
+      buf.append(f).append(" introduced in layout version ")
+          .append(f.lv).append(" (").
+      append(f.description).append(")\n");
+    }
+    
+    buf.append("\n\nLayoutVersion and supported features:\n");
+    for (Feature f : Feature.values()) {
+      buf.append(f.lv).append(": ").append(map.get(f.lv))
+          .append("\n");
+    }
+    return buf.toString();
+  }
+  
+  /**
+   * Returns true if a given feature is supported in the given layout version
+   * @param f Feature
+   * @param lv LayoutVersion
+   * @return true if {@code f} is supported in layout version {@code lv}
+   */
+  public static boolean supports(final Feature f, final int lv) {
+    final EnumSet<Feature> set =  map.get(lv);
+    return set != null && set.contains(f);
+  }
+  
+  /**
+   * Get the current layout version
+   */
+  public static int getCurrentLayoutVersion() {
+    Feature[] values = Feature.values();
+    for (int i = values.length -1; i >= 0; i--) {
+      if (!values[i].isReservedForOldRelease()) {
+        return values[i].lv;
+      }
+    }
+    throw new AssertionError("All layout versions are reserved.");
+  }
+}

src/hdfs/org/apache/hadoop/hdfs/server/namenode/FSImage.java

@@ -1820,7 +1820,7 @@ public class FSImage extends Storage {
   }
 
   static private final UTF8 U_STR = new UTF8();
-  static String readString(DataInputStream in) throws IOException {
+  public static String readString(DataInputStream in) throws IOException {
     U_STR.readFields(in);
     return U_STR.toString();
   }
@@ -1830,7 +1830,7 @@ public class FSImage extends Storage {
     return s.isEmpty()? null: s;
   }
 
-  static byte[] readBytes(DataInputStream in) throws IOException {
+  public static byte[] readBytes(DataInputStream in) throws IOException {
     U_STR.readFields(in);
     int len = U_STR.getLength();
     byte[] bytes = new byte[len];

src/hdfs/org/apache/hadoop/hdfs/server/namenode/NameNode.java

@@ -17,6 +17,9 @@
  */
 package org.apache.hadoop.hdfs.server.namenode;
 
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_DELEGATION_TOKEN_ALWAYS_USE_KEY;
+import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_DELEGATION_TOKEN_ALWAYS_USE_DEFAULT;
+
 import java.io.File;
 import java.io.IOException;
 import java.net.InetSocketAddress;
@@ -294,8 +297,15 @@ public class NameNode implements ClientProtocol, DatanodeProtocol,
     
     myMetrics = NameNodeInstrumentation.create(conf);
     this.namesystem = new FSNamesystem(this, conf);
-
-    if (UserGroupInformation.isSecurityEnabled()) {
+    
+    // For testing purposes, allow the DT secret manager to be started regardless
+    // of whether security is enabled.
+    boolean alwaysUseDelegationTokensForTests = 
+      conf.getBoolean(DFS_NAMENODE_DELEGATION_TOKEN_ALWAYS_USE_KEY,
+          DFS_NAMENODE_DELEGATION_TOKEN_ALWAYS_USE_DEFAULT);
+
+    if (UserGroupInformation.isSecurityEnabled() ||
+        alwaysUseDelegationTokensForTests) {
       namesystem.activateSecretManager();
     }
 

src/hdfs/org/apache/hadoop/hdfs/tools/offlineImageViewer/DelimitedImageVisitor.java

@@ -0,0 +1,172 @@
+/**
+ * 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.hdfs.tools.offlineImageViewer;
+
+import java.io.IOException;
+import java.util.AbstractMap;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedList;
+
+/**
+ * A DelimitedImageVisitor generates a text representation of the fsimage,
+ * with each element separated by a delimiter string.  All of the elements
+ * common to both inodes and inodes-under-construction are included. When 
+ * processing an fsimage with a layout version that did not include an 
+ * element, such as AccessTime, the output file will include a column
+ * for the value, but no value will be included.
+ * 
+ * Individual block information for each file is not currently included.
+ * 
+ * The default delimiter is tab, as this is an unlikely value to be included
+ * an inode path or other text metadata.  The delimiter value can be via the
+ * constructor.
+ */
+class DelimitedImageVisitor extends TextWriterImageVisitor {
+  private static final String defaultDelimiter = "\t"; 
+  
+  final private LinkedList<ImageElement> elemQ = new LinkedList<ImageElement>();
+  private long fileSize = 0l;
+  // Elements of fsimage we're interested in tracking
+  private final Collection<ImageElement> elementsToTrack;
+  // Values for each of the elements in elementsToTrack
+  private final AbstractMap<ImageElement, String> elements = 
+                                            new HashMap<ImageElement, String>();
+  private final String delimiter;
+
+  {
+    elementsToTrack = new ArrayList<ImageElement>();
+    
+    // This collection determines what elements are tracked and the order
+    // in which they are output
+    Collections.addAll(elementsToTrack,  ImageElement.INODE_PATH,
+                                         ImageElement.REPLICATION,
+                                         ImageElement.MODIFICATION_TIME,
+                                         ImageElement.ACCESS_TIME,
+                                         ImageElement.BLOCK_SIZE,
+                                         ImageElement.NUM_BLOCKS,
+                                         ImageElement.NUM_BYTES,
+                                         ImageElement.NS_QUOTA,
+                                         ImageElement.DS_QUOTA,
+                                         ImageElement.PERMISSION_STRING,
+                                         ImageElement.USER_NAME,
+                                         ImageElement.GROUP_NAME);
+  }
+  
+  public DelimitedImageVisitor(String filename) throws IOException {
+    this(filename, false);
+  }
+
+  public DelimitedImageVisitor(String outputFile, boolean printToScreen) 
+                                                           throws IOException {
+    this(outputFile, printToScreen, defaultDelimiter);
+  }
+  
+  public DelimitedImageVisitor(String outputFile, boolean printToScreen, 
+                               String delimiter) throws IOException {
+    super(outputFile, printToScreen);
+    this.delimiter = delimiter;
+    reset();
+  }
+
+  /**
+   * Reset the values of the elements we're tracking in order to handle
+   * the next file
+   */
+  private void reset() {
+    elements.clear();
+    for(ImageElement e : elementsToTrack) 
+      elements.put(e, null);
+    
+    fileSize = 0l;
+  }
+  
+  @Override
+  void leaveEnclosingElement() throws IOException {
+    ImageElement elem = elemQ.pop();
+
+    // If we're done with an inode, write out our results and start over
+    if(elem == ImageElement.INODE || 
+       elem == ImageElement.INODE_UNDER_CONSTRUCTION) {
+      writeLine();
+      write("\n");
+      reset();
+    }
+  }
+
+  /**
+   * Iterate through all the elements we're tracking and, if a value was
+   * recorded for it, write it out.
+   */
+  private void writeLine() throws IOException {
+    Iterator<ImageElement> it = elementsToTrack.iterator();
+    
+    while(it.hasNext()) {
+      ImageElement e = it.next();
+      
+      String v = null;
+      if(e == ImageElement.NUM_BYTES)
+        v = String.valueOf(fileSize);
+      else
+        v = elements.get(e);
+      
+      if(v != null)
+        write(v);
+      
+      if(it.hasNext())
+        write(delimiter);
+    }
+  }
+
+  @Override
+  void visit(ImageElement element, String value) throws IOException {
+    // Explicitly label the root path
+    if(element == ImageElement.INODE_PATH && value.equals(""))
+      value = "/";
+    
+    // Special case of file size, which is sum of the num bytes in each block
+    if(element == ImageElement.NUM_BYTES)
+      fileSize += Long.valueOf(value);
+    
+    if(elements.containsKey(element) && element != ImageElement.NUM_BYTES)
+      elements.put(element, value);
+    
+  }
+
+  @Override
+  void visitEnclosingElement(ImageElement element) throws IOException {
+    elemQ.push(element);
+  }
+
+  @Override
+  void visitEnclosingElement(ImageElement element, ImageElement key,
+      String value) throws IOException {
+    // Special case as numBlocks is an attribute of the blocks element
+    if(key == ImageElement.NUM_BLOCKS 
+        && elements.containsKey(ImageElement.NUM_BLOCKS))
+      elements.put(key, value);
+    
+    elemQ.push(element);
+  }
+  
+  @Override
+  void start() throws IOException { /* Nothing to do */ }
+}

src/hdfs/org/apache/hadoop/hdfs/tools/offlineImageViewer/DepthCounter.java

@@ -0,0 +1,36 @@
+/**
+ * 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.hdfs.tools.offlineImageViewer;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+/**
+ * Utility class for tracking descent into the structure of the
+ * Visitor class (ImageVisitor, EditsVisitor etc.)
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Unstable
+public class DepthCounter {
+  private int depth = 0;
+
+  public void incLevel() { depth++; }
+  public void decLevel() { if(depth >= 1) depth--; }
+  public int  getLevel() { return depth; }
+}
+

src/hdfs/org/apache/hadoop/hdfs/tools/offlineImageViewer/FileDistributionVisitor.java

@@ -0,0 +1,182 @@
+/**
+ * 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.hdfs.tools.offlineImageViewer;
+
+import java.io.IOException;
+import java.util.LinkedList;
+
+/**
+ * File size distribution visitor.
+ * 
+ * <h3>Description.</h3>
+ * This is the tool for analyzing file sizes in the namespace image.
+ * In order to run the tool one should define a range of integers
+ * <tt>[0, maxSize]</tt> by specifying <tt>maxSize</tt> and a <tt>step</tt>.
+ * The range of integers is divided into segments of size <tt>step</tt>: 
+ * <tt>[0, s<sub>1</sub>, ..., s<sub>n-1</sub>, maxSize]</tt>,
+ * and the visitor calculates how many files in the system fall into 
+ * each segment <tt>[s<sub>i-1</sub>, s<sub>i</sub>)</tt>. 
+ * Note that files larger than <tt>maxSize</tt> always fall into 
+ * the very last segment.
+ * 
+ * <h3>Input.</h3>
+ * <ul>
+ * <li><tt>filename</tt> specifies the location of the image file;</li>
+ * <li><tt>maxSize</tt> determines the range <tt>[0, maxSize]</tt> of files
+ * sizes considered by the visitor;</li>
+ * <li><tt>step</tt> the range is divided into segments of size step.</li>
+ * </ul>
+ *
+ * <h3>Output.</h3>
+ * The output file is formatted as a tab separated two column table:
+ * Size and NumFiles. Where Size represents the start of the segment,
+ * and numFiles is the number of files form the image which size falls in 
+ * this segment.
+ */
+class FileDistributionVisitor extends TextWriterImageVisitor {
+  final private LinkedList<ImageElement> elemS = new LinkedList<ImageElement>();
+
+  private final static long MAX_SIZE_DEFAULT = 0x2000000000L;   // 1/8 TB = 2^37
+  private final static int INTERVAL_DEFAULT = 0x200000;         // 2 MB = 2^21
+
+  private int[] distribution;
+  private long maxSize;
+  private int step;
+
+  private int totalFiles;
+  private int totalDirectories;
+  private int totalBlocks;
+  private long totalSpace;
+  private long maxFileSize;
+
+  private FileContext current;
+
+  private boolean inInode = false;
+
+  /**
+   * File or directory information.
+   */
+  private static class FileContext {
+    String path;
+    long fileSize;
+    int numBlocks;
+    int replication;
+  }
+
+  public FileDistributionVisitor(String filename,
+                                 long maxSize,
+                                 int step) throws IOException {
+    super(filename, false);
+    this.maxSize = (maxSize == 0 ? MAX_SIZE_DEFAULT : maxSize);
+    this.step = (step == 0 ? INTERVAL_DEFAULT : step);
+    long numIntervals = this.maxSize / this.step;
+    if(numIntervals >= Integer.MAX_VALUE)
+      throw new IOException("Too many distribution intervals " + numIntervals);
+    this.distribution = new int[1 + (int)(numIntervals)];
+    this.totalFiles = 0;
+    this.totalDirectories = 0;
+    this.totalBlocks = 0;
+    this.totalSpace = 0;
+    this.maxFileSize = 0;
+  }
+
+  @Override
+  void start() throws IOException {}
+
+  @Override
+  void finish() throws IOException {
+    // write the distribution into the output file
+    write("Size\tNumFiles\n");
+    for(int i = 0; i < distribution.length; i++)
+      write(((long)i * step) + "\t" + distribution[i] + "\n");
+    System.out.println("totalFiles = " + totalFiles);
+    System.out.println("totalDirectories = " + totalDirectories);
+    System.out.println("totalBlocks = " + totalBlocks);
+    System.out.println("totalSpace = " + totalSpace);
+    System.out.println("maxFileSize = " + maxFileSize);
+    super.finish();
+  }
+
+  @Override
+  void leaveEnclosingElement() throws IOException {
+    ImageElement elem = elemS.pop();
+
+    if(elem != ImageElement.INODE &&
+       elem != ImageElement.INODE_UNDER_CONSTRUCTION)
+      return;
+    inInode = false;
+    if(current.numBlocks < 0) {
+      totalDirectories ++;
+      return;
+    }
+    totalFiles++;
+    totalBlocks += current.numBlocks;
+    totalSpace += current.fileSize * current.replication;
+    if(maxFileSize < current.fileSize)
+      maxFileSize = current.fileSize;
+    int high;
+    if(current.fileSize > maxSize)
+      high = distribution.length-1;
+    else
+      high = (int)Math.ceil((double)current.fileSize / step);
+    distribution[high]++;
+    if(totalFiles % 1000000 == 1)
+      System.out.println("Files processed: " + totalFiles
+          + "  Current: " + current.path);
+  }
+
+  @Override
+  void visit(ImageElement element, String value) throws IOException {
+    if(inInode) {
+      switch(element) {
+      case INODE_PATH:
+        current.path = (value.equals("") ? "/" : value);
+        break;
+      case REPLICATION:
+        current.replication = Integer.valueOf(value);
+        break;
+      case NUM_BYTES:
+        current.fileSize += Long.valueOf(value);
+        break;
+      default:
+        break;
+      }
+    }
+  }
+
+  @Override
+  void visitEnclosingElement(ImageElement element) throws IOException {
+    elemS.push(element);
+    if(element == ImageElement.INODE ||
+       element == ImageElement.INODE_UNDER_CONSTRUCTION) {
+      current = new FileContext();
+      inInode = true;
+    }
+  }
+
+  @Override
+  void visitEnclosingElement(ImageElement element,
+      ImageElement key, String value) throws IOException {
+    elemS.push(element);
+    if(element == ImageElement.INODE ||
+       element == ImageElement.INODE_UNDER_CONSTRUCTION)
+      inInode = true;
+    else if(element == ImageElement.BLOCKS)
+      current.numBlocks = Integer.parseInt(value);
+  }
+}

src/hdfs/org/apache/hadoop/hdfs/tools/offlineImageViewer/ImageLoader.java

@@ -0,0 +1,83 @@
+/**
+ * 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.hdfs.tools.offlineImageViewer;
+
+import java.io.DataInputStream;
+import java.io.IOException;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+
+/**
+ * An ImageLoader can accept a DataInputStream to an Hadoop FSImage file
+ * and walk over its structure using the supplied ImageVisitor.
+ *
+ * Each implementation of ImageLoader is designed to rapidly process an
+ * image file.  As long as minor changes are made from one layout version
+ * to another, it is acceptable to tweak one implementation to read the next.
+ * However, if the layout version changes enough that it would make a
+ * processor slow or difficult to read, another processor should be created.
+ * This allows each processor to quickly read an image without getting
+ * bogged down in dealing with significant differences between layout versions.
+ */
+interface ImageLoader {
+
+  /**
+   * @param in DataInputStream pointing to an Hadoop FSImage file
+   * @param v Visit to apply to the FSImage file
+   * @param enumerateBlocks Should visitor visit each of the file blocks?
+   */
+  public void loadImage(DataInputStream in, ImageVisitor v,
+      boolean enumerateBlocks) throws IOException;
+
+  /**
+   * Can this processor handle the specified version of FSImage file?
+   *
+   * @param version FSImage version file
+   * @return True if this instance can process the file
+   */
+  public boolean canLoadVersion(int version);
+
+  /**
+   * Factory for obtaining version of image loader that can read
+   * a particular image format.
+   */
+  @InterfaceAudience.Private
+  public class LoaderFactory {
+    // Java doesn't support static methods on interfaces, which necessitates
+    // this factory class
+
+    /**
+     * Find an image loader capable of interpreting the specified
+     * layout version number.  If none, return null;
+     *
+     * @param version fsimage layout version number to be processed
+     * @return ImageLoader that can interpret specified version, or null
+     */
+    static public ImageLoader getLoader(int version) {
+      // Easy to add more image processors as they are written
+      ImageLoader[] loaders = { new ImageLoaderCurrent() };
+
+      for (ImageLoader l : loaders) {
+        if (l.canLoadVersion(version))
+          return l;
+      }
+
+      return null;
+    }
+  }
+}

src/hdfs/org/apache/hadoop/hdfs/tools/offlineImageViewer/ImageLoaderCurrent.java

@@ -0,0 +1,465 @@
+/**
+ * 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.hdfs.tools.offlineImageViewer;
+
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.text.DateFormat;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.hdfs.protocol.DatanodeInfo.AdminStates;
+import org.apache.hadoop.hdfs.protocol.LayoutVersion;
+import org.apache.hadoop.hdfs.protocol.LayoutVersion.Feature;
+import org.apache.hadoop.hdfs.security.token.delegation.DelegationTokenIdentifier;
+import org.apache.hadoop.hdfs.server.namenode.FSImage;
+import org.apache.hadoop.hdfs.tools.offlineImageViewer.ImageVisitor.ImageElement;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.WritableUtils;
+import org.apache.hadoop.io.compress.CompressionCodec;
+import org.apache.hadoop.io.compress.CompressionCodecFactory;
+import org.apache.hadoop.security.token.delegation.DelegationKey;
+
+/**
+ * ImageLoaderCurrent processes Hadoop FSImage files and walks over
+ * them using a provided ImageVisitor, calling the visitor at each element
+ * enumerated below.
+ *
+ * The only difference between v18 and v19 was the utilization of the
+ * stickybit.  Therefore, the same viewer can reader either format.
+ *
+ * Versions -19 fsimage layout (with changes from -16 up):
+ * Image version (int)
+ * Namepsace ID (int)
+ * NumFiles (long)
+ * Generation stamp (long)
+ * INodes (count = NumFiles)
+ *  INode
+ *    Path (String)
+ *    Replication (short)
+ *    Modification Time (long as date)
+ *    Access Time (long) // added in -16
+ *    Block size (long)
+ *    Num blocks (int)
+ *    Blocks (count = Num blocks)
+ *      Block
+ *        Block ID (long)
+ *        Num bytes (long)
+ *        Generation stamp (long)
+ *    Namespace Quota (long)
+ *    Diskspace Quota (long) // added in -18
+ *    Permissions
+ *      Username (String)
+ *      Groupname (String)
+ *      OctalPerms (short -> String)  // Modified in -19
+ *    Symlink (String) // added in -23
+ * NumINodesUnderConstruction (int)
+ * INodesUnderConstruction (count = NumINodesUnderConstruction)
+ *  INodeUnderConstruction
+ *    Path (bytes as string)
+ *    Replication (short)
+ *    Modification time (long as date)
+ *    Preferred block size (long)
+ *    Num blocks (int)
+ *    Blocks
+ *      Block
+ *        Block ID (long)
+ *        Num bytes (long)
+ *        Generation stamp (long)
+ *    Permissions
+ *      Username (String)
+ *      Groupname (String)
+ *      OctalPerms (short -> String)
+ *    Client Name (String)
+ *    Client Machine (String)
+ *    NumLocations (int)
+ *    DatanodeDescriptors (count = numLocations) // not loaded into memory
+ *      short                                    // but still in file
+ *      long
+ *      string
+ *      long
+ *      int
+ *      string
+ *      string
+ *      enum
+ *    CurrentDelegationKeyId (int)
+ *    NumDelegationKeys (int)
+ *      DelegationKeys (count = NumDelegationKeys)
+ *        DelegationKeyLength (vint)
+ *        DelegationKey (bytes)
+ *    DelegationTokenSequenceNumber (int)
+ *    NumDelegationTokens (int)
+ *    DelegationTokens (count = NumDelegationTokens)
+ *      DelegationTokenIdentifier
+ *        owner (String)
+ *        renewer (String)
+ *        realUser (String)
+ *        issueDate (vlong)
+ *        maxDate (vlong)
+ *        sequenceNumber (vint)
+ *        masterKeyId (vint)
+ *      expiryTime (long)     
+ *
+ */
+class ImageLoaderCurrent implements ImageLoader {
+  protected final DateFormat dateFormat = 
+                                      new SimpleDateFormat("yyyy-MM-dd HH:mm");
+  private static int[] versions = { -16, -17, -18, -19, -20, -21, -22, -23,
+      -24, -25, -26, -27, -28, -30, -31, -32, -33, -34, -41};
+  private int imageVersion = 0;
+
+  /* (non-Javadoc)
+   * @see ImageLoader#canProcessVersion(int)
+   */
+  @Override
+  public boolean canLoadVersion(int version) {
+    for(int v : versions)
+      if(v == version) return true;
+
+    return false;
+  }
+
+  /* (non-Javadoc)
+   * @see ImageLoader#processImage(java.io.DataInputStream, ImageVisitor, boolean)
+   */
+  @Override
+  public void loadImage(DataInputStream in, ImageVisitor v,
+      boolean skipBlocks) throws IOException {
+    boolean done = false;
+    try {
+      v.start();
+      v.visitEnclosingElement(ImageElement.FS_IMAGE);
+
+      imageVersion = in.readInt();
+      if( !canLoadVersion(imageVersion))
+        throw new IOException("Cannot process fslayout version " + imageVersion);
+
+      v.visit(ImageElement.IMAGE_VERSION, imageVersion);
+      v.visit(ImageElement.NAMESPACE_ID, in.readInt());
+
+      long numInodes = in.readLong();
+
+      v.visit(ImageElement.GENERATION_STAMP, in.readLong());
+
+      if (LayoutVersion.supports(Feature.FSIMAGE_COMPRESSION, imageVersion)) {
+        boolean isCompressed = in.readBoolean();
+        v.visit(ImageElement.IS_COMPRESSED, String.valueOf(isCompressed));
+        if (isCompressed) {
+          String codecClassName = Text.readString(in);
+          v.visit(ImageElement.COMPRESS_CODEC, codecClassName);
+          CompressionCodecFactory codecFac = new CompressionCodecFactory(
+              new Configuration());
+          CompressionCodec codec = codecFac.getCodecByClassName(codecClassName);
+          if (codec == null) {
+            throw new IOException("Image compression codec not supported: "
+                + codecClassName);
+          }
+          in = new DataInputStream(codec.createInputStream(in));
+        }
+      }
+      processINodes(in, v, numInodes, skipBlocks);
+
+      processINodesUC(in, v, skipBlocks);
+
+      if (LayoutVersion.supports(Feature.DELEGATION_TOKEN, imageVersion)) {
+        processDelegationTokens(in, v);
+      }
+      
+      v.leaveEnclosingElement(); // FSImage
+      done = true;
+    } finally {
+      if (done) {
+        v.finish();
+      } else {
+        v.finishAbnormally();
+      }
+    }
+  }
+
+  /**
+   * Process the Delegation Token related section in fsimage.
+   * 
+   * @param in DataInputStream to process
+   * @param v Visitor to walk over records
+   */
+  private void processDelegationTokens(DataInputStream in, ImageVisitor v)
+      throws IOException {
+    v.visit(ImageElement.CURRENT_DELEGATION_KEY_ID, in.readInt());
+    int numDKeys = in.readInt();
+    v.visitEnclosingElement(ImageElement.DELEGATION_KEYS,
+        ImageElement.NUM_DELEGATION_KEYS, numDKeys);
+    for(int i =0; i < numDKeys; i++) {
+      DelegationKey key = new DelegationKey();
+      key.readFields(in);
+      v.visit(ImageElement.DELEGATION_KEY, key.toString());
+    }
+    v.leaveEnclosingElement();
+    v.visit(ImageElement.DELEGATION_TOKEN_SEQUENCE_NUMBER, in.readInt());
+    int numDTokens = in.readInt();
+    v.visitEnclosingElement(ImageElement.DELEGATION_TOKENS,
+        ImageElement.NUM_DELEGATION_TOKENS, numDTokens);
+    for(int i=0; i<numDTokens; i++){
+      DelegationTokenIdentifier id = new  DelegationTokenIdentifier();
+      id.readFields(in);
+      long expiryTime = in.readLong();
+      v.visitEnclosingElement(ImageElement.DELEGATION_TOKEN_IDENTIFIER);
+      v.visit(ImageElement.DELEGATION_TOKEN_IDENTIFIER_KIND,
+          id.getKind().toString());
+      v.visit(ImageElement.DELEGATION_TOKEN_IDENTIFIER_SEQNO,
+          id.getSequenceNumber());
+      v.visit(ImageElement.DELEGATION_TOKEN_IDENTIFIER_RENEWER,
+          id.getRenewer().toString());
+      v.visit(ImageElement.DELEGATION_TOKEN_IDENTIFIER_ISSUE_DATE,
+          id.getIssueDate());
+      v.visit(ImageElement.DELEGATION_TOKEN_IDENTIFIER_MAX_DATE,
+          id.getMaxDate());
+      v.visit(ImageElement.DELEGATION_TOKEN_IDENTIFIER_EXPIRY_TIME,
+          expiryTime);
+      v.visit(ImageElement.DELEGATION_TOKEN_IDENTIFIER_MASTER_KEY_ID,
+          id.getMasterKeyId());
+      v.leaveEnclosingElement(); // DELEGATION_TOKEN_IDENTIFIER
+    }
+    v.leaveEnclosingElement(); // DELEGATION_TOKENS
+  }
+
+  /**
+   * Process the INodes under construction section of the fsimage.
+   *
+   * @param in DataInputStream to process
+   * @param v Visitor to walk over inodes
+   * @param skipBlocks Walk over each block?
+   */
+  private void processINodesUC(DataInputStream in, ImageVisitor v,
+      boolean skipBlocks) throws IOException {
+    int numINUC = in.readInt();
+
+    v.visitEnclosingElement(ImageElement.INODES_UNDER_CONSTRUCTION,
+                           ImageElement.NUM_INODES_UNDER_CONSTRUCTION, numINUC);
+
+    for(int i = 0; i < numINUC; i++) {
+      v.visitEnclosingElement(ImageElement.INODE_UNDER_CONSTRUCTION);
+      byte [] name = FSImage.readBytes(in);
+      String n = new String(name, "UTF8");
+      v.visit(ImageElement.INODE_PATH, n);
+      v.visit(ImageElement.REPLICATION, in.readShort());
+      v.visit(ImageElement.MODIFICATION_TIME, formatDate(in.readLong()));
+
+      v.visit(ImageElement.PREFERRED_BLOCK_SIZE, in.readLong());
+      int numBlocks = in.readInt();
+      processBlocks(in, v, numBlocks, skipBlocks);
+
+      processPermission(in, v);
+      v.visit(ImageElement.CLIENT_NAME, FSImage.readString(in));
+      v.visit(ImageElement.CLIENT_MACHINE, FSImage.readString(in));
+
+      // Skip over the datanode descriptors, which are still stored in the
+      // file but are not used by the datanode or loaded into memory
+      int numLocs = in.readInt();
+      for(int j = 0; j < numLocs; j++) {
+        in.readShort();
+        in.readLong();
+        in.readLong();
+        in.readLong();
+        in.readInt();
+        FSImage.readString(in);
+        FSImage.readString(in);
+        WritableUtils.readEnum(in, AdminStates.class);
+      }
+
+      v.leaveEnclosingElement(); // INodeUnderConstruction
+    }
+
+    v.leaveEnclosingElement(); // INodesUnderConstruction
+  }
+
+  /**
+   * Process the blocks section of the fsimage.
+   *
+   * @param in Datastream to process
+   * @param v Visitor to walk over inodes
+   * @param skipBlocks Walk over each block?
+   */
+  private void processBlocks(DataInputStream in, ImageVisitor v,
+      int numBlocks, boolean skipBlocks) throws IOException {
+    v.visitEnclosingElement(ImageElement.BLOCKS,
+                            ImageElement.NUM_BLOCKS, numBlocks);
+    
+    // directory or symlink, no blocks to process    
+    if(numBlocks == -1 || numBlocks == -2) { 
+      v.leaveEnclosingElement(); // Blocks
+      return;
+    }
+    
+    if(skipBlocks) {
+      int bytesToSkip = ((Long.SIZE * 3 /* fields */) / 8 /*bits*/) * numBlocks;
+      if(in.skipBytes(bytesToSkip) != bytesToSkip)
+        throw new IOException("Error skipping over blocks");
+      
+    } else {
+      for(int j = 0; j < numBlocks; j++) {
+        v.visitEnclosingElement(ImageElement.BLOCK);
+        v.visit(ImageElement.BLOCK_ID, in.readLong());
+        v.visit(ImageElement.NUM_BYTES, in.readLong());
+        v.visit(ImageElement.GENERATION_STAMP, in.readLong());
+        v.leaveEnclosingElement(); // Block
+      }
+    }
+    v.leaveEnclosingElement(); // Blocks
+  }
+
+  /**
+   * Extract the INode permissions stored in the fsimage file.
+   *
+   * @param in Datastream to process
+   * @param v Visitor to walk over inodes
+   */
+  private void processPermission(DataInputStream in, ImageVisitor v)
+      throws IOException {
+    v.visitEnclosingElement(ImageElement.PERMISSIONS);
+    v.visit(ImageElement.USER_NAME, Text.readString(in));
+    v.visit(ImageElement.GROUP_NAME, Text.readString(in));
+    FsPermission fsp = new FsPermission(in.readShort());
+    v.visit(ImageElement.PERMISSION_STRING, fsp.toString());
+    v.leaveEnclosingElement(); // Permissions
+  }
+
+  /**
+   * Process the INode records stored in the fsimage.
+   *
+   * @param in Datastream to process
+   * @param v Visitor to walk over INodes
+   * @param numInodes Number of INodes stored in file
+   * @param skipBlocks Process all the blocks within the INode?
+   * @throws VisitException
+   * @throws IOException
+   */
+  private void processINodes(DataInputStream in, ImageVisitor v,
+      long numInodes, boolean skipBlocks) throws IOException {
+    v.visitEnclosingElement(ImageElement.INODES,
+        ImageElement.NUM_INODES, numInodes);
+    
+    if (LayoutVersion.supports(Feature.FSIMAGE_NAME_OPTIMIZATION, imageVersion)) {
+      processLocalNameINodes(in, v, numInodes, skipBlocks);
+    } else { // full path name
+      processFullNameINodes(in, v, numInodes, skipBlocks);
+    }
+
+    
+    v.leaveEnclosingElement(); // INodes
+  }
+  
+  /**
+   * Process image with full path name
+   * 
+   * @param in image stream
+   * @param v visitor
+   * @param numInodes number of indoes to read
+   * @param skipBlocks skip blocks or not
+   * @throws IOException if there is any error occurs
+   */
+  private void processLocalNameINodes(DataInputStream in, ImageVisitor v,
+      long numInodes, boolean skipBlocks) throws IOException {
+    // process root
+    processINode(in, v, skipBlocks, "");
+    numInodes--;
+    while (numInodes > 0) {
+      numInodes -= processDirectory(in, v, skipBlocks);
+    }
+  }
+  
+  private int processDirectory(DataInputStream in, ImageVisitor v,
+     boolean skipBlocks) throws IOException {
+    String parentName = FSImage.readString(in);
+    int numChildren = in.readInt();
+    for (int i=0; i<numChildren; i++) {
+      processINode(in, v, skipBlocks, parentName);
+    }
+    return numChildren;
+  }
+  
+   /**
+    * Process image with full path name
+    * 
+    * @param in image stream
+    * @param v visitor
+    * @param numInodes number of indoes to read
+    * @param skipBlocks skip blocks or not
+    * @throws IOException if there is any error occurs
+    */
+   private void processFullNameINodes(DataInputStream in, ImageVisitor v,
+       long numInodes, boolean skipBlocks) throws IOException {
+     for(long i = 0; i < numInodes; i++) {
+       processINode(in, v, skipBlocks, null);
+     }
+   }
+   
+   /**
+    * Process an INode
+    * 
+    * @param in image stream
+    * @param v visitor
+    * @param skipBlocks skip blocks or not
+    * @param parentName the name of its parent node
+    * @throws IOException
+    */
+  private void processINode(DataInputStream in, ImageVisitor v,
+      boolean skipBlocks, String parentName) throws IOException {
+    v.visitEnclosingElement(ImageElement.INODE);
+    String pathName = FSImage.readString(in);
+    if (parentName != null) {  // local name
+      pathName = "/" + pathName;
+      if (!"/".equals(parentName)) { // children of non-root directory
+        pathName = parentName + pathName;
+      }
+    }
+
+    v.visit(ImageElement.INODE_PATH, pathName);
+    v.visit(ImageElement.REPLICATION, in.readShort());
+    v.visit(ImageElement.MODIFICATION_TIME, formatDate(in.readLong()));
+    if(LayoutVersion.supports(Feature.FILE_ACCESS_TIME, imageVersion))
+      v.visit(ImageElement.ACCESS_TIME, formatDate(in.readLong()));
+    v.visit(ImageElement.BLOCK_SIZE, in.readLong());
+    int numBlocks = in.readInt();
+
+    processBlocks(in, v, numBlocks, skipBlocks);
+
+    // File or directory
+    if (numBlocks > 0 || numBlocks == -1) {
+      v.visit(ImageElement.NS_QUOTA, numBlocks == -1 ? in.readLong() : -1);
+      if (LayoutVersion.supports(Feature.DISKSPACE_QUOTA, imageVersion))
+        v.visit(ImageElement.DS_QUOTA, numBlocks == -1 ? in.readLong() : -1);
+    }
+    if (numBlocks == -2) {
+      v.visit(ImageElement.SYMLINK, Text.readString(in));
+    }
+
+    processPermission(in, v);
+    v.leaveEnclosingElement(); // INode
+  }
+
+  /**
+   * Helper method to format dates during processing.
+   * @param date Date as read from image file
+   * @return String version of date format
+   */
+  private String formatDate(long date) {
+    return dateFormat.format(new Date(date));
+  }
+}

src/hdfs/org/apache/hadoop/hdfs/tools/offlineImageViewer/ImageVisitor.java

@@ -0,0 +1,162 @@
+/**
+ * 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.hdfs.tools.offlineImageViewer;
+
+import java.io.IOException;
+
+/**
+ * An implementation of ImageVisitor can traverse the structure of an
+ * Hadoop fsimage and respond to each of the structures within the file.
+ */
+abstract class ImageVisitor {
+
+  /**
+   * Structural elements of an FSImage that may be encountered within the
+   * file. ImageVisitors are able to handle processing any of these elements.
+   */
+  public enum ImageElement {
+    FS_IMAGE,
+    IMAGE_VERSION,
+    NAMESPACE_ID,
+    IS_COMPRESSED,
+    COMPRESS_CODEC,
+    LAYOUT_VERSION,
+    NUM_INODES,
+    GENERATION_STAMP,
+    INODES,
+    INODE,
+    INODE_PATH,
+    REPLICATION,
+    MODIFICATION_TIME,
+    ACCESS_TIME,
+    BLOCK_SIZE,
+    NUM_BLOCKS,
+    BLOCKS,
+    BLOCK,
+    BLOCK_ID,
+    NUM_BYTES,
+    NS_QUOTA,
+    DS_QUOTA,
+    PERMISSIONS,
+    SYMLINK,
+    NUM_INODES_UNDER_CONSTRUCTION,
+    INODES_UNDER_CONSTRUCTION,
+    INODE_UNDER_CONSTRUCTION,
+    PREFERRED_BLOCK_SIZE,
+    CLIENT_NAME,
+    CLIENT_MACHINE,
+    USER_NAME,
+    GROUP_NAME,
+    PERMISSION_STRING,
+    CURRENT_DELEGATION_KEY_ID,
+    NUM_DELEGATION_KEYS,
+    DELEGATION_KEYS,
+    DELEGATION_KEY,
+    DELEGATION_TOKEN_SEQUENCE_NUMBER,
+    NUM_DELEGATION_TOKENS,
+    DELEGATION_TOKENS,
+    DELEGATION_TOKEN_IDENTIFIER,
+    DELEGATION_TOKEN_IDENTIFIER_KIND,
+    DELEGATION_TOKEN_IDENTIFIER_SEQNO,
+    DELEGATION_TOKEN_IDENTIFIER_OWNER,
+    DELEGATION_TOKEN_IDENTIFIER_RENEWER,
+    DELEGATION_TOKEN_IDENTIFIER_REALUSER,
+    DELEGATION_TOKEN_IDENTIFIER_ISSUE_DATE,
+    DELEGATION_TOKEN_IDENTIFIER_MAX_DATE,
+    DELEGATION_TOKEN_IDENTIFIER_EXPIRY_TIME,
+    DELEGATION_TOKEN_IDENTIFIER_MASTER_KEY_ID
+  }
+  
+  /**
+   * Begin visiting the fsimage structure.  Opportunity to perform
+   * any initialization necessary for the implementing visitor.
+   */
+  abstract void start() throws IOException;
+
+  /**
+   * Finish visiting the fsimage structure.  Opportunity to perform any
+   * clean up necessary for the implementing visitor.
+   */
+  abstract void finish() throws IOException;
+
+  /**
+   * Finish visiting the fsimage structure after an error has occurred
+   * during the processing.  Opportunity to perform any clean up necessary
+   * for the implementing visitor.
+   */
+  abstract void finishAbnormally() throws IOException;
+
+  /**
+   * Visit non enclosing element of fsimage with specified value.
+   *
+   * @param element FSImage element
+   * @param value Element's value
+   */
+  abstract void visit(ImageElement element, String value) throws IOException;
+
+  // Convenience methods to automatically convert numeric value types to strings
+  void visit(ImageElement element, int value) throws IOException {
+    visit(element, Integer.toString(value));
+  }
+
+  void visit(ImageElement element, long value) throws IOException {
+    visit(element, Long.toString(value));
+  }
+
+  /**
+   * Begin visiting an element that encloses another element, such as
+   * the beginning of the list of blocks that comprise a file.
+   *
+   * @param element Element being visited
+   */
+  abstract void visitEnclosingElement(ImageElement element)
+     throws IOException;
+
+  /**
+   * Begin visiting an element that encloses another element, such as
+   * the beginning of the list of blocks that comprise a file.
+   *
+   * Also provide an additional key and value for the element, such as the
+   * number items within the element.
+   *
+   * @param element Element being visited
+   * @param key Key describing the element being visited
+   * @param value Value associated with element being visited
+   */
+  abstract void visitEnclosingElement(ImageElement element,
+      ImageElement key, String value) throws IOException;
+
+  // Convenience methods to automatically convert value types to strings
+  void visitEnclosingElement(ImageElement element,
+      ImageElement key, int value)
+     throws IOException {
+    visitEnclosingElement(element, key, Integer.toString(value));
+  }
+
+  void visitEnclosingElement(ImageElement element,
+      ImageElement key, long value)
+     throws IOException {
+    visitEnclosingElement(element, key, Long.toString(value));
+  }
+
+  /**
+   * Leave current enclosing element.  Called, for instance, at the end of
+   * processing the blocks that compromise a file.
+   */
+  abstract void leaveEnclosingElement() throws IOException;
+}

src/hdfs/org/apache/hadoop/hdfs/tools/offlineImageViewer/IndentedImageVisitor.java

@@ -0,0 +1,111 @@
+/**
+ * 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.hdfs.tools.offlineImageViewer;
+
+import java.io.IOException;
+import java.util.Date;
+
+/**
+ * IndentedImageVisitor walks over an FSImage and displays its structure 
+ * using indenting to organize sections within the image file.
+ */
+class IndentedImageVisitor extends TextWriterImageVisitor {
+  
+  public IndentedImageVisitor(String filename) throws IOException {
+    super(filename);
+  }
+
+  public IndentedImageVisitor(String filename, boolean printToScreen) throws IOException {
+    super(filename, printToScreen);
+  }
+
+  final private DepthCounter dc = new DepthCounter();// to track leading spacing
+
+  @Override
+  void start() throws IOException {}
+
+  @Override
+  void finish() throws IOException { super.finish(); }
+
+  @Override
+  void finishAbnormally() throws IOException {
+    System.out.println("*** Image processing finished abnormally.  Ending ***");
+    super.finishAbnormally();
+  }
+
+  @Override
+  void leaveEnclosingElement() throws IOException {
+    dc.decLevel();
+  }
+
+  @Override
+  void visit(ImageElement element, String value) throws IOException {
+    printIndents();
+    write(element + " = " + value + "\n");
+  }
+
+  @Override
+  void visit(ImageElement element, long value) throws IOException {
+    if ((element == ImageElement.DELEGATION_TOKEN_IDENTIFIER_EXPIRY_TIME) || 
+        (element == ImageElement.DELEGATION_TOKEN_IDENTIFIER_ISSUE_DATE) || 
+        (element == ImageElement.DELEGATION_TOKEN_IDENTIFIER_MAX_DATE)) {
+      visit(element, new Date(value).toString());
+    } else {
+      visit(element, Long.toString(value));
+    }
+  }
+  
+  @Override
+  void visitEnclosingElement(ImageElement element) throws IOException {
+    printIndents();
+    write(element + "\n");
+    dc.incLevel();
+  }
+
+  // Print element, along with associated key/value pair, in brackets
+  @Override
+  void visitEnclosingElement(ImageElement element,
+      ImageElement key, String value)
+      throws IOException {
+    printIndents();
+    write(element + " [" + key + " = " + value + "]\n");
+    dc.incLevel();
+  }
+
+  /**
+  * Print an appropriate number of spaces for the current level.
+  * FsImages can potentially be millions of lines long, so caching can
+  * significantly speed up output.
+  */
+  final private static String [] indents = { "",
+                                             "  ",
+                                             "    ",
+                                             "      ",
+                                             "        ",
+                                             "          ",
+                                             "            "};
+  private void printIndents() throws IOException {
+    try {
+      write(indents[dc.getLevel()]);
+    } catch (IndexOutOfBoundsException e) {
+      // There's no reason in an fsimage would need a deeper indent
+      for(int i = 0; i < dc.getLevel(); i++)
+        write(" ");
+    }
+   }
+}

src/hdfs/org/apache/hadoop/hdfs/tools/offlineImageViewer/LsImageVisitor.java

@@ -0,0 +1,178 @@
+/**
+ * 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.hdfs.tools.offlineImageViewer;
+
+import java.io.IOException;
+import java.util.Formatter;
+import java.util.LinkedList;
+
+/**
+ * LsImageVisitor displays the blocks of the namespace in a format very similar
+ * to the output of ls/lsr.  Entries are marked as directories or not,
+ * permissions listed, replication, username and groupname, along with size,
+ * modification date and full path.
+ *
+ * Note: A significant difference between the output of the lsr command
+ * and this image visitor is that this class cannot sort the file entries;
+ * they are listed in the order they are stored within the fsimage file. 
+ * Therefore, the output of this class cannot be directly compared to the
+ * output of the lsr command.
+ */
+class LsImageVisitor extends TextWriterImageVisitor {
+  final private LinkedList<ImageElement> elemQ = new LinkedList<ImageElement>();
+
+  private int numBlocks;
+  private String perms;
+  private int replication;
+  private String username;
+  private String group;
+  private long filesize;
+  private String modTime;
+  private String path;
+  private String linkTarget;
+
+  private boolean inInode = false;
+  final private StringBuilder sb = new StringBuilder();
+  final private Formatter formatter = new Formatter(sb);
+
+  public LsImageVisitor(String filename) throws IOException {
+    super(filename);
+  }
+
+  public LsImageVisitor(String filename, boolean printToScreen) throws IOException {
+    super(filename, printToScreen);
+  }
+
+  /**
+   * Start a new line of output, reset values.
+   */
+  private void newLine() {
+    numBlocks = 0;
+    perms = username = group = path = linkTarget = "";
+    filesize = 0l;
+    replication = 0;
+
+    inInode = true;
+  }
+
+  /**
+   * All the values have been gathered.  Print them to the console in an
+   * ls-style format.
+   */
+  private final static int widthRepl = 2;  
+  private final static int widthUser = 8; 
+  private final static int widthGroup = 10; 
+  private final static int widthSize = 10;
+  private final static int widthMod = 10;
+  private final static String lsStr = " %" + widthRepl + "s %" + widthUser + 
+                                       "s %" + widthGroup + "s %" + widthSize +
+                                       "d %" + widthMod + "s %s";
+  private void printLine() throws IOException {
+    sb.append(numBlocks < 0 ? "d" : "-");
+    sb.append(perms);
+
+    if (0 != linkTarget.length()) {
+      path = path + " -> " + linkTarget; 
+    }
+    formatter.format(lsStr, replication > 0 ? replication : "-",
+                           username, group, filesize, modTime, path);
+    sb.append("\n");
+
+    write(sb.toString());
+    sb.setLength(0); // clear string builder
+
+    inInode = false;
+  }
+
+  @Override
+  void start() throws IOException {}
+
+  @Override
+  void finish() throws IOException {
+    super.finish();
+  }
+
+  @Override
+  void finishAbnormally() throws IOException {
+    System.out.println("Input ended unexpectedly.");
+    super.finishAbnormally();
+  }
+
+  @Override
+  void leaveEnclosingElement() throws IOException {
+    ImageElement elem = elemQ.pop();
+
+    if(elem == ImageElement.INODE)
+      printLine();
+  }
+
+  // Maintain state of location within the image tree and record
+  // values needed to display the inode in ls-style format.
+  @Override
+  void visit(ImageElement element, String value) throws IOException {
+    if(inInode) {
+      switch(element) {
+      case INODE_PATH:
+        if(value.equals("")) path = "/";
+        else path = value;
+        break;
+      case PERMISSION_STRING:
+        perms = value;
+        break;
+      case REPLICATION:
+        replication = Integer.valueOf(value);
+        break;
+      case USER_NAME:
+        username = value;
+        break;
+      case GROUP_NAME:
+        group = value;
+        break;
+      case NUM_BYTES:
+        filesize += Long.valueOf(value);
+        break;
+      case MODIFICATION_TIME:
+        modTime = value;
+        break;
+      case SYMLINK:
+        linkTarget = value;
+        break;
+      default:
+        // This is OK.  We're not looking for all the values.
+        break;
+      }
+    }
+  }
+
+  @Override
+  void visitEnclosingElement(ImageElement element) throws IOException {
+    elemQ.push(element);
+    if(element == ImageElement.INODE)
+      newLine();
+  }
+
+  @Override
+  void visitEnclosingElement(ImageElement element,
+      ImageElement key, String value) throws IOException {
+    elemQ.push(element);
+    if(element == ImageElement.INODE)
+      newLine();
+    else if (element == ImageElement.BLOCKS)
+      numBlocks = Integer.valueOf(value);
+  }
+}

src/hdfs/org/apache/hadoop/hdfs/tools/offlineImageViewer/NameDistributionVisitor.java

@@ -0,0 +1,118 @@
+/**
+ * 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.hdfs.tools.offlineImageViewer;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map.Entry;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+
+/**
+ * File name distribution visitor. 
+ * <p>
+ * It analyzes file names in fsimage and prints the following information: 
+ * <li>Number of unique file names</li> 
+ * <li>Number file names and the corresponding number range of files that use 
+ * these same names</li>
+ * <li>Heap saved if the file name objects are reused</li>
+ */
+@InterfaceAudience.Private
+public class NameDistributionVisitor extends TextWriterImageVisitor {
+  HashMap<String, Integer> counts = new HashMap<String, Integer>();
+
+  public NameDistributionVisitor(String filename, boolean printToScreen)
+      throws IOException {
+    super(filename, printToScreen);
+  }
+
+  @Override
+  void finish() throws IOException {
+    final int BYTEARRAY_OVERHEAD = 24;
+
+    write("Total unique file names " + counts.size());
+    // Columns: Frequency of file occurrence, savings in heap, total files using
+    // the name and number of file names
+    final long stats[][] = { { 100000, 0, 0, 0 },
+                             { 10000, 0, 0, 0 },
+                             { 1000, 0, 0, 0 },
+                             { 100, 0, 0, 0 },
+                             { 10, 0, 0, 0 },
+                             { 5, 0, 0, 0 },
+                             { 4, 0, 0, 0 },
+                             { 3, 0, 0, 0 },
+                             { 2, 0, 0, 0 }};
+
+    int highbound = Integer.MIN_VALUE;
+    for (Entry<String, Integer> entry : counts.entrySet()) {
+      highbound = Math.max(highbound, entry.getValue());
+      for (int i = 0; i < stats.length; i++) {
+        if (entry.getValue() >= stats[i][0]) {
+          stats[i][1] += (BYTEARRAY_OVERHEAD + entry.getKey().length())
+              * (entry.getValue() - 1);
+          stats[i][2] += entry.getValue();
+          stats[i][3]++;
+          break;
+        }
+      }
+    }
+
+    long lowbound = 0;
+    long totalsavings = 0;
+    for (long[] stat : stats) {
+      lowbound = stat[0];
+      totalsavings += stat[1];
+      String range = lowbound == highbound ? " " + lowbound :
+          " between " + lowbound + "-" + highbound;
+      write("\n" + stat[3] + " names are used by " + stat[2] + " files"
+          + range + " times. Heap savings ~" + stat[1] + " bytes.");
+      highbound = (int) stat[0] - 1;
+    }
+    write("\n\nTotal saved heap ~" + totalsavings + "bytes.\n");
+    super.finish();
+  }
+
+  @Override
+  void visit(ImageElement element, String value) throws IOException {
+    if (element == ImageElement.INODE_PATH) {
+      String filename = value.substring(value.lastIndexOf("/") + 1);
+      if (counts.containsKey(filename)) {
+        counts.put(filename, counts.get(filename) + 1);
+      } else {
+        counts.put(filename, 1);
+      }
+    }
+  }
+
+  @Override
+  void leaveEnclosingElement() throws IOException {
+  }
+
+  @Override
+  void start() throws IOException {
+  }
+
+  @Override
+  void visitEnclosingElement(ImageElement element) throws IOException {
+  }
+
+  @Override
+  void visitEnclosingElement(ImageElement element, ImageElement key,
+      String value) throws IOException {
+  }
+}

src/hdfs/org/apache/hadoop/hdfs/tools/offlineImageViewer/OfflineImageViewer.java

@@ -0,0 +1,335 @@
+/**
+ * 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.hdfs.tools.offlineImageViewer;
+
+import java.io.BufferedInputStream;
+import java.io.DataInputStream;
+import java.io.EOFException;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FilterInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.CommandLineParser;
+import org.apache.commons.cli.OptionBuilder;
+import org.apache.commons.cli.Options;
+import org.apache.commons.cli.ParseException;
+import org.apache.commons.cli.PosixParser;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.io.IOUtils;
+
+/**
+ * OfflineImageViewer to dump the contents of an Hadoop image file to XML
+ * or the console.  Main entry point into utility, either via the
+ * command line or programatically.
+ */
+@InterfaceAudience.Private
+public class OfflineImageViewer {
+  public static final Log LOG = LogFactory.getLog(OfflineImageViewer.class);
+  
+  private final static String usage = 
+    "Usage: bin/hadoop oiv [OPTIONS] -i INPUTFILE -o OUTPUTFILE\n" +
+    "Offline Image Viewer\n" + 
+    "View a Hadoop fsimage INPUTFILE using the specified PROCESSOR,\n" +
+    "saving the results in OUTPUTFILE.\n" +
+    "\n" +
+    "The oiv utility will attempt to parse correctly formed image files\n" +
+    "and will abort fail with mal-formed image files.\n" +
+    "\n" +
+    "The tool works offline and does not require a running cluster in\n" +
+    "order to process an image file.\n" +
+    "\n" +
+    "The following image processors are available:\n" +
+    "  * Ls: The default image processor generates an lsr-style listing\n" +
+    "    of the files in the namespace, with the same fields in the same\n" +
+    "    order.  Note that in order to correctly determine file sizes,\n" +
+    "    this formatter cannot skip blocks and will override the\n" +
+    "    -skipBlocks option.\n" +
+    "  * Indented: This processor enumerates over all of the elements in\n" +
+    "    the fsimage file, using levels of indentation to delineate\n" +
+    "    sections within the file.\n" +
+    "  * Delimited: Generate a text file with all of the elements common\n" +
+    "    to both inodes and inodes-under-construction, separated by a\n" +
+    "    delimiter. The default delimiter is \u0001, though this may be\n" +
+    "    changed via the -delimiter argument. This processor also overrides\n" +
+    "    the -skipBlocks option for the same reason as the Ls processor\n" +
+    "  * XML: This processor creates an XML document with all elements of\n" +
+    "    the fsimage enumerated, suitable for further analysis by XML\n" +
+    "    tools.\n" +
+    "  * FileDistribution: This processor analyzes the file size\n" +
+    "    distribution in the image.\n" +
+    "    -maxSize specifies the range [0, maxSize] of file sizes to be\n" +
+    "     analyzed (128GB by default).\n" +
+    "    -step defines the granularity of the distribution. (2MB by default)\n" +
+    "  * NameDistribution: This processor analyzes the file names\n" +
+    "    in the image and prints total number of file names and how frequently" +
+    "    file names are reused.\n" +
+    "\n" + 
+    "Required command line arguments:\n" +
+    "-i,--inputFile <arg>   FSImage file to process.\n" +
+    "-o,--outputFile <arg>  Name of output file. If the specified\n" +
+    "                       file exists, it will be overwritten.\n" +
+    "\n" + 
+    "Optional command line arguments:\n" +
+    "-p,--processor <arg>   Select which type of processor to apply\n" +
+    "                       against image file." +
+    " (Ls|XML|Delimited|Indented|FileDistribution).\n" +
+    "-h,--help              Display usage information and exit\n" +
+    "-printToScreen         For processors that write to a file, also\n" +
+    "                       output to screen. On large image files this\n" +
+    "                       will dramatically increase processing time.\n" +
+    "-skipBlocks            Skip inodes' blocks information. May\n" +
+    "                       significantly decrease output.\n" +
+    "                       (default = false).\n" +
+    "-delimiter <arg>       Delimiting string to use with Delimited processor\n";
+
+  private final boolean skipBlocks;
+  private final String inputFile;
+  private final ImageVisitor processor;
+  
+  public OfflineImageViewer(String inputFile, ImageVisitor processor, 
+             boolean skipBlocks) {
+    this.inputFile = inputFile;
+    this.processor = processor;
+    this.skipBlocks = skipBlocks;
+  }
+
+  /**
+   * Process image file.
+   */
+  public void go() throws IOException  {
+    DataInputStream in = null;
+    PositionTrackingInputStream tracker = null;
+    ImageLoader fsip = null;
+    boolean done = false;
+    try {
+      tracker = new PositionTrackingInputStream(new BufferedInputStream(
+               new FileInputStream(new File(inputFile))));
+      in = new DataInputStream(tracker);
+
+      int imageVersionFile = findImageVersion(in);
+
+      fsip = ImageLoader.LoaderFactory.getLoader(imageVersionFile);
+
+      if(fsip == null) 
+        throw new IOException("No image processor to read version " +
+            imageVersionFile + " is available.");
+      fsip.loadImage(in, processor, skipBlocks);
+      done = true;
+    } finally {
+      if (!done) {
+        LOG.error("image loading failed at offset " + tracker.getPos());
+      }
+      IOUtils.cleanup(LOG, in, tracker);
+    }
+  }
+
+  /**
+   * Check an fsimage datainputstream's version number.
+   *
+   * The datainput stream is returned at the same point as it was passed in;
+   * this method has no effect on the datainputstream's read pointer.
+   *
+   * @param in Datainputstream of fsimage
+   * @return Filesystem layout version of fsimage represented by stream
+   * @throws IOException If problem reading from in
+   */
+  private int findImageVersion(DataInputStream in) throws IOException {
+    in.mark(42); // arbitrary amount, resetting immediately
+
+    int version = in.readInt();
+    in.reset();
+
+    return version;
+  }
+  
+  /**
+   * Build command-line options and descriptions
+   */
+  public static Options buildOptions() {
+    Options options = new Options();
+
+    // Build in/output file arguments, which are required, but there is no 
+    // addOption method that can specify this
+    OptionBuilder.isRequired();
+    OptionBuilder.hasArgs();
+    OptionBuilder.withLongOpt("outputFile");
+    options.addOption(OptionBuilder.create("o"));
+    
+    OptionBuilder.isRequired();
+    OptionBuilder.hasArgs();
+    OptionBuilder.withLongOpt("inputFile");
+    options.addOption(OptionBuilder.create("i"));
+    
+    options.addOption("p", "processor", true, "");
+    options.addOption("h", "help", false, "");
+    options.addOption("skipBlocks", false, "");
+    options.addOption("printToScreen", false, "");
+    options.addOption("delimiter", true, "");
+
+    return options;
+  }
+  
+  /**
+   * Entry point to command-line-driven operation.  User may specify
+   * options and start fsimage viewer from the command line.  Program
+   * will process image file and exit cleanly or, if an error is
+   * encountered, inform user and exit.
+   *
+   * @param args Command line options
+   * @throws IOException 
+   */
+  public static void main(String[] args) throws IOException {
+    Options options = buildOptions();
+    if(args.length == 0) {
+      printUsage();
+      return;
+    }
+    
+    CommandLineParser parser = new PosixParser();
+    CommandLine cmd;
+
+    try {
+      cmd = parser.parse(options, args);
+    } catch (ParseException e) {
+      System.out.println("Error parsing command-line options: ");
+      printUsage();
+      return;
+    }
+
+    if(cmd.hasOption("h")) { // print help and exit
+      printUsage();
+      return;
+    }
+
+    boolean skipBlocks = cmd.hasOption("skipBlocks");
+    boolean printToScreen = cmd.hasOption("printToScreen");
+    String inputFile = cmd.getOptionValue("i");
+    String processor = cmd.getOptionValue("p", "Ls");
+    String outputFile = cmd.getOptionValue("o");
+    String delimiter = cmd.getOptionValue("delimiter");
+    
+    if( !(delimiter == null || processor.equals("Delimited")) ) {
+      System.out.println("Can only specify -delimiter with Delimited processor");
+      printUsage();
+      return;
+    }
+    
+    ImageVisitor v;
+    if(processor.equals("Indented")) {
+      v = new IndentedImageVisitor(outputFile, printToScreen);
+    } else if (processor.equals("XML")) {
+      v = new XmlImageVisitor(outputFile, printToScreen);
+    } else if (processor.equals("Delimited")) {
+      v = delimiter == null ?  
+                 new DelimitedImageVisitor(outputFile, printToScreen) :
+                 new DelimitedImageVisitor(outputFile, printToScreen, delimiter);
+      skipBlocks = false;
+    } else if (processor.equals("FileDistribution")) {
+      long maxSize = Long.parseLong(cmd.getOptionValue("maxSize", "0"));
+      int step = Integer.parseInt(cmd.getOptionValue("step", "0"));
+      v = new FileDistributionVisitor(outputFile, maxSize, step);
+    } else if (processor.equals("NameDistribution")) {
+      v = new NameDistributionVisitor(outputFile, printToScreen);
+    } else {
+      v = new LsImageVisitor(outputFile, printToScreen);
+      skipBlocks = false;
+    }
+    
+    try {
+      OfflineImageViewer d = new OfflineImageViewer(inputFile, v, skipBlocks);
+      d.go();
+    } catch (EOFException e) {
+      System.err.println("Input file ended unexpectedly.  Exiting");
+    } catch(IOException e) {
+      System.err.println("Encountered exception.  Exiting: " + e.getMessage());
+    }
+  }
+
+  /**
+   * Print application usage instructions.
+   */
+  private static void printUsage() {
+    System.out.println(usage);
+  }
+
+  /**
+   * Stream wrapper that keeps track of the current stream position.
+   */
+  private static class PositionTrackingInputStream extends FilterInputStream {
+    private long curPos = 0;
+    private long markPos = -1;
+
+    public PositionTrackingInputStream(InputStream is) {
+      super(is);
+    }
+    
+    @Override
+    public int read() throws IOException {
+      int ret = super.read();
+      if (ret != -1) curPos++;
+      return ret;
+    }
+
+    @Override
+    public int read(byte[] data) throws IOException {
+      int ret = super.read(data);
+      if (ret > 0) curPos += ret;
+      return ret;
+    }
+
+    @Override
+    public int read(byte[] data, int offset, int length) throws IOException {
+      int ret = super.read(data, offset, length);
+      if (ret > 0) curPos += ret;
+      return ret;
+    }
+
+    @Override
+    public void mark(int limit) {
+      super.mark(limit);
+      markPos = curPos;
+    }
+
+    @Override
+    public void reset() throws IOException {
+      if (markPos == -1) {
+        throw new IOException("Not marked!");
+      }
+      super.reset();
+      curPos = markPos;
+      markPos = -1;
+    }
+
+    public long getPos() {
+      return curPos;
+    }
+    
+    @Override
+    public long skip(long amt) throws IOException {
+      long ret = super.skip(amt);
+      curPos += ret;
+      return ret;
+    }
+  }
+}

src/hdfs/org/apache/hadoop/hdfs/tools/offlineImageViewer/TextWriterImageVisitor.java

@@ -0,0 +1,110 @@
+/**
+ * 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.hdfs.tools.offlineImageViewer;
+
+import java.io.FileOutputStream;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.io.OutputStreamWriter;
+import java.nio.charset.Charset;
+
+/**
+ * TextWriterImageProcessor mixes in the ability for ImageVisitor
+ * implementations to easily write their output to a text file.
+ *
+ * Implementing classes should be sure to call the super methods for the
+ * constructors, finish and finishAbnormally methods, in order that the
+ * underlying file may be opened and closed correctly.
+ *
+ * Note, this class does not add newlines to text written to file or (if
+ * enabled) screen.  This is the implementing class' responsibility.
+ */
+abstract class TextWriterImageVisitor extends ImageVisitor {
+  private static final Charset UTF_8 = Charset.forName("UTF-8");
+  private boolean printToScreen = false;
+  private boolean okToWrite = false;
+  final private OutputStreamWriter fw;
+
+  /**
+   * Create a processor that writes to the file named.
+   *
+   * @param filename Name of file to write output to
+   */
+  public TextWriterImageVisitor(String filename) throws IOException {
+    this(filename, false);
+  }
+
+  /**
+   * Create a processor that writes to the file named and may or may not
+   * also output to the screen, as specified.
+   *
+   * @param filename Name of file to write output to
+   * @param printToScreen Mirror output to screen?
+   */
+  public TextWriterImageVisitor(String filename, boolean printToScreen)
+         throws IOException {
+    super();
+    this.printToScreen = printToScreen;
+    fw = new OutputStreamWriter(new FileOutputStream(filename), UTF_8);
+    okToWrite = true;
+  }
+  
+  /* (non-Javadoc)
+   * @see org.apache.hadoop.hdfs.tools.offlineImageViewer.ImageVisitor#finish()
+   */
+  @Override
+  void finish() throws IOException {
+    close();
+  }
+
+  /* (non-Javadoc)
+   * @see org.apache.hadoop.hdfs.tools.offlineImageViewer.ImageVisitor#finishAbnormally()
+   */
+  @Override
+  void finishAbnormally() throws IOException {
+    close();
+  }
+
+  /**
+   * Close output stream and prevent further writing
+   */
+  private void close() throws IOException {
+    fw.close();
+    okToWrite = false;
+  }
+
+  /**
+   * Write parameter to output file (and possibly screen).
+   *
+   * @param toWrite Text to write to file
+   */
+  protected void write(String toWrite) throws IOException  {
+    if(!okToWrite)
+      throw new IOException("file not open for writing.");
+
+    if(printToScreen)
+      System.out.print(toWrite);
+
+    try {
+      fw.write(toWrite);
+    } catch (IOException e) {
+      okToWrite = false;
+      throw e;
+    }
+  }
+}

src/hdfs/org/apache/hadoop/hdfs/tools/offlineImageViewer/XmlImageVisitor.java

@@ -0,0 +1,88 @@
+/**
+ * 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.hdfs.tools.offlineImageViewer;
+
+import java.io.IOException;
+import java.util.LinkedList;
+
+/**
+ * An XmlImageVisitor walks over an fsimage structure and writes out
+ * an equivalent XML document that contains the fsimage's components.
+ */
+class XmlImageVisitor extends TextWriterImageVisitor {
+  final private LinkedList<ImageElement> tagQ =
+                                          new LinkedList<ImageElement>();
+
+  public XmlImageVisitor(String filename) throws IOException {
+    super(filename, false);
+  }
+
+  public XmlImageVisitor(String filename, boolean printToScreen)
+       throws IOException {
+    super(filename, printToScreen);
+  }
+
+  @Override
+  void finish() throws IOException {
+    super.finish();
+  }
+
+  @Override
+  void finishAbnormally() throws IOException {
+    write("\n<!-- Error processing image file.  Exiting -->\n");
+    super.finishAbnormally();
+  }
+
+  @Override
+  void leaveEnclosingElement() throws IOException {
+    if(tagQ.size() == 0)
+      throw new IOException("Tried to exit non-existent enclosing element " +
+                "in FSImage file");
+
+    ImageElement element = tagQ.pop();
+    write("</" + element.toString() + ">\n");
+  }
+
+  @Override
+  void start() throws IOException {
+    write("<?xml version=\"1.0\" ?>\n");
+  }
+
+  @Override
+  void visit(ImageElement element, String value) throws IOException {
+    writeTag(element.toString(), value);
+  }
+
+  @Override
+  void visitEnclosingElement(ImageElement element) throws IOException {
+    write("<" + element.toString() + ">\n");
+    tagQ.push(element);
+  }
+
+  @Override
+  void visitEnclosingElement(ImageElement element,
+      ImageElement key, String value)
+       throws IOException {
+    write("<" + element.toString() + " " + key + "=\"" + value +"\">\n");
+    tagQ.push(element);
+  }
+
+  private void writeTag(String tag, String value) throws IOException {
+    write("<" + tag + ">" + value + "</" + tag + ">\n");
+  }
+}

src/test/org/apache/hadoop/hdfs/protocol/TestLayoutVersion.java

@@ -0,0 +1,86 @@
+/**
+ * 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.hdfs.protocol;
+
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import java.util.EnumSet;
+
+import org.apache.hadoop.hdfs.protocol.LayoutVersion.Feature;
+import org.junit.Test;
+
+/**
+ * Test for {@link LayoutVersion}
+ */
+public class TestLayoutVersion {
+  
+  /**
+   * Tests to make sure a given layout version supports all the
+   * features from the ancestor
+   */
+  @Test
+  public void testFeaturesFromAncestorSupported() {
+    for (Feature f : Feature.values()) {
+      validateFeatureList(f);
+    }
+  }
+  
+  /**
+   * Test to make sure 0.20.203 supports delegation token
+   */
+  @Test
+  public void testRelease203() {
+    assertTrue(LayoutVersion.supports(Feature.DELEGATION_TOKEN, 
+        Feature.RESERVED_REL20_203.lv));
+  }
+  
+  /**
+   * Test to make sure 0.20.204 supports delegation token
+   */
+  @Test
+  public void testRelease204() {
+    assertTrue(LayoutVersion.supports(Feature.DELEGATION_TOKEN, 
+        Feature.RESERVED_REL20_204.lv));
+  }
+  
+  /**
+   * Test to make sure release 1.2.0 support CONCAT
+   */
+  @Test
+  public void testRelease1_2_0() {
+    assertTrue(LayoutVersion.supports(Feature.CONCAT, 
+        Feature.RESERVED_REL1_2_0.lv));
+  }
+  
+  /**
+   * Given feature {@code f}, ensures the layout version of that feature
+   * supports all the features supported by it's ancestor.
+   */
+  private void validateFeatureList(Feature f) {
+    int lv = f.lv;
+    int ancestorLV = f.ancestorLV;
+    EnumSet<Feature> ancestorSet = LayoutVersion.map.get(ancestorLV);
+    assertNotNull(ancestorSet);
+    for (Feature  feature : ancestorSet) {
+      assertTrue("LV " + lv + " does nto support " + feature
+          + " supported by the ancestor LV " + f.ancestorLV,
+          LayoutVersion.supports(feature, lv));
+    }
+  }
+}

src/test/org/apache/hadoop/hdfs/tools/offlineImageViewer/SpotCheckImageVisitor.java

@@ -0,0 +1,89 @@
+/**
+ * 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.hdfs.tools.offlineImageViewer;
+
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.Set;
+
+/**
+ * ImageVisitor to spot check an fsimage and generate several statistics
+ * about it that we can compare with known values to give a reasonable
+ * assertion that the image was processed correctly.
+ */
+class SpotCheckImageVisitor extends ImageVisitor {
+  
+  // Statistics gathered by the visitor for Inodes and InodesUnderConstruction
+  static public class ImageInfo {
+    public long totalNumBlocks = 0; // Total number of blocks in section
+    public Set<String> pathNames = new HashSet<String>(); // All path names
+    public long totalFileSize = 0; // Total size of all the files
+    public long totalReplications = 0; // Sum of all the replications
+  }
+
+  final private ImageInfo inodes = new ImageInfo();
+  final private ImageInfo INUCs = new ImageInfo();
+  private ImageInfo current = null;
+  
+  @Override
+  void visit(ImageElement element, String value) throws IOException {
+    if(element == ImageElement.NUM_BYTES) 
+      current.totalFileSize += Long.valueOf(value);
+    else if (element == ImageElement.REPLICATION)
+      current.totalReplications += Long.valueOf(value);
+    else if (element == ImageElement.INODE_PATH)
+      current.pathNames.add(value);
+  }
+
+  @Override
+  void visitEnclosingElement(ImageElement element, ImageElement key,
+      String value) throws IOException {
+    switch(element) {
+    case INODES:
+      current = inodes;
+      break;
+    case INODES_UNDER_CONSTRUCTION:
+      current = INUCs;
+      break;
+    case BLOCKS:
+      current.totalNumBlocks += Long.valueOf(value);
+      break;
+      // OK to not have a default, we're skipping most of the values
+    }
+  }
+  
+  public ImageInfo getINodesInfo() { return inodes; }
+  
+  public ImageInfo getINUCsInfo() { return INUCs; }
+  
+  // Unnecessary visitor methods
+  @Override
+  void finish() throws IOException {}
+
+  @Override
+  void finishAbnormally() throws IOException {}
+
+  @Override
+  void leaveEnclosingElement() throws IOException {}
+
+  @Override
+  void start() throws IOException {}
+
+  @Override
+  void visitEnclosingElement(ImageElement element) throws IOException {}
+}

src/test/org/apache/hadoop/hdfs/tools/offlineImageViewer/TestDelimitedImageVisitor.java

@@ -0,0 +1,100 @@
+/**
+ * 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.hdfs.tools.offlineImageViewer;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.fail;
+
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileReader;
+import java.io.IOException;
+
+import org.apache.hadoop.hdfs.tools.offlineImageViewer.ImageVisitor.ImageElement;
+import org.junit.Test;
+
+/**
+ * Test that the DelimitedImageVisistor gives the expected output based
+ * on predetermined inputs
+ */
+public class TestDelimitedImageVisitor {
+  private static String ROOT = System.getProperty("test.build.data","/tmp");
+  private static final String delim = "--";
+  
+  // Record an element in the visitor and build the expected line in the output
+  private void build(DelimitedImageVisitor div, ImageElement elem, String val, 
+                     StringBuilder sb, boolean includeDelim) throws IOException {
+    div.visit(elem, val);
+    sb.append(val);
+    
+    if(includeDelim)
+      sb.append(delim);
+  }
+  
+  @Test
+  public void testDelimitedImageVisistor() {
+    String filename = ROOT + "/testDIV";
+    File f = new File(filename);
+    BufferedReader br = null;
+    StringBuilder sb = new StringBuilder();
+    
+    try {
+      DelimitedImageVisitor div = new DelimitedImageVisitor(filename, true, delim);
+
+      div.visit(ImageElement.FS_IMAGE, "Not in ouput");
+      div.visitEnclosingElement(ImageElement.INODE);
+      div.visit(ImageElement.LAYOUT_VERSION, "not in");
+      div.visit(ImageElement.LAYOUT_VERSION, "the output");
+      
+      build(div, ImageElement.INODE_PATH,        "hartnell", sb, true);
+      build(div, ImageElement.REPLICATION,       "99", sb, true);
+      build(div, ImageElement.MODIFICATION_TIME, "troughton", sb, true);
+      build(div, ImageElement.ACCESS_TIME,       "pertwee", sb, true);
+      build(div, ImageElement.BLOCK_SIZE,        "baker", sb, true);
+      build(div, ImageElement.NUM_BLOCKS,        "davison", sb, true);
+      build(div, ImageElement.NUM_BYTES,         "55", sb, true);
+      build(div, ImageElement.NS_QUOTA,          "baker2", sb, true);
+      build(div, ImageElement.DS_QUOTA,          "mccoy", sb, true);
+      build(div, ImageElement.PERMISSION_STRING, "eccleston", sb, true);
+      build(div, ImageElement.USER_NAME,         "tennant", sb, true);
+      build(div, ImageElement.GROUP_NAME,        "smith", sb, false);
+      
+      div.leaveEnclosingElement(); // INode
+      div.finish();
+      
+      br = new BufferedReader(new FileReader(f));
+      String actual = br.readLine();
+      
+      // Should only get one line
+      assertNull(br.readLine());
+      br.close();
+      
+      String exepcted = sb.toString();
+      System.out.println("Expect to get: " + exepcted);
+      System.out.println("Actually got:  " + actual);
+      assertEquals(exepcted, actual);
+      
+    } catch (IOException e) {
+      fail("Error while testing delmitedImageVisitor" + e.getMessage());
+    } finally {
+      if(f.exists())
+        f.delete();
+    }
+  }
+}

src/test/org/apache/hadoop/hdfs/tools/offlineImageViewer/TestOIVCanReadOldVersions.java

@@ -0,0 +1,135 @@
+/**
+ * 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.hdfs.tools.offlineImageViewer;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.apache.hadoop.hdfs.tools.offlineImageViewer.SpotCheckImageVisitor.ImageInfo;
+import org.junit.Test;
+
+public class TestOIVCanReadOldVersions {
+  // Location of fsimage files during testing.
+  public static final String TEST_CACHE_DATA_DIR =
+    System.getProperty("test.cache.data", "build/test/cache");
+  
+  // Verify that the image processor can correctly process prior Hadoop
+  // layout versions.  These fsimages were previously generated and stored
+  // with the test.  Test success indicates that no changes have been made
+  // to the OIV that causes older fsimages to be incorrectly processed.
+  @Test
+  public void testOldFSImages() {
+    // Define the expected values from the prior versions, as they were created
+    // and verified at time of creation
+    Set<String> pathNames = new HashSet<String>();
+    Collections.addAll(pathNames, "", /* root */
+                                  "/bar",
+                                  "/bar/dir0",
+                                  "/bar/dir0/file0",
+                                  "/bar/dir0/file1",
+                                  "/bar/dir1",
+                                  "/bar/dir1/file0",
+                                  "/bar/dir1/file1",
+                                  "/bar/dir2",
+                                  "/bar/dir2/file0",
+                                  "/bar/dir2/file1",
+                                  "/foo",
+                                  "/foo/dir0",
+                                  "/foo/dir0/file0",
+                                  "/foo/dir0/file1",
+                                  "/foo/dir0/file2",
+                                  "/foo/dir0/file3",
+                                  "/foo/dir1",
+                                  "/foo/dir1/file0",
+                                  "/foo/dir1/file1",
+                                  "/foo/dir1/file2",
+                                  "/foo/dir1/file3");
+    
+    Set<String> INUCpaths = new HashSet<String>();
+    Collections.addAll(INUCpaths, "/bar/dir0/file0",
+                                  "/bar/dir0/file1",
+                                  "/bar/dir1/file0",
+                                  "/bar/dir1/file1",
+                                  "/bar/dir2/file0",
+                                  "/bar/dir2/file1");
+    
+    ImageInfo v18Inodes = new ImageInfo(); // Hadoop version 18 inodes
+    v18Inodes.totalNumBlocks = 12;
+    v18Inodes.totalFileSize = 1069548540l;
+    v18Inodes.pathNames = pathNames;
+    v18Inodes.totalReplications = 14;
+    
+    ImageInfo v18INUCs = new ImageInfo(); // Hadoop version 18 inodes under construction
+    v18INUCs.totalNumBlocks = 0;
+    v18INUCs.totalFileSize = 0;
+    v18INUCs.pathNames = INUCpaths;
+    v18INUCs.totalReplications = 6;
+    
+    ImageInfo v19Inodes = new ImageInfo(); // Hadoop version 19 inodes
+    v19Inodes.totalNumBlocks = 12;
+    v19Inodes.totalFileSize = 1069548540l;
+    v19Inodes.pathNames = pathNames;
+    v19Inodes.totalReplications = 14;
+    
+    ImageInfo v19INUCs = new ImageInfo(); // Hadoop version 19 inodes under construction
+    v19INUCs.totalNumBlocks = 0;
+    v19INUCs.totalFileSize = 0;
+    v19INUCs.pathNames = INUCpaths;
+    v19INUCs.totalReplications = 6;
+    
+
+    spotCheck("18", TEST_CACHE_DATA_DIR + "/fsimageV18", v18Inodes, v18INUCs);
+    spotCheck("19", TEST_CACHE_DATA_DIR + "/fsimageV19", v19Inodes, v19INUCs);
+  }
+
+  // Check that running the processor now gives us the same values as before
+  private void spotCheck(String hadoopVersion, String input, 
+       ImageInfo inodes, ImageInfo INUCs) {
+    SpotCheckImageVisitor v = new SpotCheckImageVisitor();
+    OfflineImageViewer oiv = new OfflineImageViewer(input, v, false);
+    try {
+      oiv.go();
+    } catch (IOException e) {
+      fail("Error processing file: " + input);
+    }
+
+    compareSpotCheck(hadoopVersion, v.getINodesInfo(), inodes);
+    compareSpotCheck(hadoopVersion, v.getINUCsInfo(), INUCs);
+    System.out.println("Successfully processed fsimage file from Hadoop version " +
+                                                    hadoopVersion);
+  }
+
+  // Compare the spot check results of what we generated from the image
+  // processor and what we expected to receive
+  private void compareSpotCheck(String hadoopVersion, 
+                     ImageInfo generated, ImageInfo expected) {
+    assertEquals("Version " + hadoopVersion + ": Same number of total blocks", 
+                     expected.totalNumBlocks, generated.totalNumBlocks);
+    assertEquals("Version " + hadoopVersion + ": Same total file size", 
+                     expected.totalFileSize, generated.totalFileSize);
+    assertEquals("Version " + hadoopVersion + ": Same total replication factor", 
+                     expected.totalReplications, generated.totalReplications);
+    assertEquals("Version " + hadoopVersion + ": One-to-one matching of path names", 
+                     expected.pathNames, generated.pathNames);
+  }
+}

src/test/org/apache/hadoop/hdfs/tools/offlineImageViewer/TestOfflineImageViewer.java

@@ -0,0 +1,470 @@
+/**
+ * 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.hdfs.tools.offlineImageViewer;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.io.BufferedReader;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.EOFException;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.FileReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Set;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hdfs.DFSConfigKeys;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.hadoop.hdfs.protocol.FSConstants.SafeModeAction;
+import org.apache.hadoop.security.token.Token;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+
+/**
+ * Test function of OfflineImageViewer by:
+ *   * confirming it can correctly process a valid fsimage file and that
+ *     the processing generates a correct representation of the namespace
+ *   * confirming it correctly fails to process an fsimage file with a layout
+ *     version it shouldn't be able to handle
+ *   * confirm it correctly bails on malformed image files, in particular, a
+ *     file that ends suddenly.
+ */
+public class TestOfflineImageViewer {
+  private static final Log LOG = LogFactory.getLog(OfflineImageViewer.class);
+  private static final int NUM_DIRS = 3;
+  private static final int FILES_PER_DIR = 4;
+  private static final String TEST_RENEWER = "JobTracker";
+  private static File originalFsimage = null;
+
+  // Elements of lines of ls-file output to be compared to FileStatus instance
+  private static class LsElements {
+    public String perms;
+    public int replication;
+    public String username;
+    public String groupname;
+    public long filesize;
+    public char dir; // d if dir, - otherwise
+  }
+  
+  // namespace as written to dfs, to be compared with viewer's output
+  final static HashMap<String, FileStatus> writtenFiles = 
+      new HashMap<String, FileStatus>();
+  
+  private static String ROOT = System.getProperty("test.build.data",
+                                                  "build/test/data");
+  
+  // Create a populated namespace for later testing.  Save its contents to a
+  // data structure and store its fsimage location.
+  // We only want to generate the fsimage file once and use it for
+  // multiple tests.
+  @BeforeClass
+  public static void createOriginalFSImage() throws IOException {
+    MiniDFSCluster cluster = null;
+    try {
+      Configuration conf = new Configuration();
+      conf.setLong(DFSConfigKeys.DFS_NAMENODE_DELEGATION_TOKEN_MAX_LIFETIME_KEY, 10000);
+      conf.setLong(DFSConfigKeys.DFS_NAMENODE_DELEGATION_TOKEN_RENEW_INTERVAL_KEY, 5000);
+      conf.setBoolean(DFSConfigKeys.DFS_NAMENODE_DELEGATION_TOKEN_ALWAYS_USE_KEY, true);
+      conf.set("hadoop.security.auth_to_local",
+          "RULE:[2:$1@$0](JobTracker@.*FOO.COM)s/@.*//" + "DEFAULT");
+      cluster = new MiniDFSCluster(conf, 4, true, null);
+      cluster.waitActive();
+      FileSystem hdfs = cluster.getFileSystem();
+      
+      int filesize = 256;
+      
+      // Create a reasonable namespace 
+      for(int i = 0; i < NUM_DIRS; i++)  {
+        Path dir = new Path("/dir" + i);
+        hdfs.mkdirs(dir);
+        writtenFiles.put(dir.toString(), pathToFileEntry(hdfs, dir.toString()));
+        for(int j = 0; j < FILES_PER_DIR; j++) {
+          Path file = new Path(dir, "file" + j);
+          FSDataOutputStream o = hdfs.create(file);
+          o.write(new byte[ filesize++ ]);
+          o.close();
+          
+          writtenFiles.put(file.toString(), pathToFileEntry(hdfs, file.toString()));
+        }
+      }
+
+      // Get delegation tokens so we log the delegation token op
+      Token<?> delegationToken = hdfs.getDelegationToken(TEST_RENEWER);
+      LOG.debug("got token " + delegationToken);
+
+      // Write results to the fsimage file
+      cluster.getNameNode().setSafeMode(SafeModeAction.SAFEMODE_ENTER);
+      cluster.getNameNode().saveNamespace();
+      
+      // Determine location of fsimage file
+      File [] files = cluster.getNameDirs().toArray(new File[0]);
+      originalFsimage =  new File(files[0], "current/fsimage");
+      
+      if(!originalFsimage.exists())
+        fail("Didn't generate or can't find fsimage.");
+
+    } finally {
+      if(cluster != null)
+        cluster.shutdown();
+    }
+  }
+  
+  @AfterClass
+  public static void deleteOriginalFSImage() throws IOException {
+    if(originalFsimage != null && originalFsimage.exists()) {
+      originalFsimage.delete();
+    }
+  }
+  
+  // Convenience method to generate a file status from file system for 
+  // later comparison
+  private static FileStatus pathToFileEntry(FileSystem hdfs, String file) 
+        throws IOException {
+    return hdfs.getFileStatus(new Path(file));
+  }
+  
+  // Verify that we can correctly generate an ls-style output for a valid 
+  // fsimage
+  @Test
+  public void outputOfLSVisitor() throws IOException {
+    File testFile = new File(ROOT, "/basicCheck");
+    File outputFile = new File(ROOT, "/basicCheckOutput");
+    
+    try {
+      copyFile(originalFsimage, testFile);
+      
+      ImageVisitor v = new LsImageVisitor(outputFile.getPath(), true);
+      OfflineImageViewer oiv = new OfflineImageViewer(testFile.getPath(), v, false);
+
+      oiv.go();
+      
+      HashMap<String, LsElements> fileOutput = readLsfile(outputFile);
+      
+      compareNamespaces(writtenFiles, fileOutput);
+    } finally {
+      if(testFile.exists()) testFile.delete();
+      if(outputFile.exists()) outputFile.delete();
+    }
+    LOG.debug("Correctly generated ls-style output.");
+  }
+  
+  // Confirm that attempting to read an fsimage file with an unsupported
+  // layout results in an error
+  @Test
+  public void unsupportedFSLayoutVersion() throws IOException {
+    File testFile = new File(ROOT, "/invalidLayoutVersion");
+    File outputFile = new File(ROOT, "invalidLayoutVersionOutput");
+    
+    try {
+      int badVersionNum = -432;
+      changeLayoutVersion(originalFsimage, testFile, badVersionNum);
+      ImageVisitor v = new LsImageVisitor(outputFile.getPath(), true);
+      OfflineImageViewer oiv = new OfflineImageViewer(testFile.getPath(), v, false);
+      
+      try {
+        oiv.go();
+        fail("Shouldn't be able to read invalid laytout version");
+      } catch(IOException e) {
+        if(!e.getMessage().contains(Integer.toString(badVersionNum)))
+          throw e; // wasn't error we were expecting
+        LOG.debug("Correctly failed at reading bad image version.");
+      }
+    } finally {
+      if(testFile.exists()) testFile.delete();
+      if(outputFile.exists()) outputFile.delete();
+    }
+  }
+  
+  // Verify that image viewer will bail on a file that ends unexpectedly
+  @Test
+  public void truncatedFSImage() throws IOException {
+    File testFile = new File(ROOT, "/truncatedFSImage");
+    File outputFile = new File(ROOT, "/trucnatedFSImageOutput");
+    try {
+      copyPartOfFile(originalFsimage, testFile);
+      assertTrue("Created truncated fsimage", testFile.exists());
+      
+      ImageVisitor v = new LsImageVisitor(outputFile.getPath(), true);
+      OfflineImageViewer oiv = new OfflineImageViewer(testFile.getPath(), v, false);
+
+      try {
+        oiv.go();
+        fail("Managed to process a truncated fsimage file");
+      } catch (EOFException e) {
+        LOG.debug("Correctly handled EOF");
+      }
+
+    } finally {
+      if(testFile.exists()) testFile.delete();
+      if(outputFile.exists()) outputFile.delete();
+    }
+  }
+  
+  // Test that our ls file has all the same compenents of the original namespace
+  private void compareNamespaces(HashMap<String, FileStatus> written,
+      HashMap<String, LsElements> fileOutput) {
+    assertEquals( "Should be the same number of files in both, plus one for root"
+            + " in fileoutput", fileOutput.keySet().size(), 
+                                written.keySet().size() + 1);
+    Set<String> inFile = fileOutput.keySet();
+
+    // For each line in the output file, verify that the namespace had a
+    // filestatus counterpart 
+    for (String path : inFile) {
+      if (path.equals("/")) // root's not included in output from system call
+        continue;
+
+      assertTrue("Path in file (" + path + ") was written to fs", written
+          .containsKey(path));
+      
+      compareFiles(written.get(path), fileOutput.get(path));
+      
+      written.remove(path);
+    }
+
+    assertEquals("No more files were written to fs", 0, written.size());
+  }
+  
+  // Compare two files as listed in the original namespace FileStatus and
+  // the output of the ls file from the image processor
+  private void compareFiles(FileStatus fs, LsElements elements) {
+    assertEquals("directory listed as such",  
+                 fs.isDir() ? 'd' : '-', elements.dir);
+    assertEquals("perms string equal", 
+                                fs.getPermission().toString(), elements.perms);
+    assertEquals("replication equal", fs.getReplication(), elements.replication);
+    assertEquals("owner equal", fs.getOwner(), elements.username);
+    assertEquals("group equal", fs.getGroup(), elements.groupname);
+    assertEquals("lengths equal", fs.getLen(), elements.filesize);
+  }
+
+  // Read the contents of the file created by the Ls processor
+  private HashMap<String, LsElements> readLsfile(File lsFile) throws IOException {
+    BufferedReader br = new BufferedReader(new FileReader(lsFile));
+    String line = null;
+    HashMap<String, LsElements> fileContents = new HashMap<String, LsElements>();
+    
+    while((line = br.readLine()) != null) 
+      readLsLine(line, fileContents);
+    
+    return fileContents;
+  }
+  
+  // Parse a line from the ls output.  Store permissions, replication, 
+  // username, groupname and filesize in hashmap keyed to the path name
+  private void readLsLine(String line, HashMap<String, LsElements> fileContents) {
+    String elements [] = line.split("\\s+");
+    
+    assertEquals("Not enough elements in ls output", 8, elements.length);
+    
+    LsElements lsLine = new LsElements();
+    
+    lsLine.dir = elements[0].charAt(0);
+    lsLine.perms = elements[0].substring(1);
+    lsLine.replication = elements[1].equals("-") 
+                                             ? 0 : Integer.valueOf(elements[1]);
+    lsLine.username = elements[2];
+    lsLine.groupname = elements[3];
+    lsLine.filesize = Long.valueOf(elements[4]);
+    // skipping date and time 
+    
+    String path = elements[7];
+    
+    // Check that each file in the ls output was listed once
+    assertFalse("LS file had duplicate file entries", 
+        fileContents.containsKey(path));
+    
+    fileContents.put(path, lsLine);
+  }
+  
+  // Copy one fsimage to another, changing the layout version in the process
+  private void changeLayoutVersion(File src, File dest, int newVersion) 
+         throws IOException {
+    DataInputStream in = null; 
+    DataOutputStream out = null; 
+    
+    try {
+      in = new DataInputStream(new FileInputStream(src));
+      out = new DataOutputStream(new FileOutputStream(dest));
+      
+      in.readInt();
+      out.writeInt(newVersion);
+      
+      byte [] b = new byte[1024];
+      while( in.read(b)  > 0 ) {
+        out.write(b);
+      }
+    } finally {
+      if(in != null) in.close();
+      if(out != null) out.close();
+    }
+  }
+  
+  // Only copy part of file into the other.  Used for testing truncated fsimage
+  private void copyPartOfFile(File src, File dest) throws IOException {
+    InputStream in = null;
+    OutputStream out = null;
+    
+    byte [] b = new byte[256];
+    int bytesWritten = 0;
+    int count;
+    int maxBytes = 700;
+    
+    try {
+      in = new FileInputStream(src);
+      out = new FileOutputStream(dest);
+      
+      while( (count = in.read(b))  > 0 && bytesWritten < maxBytes ) {
+        out.write(b);
+        bytesWritten += count;
+      } 
+    } finally {
+      if(in != null) in.close();
+      if(out != null) out.close();
+    }
+  }
+  
+  // Copy one file's contents into the other
+  private void copyFile(File src, File dest) throws IOException {
+    InputStream in = null;
+    OutputStream out = null;
+    
+    try {
+      in = new FileInputStream(src);
+      out = new FileOutputStream(dest);
+
+      byte [] b = new byte[1024];
+      while( in.read(b)  > 0 ) {
+        out.write(b);
+      }
+    } finally {
+      if(in != null) in.close();
+      if(out != null) out.close();
+    }
+  }
+
+  @Test
+  public void outputOfFileDistributionVisitor() throws IOException {
+    File testFile = new File(ROOT, "/basicCheck");
+    File outputFile = new File(ROOT, "/fileDistributionCheckOutput");
+
+    int totalFiles = 0;
+    try {
+      copyFile(originalFsimage, testFile);
+      ImageVisitor v = new FileDistributionVisitor(outputFile.getPath(), 0, 0);
+      OfflineImageViewer oiv = 
+        new OfflineImageViewer(testFile.getPath(), v, false);
+
+      oiv.go();
+
+      BufferedReader reader = new BufferedReader(new FileReader(outputFile));
+      String line = reader.readLine();
+      assertEquals(line, "Size\tNumFiles");
+      while((line = reader.readLine()) != null) {
+        String[] row = line.split("\t");
+        assertEquals(row.length, 2);
+        totalFiles += Integer.parseInt(row[1]);
+      }
+    } finally {
+      if(testFile.exists()) testFile.delete();
+      if(outputFile.exists()) outputFile.delete();
+    }
+    assertEquals(totalFiles, NUM_DIRS * FILES_PER_DIR);
+  }
+  
+  private static class TestImageVisitor extends ImageVisitor {
+    private List<String> delegationTokenRenewers = new LinkedList<String>();
+    TestImageVisitor() {
+    }
+    
+    List<String> getDelegationTokenRenewers() {
+      return delegationTokenRenewers;
+    }
+
+    @Override
+    void start() throws IOException {
+    }
+
+    @Override
+    void finish() throws IOException {
+    }
+
+    @Override
+    void finishAbnormally() throws IOException {
+    }
+
+    @Override
+    void visit(ImageElement element, String value) throws IOException {
+      if (element == ImageElement.DELEGATION_TOKEN_IDENTIFIER_RENEWER) {
+        delegationTokenRenewers.add(value);
+      }
+    }
+
+    @Override
+    void visitEnclosingElement(ImageElement element) throws IOException {
+    }
+
+    @Override
+    void visitEnclosingElement(ImageElement element, ImageElement key,
+        String value) throws IOException {
+    }
+
+    @Override
+    void leaveEnclosingElement() throws IOException {
+    }
+  }
+
+  @Test
+  public void outputOfTestVisitor() throws IOException {
+    File testFile = new File(ROOT, "/basicCheck");
+
+    try {
+      copyFile(originalFsimage, testFile);
+      TestImageVisitor v = new TestImageVisitor();
+      OfflineImageViewer oiv = new OfflineImageViewer(testFile.getPath(), v, true);
+      oiv.go();
+
+      // Validated stored delegation token identifiers.
+      List<String> dtrs = v.getDelegationTokenRenewers();
+      assertEquals(1, dtrs.size());
+      assertEquals(TEST_RENEWER, dtrs.get(0));
+    } finally {
+      if(testFile.exists()) testFile.delete();
+    }
+    LOG.debug("Passed TestVisitor validation.");
+  }
+}