HADOOP-3981. Implement distributed file checksum and change DistCp to use it.

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

CHANGES.txt

@@ -156,6 +156,10 @@ Trunk (unreleased changes)
     HADOOP-3992. Add a synthetic load generation facility to the test
     directory. (hairong via szetszwo)
 
+    HADOOP-3981. Implement a distributed file checksum algorithm in HDFS
+    and change DistCp to use file checksum for comparing src and dst files
+    (szetszwo)
+
   IMPROVEMENTS
 
     HADOOP-3908. Fuse-dfs: better error message if llibhdfs.so doesn't exist.

src/core/org/apache/hadoop/fs/FileSystem.java

@@ -1289,10 +1289,12 @@ public abstract class FileSystem extends Configured implements Closeable {
    * Get the checksum of a file.
    *
    * @param f The file path
-   * @return The checksum 
+   * @return The file checksum.  The default return value is null,
+   *  which indicates that no checksum algorithm is implemented
+   *  in the corresponding FileSystem.
    */
   public FileChecksum getFileChecksum(Path f) throws IOException {
-    return new LengthFileChecksum(getFileStatus(f).getLen());
+    return null;
   }
 
   /**

src/core/org/apache/hadoop/fs/LengthFileChecksum.java

@@ -1,66 +0,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.
- */
-package org.apache.hadoop.fs;
-
-import java.io.DataInput;
-import java.io.DataOutput;
-import java.io.IOException;
-
-/** An abstract class representing file checksums for files. */
-public class LengthFileChecksum extends FileChecksum {
-  public static final String ALGORITHM_NAME = "FILE-LENGTH";
-  
-  private long value;
-
-  /** Constructor */
-  public LengthFileChecksum() {}
-
-  public LengthFileChecksum(long value) {
-    this.value = value;
-  }
-
-  /** {@inheritDoc} */ 
-  public String getAlgorithmName() {return ALGORITHM_NAME;}
-
-  /** {@inheritDoc} */ 
-  public int getLength() {return Long.SIZE/Byte.SIZE;}
-
-  /** {@inheritDoc} */ 
-  public byte[] getBytes() {
-    final byte[] b = new byte[getLength()];
-    for(int i = 0; i < b.length; i++) {
-      b[i] = (byte)(value >>> (i*Byte.SIZE));
-    }
-    return b;
-  }
-
-  /** {@inheritDoc} */ 
-  public void readFields(DataInput in) throws IOException {
-    value = in.readLong();
-  }
-
-  /** {@inheritDoc} */ 
-  public void write(DataOutput out) throws IOException {
-    out.writeLong(value);    
-  }
-
-  /** {@inheritDoc} */ 
-  public String toString() {
-    return getClass().getSimpleName() + ": " + value;
-  }
-}

src/core/org/apache/hadoop/fs/MD5MD5CRC32FileChecksum.java

@@ -0,0 +1,79 @@
+/**
+ * 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.fs;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+import org.apache.hadoop.io.MD5Hash;
+import org.apache.hadoop.io.WritableUtils;
+
+/** MD5 of MD5 of CRC32. */
+public class MD5MD5CRC32FileChecksum extends FileChecksum {
+  public static final int LENGTH = MD5Hash.MD5_LEN
+      + (Integer.SIZE + Long.SIZE)/Byte.SIZE;
+
+  private int bytesPerCRC;
+  private long crcPerBlock;
+  private MD5Hash md5;
+
+  /** Same as this(0, 0, null) */
+  public MD5MD5CRC32FileChecksum() {
+    this(0, 0, null);
+  }
+
+  /** Create a MD5FileChecksum */
+  public MD5MD5CRC32FileChecksum(int bytesPerCRC, long crcPerBlock, MD5Hash md5) {
+    this.bytesPerCRC = bytesPerCRC;
+    this.crcPerBlock = crcPerBlock;
+    this.md5 = md5;
+  }
+  
+  /** {@inheritDoc} */ 
+  public String getAlgorithmName() {
+    return "MD5-of-" + crcPerBlock + "MD5-of-" + bytesPerCRC + "CRC32";
+  }
+
+  /** {@inheritDoc} */ 
+  public int getLength() {return LENGTH;}
+
+  /** {@inheritDoc} */ 
+  public byte[] getBytes() {
+    return WritableUtils.toByteArray(this);
+  }
+
+  /** {@inheritDoc} */ 
+  public void readFields(DataInput in) throws IOException {
+    bytesPerCRC = in.readInt();
+    crcPerBlock = in.readLong();
+    md5 = MD5Hash.read(in);
+  }
+
+  /** {@inheritDoc} */ 
+  public void write(DataOutput out) throws IOException {
+    out.writeInt(bytesPerCRC);
+    out.writeLong(crcPerBlock);
+    md5.write(out);    
+  }
+
+  /** {@inheritDoc} */ 
+  public String toString() {
+    return getAlgorithmName() + ":" + md5;
+  }
+}

src/core/org/apache/hadoop/io/MD5Hash.java

@@ -21,13 +21,15 @@ package org.apache.hadoop.io;
 import java.io.IOException;
 import java.io.DataInput;
 import java.io.DataOutput;
+import java.io.InputStream;
 import java.util.Arrays;
 import java.security.*;
 
 /** A Writable for MD5 hash values.
  */
-public class MD5Hash implements WritableComparable {
+public class MD5Hash implements WritableComparable<MD5Hash> {
   public static final int MD5_LEN = 16;
+
   private static ThreadLocal<MessageDigest> DIGESTER_FACTORY = new ThreadLocal<MessageDigest>() {
     protected MessageDigest initialValue() {
       try {
@@ -87,6 +89,18 @@ public class MD5Hash implements WritableComparable {
     return digest(data, 0, data.length);
   }
 
+  /** Construct a hash value for the content from the InputStream. */
+  public static MD5Hash digest(InputStream in) throws IOException {
+    final byte[] buffer = new byte[4*1024]; 
+
+    final MessageDigest digester = DIGESTER_FACTORY.get();
+    for(int n; (n = in.read(buffer)) != -1; ) {
+      digester.update(buffer, 0, n);
+    }
+
+    return new MD5Hash(digester.digest());
+  }
+
   /** Construct a hash value for a byte array. */
   public static MD5Hash digest(byte[] data, int start, int len) {
     byte[] digest;
@@ -143,8 +157,7 @@ public class MD5Hash implements WritableComparable {
 
 
   /** Compares this object with the specified object for order.*/
-  public int compareTo(Object o) {
-    MD5Hash that = (MD5Hash)o;
+  public int compareTo(MD5Hash that) {
     return WritableComparator.compareBytes(this.digest, 0, MD5_LEN,
                                            that.digest, 0, MD5_LEN);
   }

src/core/org/apache/hadoop/io/WritableUtils.java

@@ -428,4 +428,18 @@ public final class WritableUtils  {
                             "due to end of input.");
     }
   }
+
+  /** Convert writables to a byte array */
+  public static byte[] toByteArray(Writable... writables) {
+    final DataOutputBuffer out = new DataOutputBuffer();
+    try {
+      for(Writable w : writables) {
+        w.write(out);
+      }
+      out.close();
+    } catch (IOException e) {
+      throw new RuntimeException("Fail to convert writables to a byte array",e);
+    }
+    return out.getData();
+  }
 }

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

@@ -579,6 +579,109 @@ public class DFSClient implements FSConstants, java.io.Closeable {
     }
   }
 
+  /**
+   * Get the checksum of a file.
+   * @param src The file path
+   * @return The checksum 
+   */
+  MD5MD5CRC32FileChecksum getFileChecksum(String src) throws IOException {
+    checkOpen();
+    
+    //get all block locations
+    final List<LocatedBlock> locatedblocks
+        = callGetBlockLocations(src, 0, Long.MAX_VALUE).getLocatedBlocks();
+    final DataOutputBuffer md5out = new DataOutputBuffer();
+    int bytesPerCRC = 0;
+    long crcPerBlock = 0;
+
+    //get block checksum for each block
+    for(int i = 0; i < locatedblocks.size(); i++) {
+      LocatedBlock lb = locatedblocks.get(i);
+      final Block block = lb.getBlock();
+      final DatanodeInfo[] datanodes = lb.getLocations();
+      
+      //try each datanode location of the block
+      final int timeout = 3000 * datanodes.length + socketTimeout;
+      boolean done = false;
+      for(int j = 0; !done && j < datanodes.length; j++) {
+        //connect to a datanode
+        final Socket sock = socketFactory.createSocket();
+        sock.connect(NetUtils.createSocketAddr(datanodes[j].getName()), timeout);
+        sock.setSoTimeout(timeout);
+
+        DataOutputStream out = new DataOutputStream(
+            new BufferedOutputStream(NetUtils.getOutputStream(sock), 
+                                     DataNode.SMALL_BUFFER_SIZE));
+        DataInputStream in = new DataInputStream(NetUtils.getInputStream(sock));
+
+        // get block MD5
+        try {
+          if (LOG.isDebugEnabled()) {
+            LOG.debug("write to " + datanodes[j].getName() + ": "
+                + OP_BLOCK_CHECKSUM + ", block=" + block);
+          }
+          out.writeShort(DATA_TRANSFER_VERSION);
+          out.write(OP_BLOCK_CHECKSUM);
+          out.writeLong(block.getBlockId());
+          out.writeLong(block.getGenerationStamp());
+          out.flush();
+         
+          final short reply = in.readShort();
+          if (reply != OP_STATUS_SUCCESS) {
+            throw new IOException("Bad response " + reply + " for block "
+                + block + " from datanode " + datanodes[j].getName());
+          }
+
+          //read byte-per-checksum
+          final int bpc = in.readInt(); 
+          if (i == 0) { //first block
+            bytesPerCRC = bpc;
+          }
+          else if (bpc != bytesPerCRC) {
+            throw new IOException("Byte-per-checksum not matched: bpc=" + bpc
+                + " but bytesPerCRC=" + bytesPerCRC);
+          }
+          
+          //read crc-per-block
+          final long cpb = in.readLong();
+          if (locatedblocks.size() > 1 && i == 0) {
+            crcPerBlock = cpb;
+          }
+
+          //read md5
+          final MD5Hash md5 = MD5Hash.read(in);
+          md5.write(md5out);
+          
+          done = true;
+
+          if (LOG.isDebugEnabled()) {
+            if (i == 0) {
+              LOG.debug("set bytesPerCRC=" + bytesPerCRC
+                  + ", crcPerBlock=" + crcPerBlock);
+            }
+            LOG.debug("got reply from " + datanodes[j].getName()
+                + ": md5=" + md5);
+          }
+        } catch (IOException ie) {
+          LOG.warn("src=" + src + ", datanodes[" + j + "].getName()="
+              + datanodes[j].getName(), ie);
+        } finally {
+          IOUtils.closeStream(in);
+          IOUtils.closeStream(out);
+          IOUtils.closeSocket(sock);        
+        }
+      }
+
+      if (!done) {
+        throw new IOException("Fail to get block MD5 for " + block);
+      }
+    }
+
+    //compute file MD5
+    final MD5Hash fileMD5 = MD5Hash.digest(md5out.getData()); 
+    return new MD5MD5CRC32FileChecksum(bytesPerCRC, crcPerBlock, fileMD5);
+  }
+
   /**
    * Set permissions to a file or directory.
    * @param src path name.

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

@@ -406,6 +406,11 @@ public class DistributedFileSystem extends FileSystem {
     }
   }
 
+  /** {@inheritDoc} */
+  public MD5MD5CRC32FileChecksum getFileChecksum(Path f) throws IOException {
+    return dfs.getFileChecksum(getPathName(f));
+  }
+
   /** {@inheritDoc }*/
   public void setPermission(Path p, FsPermission permission
       ) throws IOException {

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

@@ -87,6 +87,7 @@ public interface FSConstants {
   public static final byte OP_READ_METADATA = (byte) 82;
   public static final byte OP_REPLACE_BLOCK = (byte) 83;
   public static final byte OP_COPY_BLOCK = (byte) 84;
+  public static final byte OP_BLOCK_CHECKSUM = (byte) 85;
   
   public static final int OP_STATUS_SUCCESS = 0;  
   public static final int OP_STATUS_ERROR = 1;  

src/hdfs/org/apache/hadoop/hdfs/server/datanode/DataXceiver.java

@@ -33,8 +33,10 @@ import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
 import org.apache.hadoop.hdfs.protocol.FSConstants;
 import org.apache.hadoop.hdfs.server.datanode.FSDatasetInterface.MetaDataInputStream;
 import org.apache.hadoop.io.IOUtils;
+import org.apache.hadoop.io.MD5Hash;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.util.DataChecksum;
 import org.apache.hadoop.util.StringUtils;
 import static org.apache.hadoop.hdfs.server.datanode.DataNode.DN_CLIENTTRACE_FORMAT;
 
@@ -114,6 +116,10 @@ class DataXceiver implements Runnable, FSConstants {
         copyBlock(in);
         datanode.myMetrics.copyBlockOp.inc(DataNode.now() - startTime);
         break;
+      case OP_BLOCK_CHECKSUM: //get the checksum of a block
+        getBlockChecksum(in);
+        datanode.myMetrics.blockChecksumOp.inc(DataNode.now() - startTime);
+        break;
       default:
         throw new IOException("Unknown opcode " + op + " in data stream");
       }
@@ -413,6 +419,49 @@ class DataXceiver implements Runnable, FSConstants {
     }
   }
   
+  /**
+   * Get block checksum (MD5 of CRC32).
+   * @param in
+   */
+  void getBlockChecksum(DataInputStream in) throws IOException {
+    final Block block = new Block(in.readLong(), 0 , in.readLong());
+
+    DataOutputStream out = null;
+    final MetaDataInputStream metadataIn = datanode.data.getMetaDataInputStream(block);
+    final DataInputStream checksumIn = new DataInputStream(new BufferedInputStream(
+        metadataIn, BUFFER_SIZE));
+
+    try {
+      //read metadata file
+      final BlockMetadataHeader header = BlockMetadataHeader.readHeader(checksumIn);
+      final DataChecksum checksum = header.getChecksum(); 
+      final int bytesPerCRC = checksum.getBytesPerChecksum();
+      final long crcPerBlock = (metadataIn.getLength()
+          - BlockMetadataHeader.getHeaderSize())/checksum.getChecksumSize();
+      
+      //compute block checksum
+      final MD5Hash md5 = MD5Hash.digest(checksumIn);
+
+      if (LOG.isDebugEnabled()) {
+        LOG.debug("block=" + block + ", bytesPerCRC=" + bytesPerCRC
+            + ", crcPerBlock=" + crcPerBlock + ", md5=" + md5);
+      }
+
+      //write reply
+      out = new DataOutputStream(
+          NetUtils.getOutputStream(s, datanode.socketWriteTimeout));
+      out.writeShort(OP_STATUS_SUCCESS);
+      out.writeInt(bytesPerCRC);
+      out.writeLong(crcPerBlock);
+      md5.write(out);
+      out.flush();
+    } finally {
+      IOUtils.closeStream(out);
+      IOUtils.closeStream(checksumIn);
+      IOUtils.closeStream(metadataIn);
+    }
+  }
+
   /**
    * Read a block from the disk and then sends it to a destination.
    * 

src/hdfs/org/apache/hadoop/hdfs/server/datanode/metrics/DataNodeMetrics.java

@@ -76,6 +76,8 @@ public class DataNodeMetrics implements Updater {
                 new MetricsTimeVaryingRate("writeBlockOp");
   public MetricsTimeVaryingRate readMetadataOp = 
                 new MetricsTimeVaryingRate("readMetadataOp");
+  public MetricsTimeVaryingRate blockChecksumOp = 
+                new MetricsTimeVaryingRate("blockChecksumOp");
   public MetricsTimeVaryingRate copyBlockOp = 
                 new MetricsTimeVaryingRate("copyBlockOp");
   public MetricsTimeVaryingRate replaceBlockOp = 
@@ -130,6 +132,7 @@ public class DataNodeMetrics implements Updater {
       readBlockOp.pushMetric(metricsRecord);
       writeBlockOp.pushMetric(metricsRecord);
       readMetadataOp.pushMetric(metricsRecord);
+      blockChecksumOp.pushMetric(metricsRecord);
       copyBlockOp.pushMetric(metricsRecord);
       replaceBlockOp.pushMetric(metricsRecord);
       heartbeats.pushMetric(metricsRecord);
@@ -141,6 +144,7 @@ public class DataNodeMetrics implements Updater {
     readBlockOp.resetMinMax();
     writeBlockOp.resetMinMax();
     readMetadataOp.resetMinMax();
+    blockChecksumOp.resetMinMax();
     copyBlockOp.resetMinMax();
     replaceBlockOp.resetMinMax();
     heartbeats.resetMinMax();

src/hdfs/org/apache/hadoop/hdfs/server/datanode/metrics/DataNodeStatistics.java

@@ -194,6 +194,34 @@ public class DataNodeStatistics implements DataNodeStatisticsMBean {
     return myMetrics.readMetadataOp.getPreviousIntervalNumOps();
   }
 
+  /**
+   * @inheritDoc
+   */
+  public long getBlockChecksumOpAverageTime() {
+    return myMetrics.blockChecksumOp.getPreviousIntervalAverageTime();
+  }
+
+  /**
+   * @inheritDoc
+   */
+  public long getBlockChecksumOpMaxTime() {
+    return myMetrics.blockChecksumOp.getMaxTime();
+  }
+
+  /**
+   * @inheritDoc
+   */
+  public long getBlockChecksumOpMinTime() {
+    return myMetrics.blockChecksumOp.getMinTime();
+  }
+
+  /**
+   * @inheritDoc
+   */
+  public int getBlockChecksumOpNum() {
+    return myMetrics.blockChecksumOp.getPreviousIntervalNumOps();
+  }
+
   /**
    * @inheritDoc
    */

src/hdfs/org/apache/hadoop/hdfs/server/datanode/metrics/DataNodeStatisticsMBean.java

@@ -185,6 +185,29 @@ public interface DataNodeStatisticsMBean {
    */
   long getReadMetadataOpMaxTime();
   
+  /**
+   * Number of block BlockChecksum in last interval
+   * @return number of operations
+   */
+  int getBlockChecksumOpNum(); 
+
+  /**
+   * Average time for BlockChecksum Operation in last interval
+   * @return time in msec
+   */
+  long getBlockChecksumOpAverageTime();
+  
+  /**
+   *   The Minimum BlockChecksum Operation Time since reset was called
+   * @return time in msec
+   */
+  long getBlockChecksumOpMinTime();
+  
+  /**
+   *   The Maximum BlockChecksum Operation Time since reset was called
+   * @return time in msec
+   */
+  long getBlockChecksumOpMaxTime();
   
   /**
    * Number of CopyBlock Operation in last interval

src/test/org/apache/hadoop/fs/TestFileSystem.java

@@ -58,7 +58,6 @@ import org.apache.hadoop.security.UnixUserGroupInformation;
 
 public class TestFileSystem extends TestCase {
   private static final Log LOG = FileSystem.LOG;
-  private static final Random RAN = new Random();
 
   private static Configuration conf = new Configuration();
   private static int BUFFER_SIZE = conf.getInt("io.file.buffer.size", 4096);
@@ -619,38 +618,4 @@ public class TestFileSystem extends TestCase {
     assertTrue(map.containsKey(lowercaseCachekey2));    
 
   }
-  
-  public void testFileChecksum() throws IOException {
-    final long seed = RAN.nextLong();
-    System.out.println("seed=" + seed);
-    RAN.setSeed(seed);
-
-    final Configuration conf = new Configuration();
-    final String dir = ROOT + "/fileChecksum";
-    final LocalFileSystem fs = FileSystem.getLocal(conf);
-    final Path foo = new Path(dir, "foo");
-
-    //generate random data
-    final byte[] data = new byte[RAN.nextInt(3*1024) + 10*1024];
-    RAN.nextBytes(data);
-
-    //write data to a file
-    final FSDataOutputStream out = fs.create(foo);
-    out.write(data);
-    out.close();
-    
-    //compute checksum
-    final FileChecksum cs1 = fs.getFileChecksum(foo);
-    System.out.println("cs1=" + cs1);
-    
-    //rename the file and verify again
-    final Path bar = new Path(dir, "bar");
-    fs.rename(foo, bar);
-
-    { //verify checksum
-      final FileChecksum cs2 = fs.getFileChecksum(bar);
-      assertEquals(cs1.hashCode(), cs2.hashCode());
-      assertEquals(cs1, cs2);
-    }
-  }
 }

src/test/org/apache/hadoop/hdfs/TestDistributedFileSystem.java

@@ -18,15 +18,20 @@
 
 package org.apache.hadoop.hdfs;
 
+import java.io.IOException;
 import java.net.URI;
+import java.util.Random;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FSDataInputStream;
 import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileChecksum;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 
 public class TestDistributedFileSystem extends junit.framework.TestCase {
+  private static final Random RAN = new Random();
+
   public void testFileSystemCloseAll() throws Exception {
     Configuration conf = new Configuration();
     MiniDFSCluster cluster = new MiniDFSCluster(conf, 0, true, null);
@@ -110,4 +115,55 @@ public class TestDistributedFileSystem extends junit.framework.TestCase {
       if (cluster != null) {cluster.shutdown();}
     }
   }
+  
+  public void testFileChecksum() throws IOException {
+    final long seed = RAN.nextLong();
+    System.out.println("seed=" + seed);
+    RAN.setSeed(seed);
+
+    final Configuration conf = new Configuration();
+    final MiniDFSCluster cluster = new MiniDFSCluster(conf, 2, true, null);
+    final FileSystem fs = cluster.getFileSystem();
+
+    final String dir = "/fileChecksum";
+    final int block_size = 1024;
+    final int buffer_size = conf.getInt("io.file.buffer.size", 4096);
+    conf.setInt("io.bytes.per.checksum", 512);
+
+    //try different number of blocks
+    for(int n = 0; n < 5; n++) {
+      //generate random data
+      final byte[] data = new byte[RAN.nextInt(block_size/2-1)+n*block_size+1];
+      RAN.nextBytes(data);
+      System.out.println("data.length=" + data.length);
+  
+      //write data to a file
+      final Path foo = new Path(dir, "foo" + n);
+      {
+        final FSDataOutputStream out = fs.create(foo, false, buffer_size,
+            (short)2, block_size);
+        out.write(data);
+        out.close();
+      }
+      
+      //compute checksum
+      final FileChecksum foocs = fs.getFileChecksum(foo);
+      System.out.println("foocs=" + foocs);
+      
+      //write another file
+      final Path bar = new Path(dir, "bar" + n);
+      {
+        final FSDataOutputStream out = fs.create(bar, false, buffer_size,
+            (short)2, block_size);
+        out.write(data);
+        out.close();
+      }
+  
+      { //verify checksum
+        final FileChecksum barcs = fs.getFileChecksum(bar);
+        assertEquals(foocs.hashCode(), barcs.hashCode());
+        assertEquals(foocs, barcs);
+      }
+    }
+  }
 }

src/tools/org/apache/hadoop/tools/DistCp.java

@@ -336,9 +336,10 @@ public class DistCp implements Tool {
      * be meaningful in this context.
      * @throws IOException 
      */
-    private boolean needsUpdate(FileSystem srcfs, Path srcpath,
+    private boolean needsUpdate(FileStatus srcstatus,
         FileSystem dstfs, Path dstpath) throws IOException {
-      return update && !sameFile(srcfs, srcpath, dstfs, dstpath);
+      return update && !sameFile(srcstatus.getPath().getFileSystem(job),
+          srcstatus, dstfs, dstpath);
     }
     
     private FSDataOutputStream create(Path f, Reporter reporter,
@@ -387,10 +388,8 @@ public class DistCp implements Tool {
         return;
       }
 
-      final Path srcpath = srcstat.getPath();
-      final FileSystem srcfs = srcpath.getFileSystem(job);
       if (destFileSys.exists(absdst) && !overwrite
-          && !needsUpdate(srcfs, srcpath, destFileSys, absdst)) {
+          && !needsUpdate(srcstat, destFileSys, absdst)) {
         outc.collect(null, new Text("SKIP: " + srcstat.getPath()));
         ++skipcount;
         reporter.incrCounter(Counter.SKIP, 1);
@@ -1050,8 +1049,7 @@ public class DistCp implements Tool {
             }
             else {
               //skip file if the src and the dst files are the same.
-              final Path absdst = new Path(args.dst, dst);
-              skipfile = update && sameFile(srcfs,child.getPath(),dstfs,absdst);
+              skipfile = update && sameFile(srcfs, child, dstfs, new Path(args.dst, dst));
               
               if (!skipfile) {
                 ++fileCount;
@@ -1134,15 +1132,38 @@ public class DistCp implements Tool {
   }
 
   /**
-   * Check whether the src and the dst are the same.
-   * Two files are considered as the same if they have the same size.
+   * Check whether the contents of src and dst are the same.
+   * 
+   * Return false if dstpath does not exist
+   * 
+   * If the files have different sizes, return false.
+   * 
+   * If the files have the same sizes, the file checksums will be compared.
+   * 
+   * When file checksum is not supported in any of file systems,
+   * two files are considered as the same if they have the same size.
    */
-  static private boolean sameFile(FileSystem srcfs, Path srcpath,
+  static private boolean sameFile(FileSystem srcfs, FileStatus srcstatus,
       FileSystem dstfs, Path dstpath) throws IOException {
+    FileStatus dststatus;
     try {
-      final FileChecksum srccs = srcfs.getFileChecksum(srcpath);
-      final FileChecksum dstcs = dstfs.getFileChecksum(dstpath);
-      return srccs != null && srccs.equals(dstcs);
+      dststatus = dstfs.getFileStatus(dstpath);
+    } catch(FileNotFoundException fnfe) {
+      return false;
+    }
+
+    //same length?
+    if (srcstatus.getLen() != dststatus.getLen()) {
+      return false;
+    }
+
+    //compare checksums
+    try {
+      final FileChecksum srccs = srcfs.getFileChecksum(srcstatus.getPath());
+      final FileChecksum dstcs = dstfs.getFileChecksum(dststatus.getPath());
+      //return true if checksum is not supported
+      //(i.e. some of the checksums is null)
+      return srccs == null || dstcs == null || srccs.equals(dstcs);
     } catch(FileNotFoundException fnfe) {
       return false;
     }