package com.airbnb.reair.batch;

import com.airbnb.reair.common.FsUtils;
import com.airbnb.reair.incremental.deploy.ConfigurationKeys;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.util.Progressable;

import java.io.IOException;

/**
 * Utilities for batch replication.
 */
public class BatchUtils {
  private static final Log LOG = LogFactory.getLog(BatchUtils.class);

  /**
   * Executes a file copy.
   *
   * @param conf Hadoop configuration object
   * @param srcFileStatus Status of the source file
   * @param srcFs Source FileSystem
   * @param dstDir Destination directory
   * @param dstFs Destination FileSystem
   * @param tmpDirPath Temporary copy staging location.
   * @param progressable A progressable object to progress during long file copies
   * @param forceUpdate Whether to force a copy
   * @param identifier Identifier to use in the temporary file
   * @return An error string or null if successful
   */
  public static String doCopyFileAction(
      Configuration conf,
      SimpleFileStatus srcFileStatus,
      FileSystem srcFs,
      String dstDir,
      FileSystem dstFs,
      Path tmpDirPath,
      Progressable progressable,
      boolean forceUpdate,
      String identifier) {
    // TODO: Should be configurable
    int retry = 3;
    String lastError = null;

    while (retry > 0) {
      try {
        Path srcPath = new Path(srcFileStatus.getFullPath());
        if (!srcFs.exists(srcPath)) {
          LOG.info("Src does not exist. " + srcFileStatus.getFullPath());
          return "Src does not exist. " + srcFileStatus.getFullPath();
        }
        FileStatus srcStatus = srcFs.getFileStatus(srcPath);

        final FSDataInputStream inputStream = srcFs.open(srcPath);
        Path dstPath = new Path(dstDir, srcFileStatus.getFileName());
        // if dst already exists.
        if (dstFs.exists(dstPath)) {
          FileStatus dstStatus = dstFs.getFileStatus(dstPath);
          // If it is not force update, and the file size are same we will not recopy.
          // This normally happens when we do retry run.
          if (!forceUpdate && srcStatus.getLen() == dstStatus.getLen()) {
            LOG.info("dst already exists. " + dstPath.toString());
            return "dst already exists. " + dstPath.toString();
          }
        }

        Path dstParentPath = new Path(dstDir);
        if (!dstFs.exists(dstParentPath) && !dstFs.mkdirs(dstParentPath)) {
          LOG.info("Could not create directory: " + dstDir);
          return "Could not create directory: " + dstDir;
        }

        Path tmpDstPath = new Path(
            tmpDirPath,
            "__tmp__copy__file_" + identifier + "_" + srcFileStatus.getFileName()
                + "." + System.currentTimeMillis());
        if (dstFs.exists(tmpDstPath)) {
          dstFs.delete(tmpDstPath, false);
        }

        // Keep the same replication factor and block size as the source file.
        FSDataOutputStream outputStream = dstFs.create(
            tmpDstPath,
            srcStatus.getPermission(),
            true,
            conf.getInt("io.file.buffer.size", 4096),
            srcStatus.getReplication(),
            srcStatus.getBlockSize(),
            progressable);

        IOUtils.copyBytes(inputStream, outputStream, conf);
        inputStream.close();
        outputStream.close();
        if (forceUpdate && dstFs.exists(dstPath)) {
          dstFs.delete(dstPath, false);
        }

        // If checksums exist and don't match, re-do the copy. If checksums do not exist, assume
        // that they match.
        if (conf.getBoolean(ConfigurationKeys.BATCH_JOB_VERIFY_COPY_CHECKSUM, true)
            && !FsUtils.checksumsMatch(conf, srcPath, tmpDstPath)
            .map(Boolean::booleanValue)
            .orElse(true)) {
          throw new IOException(String.format("Not renaming %s to %s since checksums do not match "
                  + "between %s and %s",
              tmpDstPath,
              dstPath,
              srcPath,
              tmpDstPath));
        }

        dstFs.rename(tmpDstPath, dstPath);
        dstFs.setTimes(dstPath, srcStatus.getModificationTime(), srcStatus.getAccessTime());
        LOG.info(dstPath.toString() + " file copied");
        progressable.progress();
        return null;
      } catch (IOException e) {
        LOG.info("Got an exception!", e);
        lastError = e.getMessage();
        --retry;
      }
    }

    return lastError;
  }
}

@@ -2006,75 +2006,8 @@
 );%0A%0A
-        final FSDataInputStream inputStream = srcFs.open(srcPath);%0A
     
@@ -3199,16 +3199,84 @@
 e file.%0A
+        try (FSDataInputStream inputStream = srcFs.open(srcPath);%0A  
         
@@ -3561,19 +3561,22 @@
 essable)
-;%0A%0A
+) %7B%0A  
         
@@ -3639,58 +3639,10 @@
     
-inputStream.close();%0A        outputStream.close();
+%7D%0A
 %0A   

Make sure inputStream is closed (#71)