svn commit: r1552393 - /hive/trunk/hcatalog/webhcat/svr/src/main/java/org/apache/hive/hcatalog/templeton/tool/LaunchMapper.java

[daijy]

Author: daijy
Date: Thu Dec 19 18:38:03 2013
New Revision: 1552393

URL: http://svn.apache.org/r1552393
Log:
HIVE-5540: webhcat e2e test failures: "Expect 1 jobs in logs, but get 1"

Modified:
    
hive/trunk/hcatalog/webhcat/svr/src/main/java/org/apache/hive/hcatalog/templeton/tool/LaunchMapper.java

Modified: 
hive/trunk/hcatalog/webhcat/svr/src/main/java/org/apache/hive/hcatalog/templeton/tool/LaunchMapper.java
URL: 
http://svn.apache.org/viewvc/hive/trunk/hcatalog/webhcat/svr/src/main/java/org/apache/hive/hcatalog/templeton/tool/LaunchMapper.java?rev=1552393&r1=1552392&r2=1552393&view=diff
==============================================================================
--- 
hive/trunk/hcatalog/webhcat/svr/src/main/java/org/apache/hive/hcatalog/templeton/tool/LaunchMapper.java
 (original)
+++ 
hive/trunk/hcatalog/webhcat/svr/src/main/java/org/apache/hive/hcatalog/templeton/tool/LaunchMapper.java
 Thu Dec 19 18:38:03 2013
@@ -241,6 +241,7 @@ public class LaunchMapper extends Mapper
     private OutputStream out;
     private final JobID jobid;
     private final Configuration conf;
+    boolean needCloseOutput = false;
 
     public Watcher(Configuration conf, JobID jobid, InputStream in, String 
statusdir, String name)
       throws IOException {
@@ -258,16 +259,18 @@ public class LaunchMapper extends Mapper
         Path p = new Path(statusdir, name);
         FileSystem fs = p.getFileSystem(conf);
         out = fs.create(p);
+        needCloseOutput = true;
         LOG.info("templeton: Writing status to " + p);
       }
     }
 
     @Override
     public void run() {
+      PrintWriter writer = null;
       try {
         InputStreamReader isr = new InputStreamReader(in);
         BufferedReader reader = new BufferedReader(isr);
-        PrintWriter writer = new PrintWriter(out);
+        writer = new PrintWriter(out);
 
         String line;
         while ((line = reader.readLine()) != null) {
@@ -308,6 +311,15 @@ public class LaunchMapper extends Mapper
         }
       } catch (IOException e) {
         LOG.error("templeton: execute error: ", e);
+      } finally {
+        // Need to close() because in some FileSystem
+        // implementations flush() is no-op.
+        // Close the file handle if it is a hdfs file.
+        // But if it is stderr/stdout, skip it since
+        // WebHCat is not supposed to close it
+        if (needCloseOutput && writer!=null) {
+          writer.close();
+        }
       }
     }
   }