HADOOP-2063. A new parameter to dfs -get command to fetch a file

even if it is corrupted.  (Tsz Wo (Nicholas), SZE via dhruba)



git-svn-id: https://svn.apache.org/repos/asf/hadoop/core/trunk@631986 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -26,6 +26,9 @@ Trunk (unreleased changes)
 
     HADOOP-2178.  Job History on DFS. (Amareshwari Sri Ramadasu via ddas)
 
+    HADOOP-2063. A new parameter to dfs -get command to fetch a file 
+    even if it is corrupted.  (Tsz Wo (Nicholas), SZE via dhruba)
+
   IMPROVEMENTS
 
     HADOOP-2655. Copy on write for data and metadata files in the 

src/java/org/apache/hadoop/dfs/DFSClient.java

@@ -252,7 +252,7 @@ class DFSClient implements FSConstants {
   }
 
   public DFSInputStream open(String src) throws IOException {
-    return open(src, conf.getInt("io.file.buffer.size", 4096));
+    return open(src, conf.getInt("io.file.buffer.size", 4096), true);
   }
   /**
    * Create an input stream that obtains a nodelist from the
@@ -260,10 +260,11 @@ class DFSClient implements FSConstants {
    * inner subclass of InputStream that does the right out-of-band
    * work.
    */
-  public DFSInputStream open(String src, int buffersize) throws IOException {
+  DFSInputStream open(String src, int buffersize, boolean verifyChecksum
+      ) throws IOException {
     checkOpen();
     //    Get block info from namenode
-    return new DFSInputStream(src, buffersize);
+    return new DFSInputStream(src, buffersize, verifyChecksum);
   }
 
   /**
@@ -773,10 +774,11 @@ class DFSClient implements FSConstants {
     }
     
     private BlockReader( String file, long blockId, DataInputStream in, 
-                         DataChecksum checksum, long startOffset,
-                         long firstChunkOffset ) {
+                         DataChecksum checksum, boolean verifyChecksum,
+                         long startOffset, long firstChunkOffset ) {
       super(new Path("/blk_" + blockId + ":of:" + file)/*too non path-like?*/,
-            1, (checksum.getChecksumSize() > 0) ? checksum : null, 
+            1, verifyChecksum,
+            checksum.getChecksumSize() > 0? checksum : null, 
             checksum.getBytesPerChecksum(),
             checksum.getChecksumSize());
       
@@ -792,12 +794,17 @@ class DFSClient implements FSConstants {
       checksumSize = this.checksum.getChecksumSize();
     }
 
+    static BlockReader newBlockReader(Socket sock, String file, long blockId, 
+        long startOffset, long len, int bufferSize) throws IOException {
+      return newBlockReader(sock, file, blockId, startOffset, len, bufferSize,
+          true);
+    }
+
     /** Java Doc required */
     static BlockReader newBlockReader( Socket sock, String file, long blockId, 
                                        long startOffset, long len,
-                                       int bufferSize)
+                                       int bufferSize, boolean verifyChecksum)
                                        throws IOException {
-      
       // in and out will be closed when sock is closed (by the caller)
       DataOutputStream out = new DataOutputStream(
                        new BufferedOutputStream(sock.getOutputStream()));
@@ -835,7 +842,7 @@ class DFSClient implements FSConstants {
                               startOffset + " for file " + file);
       }
 
-      return new BlockReader( file, blockId, in, checksum,
+      return new BlockReader( file, blockId, in, checksum, verifyChecksum,
                               startOffset, firstChunkOffset );
     }
 
@@ -882,7 +889,8 @@ class DFSClient implements FSConstants {
 
     private String src;
     private long prefetchSize = 10 * defaultBlockSize;
-    private BlockReader blockReader;
+    private BlockReader blockReader = null;
+    private boolean verifyChecksum;
     private LocatedBlocks locatedBlocks = null;
     private DatanodeInfo currentNode = null;
     private Block currentBlock = null;
@@ -900,14 +908,13 @@ class DFSClient implements FSConstants {
       deadNodes.put(dnInfo, dnInfo);
     }
     
-    /**
-     */
-    public DFSInputStream(String src, int buffersize) throws IOException {
+    DFSInputStream(String src, int buffersize, boolean verifyChecksum
+        ) throws IOException {
+      this.verifyChecksum = verifyChecksum;
       this.buffersize = buffersize;
       this.src = src;
       prefetchSize = conf.getLong("dfs.read.prefetch.size", prefetchSize);
       openInfo();
-      blockReader = null;
     }
 
     /**
@@ -1065,10 +1072,8 @@ class DFSClient implements FSConstants {
           Block blk = targetBlock.getBlock();
           
           blockReader = BlockReader.newBlockReader(s, src, blk.getBlockId(), 
-                                                   offsetIntoBlock,
-                                                   (blk.getNumBytes() - 
-                                                    offsetIntoBlock),
-                                                   buffersize);
+              offsetIntoBlock, blk.getNumBytes() - offsetIntoBlock,
+              buffersize, verifyChecksum);
           return chosenNode;
         } catch (IOException ex) {
           // Put chosen node into dead list, continue
@@ -1248,7 +1253,7 @@ class DFSClient implements FSConstants {
               
           BlockReader reader = 
             BlockReader.newBlockReader(dn, src, block.getBlock().getBlockId(),
-                                       start, len, buffersize);
+                                       start, len, buffersize, verifyChecksum);
           int nread = reader.readAll(buf, offset, len);
           if (nread != len) {
             throw new IOException("truncated return from reader.read(): " +

src/java/org/apache/hadoop/dfs/DistributedFileSystem.java

@@ -40,6 +40,7 @@ public class DistributedFileSystem extends FileSystem {
   private URI uri;
 
   DFSClient dfs;
+  private boolean verifyChecksum = true;
 
   public DistributedFileSystem() {
   }
@@ -111,9 +112,14 @@ public class DistributedFileSystem extends FileSystem {
     return dfs.getHints(getPathName(f), start, len);
   }
 
+  public void setVerifyChecksum(boolean verifyChecksum) {
+    this.verifyChecksum = verifyChecksum;
+  }
+
   public FSDataInputStream open(Path f, int bufferSize) throws IOException {
     try {
-      return new DFSClient.DFSDataInputStream(dfs.open(getPathName(f),bufferSize));
+      return new DFSClient.DFSDataInputStream(
+          dfs.open(getPathName(f), bufferSize, verifyChecksum));
     } catch(RemoteException e) {
       if (IOException.class.getName().equals(e.getClassName()) &&
           e.getMessage().startsWith(

src/java/org/apache/hadoop/fs/FSInputChecker.java

@@ -37,6 +37,7 @@ abstract public class FSInputChecker extends FSInputStream {
   /** The file name from which data is read from */
   protected Path file;
   private Checksum sum;
+  private boolean verifyChecksum;
   private byte[] buf;
   private byte[] checksum;
   private int pos;
@@ -66,8 +67,9 @@ abstract public class FSInputChecker extends FSInputStream {
    * @param checksumSize the number byte of each checksum
    */
   protected FSInputChecker( Path file, int numOfRetries, 
-      Checksum sum, int chunkSize, int checksumSize ) {
+      boolean verifyChecksum, Checksum sum, int chunkSize, int checksumSize ) {
     this(file, numOfRetries);
+    this.verifyChecksum = verifyChecksum;
     set(sum, chunkSize, checksumSize);
   }
   
@@ -93,7 +95,7 @@ abstract public class FSInputChecker extends FSInputStream {
 
   /** Return true if there is a need for checksum verification */
   protected synchronized boolean needChecksum() {
-    return sum != null;
+    return verifyChecksum && sum != null;
   }
   
   /**
@@ -163,7 +165,7 @@ abstract public class FSInputChecker extends FSInputStream {
     }
   }
   
-  /*
+  /**
    * Fills the buffer with a chunk data. 
    * No mark is supported.
    * This method assumes that all data in the buffer has already been read in,

src/java/org/apache/hadoop/fs/FsShell.java

@@ -39,9 +39,9 @@ import org.apache.hadoop.io.WritableComparable;
 import org.apache.hadoop.ipc.RPC;
 import org.apache.hadoop.ipc.RemoteException;
 import org.apache.hadoop.util.ReflectionUtils;
-import org.apache.hadoop.util.StringUtils;
 import org.apache.hadoop.util.Tool;
 import org.apache.hadoop.util.ToolRunner;
+import org.apache.hadoop.dfs.DistributedFileSystem;
 
 /** Provide command line access to a FileSystem. */
 public class FsShell extends Configured implements Tool {
@@ -56,6 +56,9 @@ public class FsShell extends Configured implements Tool {
     modifFmt.setTimeZone(TimeZone.getTimeZone("UTC"));
   }
   static final String SETREP_SHORT_USAGE="-setrep [-R] [-w] <rep> <path/file>";
+  static final String GET_SHORT_USAGE = "-get [-ignoreCrc] [-crc] <src> <localdst>";
+  static final String COPYTOLOCAL_SHORT_USAGE = GET_SHORT_USAGE.replace(
+      "-get", "-copyToLocal");
   static final String TAIL_USAGE="-tail [-f] <file>";
   private static final DecimalFormat decimalFormat;
   static {
@@ -142,7 +145,7 @@ public class FsShell extends Configured implements Tool {
   /**
    * Obtain the indicated files that match the file pattern <i>srcf</i>
    * and copy them to the local name. srcf is kept.
-   * When copying mutiple files, the destination must be a directory. 
+   * When copying multiple files, the destination must be a directory. 
    * Otherwise, IOException is thrown.
    * @param argv: arguments
    * @param pos: Ignore everything before argv[pos]  
@@ -150,39 +153,58 @@ public class FsShell extends Configured implements Tool {
    * @see org.apache.hadoop.fs.FileSystem.globPaths 
    */
   void copyToLocal(String[]argv, int pos) throws IOException {
-    if (argv.length-pos<2 || (argv.length-pos==2 && argv[pos].equalsIgnoreCase("-crc"))) {
-      System.err.println("Usage: -get [-crc] <src> <dst>");
-      throw new RuntimeException("Usage: -get [-crc] <src> <dst>");
-    }
-    boolean copyCrc = false;
-    if ("-crc".equalsIgnoreCase(argv[pos])) {
-      pos++;
-      copyCrc = true;
-    }
-    String srcf = argv[pos++];
-    String dstf = argv[pos++];
-    if (dstf.equals("-")) {
+    CommandFormat cf = new CommandFormat("copyToLocal", 2,2,"crc","ignoreCrc");
+    
+    String srcstr = null;
+    String dststr = null;
+    try {
+      List<String> parameters = cf.parse(argv, pos);
+      srcstr = parameters.get(0);
+      dststr = parameters.get(1);
+    }
+    catch(IllegalArgumentException iae) {
+      System.err.println("Usage: java FsShell " + GET_SHORT_USAGE);
+      throw iae;
+    }
+    final boolean copyCrc = cf.options.get("crc");
+    final boolean verifyChecksum = !cf.options.get("ignoreCrc");
+
+    if (dststr.equals("-")) {
       if (copyCrc) {
         System.err.println("-crc option is not valid when destination is stdout.");
       }
-      cat(srcf);
+      cat(srcstr, verifyChecksum);
     } else {
-      File dst = new File(dstf);      
-      Path src = new Path(srcf);
-      FileSystem srcFs = src.getFileSystem(getConf());      
-      Path [] srcs = srcFs.globPaths(src);
+      File dst = new File(dststr);      
+      Path srcpath = new Path(srcstr);
+      FileSystem srcFS = getSrcFileSystem(srcpath, verifyChecksum);
+      FileStatus[] srcs = srcFS.globStatus(srcpath);
       boolean dstIsDir = dst.isDirectory(); 
       if (srcs.length > 1 && !dstIsDir) {
         throw new IOException("When copying multiple files, "
                               + "destination should be a directory.");
       }
-      for (Path srcPath : srcs) {
-        File dstFile = (dstIsDir ? new File(dst, srcPath.getName()) : dst);
-        copyToLocal(srcFs, srcPath, dstFile, copyCrc);
+      for (FileStatus status : srcs) {
+        Path p = status.getPath();
+        File f = dstIsDir? new File(dst, p.getName()): dst;
+        copyToLocal(srcFS, p, f, copyCrc);
       }
     }
   }
 
+  /**
+   * Return the {@link FileSystem} specified by src and the conf.
+   * It the {@link FileSystem} supports checksum, set verifyChecksum.
+   */
+  private FileSystem getSrcFileSystem(Path src, boolean verifyChecksum
+      ) throws IOException { 
+    FileSystem srcFs = src.getFileSystem(getConf());
+    if (srcFs instanceof DistributedFileSystem) {
+      ((DistributedFileSystem)srcFs).setVerifyChecksum(verifyChecksum);
+    }
+    return srcFs;
+  }
+
   /**
    * The prefix for the tmp file used in copyToLocal.
    * It must be at least three characters long, required by
@@ -207,7 +229,7 @@ public class FsShell extends Configured implements Tool {
      * copyCrc and useTmpFile (may be useTmpFile need not be an option).
      */
     
-    if (!srcFS.isDirectory(src)) {
+    if (!srcFS.getFileStatus(src).isDir()) {
       if (dst.exists()) {
         // match the error message in FileUtil.checkDest():
         throw new IOException("Target " + dst + " already exists");
@@ -298,7 +320,7 @@ public class FsShell extends Configured implements Tool {
    * @exception: IOException
    * @see org.apache.hadoop.fs.FileSystem.globPaths 
    */
-  void cat(String srcf) throws IOException {
+  void cat(String src, boolean verifyChecksum) throws IOException {
     //cat behavior in Linux
     //  [~/1207]$ ls ?.txt
     //  x.txt  z.txt
@@ -307,6 +329,7 @@ public class FsShell extends Configured implements Tool {
     //  cat: y.txt: No such file or directory
     //  zzz
 
+    Path srcPattern = new Path(src);
     new DelayedExceptionThrowing() {
       @Override
       void process(Path p, FileSystem srcFs) throws IOException {
@@ -315,7 +338,7 @@ public class FsShell extends Configured implements Tool {
         }
         printToStdout(srcFs.open(p));
       }
-    }.process(srcf);
+    }.process(srcPattern, getSrcFileSystem(srcPattern, verifyChecksum));
   }
 
   private class TextRecordInputStream extends InputStream {
@@ -374,6 +397,7 @@ public class FsShell extends Configured implements Tool {
   }
 
   void text(String srcf) throws IOException {
+    Path srcPattern = new Path(srcf);
     new DelayedExceptionThrowing() {
       @Override
       void process(Path p, FileSystem srcFs) throws IOException {
@@ -382,7 +406,7 @@ public class FsShell extends Configured implements Tool {
         }
         printToStdout(forMagic(p, srcFs));
       }
-    }.process(srcf);
+    }.process(srcPattern, srcPattern.getFileSystem(getConf()));
   }
 
   /**
@@ -1002,12 +1026,13 @@ public class FsShell extends Configured implements Tool {
     //  [~/1207]$ rm x.txt y.txt z.txt 
     //  rm: cannot remove `y.txt': No such file or directory
 
+    Path srcPattern = new Path(srcf);
     new DelayedExceptionThrowing() {
       @Override
       void process(Path p, FileSystem srcFs) throws IOException {
         delete(p, srcFs, recursive);
       }
-    }.process(srcf);
+    }.process(srcPattern, srcPattern.getFileSystem(getConf()));
   }
     
   /* delete a file */
@@ -1222,9 +1247,9 @@ public class FsShell extends Configured implements Tool {
       "[-D <property=value>] [-ls <path>] [-lsr <path>] [-du <path>]\n\t" + 
       "[-dus <path>] [-mv <src> <dst>] [-cp <src> <dst>] [-rm <src>]\n\t" + 
       "[-rmr <src>] [-put <localsrc> <dst>] [-copyFromLocal <localsrc> <dst>]\n\t" +
-      "[-moveFromLocal <localsrc> <dst>] [-get <src> <localdst>]\n\t" +
+      "[-moveFromLocal <localsrc> <dst>] [" + GET_SHORT_USAGE + "\n\t" +
       "[-getmerge <src> <localdst> [addnl]] [-cat <src>]\n\t" +
-      "[-copyToLocal <src><localdst>] [-moveToLocal <src> <localdst>]\n\t" +
+      "[" + COPYTOLOCAL_SHORT_USAGE + "] [-moveToLocal <src> <localdst>]\n\t" +
       "[-mkdir <path>] [-report] [" + SETREP_SHORT_USAGE + "]\n\t" +
       "[-touchz <path>] [-test -[ezd] <path>] [-stat [format] <path>]\n\t" +
       "[-tail [-f] <path>] [-text <path>]\n\t" +
@@ -1298,7 +1323,8 @@ public class FsShell extends Configured implements Tool {
     String moveFromLocal = "-moveFromLocal <localsrc> <dst>:  Same as -put, except that the source is\n" +
       "\t\tdeleted after it's copied.\n"; 
 
-    String get = "-get <src> <localdst>:  Copy files that match the file pattern <src> \n" +
+    String get = GET_SHORT_USAGE
+      + ":  Copy files that match the file pattern <src> \n" +
       "\t\tto the local name.  <src> is kept.  When copying mutiple, \n" +
       "\t\tfiles, the destination must be a directory. \n";
 
@@ -1313,7 +1339,8 @@ public class FsShell extends Configured implements Tool {
       "\t\tmatch a magic number associated with a known format\n" +
       "\t\t(gzip, SequenceFile)\n";
         
-    String copyToLocal = "-copyToLocal <src> <localdst>:  Identical to the -get command.\n";
+    String copyToLocal = COPYTOLOCAL_SHORT_USAGE
+                         + ":  Identical to the -get command.\n";
 
     String moveToLocal = "-moveToLocal <src> <localdst>:  Not implemented yet \n";
         
@@ -1473,7 +1500,7 @@ public class FsShell extends Configured implements Tool {
         // issue the command to the fs
         //
         if ("-cat".equals(cmd)) {
-          cat(argv[i]);
+          cat(argv[i], true);
         } else if ("-mkdir".equals(cmd)) {
           mkdir(argv[i]);
         } else if ("-rm".equals(cmd)) {
@@ -1496,7 +1523,7 @@ public class FsShell extends Configured implements Tool {
       } catch (RemoteException e) {
         //
         // This is a error returned by hadoop server. Print
-        // out the first line of the error mesage.
+        // out the first line of the error message.
         //
         exitCode = -1;
         try {
@@ -1548,16 +1575,16 @@ public class FsShell extends Configured implements Tool {
                "-moveFromLocal".equals(cmd)) {
       System.err.println("Usage: java FsShell" + 
                          " [" + cmd + " <localsrc> <dst>]");
-    } else if ("-get".equals(cmd) || "-copyToLocal".equals(cmd) ||
-               "-moveToLocal".equals(cmd)) {
+    } else if ("-get".equals(cmd)) {
+      System.err.println("Usage: java FsShell [" + GET_SHORT_USAGE + "]"); 
+    } else if ("-copyToLocal".equals(cmd)) {
+      System.err.println("Usage: java FsShell [" + COPYTOLOCAL_SHORT_USAGE+ "]"); 
+    } else if ("-moveToLocal".equals(cmd)) {
       System.err.println("Usage: java FsShell" + 
                          " [" + cmd + " [-crc] <src> <localdst>]");
     } else if ("-cat".equals(cmd)) {
       System.err.println("Usage: java FsShell" + 
                          " [" + cmd + " <src>]");
-    } else if ("-get".equals(cmd)) {
-      System.err.println("Usage: java FsShell" + 
-                         " [" + cmd + " <src> <localdst> [addnl]]");
     } else if ("-setrep".equals(cmd)) {
       System.err.println("Usage: java FsShell [" + SETREP_SHORT_USAGE + "]");
     } else if ("-test".equals(cmd)) {
@@ -1582,11 +1609,11 @@ public class FsShell extends Configured implements Tool {
       System.err.println("           [-put <localsrc> <dst>]");
       System.err.println("           [-copyFromLocal <localsrc> <dst>]");
       System.err.println("           [-moveFromLocal <localsrc> <dst>]");
-      System.err.println("           [-get [-crc] <src> <localdst>]");
+      System.err.println("           [" + GET_SHORT_USAGE + "]");
       System.err.println("           [-getmerge <src> <localdst> [addnl]]");
       System.err.println("           [-cat <src>]");
       System.err.println("           [-text <src>]");
-      System.err.println("           [-copyToLocal [-crc] <src> <localdst>]");
+      System.err.println("           [" + COPYTOLOCAL_SHORT_USAGE + "]");
       System.err.println("           [-moveToLocal [-crc] <src> <localdst>]");
       System.err.println("           [-mkdir <path>]");
       System.err.println("           [" + SETREP_SHORT_USAGE + "]");
@@ -1793,17 +1820,16 @@ public class FsShell extends Configured implements Tool {
     System.exit(res);
   }
 
-  /*
+  /**
    * Accumulate exceptions if there is any.  Throw them at last.
    */
   private abstract class DelayedExceptionThrowing {
     abstract void process(Path p, FileSystem srcFs) throws IOException;
 
-    void process(String srcf) throws IOException {
+    final void processSrc(Path srcPattern, FileSystem srcFs
+        ) throws IOException {
       List<IOException> exceptions = new ArrayList<IOException>();
-      Path srcPath = new Path(srcf);
-      FileSystem srcFs = srcPath.getFileSystem(getConf());
-      for(Path p : srcFs.globPaths(new Path(srcf)))
+      for(Path p : srcFs.globPaths(srcPattern))
         try { process(p, srcFs); } 
         catch(IOException ioe) { exceptions.add(ioe); }
     

src/test/org/apache/hadoop/dfs/DFSTestUtil.java

@@ -18,8 +18,10 @@
 
 package org.apache.hadoop.dfs;
 
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileReader;
 import java.io.IOException;
-import java.net.URI;
 import java.util.Random;
 import junit.framework.TestCase;
 import org.apache.hadoop.conf.Configuration;
@@ -37,7 +39,6 @@ public class DFSTestUtil extends TestCase {
   private static String[] dirNames = {
     "zero", "one", "two", "three", "four", "five", "six", "seven", "eight", "nine"
   };
-  private static Configuration conf = new Configuration();
   
   private int maxLevels;// = 3;
   private int maxSize;// = 8*1024;
@@ -220,4 +221,17 @@ public class DFSTestUtil extends TestCase {
     in.readByte();
     return in.getCurrentBlock();
   }  
+
+  static void setLogLevel2All(org.apache.commons.logging.Log log) {
+    ((org.apache.commons.logging.impl.Log4JLogger)log
+        ).getLogger().setLevel(org.apache.log4j.Level.ALL);
+  }
+
+  static String readFile(File f) throws IOException {
+    StringBuilder b = new StringBuilder();
+    BufferedReader in = new BufferedReader(new FileReader(f));
+    for(int c; (c = in.read()) != -1; b.append((char)c));
+    in.close();      
+    return b.toString();
+  }
 }

src/test/org/apache/hadoop/dfs/TestDFSShell.java

@@ -53,8 +53,9 @@ public class TestDFSShell extends TestCase {
 
   static File createLocalFile(File f) throws IOException {
     assertTrue(!f.exists());
-    PrintWriter out = new PrintWriter(new BufferedWriter(new FileWriter(f)));
-    out.println(f.getAbsolutePath());
+    PrintWriter out = new PrintWriter(f);
+    out.print("createLocalFile: " + f.getAbsolutePath());
+    out.flush();
     out.close();
     assertTrue(f.exists());
     assertTrue(f.isFile());
@@ -866,4 +867,92 @@ public class TestDFSShell extends TestCase {
       cluster.shutdown();
     }
   }
+
+  static List<File> getBlockFiles(MiniDFSCluster cluster) throws IOException {
+    List<File> files = new ArrayList<File>();
+    List<DataNode> datanodes = cluster.getDataNodes();
+    Block[][] blocks = cluster.getAllBlockReports();
+    for(int i = 0; i < blocks.length; i++) {
+      FSDataset ds = (FSDataset)datanodes.get(i).getFSDataset();
+      for(Block b : blocks[i]) {
+        files.add(ds.getBlockFile(b));
+      }        
+    }
+    return files;
+  }
+
+  static void corrupt(List<File> files) throws IOException {
+    for(File f : files) {
+      StringBuilder content = new StringBuilder(DFSTestUtil.readFile(f));
+      char c = content.charAt(0);
+      content.setCharAt(0, ++c);
+      PrintWriter out = new PrintWriter(f);
+      out.print(content);
+      out.flush();
+      out.close();      
+    }
+  }
+
+  static interface TestGetRunner {
+    String run(int exitcode, String... options) throws IOException;
+  }
+
+  public void testGet() throws IOException {
+    DFSTestUtil.setLogLevel2All(FSInputChecker.LOG);
+    final Configuration conf = new Configuration();
+    MiniDFSCluster cluster = new MiniDFSCluster(conf, 2, true, null);
+    DistributedFileSystem dfs = (DistributedFileSystem)cluster.getFileSystem();
+
+    try {
+      final String fname = "testGet.txt";
+      final File localf = createLocalFile(new File(TEST_ROOT_DIR, fname));
+      final String localfcontent = DFSTestUtil.readFile(localf);
+      final Path root = mkdir(dfs, new Path("/test/get"));
+      final Path remotef = new Path(root, fname);
+      dfs.copyFromLocalFile(false, false, new Path(localf.getPath()), remotef);
+
+      final FsShell shell = new FsShell();
+      shell.setConf(conf);
+      TestGetRunner runner = new TestGetRunner() {
+        private int count = 0;
+
+        public String run(int exitcode, String... options) throws IOException {
+          String dst = TEST_ROOT_DIR + "/" + fname+ ++count;
+          String[] args = new String[options.length + 3];
+          args[0] = "-get"; 
+          args[args.length - 2] = remotef.toString();
+          args[args.length - 1] = dst;
+          for(int i = 0; i < options.length; i++) {
+            args[i + 1] = options[i];
+          }
+          show("args=" + Arrays.asList(args));
+          
+          try {
+            assertEquals(exitcode, shell.run(args));
+          } catch (Exception e) {
+            assertTrue(StringUtils.stringifyException(e), false); 
+          }
+          return exitcode == 0? DFSTestUtil.readFile(new File(dst)): null; 
+        }
+      };
+
+      assertEquals(localfcontent, runner.run(0));
+      assertEquals(localfcontent, runner.run(0, "-ignoreCrc"));
+
+      //find and modify the block files
+      List<File> files = getBlockFiles(cluster);
+      show("files=" + files);
+      corrupt(files);
+
+      assertEquals(null, runner.run(-1));
+      String corruptedcontent = runner.run(0, "-ignoreCrc");
+      assertEquals(localfcontent.substring(1), corruptedcontent.substring(1));
+      assertEquals(localfcontent.charAt(0)+1, corruptedcontent.charAt(0));
+
+      localf.delete();
+    } finally {
+      try {dfs.close();} catch (Exception e) {}
+      cluster.shutdown();
+    }
+  }
 }