HDFS-8305: HDFS INotify: the destination field of RenameOp should always end with the file name (cmccabe)

(cherry picked from commit fcd4cb7516)
(cherry picked from commit 1f01d8347a)

hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt

@@ -62,6 +62,9 @@ Release 2.7.1 - UNRELEASED
     HDFS-8091: ACLStatus and XAttributes should be presented to
     INodeAttributesProvider before returning to client (asuresh)
 
+    HDFS-8305: HDFS INotify: the destination field of RenameOp should always
+    end with the file name (cmccabe)
+
 Release 2.7.0 - 2015-04-20
 
   INCOMPATIBLE CHANGES

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSDirRenameOp.java

@@ -476,7 +476,7 @@ class FSDirRenameOp {
       fsd.writeUnlock();
     }
     if (stat) {
-      fsd.getEditLog().logRename(src, dst, mtime, logRetryCache);
+      fsd.getEditLog().logRename(src, actualDst, mtime, logRetryCache);
       return true;
     }
     return false;

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSEditLog.java

@@ -816,7 +816,9 @@ public class FSEditLog implements LogsPurgeable {
   }
   
   /** 
-   * Add rename record to edit log
+   * Add rename record to edit log.
+   *
+   * The destination should be the file name, not the destination directory.
    * TODO: use String parameters until just before writing to disk
    */
   void logRename(String src, String dst, long timestamp, boolean toLogRpcIds) {
@@ -827,9 +829,11 @@ public class FSEditLog implements LogsPurgeable {
     logRpcIds(op, toLogRpcIds);
     logEdit(op);
   }
-  
+
   /** 
-   * Add rename record to edit log
+   * Add rename record to edit log.
+   *
+   * The destination should be the file name, not the destination directory.
    */
   void logRename(String src, String dst, long timestamp, boolean toLogRpcIds,
       Options.Rename... options) {

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDFSInotifyEventInputStream.java

@@ -135,6 +135,7 @@ public class TestDFSInotifyEventInputStream {
       client.setAcl("/file5", AclEntry.parseAclSpec(
           "user::rwx,user:foo:rw-,group::r--,other::---", true));
       client.removeAcl("/file5"); // SetAclOp -> MetadataUpdateEvent
+      client.rename("/file5", "/dir"); // RenameOldOp -> RenameEvent
 
       EventBatch batch = null;
 
@@ -343,6 +344,16 @@ public class TestDFSInotifyEventInputStream {
           Event.MetadataUpdateEvent.MetadataType.ACLS);
       Assert.assertTrue(mue8.getAcls() == null);
 
+      // RenameOp (2)
+      batch = waitForNextEvents(eis);
+      Assert.assertEquals(1, batch.getEvents().length);
+      txid = checkTxid(batch, txid);
+      Assert.assertTrue(batch.getEvents()[0].getEventType() == Event.EventType.RENAME);
+      Event.RenameEvent re3 = (Event.RenameEvent) batch.getEvents()[0];
+      Assert.assertTrue(re3.getDstPath().equals("/dir/file5"));
+      Assert.assertTrue(re3.getSrcPath().equals("/file5"));
+      Assert.assertTrue(re.getTimestamp() > 0);
+
       // Returns null when there are no further events
       Assert.assertTrue(eis.poll() == null);