HADOOP-977. Send task's stdout and stderr to JobClient's stdout and stderr, respectively. Contributed by Arun.

git-svn-id: https://svn.apache.org/repos/asf/lucene/hadoop/trunk@505507 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -26,6 +26,10 @@ Trunk (unreleased changes)
  7. HADOOP-882.  Upgrade to jets3t version 0.5, used by the S3
     FileSystem.  This version supports retries.  (Michael Stack via cutting)
 
+ 8. HADOOP-977.  Send task's stdout and stderr to JobClient's stdout
+    and stderr respectively, with each line tagged by the task's name.
+    (Arun C Murthy via cutting)
+
 
 Branch 0.11 - unreleased
 

conf/hadoop-default.xml

@@ -699,6 +699,15 @@ creations/deletions), or "all".</description>
   </description>
 </property>
 
+<property>
+  <name>jobclient.output.filter</name>
+  <value>FAILED</value>
+  <description>The filter for controlling the output of the task's userlogs sent
+               to the console of the JobClient. 
+               The permissible options are: NONE, FAILED, SUCCEEDED and ALL.
+  </description>
+</property>
+
 <!-- ipc properties -->
 
 <property>

src/java/org/apache/hadoop/mapred/JobClient.java

@@ -384,7 +384,7 @@ public class JobClient extends ToolBase implements MRConstants  {
       String lastReport = null;
       final int MAX_RETRIES = 5;
       int retries = MAX_RETRIES;
-      String outputFilterName = job.get("jobclient.output.filter");
+      String outputFilterName = job.get("jobclient.output.filter", "FAILED");
 
       if (null != outputFilterName) {
         try {
@@ -428,19 +428,19 @@ public class JobClient extends ToolBase implements MRConstants  {
                   if( event.getTaskStatus() == 
                     TaskCompletionEvent.Status.SUCCEEDED){
                     LOG.info(event.toString());
-                    printHttpFile(event.getTaskTrackerHttp());
+                    displayTaskLogs(event.getTaskId(), event.getTaskTrackerHttp());
                   }
                   break; 
                 case FAILED:
                   if( event.getTaskStatus() == 
                     TaskCompletionEvent.Status.FAILED){
                     LOG.info(event.toString());
-                    printHttpFile(event.getTaskTrackerHttp());
+                    displayTaskLogs(event.getTaskId(), event.getTaskTrackerHttp());
                   }
                   break ; 
                 case ALL:
                   LOG.info(event.toString());
-                  printHttpFile(event.getTaskTrackerHttp());
+                  displayTaskLogs(event.getTaskId(), event.getTaskTrackerHttp());
                   break;
                 }
               }
@@ -467,29 +467,33 @@ public class JobClient extends ToolBase implements MRConstants  {
         jc.close();
       }
     }
+
+    private static void displayTaskLogs(String taskId, String baseUrl)
+    throws IOException {
+      // Copy tasks's stdout of the JobClient
+      getTaskLogs(taskId, new URL(baseUrl+"&filter=stdout"), System.out);
+      
+      // Copy task's stderr to stderr of the JobClient 
+      getTaskLogs(taskId, new URL(baseUrl+"&filter=stderr"), System.err);
+    }
     
-    static void printHttpFile(String httpURL ) throws IOException {
-      boolean good = false;
-      long totalBytes = 0;
-      URL path = new URL(httpURL); 
+    private static void getTaskLogs(String taskId, URL taskLogUrl, 
+            OutputStream out) {
       try {
-        URLConnection connection = path.openConnection();
-        InputStream input = connection.getInputStream();
+        URLConnection connection = taskLogUrl.openConnection();
+        BufferedReader input = 
+          new BufferedReader(new InputStreamReader(connection.getInputStream()));
+        BufferedWriter output = 
+          new BufferedWriter(new OutputStreamWriter(out));
         try {
-          byte[] buffer = new byte[64 * 1024];
-          int len = input.read(buffer);
-          while (len > 0) {
-            totalBytes += len;
-            LOG.info(new String(buffer).trim());
-            len = input.read(buffer);
-          }
-          good = ((int) totalBytes) == connection.getContentLength();
-          if (!good) {
-          LOG.warn("Incomplete task output received for " + path +
-                          " (" + totalBytes + " instead of " + 
-                          connection.getContentLength() + ")");
+          String logData = null;
+          while ((logData = input.readLine()) != null) {
+            if (logData.length() > 0) {
+              output.write(taskId + ": " + logData + "\n");
+              output.flush();
+            }
           }
-        }finally {
+        } finally {
             input.close();
         }
       }catch(IOException ioe){

src/webapps/task/tasklog.jsp

@@ -121,9 +121,6 @@
   	  }
   	  gotRequiredData = false;
   	}
-  	if( plainText ) {
-  	  response.setContentLength(bytesRead); 
-  	}
 	String logData = new String(b, 0, bytesRead);
 	out.println(logData);
   } catch (IOException ioe) {