From 2fc7e14e392f188958b9867a5d2dd563dfcc378a Mon Sep 17 00:00:00 2001
From: Chris Nauroth <cnauroth@apache.org>
Date: Thu, 15 Aug 2013 20:43:46 +0000
Subject: [PATCH] HDFS-5099. Namenode#copyEditLogSegmentsToSharedDir should
 close EditLogInputStreams upon finishing. Contributed by Chuan Liu.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/trunk@1514481 13f79535-47bb-0310-9956-ffa450edef68
---
 hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt   |  3 +
 .../hadoop/hdfs/server/namenode/NameNode.java | 64 +++++++++++--------
 2 files changed, 39 insertions(+), 28 deletions(-)

diff --git a/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt b/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
index fbd7e61eb66..02beab5c609 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
+++ b/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
@@ -339,6 +339,9 @@ Release 2.1.1-beta - UNRELEASED
     HDFS-5080. BootstrapStandby not working with QJM when the existing NN is 
     active. (jing9)
 
+    HDFS-5099. Namenode#copyEditLogSegmentsToSharedDir should close
+    EditLogInputStreams upon finishing. (Chuan Liu via cnauroth)
+
 Release 2.1.0-beta - 2013-08-22
 
   INCOMPATIBLE CHANGES
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNode.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNode.java
index a933585523a..b8a51390c11 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNode.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNode.java
@@ -956,41 +956,49 @@ private static void copyEditLogSegmentsToSharedDir(FSNamesystem fsns,
     FSEditLog sourceEditLog = fsns.getFSImage().editLog;
     
     long fromTxId = fsns.getFSImage().getMostRecentCheckpointTxId();
-    Collection<EditLogInputStream> streams = sourceEditLog.selectInputStreams(
-        fromTxId+1, 0);
-
-    // Set the nextTxid to the CheckpointTxId+1
-    newSharedEditLog.setNextTxId(fromTxId + 1);
     
-    // Copy all edits after last CheckpointTxId to shared edits dir
-    for (EditLogInputStream stream : streams) {
-      LOG.debug("Beginning to copy stream " + stream + " to shared edits");
-      FSEditLogOp op;
-      boolean segmentOpen = false;
-      while ((op = stream.readOp()) != null) {
-        if (LOG.isTraceEnabled()) {
-          LOG.trace("copying op: " + op);
-        }
-        if (!segmentOpen) {
-          newSharedEditLog.startLogSegment(op.txid, false);
-          segmentOpen = true;
-        }
-        
-        newSharedEditLog.logEdit(op);
+    Collection<EditLogInputStream> streams = null;
+    try {
+      streams = sourceEditLog.selectInputStreams(fromTxId + 1, 0);
 
-        if (op.opCode == FSEditLogOpCodes.OP_END_LOG_SEGMENT) {
+      // Set the nextTxid to the CheckpointTxId+1
+      newSharedEditLog.setNextTxId(fromTxId + 1);
+
+      // Copy all edits after last CheckpointTxId to shared edits dir
+      for (EditLogInputStream stream : streams) {
+        LOG.debug("Beginning to copy stream " + stream + " to shared edits");
+        FSEditLogOp op;
+        boolean segmentOpen = false;
+        while ((op = stream.readOp()) != null) {
+          if (LOG.isTraceEnabled()) {
+            LOG.trace("copying op: " + op);
+          }
+          if (!segmentOpen) {
+            newSharedEditLog.startLogSegment(op.txid, false);
+            segmentOpen = true;
+          }
+
+          newSharedEditLog.logEdit(op);
+
+          if (op.opCode == FSEditLogOpCodes.OP_END_LOG_SEGMENT) {
+            newSharedEditLog.logSync();
+            newSharedEditLog.endCurrentLogSegment(false);
+            LOG.debug("ending log segment because of END_LOG_SEGMENT op in "
+                + stream);
+            segmentOpen = false;
+          }
+        }
+
+        if (segmentOpen) {
+          LOG.debug("ending log segment because of end of stream in " + stream);
           newSharedEditLog.logSync();
           newSharedEditLog.endCurrentLogSegment(false);
-          LOG.debug("ending log segment because of END_LOG_SEGMENT op in " + stream);
           segmentOpen = false;
         }
       }
-      
-      if (segmentOpen) {
-        LOG.debug("ending log segment because of end of stream in " + stream);
-        newSharedEditLog.logSync();
-        newSharedEditLog.endCurrentLogSegment(false);
-        segmentOpen = false;
+    } finally {
+      if (streams != null) {
+        FSEditLog.closeAllStreams(streams);
       }
     }
   }