HADOOP-2208. Counter update frequency (from TaskTracker to JobTracker) is capped at 1 minute. Contributed by Amareshwari Sri Ramadasu.

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

CHANGES.txt

@@ -152,6 +152,9 @@ Trunk (unreleased changes)
     HADOOP-1900. Makes the heartbeat and task event queries interval 
     dependent on the cluster size.  (Amareshwari Sri Ramadasu via ddas)
 
+    HADOOP-2208. Counter update frequency (from TaskTracker to JobTracker) is 
+    capped at 1 minute.  (Amareshwari Sri Ramadasu via ddas)
+
   BUG FIXES
 
     HADOOP-2100.  Remove faulty check for existence of $HADOOP_PID_DIR and let

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

@@ -39,8 +39,9 @@ interface InterTrackerProtocol extends VersionedProtocol {
    * TaskTrackerStatus for HADOOP-1274
    * Version 8: HeartbeatResponse is added with the next heartbeat interval.
    * version 9 changes the counter representation for HADOOP-2248
+   * version 10 changes the TaskStatus representation for HADOOP-2208
    */
-  public static final long versionID = 9L;
+  public static final long versionID = 10L;
   
   public final static int TRACKERS_OK = 0;
   public final static int UNKNOWN_TASKTRACKER = 1;

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

@@ -29,6 +29,8 @@ interface MRConstants {
   
   public static final int CLUSTER_INCREMENT = 50;
 
+  public static final long COUNTER_UPDATE_INTERVAL = 60 * 1000;
+
   //for the inmemory filesystem (to do in-memory merge)
   /**
    * Constant denoting when a merge of in memory files will be triggered 

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

@@ -55,6 +55,7 @@ abstract class TaskStatus implements Writable, Cloneable {
     
   private Phase phase = Phase.STARTING; 
   private Counters counters;
+  private boolean includeCounters;
 
   public TaskStatus() {}
 
@@ -70,6 +71,7 @@ abstract class TaskStatus implements Writable, Cloneable {
     this.taskTracker = taskTracker;
     this.phase = phase;
     this.counters = counters;
+    this.includeCounters = true;
   }
   
   public String getTaskId() { return taskid; }
@@ -181,6 +183,15 @@ abstract class TaskStatus implements Writable, Cloneable {
     }
     this.phase = phase; 
   }
+  
+  public boolean getIncludeCounters() {
+    return includeCounters; 
+  }
+  
+  public void setIncludeCounters(boolean send) {
+    includeCounters = send;
+  }
+  
   /**
    * Get task's counters.
    */
@@ -281,7 +292,10 @@ abstract class TaskStatus implements Writable, Cloneable {
     WritableUtils.writeEnum(out, phase);
     out.writeLong(startTime);
     out.writeLong(finishTime);
-    counters.write(out);
+    out.writeBoolean(includeCounters);
+    if (includeCounters) {
+      counters.write(out);
+    }
   }
 
   public void readFields(DataInput in) throws IOException {
@@ -294,7 +308,10 @@ abstract class TaskStatus implements Writable, Cloneable {
     this.startTime = in.readLong(); 
     this.finishTime = in.readLong(); 
     counters = new Counters();
-    counters.readFields(in);
+    this.includeCounters = in.readBoolean();
+    if (includeCounters) {
+      counters.readFields(in);
+    }
   }
   
   //////////////////////////////////////////////////////////////////////////////

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

@@ -958,12 +958,25 @@ public class TaskTracker
     return State.NORMAL;
   }
 
+  private long previousUpdate = 0;
+
   /**
    * Build and transmit the heart beat to the JobTracker
    * @return false if the tracker was unknown
    * @throws IOException
    */
   private HeartbeatResponse transmitHeartBeat() throws IOException {
+    // Send Counters in the status once every COUNTER_UPDATE_INTERVAL
+    long now = System.currentTimeMillis();
+    boolean sendCounters;
+    if (now > (previousUpdate + COUNTER_UPDATE_INTERVAL)) {
+      sendCounters = true;
+      previousUpdate = now;
+    }
+    else {
+      sendCounters = false;
+    }
+
     // 
     // Check if the last heartbeat got through... 
     // if so then build the heartbeat information for the JobTracker;
@@ -972,7 +985,9 @@ public class TaskTracker
     if (status == null) {
       synchronized (this) {
         status = new TaskTrackerStatus(taskTrackerName, localHostname, 
-                                       httpPort, cloneAndResetRunningTaskStatuses(), 
+                                       httpPort, 
+                                       cloneAndResetRunningTaskStatuses(
+                                         sendCounters), 
                                        failures, 
                                        maxCurrentMapTasks,
                                        maxCurrentReduceTasks); 
@@ -2060,10 +2075,16 @@ public class TaskTracker
     return taskTrackerName;
   }
     
-  private synchronized List<TaskStatus> cloneAndResetRunningTaskStatuses() {
+  private synchronized List<TaskStatus> cloneAndResetRunningTaskStatuses(
+                                          boolean sendCounters) {
     List<TaskStatus> result = new ArrayList<TaskStatus>(runningTasks.size());
     for(TaskInProgress tip: runningTasks.values()) {
       TaskStatus status = tip.getStatus();
+      status.setIncludeCounters(sendCounters);
+      // send counters for finished or failed tasks.
+      if (status.getRunState() != TaskStatus.State.RUNNING) {
+        status.setIncludeCounters(true);
+      }
       result.add((TaskStatus)status.clone());
       status.clearStatus();
     }

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

@@ -37,8 +37,9 @@ interface TaskUmbilicalProtocol extends VersionedProtocol {
    *         org.apache.hadoop.mapred.TaskStatus.Phase, Counters) 
    *         with {@link #statusUpdate(String, TaskStatus)}
    * Version 5 changed counters representation for HADOOP-2248
+   * Version 6 changes the TaskStatus representation for HADOOP-2208
    * */
-  public static final long versionID = 5L;
+  public static final long versionID = 6L;
   
   /** Called when a child task process starts, to get its task.*/
   Task getTask(String taskid) throws IOException;