From 368def7c6003a715f23bdde5cd2953f3779ebc05 Mon Sep 17 00:00:00 2001
From: Brandon Li <brandonli@apache.org>
Date: Thu, 20 Feb 2014 22:32:49 +0000
Subject: [PATCH] Merging change r1570366 from trunk

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-2@1570372 13f79535-47bb-0310-9956-ffa450edef68
---
 hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt   |  3 +
 .../hdfs/server/namenode/LeaseManager.java    |  7 ++-
 .../server/namenode/TestLeaseManager.java     | 57 +++++++++++++++++++
 3 files changed, 66 insertions(+), 1 deletion(-)
 create mode 100644 hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestLeaseManager.java

diff --git a/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt b/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
index 9082ca6290d..a044696bada 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
+++ b/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
@@ -211,6 +211,9 @@ Release 2.4.0 - UNRELEASED
     HDFS-5962. Mtime and atime are not persisted for symbolic links. (Akira
     Ajisaka via kihwal)
 
+    HDFS-5944. LeaseManager:findLeaseWithPrefixPath can't handle path like /a/b/
+    and cause SecondaryNameNode failed do checkpoint (Yunjiong Zhao via brandonli)
+
   BREAKDOWN OF HDFS-5698 SUBTASKS AND RELATED JIRAS
 
     HDFS-5717. Save FSImage header in protobuf. (Haohui Mai via jing9)
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/LeaseManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/LeaseManager.java
index b9f9d10396e..43e59ff757b 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/LeaseManager.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/LeaseManager.java
@@ -339,7 +339,12 @@ static private Map<String, Lease> findLeaseWithPrefixPath(
     }
 
     final Map<String, Lease> entries = new HashMap<String, Lease>();
-    final int srclen = prefix.length();
+    int srclen = prefix.length();
+    
+    // prefix may ended with '/'
+    if (prefix.charAt(srclen - 1) == Path.SEPARATOR_CHAR) {
+      srclen -= 1;
+    }
 
     for(Map.Entry<String, Lease> entry : path2lease.tailMap(prefix).entrySet()) {
       final String p = entry.getKey();
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestLeaseManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestLeaseManager.java
new file mode 100644
index 00000000000..805f7c87ca5
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestLeaseManager.java
@@ -0,0 +1,57 @@
+/**
+ * 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.hdfs.server.namenode;
+
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdfs.HdfsConfiguration;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+
+public class TestLeaseManager {
+  Configuration conf = new HdfsConfiguration();
+  
+  @Test
+  public void testRemoveLeaseWithPrefixPath() throws Exception {
+    MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf).numDataNodes(2).build();
+    cluster.waitActive();
+
+    LeaseManager lm = NameNodeAdapter.getLeaseManager(cluster.getNamesystem());
+    lm.addLease("holder1", "/a/b");
+    lm.addLease("holder2", "/a/c");
+    assertNotNull(lm.getLeaseByPath("/a/b"));
+    assertNotNull(lm.getLeaseByPath("/a/c"));
+
+    lm.removeLeaseWithPrefixPath("/a");
+
+    assertNull(lm.getLeaseByPath("/a/b"));
+    assertNull(lm.getLeaseByPath("/a/c"));
+
+    lm.addLease("holder1", "/a/b");
+    lm.addLease("holder2", "/a/c");
+
+    lm.removeLeaseWithPrefixPath("/a/");
+
+    assertNull(lm.getLeaseByPath("/a/b"));
+    assertNull(lm.getLeaseByPath("/a/c"));
+  }
+}