public static void compareFileLengthsAndChecksums(long srcLen,
             FileSystem sourceFS, Path source, FileChecksum sourceChecksum,
             FileSystem targetFS, Path target, boolean skipCrc,
             long targetLen) throws IOException {
    if (srcLen != targetLen) {
      throw new IOException(
          DistCpConstants.LENGTH_MISMATCH_ERROR_MSG + source + " (" + srcLen
              + ") and target:" + target + " (" + targetLen + ")");
    }

    //At this point, src & dest lengths are same. if length==0, we skip checksum
    if ((srcLen != 0) && (!skipCrc)) {
      CopyMapper.ChecksumComparison
          checksumComparison = checksumsAreEqual(sourceFS, source, sourceChecksum,
              targetFS, target, srcLen);
      // If Checksum comparison is false set it to false, else set to true.
      boolean checksumResult = !checksumComparison.equals(CopyMapper.ChecksumComparison.FALSE);
      if (!checksumResult) {
        StringBuilder errorMessage =
            new StringBuilder(DistCpConstants.CHECKSUM_MISMATCH_ERROR_MSG)
                .append(source).append(" and ").append(target).append(".");
        boolean addSkipHint = false;
        String srcScheme = sourceFS.getScheme();
        String targetScheme = targetFS.getScheme();
        if (!srcScheme.equals(targetScheme)) {
          // the filesystems are different and they aren't both hdfs connectors
          errorMessage.append("Source and destination filesystems are of"
              + " different types\n")
              .append("Their checksum algorithms may be incompatible");
          addSkipHint = true;
        } else if (sourceFS.getFileStatus(source).getBlockSize() !=
            targetFS.getFileStatus(target).getBlockSize()) {
          errorMessage.append(" Source and target differ in block-size.\n")
              .append(" Use -pb to preserve block-sizes during copy.");
          addSkipHint = true;
        }
        if (addSkipHint) {
          errorMessage
              .append(" You can choose file-level checksum validation via "
                  + "-Ddfs.checksum.combine.mode=COMPOSITE_CRC when block-sizes"
                  + " or filesystems are different.")
              .append(" Or you can skip checksum-checks altogether "
                  + " with -skipcrccheck.\n")
              .append(" (NOTE: By skipping checksums, one runs the risk of " +
                  "masking data-corruption during file-transfer.)\n");
        }
        throw new IOException(errorMessage.toString());
      }
    }
  }
@Test
  public void testCompareFileLengthsAndChecksums() throws Throwable {

    String base = "/tmp/verify-checksum/";
    long srcSeed = System.currentTimeMillis();
    long dstSeed = srcSeed + rand.nextLong();
    short replFactor = 2;

    FileSystem fs = FileSystem.get(config);
    Path basePath = new Path(base);
    fs.mkdirs(basePath);

    // empty lengths comparison
    Path srcWithLen0 = new Path(base + "srcLen0");
    Path dstWithLen0 = new Path(base + "dstLen0");
    fs.create(srcWithLen0).close();
    fs.create(dstWithLen0).close();
    DistCpUtils.compareFileLengthsAndChecksums(0, fs, srcWithLen0,
        null, fs, dstWithLen0, false, 0);

    // different lengths comparison
    Path srcWithLen1 = new Path(base + "srcLen1");
    Path dstWithLen2 = new Path(base + "dstLen2");
    DFSTestUtil.createFile(fs, srcWithLen1, 1, replFactor, srcSeed);
    DFSTestUtil.createFile(fs, dstWithLen2, 2, replFactor, srcSeed);

    intercept(IOException.class, DistCpConstants.LENGTH_MISMATCH_ERROR_MSG,
        () -> DistCpUtils.compareFileLengthsAndChecksums(1, fs,
                srcWithLen1, null, fs, dstWithLen2, false, 2));

    // checksums matched
    Path srcWithChecksum1 = new Path(base + "srcChecksum1");
    Path dstWithChecksum1 = new Path(base + "dstChecksum1");
    DFSTestUtil.createFile(fs, srcWithChecksum1, 1024,
        replFactor, srcSeed);
    DFSTestUtil.createFile(fs, dstWithChecksum1, 1024,
        replFactor, srcSeed);
    DistCpUtils.compareFileLengthsAndChecksums(1024, fs, srcWithChecksum1,
        null, fs, dstWithChecksum1, false, 1024);
    DistCpUtils.compareFileLengthsAndChecksums(1024, fs, srcWithChecksum1,
        fs.getFileChecksum(srcWithChecksum1), fs, dstWithChecksum1,
        false, 1024);

    // checksums mismatched
    Path dstWithChecksum2 = new Path(base + "dstChecksum2");
    DFSTestUtil.createFile(fs, dstWithChecksum2, 1024,
        replFactor, dstSeed);
    intercept(IOException.class, DistCpConstants.CHECKSUM_MISMATCH_ERROR_MSG,
        () -> DistCpUtils.compareFileLengthsAndChecksums(1024, fs,
               srcWithChecksum1, null, fs, dstWithChecksum2,
               false, 1024));

    // checksums mismatched but skipped
    DistCpUtils.compareFileLengthsAndChecksums(1024, fs, srcWithChecksum1,
        null, fs, dstWithChecksum2, true, 1024);
  }