Fix #1820, Fix reporting file rename action success but actually fail…

…ed bug (#1821)

smart-hadoop-support/smart-hadoop/src/main/java/org/smartdata/hdfs/action/RenameFileAction.java

@@ -67,11 +67,12 @@ protected void execute() throws Exception {
     if (destPath == null) {
       throw new IllegalArgumentException("Dest File parameter is missing.");
     }
-    appendLog(
-        String.format("Action starts at %s : Rename %s to %s",
-            Utils.getFormatedCurrentTime(), srcPath, destPath));
-    //rename
-    renameSingleFile(srcPath, destPath);
+    appendLog(String.format("Action starts at %s : Rename %s to %s",
+        Utils.getFormatedCurrentTime(), srcPath, destPath));
+
+    if (!renameSingleFile(srcPath, destPath)) {
+      throw new IOException("Failed to rename " + srcPath + " -> " + destPath);
+    }
   }
 
   private boolean renameSingleFile(String src,

smart-hadoop-support/smart-hadoop/src/test/java/org/smartdata/hdfs/action/TestRenameFileAction.java

@@ -88,4 +88,45 @@ public void testRemoteFileRename() throws IOException {
     Assert.assertFalse(dfsClient.exists(srcPath + "/" + file1));
     Assert.assertTrue(dfsClient.exists(destPath + "/" + destFilename));
   }
+
+  @Test
+  public void testDestExistingFileRename() throws Exception {
+    final String srcPath = "/testRename";
+    final String destPath = "/destDir";
+    final String src = srcPath + "/file1";
+    final String dest = destPath + "/file2";
+
+    dfs.mkdirs(new Path(dfs.getUri() + srcPath));
+    dfs.mkdirs(new Path(dfs.getUri() + destPath));
+
+    // write to DISK
+    final FSDataOutputStream out1 = dfs.create(new Path(dfs.getUri() + src));
+    out1.writeChars("testCopy1");
+    out1.close();
+    final FSDataOutputStream out2 = dfs.create(new Path(dfs.getUri() + dest));
+    out2.writeChars("testCopy2");
+    out2.close();
+
+    verifyRenameFailed(src, dest);
+  }
+
+  @Test
+  public void testFileRenameFailed() throws Exception {
+    String src = new Path(dfs.getUri() + "/somesrcfile").toString();
+    String dest = new Path(dfs.getUri() + "/somedestfile").toString();
+    verifyRenameFailed(src, dest);
+    verifyRenameFailed("/somesrclocalfile", "/somedestlocalfile");
+  }
+
+  private void verifyRenameFailed(String src, String dest) throws Exception {
+    RenameFileAction renameFileAction = new RenameFileAction();
+    renameFileAction.setDfsClient(dfsClient);
+    renameFileAction.setContext(smartContext);
+    Map<String , String> args = new HashMap<>();
+    args.put(RenameFileAction.FILE_PATH , src);
+    args.put(RenameFileAction.DEST_PATH , dest);
+    renameFileAction.init(args);
+    renameFileAction.run();
+    Assert.assertTrue(renameFileAction.getActionStatus().getThrowable() != null);
+  }
 }