diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java
index 68ddfb0..7652be8 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java
@@ -3162,13 +3162,20 @@ public class FSNamesystem implements Namesystem, FSClusterStats,
       penultimateBlockMinReplication = 
         blockManager.checkMinReplication(penultimateBlock);
     }
-    assert penultimateBlockState == BlockUCState.COMPLETE ||
-           penultimateBlockState == BlockUCState.COMMITTED :
-           "Unexpected state of penultimate block in " + src;
+
+    boolean expected=(penultimateBlockState == BlockUCState.COMPLETE ||
+                      penultimateBlockState == BlockUCState.COMMITTED);
+
+    String unexpected_msg = "Unexpected state of penultimate block in " + src;
+    assert expected: unexpected_msg;
+    if (!expected)
+        LOG.error(unexpected_msg);
 
     switch(lastBlockState) {
     case COMPLETE:
-      assert false : "Already checked that the last block is incomplete";
+      String assert_msg = "Already checked that the last block is incomplete in " + src;
+      assert false : assert_msg;
+      LOG.error(assert_msg);
       break;
     case COMMITTED:
       // Close file if committed blocks are minimally replicated
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 d49311f..1da735f 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
@@ -65,12 +65,17 @@ import static org.apache.hadoop.util.Time.now;
  */
 @InterfaceAudience.Private
 public class LeaseManager {
+
+  private static final String NAMENODE_LEASEMANAGER_ITERMAX_KEY = "dfs.namenode.leasemanager.itermax";
+  private static final long NAMENODE_LEASEMANAGER_ITERMAX_DEFAULT = 1000;
+
   public static final Log LOG = LogFactory.getLog(LeaseManager.class);
 
   private final FSNamesystem fsnamesystem;
 
   private long softLimit = HdfsConstants.LEASE_SOFTLIMIT_PERIOD;
   private long hardLimit = HdfsConstants.LEASE_HARDLIMIT_PERIOD;
+  private final long lease_iters_max;
 
   //
   // Used for handling lock-leases
@@ -89,7 +94,12 @@ public class LeaseManager {
   private Daemon lmthread;
   private volatile boolean shouldRunMonitor;
 
-  LeaseManager(FSNamesystem fsnamesystem) {this.fsnamesystem = fsnamesystem;}
+  LeaseManager(FSNamesystem fsnamesystem) {
+    this.fsnamesystem = fsnamesystem;
+    org.apache.hadoop.conf.Configuration conf = new org.apache.hadoop.conf.Configuration();
+    this.lease_iters_max = conf.getLong(NAMENODE_LEASEMANAGER_ITERMAX_KEY,
+                                        NAMENODE_LEASEMANAGER_ITERMAX_DEFAULT);
+  }
 
   Lease getLease(String holder) {
     return leases.get(holder);
@@ -431,6 +441,7 @@ public class LeaseManager {
    */
   private synchronized boolean checkLeases() {
     boolean needSync = false;
+    long iters = 0;
     assert fsnamesystem.hasWriteLock();
     for(; sortedLeases.size() > 0; ) {
       final Lease oldest = sortedLeases.first();
@@ -438,7 +449,7 @@ public class LeaseManager {
         return needSync;
       }
 
-      LOG.info(oldest + " has expired hard limit");
+      LOG.warn(oldest + " has expired hard limit, " + oldest.getPaths().size() + " paths to process");
 
       final List<String> removing = new ArrayList<String>();
       // need to create a copy of the oldest lease paths, becuase 
@@ -451,6 +462,7 @@ public class LeaseManager {
         try {
           boolean completed = fsnamesystem.internalReleaseLease(oldest, p,
               HdfsServerConstants.NAMENODE_LEASE_HOLDER);
+
           if (LOG.isDebugEnabled()) {
             if (completed) {
               LOG.debug("Lease recovery for " + p + " is complete. File closed.");
@@ -462,11 +474,22 @@ public class LeaseManager {
           if (!needSync && !completed) {
             needSync = true;
           }
+
         } catch (IOException e) {
           LOG.error("Cannot release the path " + p + " in the lease "
               + oldest, e);
           removing.add(p);
         }
+
+        /* We don't want to be trapped here endlessly looping,
+         * because we're holding a lock. Get away as needed after a
+         * few iterations.
+         */
+        if (iters++ >= lease_iters_max) {
+          LOG.warn("Releasing lock after " + iters + " iterations, breaking out");
+          break;
+        }
+
       }
 
       for(String p : removing) {