HADOOP-13600 moving parallel copy into its own class

Change-Id: Ibfe08fb289deaa3b2fa21a8281f58651edb6d2ab

Author: steveloughran

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/ParallelCopier.java

@@ -0,0 +1,26 @@
+/*
+ * 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.s3a;
+
+/**
+ * Execute copy operations in parallel;  blocks until all operations
+ * are completed.
+ */
+class ParallelCopier {
+}

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java

@@ -663,9 +663,6 @@ public boolean rename(Path src, Path dst) throws IOException {
       return innerRename(src, dst);
     } catch (AmazonClientException e) {
       throw translateException("rename(" + src +", " + dst + ")", src, e);
-    } catch (FileNotFoundException e) {
-      LOG.error("rename: src not found {}", src);
-      return false;
     } catch (RenameFailedException e) {
       LOG.debug(e.getMessage());
       return e.getExitCode();
@@ -701,6 +698,8 @@ private boolean innerRename(Path src, Path dst) throws IOException,
       throw new RenameFailedException(src, dst, "dest is root directory");
     }
 
+    // get the source file status; this raises a FNFE if there is no source
+    // file.
     S3AFileStatus srcStatus = getFileStatus(src);
 
     if (srcKey.equals(dstKey)) {
@@ -714,15 +713,31 @@ private boolean innerRename(Path src, Path dst) throws IOException,
     S3AFileStatus dstStatus = null;
     try {
       dstStatus = getFileStatus(dst);
-
-      if (srcStatus.isDirectory() && dstStatus.isFile()) {
-        throw new RenameFailedException(src, dst,
-            "source is a directory and dest is a file")
-            .withExitCode(srcStatus.isFile());
-      }
-      if (dstStatus.isDirectory() && !dstStatus.isEmptyDirectory()) {
-        return false;
+      // if there is no destination entry, an exception is raised.
+      // hence this code sequence can assume that there is something
+      // at the end of the path; the only detail being what it is and
+      // whether or not it can be the destination of the rename.
+      if (srcStatus.isDirectory()) {
+        if (dstStatus.isFile()) {
+          throw new RenameFailedException(src, dst,
+              "source is a directory and dest is a file")
+              .withExitCode(srcStatus.isFile());
+        } else if (!dstStatus.isEmptyDirectory()) {
+          throw new RenameFailedException(src, dst,
+              "Destination is a non-empty directory")
+              .withExitCode(false);
+        }
+        // at this point the destination is an empty directory
+      } else {
+        // source is a file. Look at the destination
+        if (dstStatus.isFile()) {
+          throw new RenameFailedException(src, dst,
+              "destination file for rename operations already exists")
+              .withExitCode(false);
+        }
+        //
       }
+
     } catch (FileNotFoundException e) {
       LOG.debug("rename: destination path {} not found", dst);
       // Parent must exist
@@ -880,6 +895,7 @@ public CopyResult call() throws Exception {
               throw new IOException(ex);
             }
           } else {
+            //no cause
             throw new IOException(e);
           }
         }