@Test
public void testReportChecksumFailure() throws IOException {

  base.mkdirs();
  assertTrue(base.exists() && base.isDirectory());
  
  final File dir1 = new File(base, "dir1");
  final File dir2 = new File(dir1, "dir2");
  dir2.mkdirs();
  assertTrue(dir2.exists() && FileUtil.canWrite(dir2));
  
  final String dataFileName = "corruptedData";
  final Path dataPath = new Path(new File(dir2, dataFileName).toURI());
  final Path checksumPath = fileSys.getChecksumFile(dataPath);
  final FSDataOutputStream fsdos = fileSys.create(dataPath);
  try {
    fsdos.writeUTF("foo");
  } finally {
    fsdos.close();
  }
  assertTrue(fileSys.pathToFile(dataPath).exists());
  final long dataFileLength = fileSys.getFileStatus(dataPath).getLen();
  assertTrue(dataFileLength > 0);
  
  // check the the checksum file is created and not empty:
  assertTrue(fileSys.pathToFile(checksumPath).exists());
  final long checksumFileLength = fileSys.getFileStatus(checksumPath).getLen();
  assertTrue(checksumFileLength > 0);
  
  // this is a hack to force the #reportChecksumFailure() method to stop
  // climbing up at the 'base' directory and use 'dir1/bad_files' as the 
  // corrupted files storage:
  FileUtil.setWritable(base, false);
  
  FSDataInputStream dataFsdis = fileSys.open(dataPath);
  FSDataInputStream checksumFsdis = fileSys.open(checksumPath);
  
  boolean retryIsNecessary = fileSys.reportChecksumFailure(dataPath, dataFsdis, 0, checksumFsdis, 0);
  assertTrue(!retryIsNecessary);
  
  // the data file should be moved:
  assertTrue(!fileSys.pathToFile(dataPath).exists());
  // the checksum file should be moved:
  assertTrue(!fileSys.pathToFile(checksumPath).exists());
  
  // check that the files exist in the new location where they were moved:
  File[] dir1files = dir1.listFiles(new FileFilter() {
    @Override
    public boolean accept(File pathname) {
      return pathname != null && !pathname.getName().equals("dir2");
    }
  });
  assertTrue(dir1files != null);
  assertTrue(dir1files.length == 1);
  File badFilesDir = dir1files[0];
  
  File[] badFiles = badFilesDir.listFiles();
  assertTrue(badFiles != null);
  assertTrue(badFiles.length == 2);
  boolean dataFileFound = false;
  boolean checksumFileFound = false;
  for (File badFile: badFiles) {
    if (badFile.getName().startsWith(dataFileName)) {
      assertTrue(dataFileLength == badFile.length());
      dataFileFound = true;
    } else if (badFile.getName().contains(dataFileName + ".crc")) {
      assertTrue(checksumFileLength == badFile.length());
      checksumFileFound = true;
    }
  }
  assertTrue(dataFileFound);
  assertTrue(checksumFileFound);
}