/**
 * Write last checkpoint time and version file into the storage directory.
 * 
 * The version file should always be written last.
 * Missing or corrupted version file indicates that 
 * the checkpoint is not valid.
 * 
 * @param sd storage directory
 * @throws IOException
 */
protected void setFields(Properties props, 
                         StorageDirectory sd 
                         ) throws IOException {
  super.setFields(props, sd);
  boolean uState = getDistributedUpgradeState();
  int uVersion = getDistributedUpgradeVersion();
  if(uState && uVersion != getLayoutVersion()) {
    props.setProperty("distributedUpgradeState", Boolean.toString(uState));
    props.setProperty("distributedUpgradeVersion", Integer.toString(uVersion)); 
  }
  if (this.newImageDigest) {
    this.setImageDigest(MD5Hash.digest(
        new FileInputStream(getImageFile(sd, NameNodeFile.IMAGE))));
  }
  props.setProperty(MESSAGE_DIGEST_PROPERTY, 
      this.getImageDigest().toString());
  writeCheckpointTime(sd);
}
 

public static MD5Hash downloadImageToStorage(URL fsName, long imageTxId,
    Storage dstStorage, boolean needDigest) throws IOException {
  String fileid = ImageServlet.getParamStringForImage(null,
      imageTxId, dstStorage);
  String fileName = NNStorage.getCheckpointImageFileName(imageTxId);
  
  List<File> dstFiles = dstStorage.getFiles(
      NameNodeDirType.IMAGE, fileName);
  if (dstFiles.isEmpty()) {
    throw new IOException("No targets in destination storage!");
  }
  
  MD5Hash hash = getFileClient(fsName, fileid, dstFiles, dstStorage, needDigest);
  LOG.info("Downloaded file " + dstFiles.get(0).getName() + " size " +
      dstFiles.get(0).length() + " bytes.");
  return hash;
}
 

static MD5Hash handleUploadImageRequest(HttpServletRequest request,
    long imageTxId, Storage dstStorage, InputStream stream,
    long advertisedSize, DataTransferThrottler throttler) throws IOException {

  String fileName = NNStorage.getCheckpointImageFileName(imageTxId);

  List<File> dstFiles = dstStorage.getFiles(NameNodeDirType.IMAGE, fileName);
  if (dstFiles.isEmpty()) {
    throw new IOException("No targets in destination storage!");
  }

  MD5Hash advertisedDigest = parseMD5Header(request);
  MD5Hash hash = receiveFile(fileName, dstFiles, dstStorage, true,
      advertisedSize, advertisedDigest, fileName, stream, throttler);
  LOG.info("Downloaded file " + dstFiles.get(0).getName() + " size "
      + dstFiles.get(0).length() + " bytes.");
  return hash;
}
 

/**
 * Load in the filesystem image from file. It's a big list of
 * filenames and blocks.
 */
private void loadFSImage(File curFile, MD5Hash expectedMd5,
    FSNamesystem target, MetaRecoveryContext recovery,
    boolean requireSameLayoutVersion) throws IOException {
  // BlockPoolId is required when the FsImageLoader loads the rolling upgrade
  // information. Make sure the ID is properly set.
  target.setBlockPoolId(this.getBlockPoolID());

  FSImageFormat.LoaderDelegator loader = FSImageFormat.newLoader(conf, target);
  loader.load(curFile, requireSameLayoutVersion);

  // Check that the image digest we loaded matches up with what
  // we expected
  MD5Hash readImageMd5 = loader.getLoadedImageMd5();
  if (expectedMd5 != null &&
      !expectedMd5.equals(readImageMd5)) {
    throw new IOException("Image file " + curFile +
        " is corrupt with MD5 checksum of " + readImageMd5 +
        " but expecting " + expectedMd5);
  }

  long txId = loader.getLoadedImageTxId();
  LOG.info("Loaded image for txid " + txId + " from " + curFile);
  lastAppliedTxId = txId;
  storage.setMostRecentCheckpointInfo(txId, curFile.lastModified());
}
 

/**
 * Read the md5 checksum stored alongside the given data file.
 * @param dataFile the file containing data
 * @return the checksum stored in dataFile.md5
 */
public static MD5Hash readStoredMd5ForFile(File dataFile) throws IOException {
  final File md5File = getDigestFileForFile(dataFile);
  if (!md5File.exists()) {
    return null;
  }

  final Matcher matcher = readStoredMd5(md5File);
  String storedHash = matcher.group(1);
  File referencedFile = new File(matcher.group(2));

  // Sanity check: Make sure that the file referenced in the .md5 file at
  // least has the same name as the file we expect
  if (!referencedFile.getName().equals(dataFile.getName())) {
    throw new IOException(
        "MD5 file at " + md5File + " references file named " +
        referencedFile.getName() + " but we expected it to reference " +
        dataFile);
  }
  return new MD5Hash(storedHash);
}
 

/**
 * Corrupts the MD5 sum of the fsimage.
 * 
 * @param corruptAll
 *          whether to corrupt one or all of the MD5 sums in the configured
 *          namedirs
 * @throws IOException
 */
private void corruptFSImageMD5(boolean corruptAll) throws IOException {
  List<URI> nameDirs = (List<URI>)FSNamesystem.getNamespaceDirs(config);
  // Corrupt the md5 files in all the namedirs
  for (URI uri: nameDirs) {
    // Directory layout looks like:
    // test/data/dfs/nameN/current/{fsimage,edits,...}
    File nameDir = new File(uri.getPath());
    File dfsDir = nameDir.getParentFile();
    assertEquals(dfsDir.getName(), "dfs"); // make sure we got right dir
    // Set the md5 file to all zeros
    File imageFile = new File(nameDir,
        Storage.STORAGE_DIR_CURRENT + "/"
        + NNStorage.getImageFileName(0));
    MD5FileUtils.saveMD5File(imageFile, new MD5Hash(new byte[16]));
    // Only need to corrupt one if !corruptAll
    if (!corruptAll) {
      break;
    }
  }
}
 

/** Return the object represented in the attributes. */
public static MD5MD5CRC32FileChecksum valueOf(Attributes attrs
    ) throws SAXException {
  final String bytesPerCRC = attrs.getValue("bytesPerCRC");
  final String crcPerBlock = attrs.getValue("crcPerBlock");
  final String md5 = attrs.getValue("md5");
  if (bytesPerCRC == null || crcPerBlock == null || md5 == null) {
    return null;
  }

  try {
    return new MD5MD5CRC32FileChecksum(Integer.valueOf(bytesPerCRC),
        Integer.valueOf(crcPerBlock), new MD5Hash(md5));
  } catch(Exception e) {
    throw new SAXException("Invalid attributes: bytesPerCRC=" + bytesPerCRC
        + ", crcPerBlock=" + crcPerBlock + ", md5=" + md5, e);
  }
}
 

public static MD5Hash downloadImageToStorage(URL fsName, long imageTxId,
    Storage dstStorage, boolean needDigest) throws IOException {
  String fileid = ImageServlet.getParamStringForImage(null,
      imageTxId, dstStorage);
  String fileName = NNStorage.getCheckpointImageFileName(imageTxId);
  
  List<File> dstFiles = dstStorage.getFiles(
      NameNodeDirType.IMAGE, fileName);
  if (dstFiles.isEmpty()) {
    throw new IOException("No targets in destination storage!");
  }
  
  MD5Hash hash = getFileClient(fsName, fileid, dstFiles, dstStorage, needDigest);
  LOG.info("Downloaded file " + dstFiles.get(0).getName() + " size " +
      dstFiles.get(0).length() + " bytes.");
  return hash;
}
 

static MD5Hash handleUploadImageRequest(HttpServletRequest request,
    long imageTxId, Storage dstStorage, InputStream stream,
    long advertisedSize, DataTransferThrottler throttler) throws IOException {

  String fileName = NNStorage.getCheckpointImageFileName(imageTxId);

  List<File> dstFiles = dstStorage.getFiles(NameNodeDirType.IMAGE, fileName);
  if (dstFiles.isEmpty()) {
    throw new IOException("No targets in destination storage!");
  }

  MD5Hash advertisedDigest = parseMD5Header(request);
  MD5Hash hash = receiveFile(fileName, dstFiles, dstStorage, true,
      advertisedSize, advertisedDigest, fileName, stream, throttler);
  LOG.info("Downloaded file " + dstFiles.get(0).getName() + " size "
      + dstFiles.get(0).length() + " bytes.");
  return hash;
}
 

/**
 * Load in the filesystem image from file. It's a big list of
 * filenames and blocks.
 */
private void loadFSImage(File curFile, MD5Hash expectedMd5,
    FSNamesystem target, MetaRecoveryContext recovery,
    boolean requireSameLayoutVersion) throws IOException {
  // BlockPoolId is required when the FsImageLoader loads the rolling upgrade
  // information. Make sure the ID is properly set.
  target.setBlockPoolId(this.getBlockPoolID());

  FSImageFormat.LoaderDelegator loader = FSImageFormat.newLoader(conf, target);
  loader.load(curFile, requireSameLayoutVersion);

  // Check that the image digest we loaded matches up with what
  // we expected
  MD5Hash readImageMd5 = loader.getLoadedImageMd5();
  if (expectedMd5 != null &&
      !expectedMd5.equals(readImageMd5)) {
    throw new IOException("Image file " + curFile +
        " is corrupt with MD5 checksum of " + readImageMd5 +
        " but expecting " + expectedMd5);
  }

  long txId = loader.getLoadedImageTxId();
  LOG.info("Loaded image for txid " + txId + " from " + curFile);
  lastAppliedTxId = txId;
  storage.setMostRecentCheckpointInfo(txId, curFile.lastModified());
}
 

/**
 * Corrupts the MD5 sum of the fsimage.
 * 
 * @param corruptAll
 *          whether to corrupt one or all of the MD5 sums in the configured
 *          namedirs
 * @throws IOException
 */
private void corruptFSImageMD5(boolean corruptAll) throws IOException {
  List<URI> nameDirs = (List<URI>)FSNamesystem.getNamespaceDirs(config);
  // Corrupt the md5 files in all the namedirs
  for (URI uri: nameDirs) {
    // Directory layout looks like:
    // test/data/dfs/nameN/current/{fsimage,edits,...}
    File nameDir = new File(uri.getPath());
    File dfsDir = nameDir.getParentFile();
    assertEquals(dfsDir.getName(), "dfs"); // make sure we got right dir
    // Set the md5 file to all zeros
    File imageFile = new File(nameDir,
        Storage.STORAGE_DIR_CURRENT + "/"
        + NNStorage.getImageFileName(0));
    MD5FileUtils.saveMD5File(imageFile, new MD5Hash(new byte[16]));
    // Only need to corrupt one if !corruptAll
    if (!corruptAll) {
      break;
    }
  }
}
 

/** Return the object represented in the attributes. */
public static MD5MD5CRC32FileChecksum valueOf(Attributes attrs
    ) throws SAXException {
  final String bytesPerCRC = attrs.getValue("bytesPerCRC");
  final String crcPerBlock = attrs.getValue("crcPerBlock");
  final String md5 = attrs.getValue("md5");
  if (bytesPerCRC == null || crcPerBlock == null || md5 == null) {
    return null;
  }

  try {
    return new MD5MD5CRC32FileChecksum(Integer.valueOf(bytesPerCRC),
        Integer.valueOf(crcPerBlock), new MD5Hash(md5));
  } catch(Exception e) {
    throw new SAXException("Invalid attributes: bytesPerCRC=" + bytesPerCRC
        + ", crcPerBlock=" + crcPerBlock + ", md5=" + md5, e);
  }
}
 

void loadFSImageFile(FSNamesystem target, MetaRecoveryContext recovery,
    FSImageFile imageFile, StartupOption startupOption) throws IOException {
  LOG.debug("Planning to load image :\n" + imageFile);
  StorageDirectory sdForProperties = imageFile.sd;
  storage.readProperties(sdForProperties, startupOption);

  if (NameNodeLayoutVersion.supports(
      LayoutVersion.Feature.TXID_BASED_LAYOUT, getLayoutVersion())) {
    // For txid-based layout, we should have a .md5 file
    // next to the image file
    boolean isRollingRollback = RollingUpgradeStartupOption.ROLLBACK
        .matches(startupOption);
    loadFSImage(imageFile.getFile(), target, recovery, isRollingRollback);
  } else if (NameNodeLayoutVersion.supports(
      LayoutVersion.Feature.FSIMAGE_CHECKSUM, getLayoutVersion())) {
    // In 0.22, we have the checksum stored in the VERSION file.
    String md5 = storage.getDeprecatedProperty(
        NNStorage.DEPRECATED_MESSAGE_DIGEST_PROPERTY);
    if (md5 == null) {
      throw new InconsistentFSStateException(sdForProperties.getRoot(),
          "Message digest property " +
          NNStorage.DEPRECATED_MESSAGE_DIGEST_PROPERTY +
          " not set for storage directory " + sdForProperties.getRoot());
    }
    loadFSImage(imageFile.getFile(), new MD5Hash(md5), target, recovery,
        false);
  } else {
    // We don't have any record of the md5sum
    loadFSImage(imageFile.getFile(), null, target, recovery, false);
  }
}
 

/**
 * Load the image namespace from the given image file, verifying
 * it against the MD5 sum stored in its associated .md5 file.
 */
private void loadFSImage(File imageFile, FSNamesystem target,
    MetaRecoveryContext recovery, boolean requireSameLayoutVersion)
    throws IOException {
  MD5Hash expectedMD5 = MD5FileUtils.readStoredMd5ForFile(imageFile);
  if (expectedMD5 == null) {
    throw new IOException("No MD5 file found corresponding to image file "
        + imageFile);
  }
  loadFSImage(imageFile, expectedMD5, target, recovery,
      requireSameLayoutVersion);
}
 

/**
 * This is called by the 2NN after having downloaded an image, and by
 * the NN after having received a new image from the 2NN. It
 * renames the image from fsimage_N.ckpt to fsimage_N and also
 * saves the related .md5 file into place.
 */
public synchronized void saveDigestAndRenameCheckpointImage(NameNodeFile nnf,
    long txid, MD5Hash digest) throws IOException {
  // Write and rename MD5 file
  List<StorageDirectory> badSds = Lists.newArrayList();
  
  for (StorageDirectory sd : storage.dirIterable(NameNodeDirType.IMAGE)) {
    File imageFile = NNStorage.getImageFile(sd, nnf, txid);
    try {
      MD5FileUtils.saveMD5File(imageFile, digest);
    } catch (IOException ioe) {
      badSds.add(sd);
    }
  }
  storage.reportErrorsOnDirectories(badSds);
  
  CheckpointFaultInjector.getInstance().afterMD5Rename();
  
  // Rename image from tmp file
  renameCheckpoint(txid, NameNodeFile.IMAGE_NEW, nnf, false);
  // So long as this is the newest image available,
  // advertise it as such to other checkpointers
  // from now on
  if (txid > storage.getMostRecentCheckpointTxId()) {
    storage.setMostRecentCheckpointInfo(txid, Time.now());
  }
}
 

/**
 * Set headers for content length, and, if available, md5.
 * @throws IOException 
 */
public static void setVerificationHeadersForGet(HttpServletResponse response,
    File file) throws IOException {
  response.setHeader(TransferFsImage.CONTENT_LENGTH,
      String.valueOf(file.length()));
  MD5Hash hash = MD5FileUtils.readStoredMd5ForFile(file);
  if (hash != null) {
    response.setHeader(TransferFsImage.MD5_HEADER, hash.toString());
  }
}
 

/**
 * Set headers for image length and if available, md5.
 * 
 * @throws IOException
 */
static void setVerificationHeadersForPut(HttpURLConnection connection,
    File file) throws IOException {
  connection.setRequestProperty(TransferFsImage.CONTENT_LENGTH,
      String.valueOf(file.length()));
  MD5Hash hash = MD5FileUtils.readStoredMd5ForFile(file);
  if (hash != null) {
    connection
        .setRequestProperty(TransferFsImage.MD5_HEADER, hash.toString());
  }
}
 

private MD5Hash calcPartialBlockChecksum(ExtendedBlock block,
    long requestLength, DataChecksum checksum, DataInputStream checksumIn)
    throws IOException {
  final int bytesPerCRC = checksum.getBytesPerChecksum();
  final int csize = checksum.getChecksumSize();
  final byte[] buffer = new byte[4*1024];
  MessageDigest digester = MD5Hash.getDigester();

  long remaining = requestLength / bytesPerCRC * csize;
  for (int toDigest = 0; remaining > 0; remaining -= toDigest) {
    toDigest = checksumIn.read(buffer, 0,
        (int) Math.min(remaining, buffer.length));
    if (toDigest < 0) {
      break;
    }
    digester.update(buffer, 0, toDigest);
  }
  
  int partialLength = (int) (requestLength % bytesPerCRC);
  if (partialLength > 0) {
    byte[] buf = new byte[partialLength];
    final InputStream blockIn = datanode.data.getBlockInputStream(block,
        requestLength - partialLength);
    try {
      // Get the CRC of the partialLength.
      IOUtils.readFully(blockIn, buf, 0, partialLength);
    } finally {
      IOUtils.closeStream(blockIn);
    }
    checksum.update(buf, 0, partialLength);
    byte[] partialCrc = new byte[csize];
    checksum.writeValue(partialCrc, 0, true);
    digester.update(partialCrc);
  }
  return new MD5Hash(digester.digest());
}
 

protected void getFields(Properties props, 
                         StorageDirectory sd 
                         ) throws IOException {
  super.getFields(props, sd);
  if (layoutVersion == 0)
    throw new IOException("NameNode directory " 
                          + sd.getRoot() + " is not formatted.");
  String sDUS, sDUV;
  sDUS = props.getProperty("distributedUpgradeState"); 
  sDUV = props.getProperty("distributedUpgradeVersion");
  setDistributedUpgradeState(
      sDUS == null? false : Boolean.parseBoolean(sDUS),
      sDUV == null? getLayoutVersion() : Integer.parseInt(sDUV));
  String sMd5 = props.getProperty(MESSAGE_DIGEST_PROPERTY);
  if (layoutVersion <= -26) {
    if (sMd5 == null) {
      throw new InconsistentFSStateException(sd.getRoot(),
          "file " + STORAGE_FILE_VERSION + " does not have MD5 image digest.");
    }
    this.setImageDigest(new MD5Hash(sMd5));
  } else if (sMd5 != null) {
    throw new InconsistentFSStateException(sd.getRoot(),
        "file " + STORAGE_FILE_VERSION + 
        " has image MD5 digest when version is " + layoutVersion);
  }
  this.checkpointTime = readCheckpointTime(sd);
}
 

/**
 * Read dataFile and compute its MD5 checksum.
 */
public static MD5Hash computeMd5ForFile(File dataFile) throws IOException {
  InputStream in = new FileInputStream(dataFile);
  try {
    MessageDigest digester = MD5Hash.getDigester();
    DigestInputStream dis = new DigestInputStream(in, digester);
    IOUtils.copyBytes(dis, new IOUtils.NullOutputStream(), 128*1024);
    
    return new MD5Hash(digester.digest());
  } finally {
    IOUtils.closeStream(in);
  }
}
 

/**
 * Run `hdfs dfsadmin -fetchImage ...' and verify that the downloaded image is
 * correct.
 */
private static void runFetchImage(DFSAdmin dfsAdmin, MiniDFSCluster cluster)
    throws Exception {
  int retVal = dfsAdmin.run(new String[]{"-fetchImage",
      FETCHED_IMAGE_FILE.getPath() });
  
  assertEquals(0, retVal);
  
  File highestImageOnNn = getHighestFsImageOnCluster(cluster);
  MD5Hash expected = MD5FileUtils.computeMd5ForFile(highestImageOnNn);
  MD5Hash actual = MD5FileUtils.computeMd5ForFile(
      new File(FETCHED_IMAGE_FILE, highestImageOnNn.getName()));
  
  assertEquals(expected, actual);
}
 

/**
 * Test when .md5 file exists but incorrect checksum
 */
@Test
public void testVerifyMD5FileBadDigest() throws Exception {
  MD5FileUtils.saveMD5File(TEST_FILE, MD5Hash.digest(new byte[0]));
  try {
    MD5FileUtils.verifySavedMD5(TEST_FILE, TEST_MD5);
    fail("Did not throw");
  } catch (IOException ioe) {
    // Expected
  }
}
 

protected static String getTraceSignature(String input) throws IOException {
  Path inputPath = new Path(input);
  FileSystem fs = inputPath.getFileSystem(new Configuration());
  FileStatus status = fs.getFileStatus(inputPath);
  Path qPath = fs.makeQualified(status.getPath());
  String traceID = status.getModificationTime() + qPath.toString()
                   + status.getOwner() + status.getLen();
  return MD5Hash.digest(traceID).toString();
}
 

/**
 * Moves fsimage.ckpt to fsImage and edits.new to edits
 * Reopens the new edits file.
 * 
 * @param newImageSignature the signature of the new image
 */
void rollFSImage(CheckpointSignature newImageSignature) throws IOException {
  MD5Hash newImageDigest = newImageSignature.getImageDigest();
  if (!newImageDigest.equals(checkpointImageDigest)) {
    throw new IOException(
        "Checkpoint image is corrupt: expecting an MD5 checksum of" +
        newImageDigest + " but is " + checkpointImageDigest);
  }
  rollFSImage(newImageSignature.getImageDigest());
}
 

/**
 * Client-side Method to fetch file from a server
 * Copies the response from the URL to a list of local files.
 * @param dstStorage if an error occurs writing to one of the files,
 *                   this storage object will be notified. 
 * @Return a digest of the received file if getChecksum is true
 */
static MD5Hash getFileClient(URL infoServer,
    String queryString, List<File> localPaths,
    Storage dstStorage, boolean getChecksum) throws IOException {
  URL url = new URL(infoServer, ImageServlet.PATH_SPEC + "?" + queryString);
  LOG.info("Opening connection to " + url);
  return doGetUrl(url, localPaths, dstStorage, getChecksum);
}
 

public void readFields(DataInput in) throws IOException {
  layoutVersion = in.readInt();
  namespaceID = in.readInt();
  cTime = in.readLong();
  editsTime = in.readLong();
  checkpointTime = in.readLong();
  imageDigest = new MD5Hash();
  imageDigest.readFields(in);
  checkpointState = CheckpointStates.deserialize(in.readInt());
}
 

/**
 * Load the image namespace from the given image file, verifying
 * it against the MD5 sum stored in its associated .md5 file.
 */
private void loadFSImage(File imageFile, FSNamesystem target,
    MetaRecoveryContext recovery, boolean requireSameLayoutVersion)
    throws IOException {
  MD5Hash expectedMD5 = MD5FileUtils.readStoredMd5ForFile(imageFile);
  if (expectedMD5 == null) {
    throw new IOException("No MD5 file found corresponding to image file "
        + imageFile);
  }
  loadFSImage(imageFile, expectedMD5, target, recovery,
      requireSameLayoutVersion);
}
 

/**
 * This is called by the 2NN after having downloaded an image, and by
 * the NN after having received a new image from the 2NN. It
 * renames the image from fsimage_N.ckpt to fsimage_N and also
 * saves the related .md5 file into place.
 */
public synchronized void saveDigestAndRenameCheckpointImage(NameNodeFile nnf,
    long txid, MD5Hash digest) throws IOException {
  // Write and rename MD5 file
  List<StorageDirectory> badSds = Lists.newArrayList();
  
  for (StorageDirectory sd : storage.dirIterable(NameNodeDirType.IMAGE)) {
    File imageFile = NNStorage.getImageFile(sd, nnf, txid);
    try {
      MD5FileUtils.saveMD5File(imageFile, digest);
    } catch (IOException ioe) {
      badSds.add(sd);
    }
  }
  storage.reportErrorsOnDirectories(badSds);
  
  CheckpointFaultInjector.getInstance().afterMD5Rename();
  
  // Rename image from tmp file
  renameCheckpoint(txid, NameNodeFile.IMAGE_NEW, nnf, false);
  // So long as this is the newest image available,
  // advertise it as such to other checkpointers
  // from now on
  if (txid > storage.getMostRecentCheckpointTxId()) {
    storage.setMostRecentCheckpointInfo(txid, Time.now());
  }
}
 

/**
 * Set headers for content length, and, if available, md5.
 * @throws IOException 
 */
public static void setVerificationHeadersForGet(HttpServletResponse response,
    File file) throws IOException {
  response.setHeader(TransferFsImage.CONTENT_LENGTH,
      String.valueOf(file.length()));
  MD5Hash hash = MD5FileUtils.readStoredMd5ForFile(file);
  if (hash != null) {
    response.setHeader(TransferFsImage.MD5_HEADER, hash.toString());
  }
}
 

/**
 * Set headers for image length and if available, md5.
 * 
 * @throws IOException
 */
static void setVerificationHeadersForPut(HttpURLConnection connection,
    File file) throws IOException {
  connection.setRequestProperty(TransferFsImage.CONTENT_LENGTH,
      String.valueOf(file.length()));
  MD5Hash hash = MD5FileUtils.readStoredMd5ForFile(file);
  if (hash != null) {
    connection
        .setRequestProperty(TransferFsImage.MD5_HEADER, hash.toString());
  }
}