commit db8f1bf2ddc04574c8a7a1b9844ddfdacafcfa4c

Author: Hemanth Yamijala <yhemanth@friendchild-lm.(none)>
Date:   Sun Feb 14 14:50:27 2010 +0530

    HADOOP:2141 from https://issues.apache.org/jira/secure/attachment/12435253/hadoop-2141-yahoo-v1.4.8.patch (only test related changes)
    
    +++ b/YAHOO-CHANGES.txt
    +    HADOOP-2141. Backport changes made in the original JIRA to aid
    +    fast unit tests in Map/Reduce. (Amar Kamat via yhemanth)
    +


git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-0.20-security-patches@1077167 13f79535-47bb-0310-9956-ffa450edef68

src/contrib/capacity-scheduler/src/test/org/apache/hadoop/mapred/TestCapacityScheduler.java

@@ -338,7 +338,8 @@ public class TestCapacityScheduler extends TestCase {
     
     FakeTaskInProgress(JobID jId, JobConf jobConf, Task t, 
         boolean isMap, FakeJobInProgress job) {
-      super(jId, "", JobSplit.EMPTY_TASK_SPLIT, null, jobConf, job, 0, 1);
+      super(jId, "", JobSplit.EMPTY_TASK_SPLIT, job.jobtracker, jobConf, job, 
+            0, 1);
       this.isMap = isMap;
       this.fakeJob = job;
       activeTasks = new TreeMap<TaskAttemptID, String>();

src/contrib/fairscheduler/src/java/org/apache/hadoop/mapred/FairScheduler.java

@@ -87,15 +87,6 @@ public class FairScheduler extends TaskScheduler {
     double reduceFairShare = 0; // Fair share of reduce slots at last update
   }
   
-  /**
-   * A clock class - can be mocked out for testing.
-   */
-  static class Clock {
-    long getTime() {
-      return System.currentTimeMillis();
-    }
-  }
-  
   public FairScheduler() {
     this(new Clock(), true);
   }

src/contrib/fairscheduler/src/test/org/apache/hadoop/mapred/TestFairScheduler.java

@@ -35,6 +35,7 @@ import org.apache.hadoop.mapred.JobStatus;
 import org.apache.hadoop.mapred.FairScheduler.JobInfo;
 import org.apache.hadoop.mapreduce.server.jobtracker.TaskTracker;
 import org.apache.hadoop.mapreduce.split.JobSplit;
+import org.apache.hadoop.mapred.UtilsForTests.FakeClock;
 
 public class TestFairScheduler extends TestCase {
   final static String TEST_DIR = new File(System.getProperty("test.build.data",
@@ -236,19 +237,6 @@ public class TestFairScheduler extends TestCase {
     }
   }
   
-  protected class FakeClock extends FairScheduler.Clock {
-    private long time = 0;
-    
-    public void advance(long millis) {
-      time += millis;
-    }
-
-    @Override
-    long getTime() {
-      return time;
-    }
-  }
-  
   protected JobConf conf;
   protected FairScheduler scheduler;
   private FakeTaskTrackerManager taskTrackerManager;

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

@@ -0,0 +1,28 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.mapred;
+
+/**
+ * A clock class - can be mocked out for testing.
+ */
+class Clock {
+  long getTime() {
+    return System.currentTimeMillis();
+  }
+}

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

@@ -111,6 +111,15 @@ public class ClusterStatus implements Writable {
   }
 
   /**
+   * Construct a new cluster status.
+   * @param trackers no. of tasktrackers in the cluster
+   * @param blacklists no of blacklisted task trackers in the cluster
+   * @param ttExpiryInterval the tasktracker expiry interval
+   * @param maps no. of currently running map-tasks in the cluster
+   * @param reduces no. of currently running reduce-tasks in the cluster
+   * @param maxMaps the maximum no. of map tasks in the cluster
+   * @param maxReduces the maximum no. of reduce tasks in the cluster
+   * @param state the {@link JobTracker.State} of the <code>JobTracker</code>
    * @param numDecommissionedNodes number of decommission trackers
    */
   ClusterStatus(int trackers, int blacklists, long ttExpiryInterval, 
@@ -151,6 +160,15 @@ public class ClusterStatus implements Writable {
   }
 
   /**
+   * Construct a new cluster status. 
+   * @param activeTrackers active tasktrackers in the cluster
+   * @param blacklistedTrackers blacklisted tasktrackers in the cluster
+   * @param ttExpiryInterval the tasktracker expiry interval
+   * @param maps no. of currently running map-tasks in the cluster
+   * @param reduces no. of currently running reduce-tasks in the cluster
+   * @param maxMaps the maximum no. of map tasks in the cluster
+   * @param maxReduces the maximum no. of reduce tasks in the cluster
+   * @param state the {@link JobTracker.State} of the <code>JobTracker</code>
    * @param numDecommissionNodes number of decommission trackers
    */
   ClusterStatus(Collection<String> activeTrackers, 

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

@@ -21,6 +21,8 @@ import java.io.IOException;
 import java.security.PrivilegedExceptionAction;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.IdentityHashMap;
@@ -286,6 +288,17 @@ class JobInProgress {
     this.anyCacheLevel = this.maxLevel+1;
     this.jobtracker = tracker;
     this.restartCount = 0;
+    hasSpeculativeMaps = conf.getMapSpeculativeExecution();
+    hasSpeculativeReduces = conf.getReduceSpeculativeExecution();
+    this.nonLocalMaps = new LinkedList<TaskInProgress>();
+    this.nonLocalRunningMaps = new LinkedHashSet<TaskInProgress>();
+    this.runningMapCache = new IdentityHashMap<Node, Set<TaskInProgress>>();
+    this.nonRunningReduces = new LinkedList<TaskInProgress>();
+    this.runningReduces = new LinkedHashSet<TaskInProgress>();
+    this.resourceEstimator = new ResourceEstimator(this);
+    this.status = new JobStatus(jobid, 0.0f, 0.0f, JobStatus.PREP);
+    this.taskCompletionEvents = new ArrayList<TaskCompletionEvent>
+      (numMapTasks + numReduceTasks + 10);
     try {
       this.userUGI = UserGroupInformation.getCurrentUser();
     } catch (IOException ie){
@@ -319,7 +332,7 @@ class JobInProgress {
     this.jobtracker = jobtracker;
     this.status = new JobStatus(jobId, 0.0f, 0.0f, JobStatus.PREP);
     this.jobtracker.getInstrumentation().addPrepJob(conf, jobId);
-    this.startTime = System.currentTimeMillis();
+    this.startTime = jobtracker.getClock().getTime();
     status.setStartTime(startTime);
     this.localFs = jobtracker.getLocalFileSystem();
 
@@ -430,7 +443,7 @@ class JobInProgress {
     
     for (int i = 0; i < splits.length; i++) {
       String[] splitLocations = splits[i].getLocations();
-      if (splitLocations.length == 0) {
+      if (splitLocations == null || splitLocations.length == 0) {
         nonLocalMaps.add(maps[i]);
         continue;
       }
@@ -588,7 +601,7 @@ class JobInProgress {
     }
         
     // set the launch time
-    this.launchTime = System.currentTimeMillis();
+    this.launchTime = jobtracker.getClock().getTime();
 
     //
     // Create reduce tasks
@@ -1586,7 +1599,7 @@ class JobInProgress {
     Map<TaskTracker, FallowSlotInfo> map =
       (type == TaskType.MAP) ? trackersReservedForMaps : trackersReservedForReduces;
     
-    long now = System.currentTimeMillis();
+    long now = jobtracker.getClock().getTime();
     
     FallowSlotInfo info = map.get(taskTracker);
     int reservedSlots = 0;
@@ -1632,7 +1645,7 @@ class JobInProgress {
       return;
     }
     
-    long now = System.currentTimeMillis();
+    long now = jobtracker.getClock().getTime();
 
     Enum<Counter> counter = 
       (type == TaskType.MAP) ? 
@@ -1720,7 +1733,7 @@ class JobInProgress {
     String[] splitLocations = tip.getSplitLocations();
 
     // Remove the TIP from the list for running non-local maps
-    if (splitLocations.length == 0) {
+    if (splitLocations == null || splitLocations.length == 0) {
       nonLocalRunningMaps.remove(tip);
       return;
     }
@@ -1760,7 +1773,7 @@ class JobInProgress {
    * Adds a map tip to the list of running maps.
    * @param tip the tip that needs to be scheduled as running
    */
-  private synchronized void scheduleMap(TaskInProgress tip) {
+  protected synchronized void scheduleMap(TaskInProgress tip) {
     
     if (runningMapCache == null) {
       LOG.warn("Running cache for maps is missing!! " 
@@ -1770,7 +1783,7 @@ class JobInProgress {
     String[] splitLocations = tip.getSplitLocations();
 
     // Add the TIP to the list of non-local running TIPs
-    if (splitLocations.length == 0) {
+    if (splitLocations == null || splitLocations.length == 0) {
       nonLocalRunningMaps.add(tip);
       return;
     }
@@ -1795,7 +1808,7 @@ class JobInProgress {
    * Adds a reduce tip to the list of running reduces
    * @param tip the tip that needs to be scheduled as running
    */
-  private synchronized void scheduleReduce(TaskInProgress tip) {
+  protected synchronized void scheduleReduce(TaskInProgress tip) {
     if (runningReduces == null) {
       LOG.warn("Running cache for reducers missing!! "
                + "Job details are missing.");
@@ -1822,7 +1835,7 @@ class JobInProgress {
     String[] splitLocations = tip.getSplitLocations();
 
     // Add the TIP in the front of the list for non-local non-running maps
-    if (splitLocations.length == 0) {
+    if (splitLocations == null || splitLocations.length == 0) {
       nonLocalMaps.add(0, tip);
       return;
     }
@@ -2106,7 +2119,7 @@ class JobInProgress {
     // 
  
     if (hasSpeculativeMaps) {
-      long currentTime = System.currentTimeMillis();
+      long currentTime = jobtracker.getClock().getTime();
 
       // 1. Check bottom up for speculative tasks from the running cache
       if (node != null) {
@@ -2214,7 +2227,7 @@ class JobInProgress {
     // 2. check for a reduce tip to be speculated
     if (hasSpeculativeReduces) {
       tip = findSpeculativeTask(runningReduces, tts, avgProgress, 
-                                System.currentTimeMillis(), false);
+                                jobtracker.getClock().getTime(), false);
       if (tip != null) {
         scheduleReduce(tip);
         return tip.getIdWithinJob();
@@ -2435,7 +2448,7 @@ class JobInProgress {
       if (reduces.length == 0) {
         this.status.setReduceProgress(1.0f);
       }
-      this.finishTime = System.currentTimeMillis();
+      this.finishTime = jobtracker.getClock().getTime();
       LOG.info("Job " + this.status.getJobID() + 
                " has completed successfully.");
       
@@ -2460,7 +2473,7 @@ class JobInProgress {
   private synchronized void terminateJob(int jobTerminationState) {
     if ((status.getRunState() == JobStatus.RUNNING) ||
         (status.getRunState() == JobStatus.PREP)) {
-      this.finishTime = System.currentTimeMillis();
+      this.finishTime = jobtracker.getClock().getTime();
       this.status.setMapProgress(1.0f);
       this.status.setReduceProgress(1.0f);
       this.status.setCleanupProgress(1.0f);
@@ -2848,10 +2861,10 @@ class JobInProgress {
     // update the actual start-time of the attempt
     TaskStatus oldStatus = tip.getTaskStatus(taskid); 
     long startTime = oldStatus == null
-                     ? System.currentTimeMillis()
+                     ? jobtracker.getClock().getTime()
                      : oldStatus.getStartTime();
     status.setStartTime(startTime);
-    status.setFinishTime(System.currentTimeMillis());
+    status.setFinishTime(jobtracker.getClock().getTime());
     boolean wasComplete = tip.isComplete();
     updateTaskStatus(tip, status);
     boolean isComplete = tip.isComplete();

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

@@ -191,6 +191,8 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
   // system files should have 700 permission
   final static FsPermission SYSTEM_FILE_PERMISSION =
     FsPermission.createImmutable((short) 0700); // rwx------
+  
+  private Clock clock;
 
   private TokenStorage tokenStorage;
   private final JobTokenSecretManager jobTokenSecretManager
@@ -225,6 +227,10 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
   private int nextJobId = 1;
 
   public static final Log LOG = LogFactory.getLog(JobTracker.class);
+  
+  public Clock getClock() {
+    return clock;
+  }
     
   /**
    * Start the JobTracker with given configuration.
@@ -314,7 +320,7 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
         try {
           // Every 3 minutes check for any tasks that are overdue
           Thread.sleep(TASKTRACKER_EXPIRY_INTERVAL/3);
-          long now = System.currentTimeMillis();
+          long now = clock.getTime();
           LOG.debug("Starting launching task sweep");
           synchronized (JobTracker.this) {
             synchronized (launchingTasks) {
@@ -368,7 +374,7 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
     public void addNewTask(TaskAttemptID taskName) {
       synchronized (launchingTasks) {
         launchingTasks.put(taskName, 
-                           System.currentTimeMillis());
+                           clock.getTime());
       }
     }
       
@@ -412,7 +418,7 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
           synchronized (JobTracker.this) {
             synchronized (taskTrackers) {
               synchronized (trackerExpiryQueue) {
-                long now = System.currentTimeMillis();
+                long now = clock.getTime();
                 TaskTrackerStatus leastRecent = null;
                 while ((trackerExpiryQueue.size() > 0) &&
                        (leastRecent = trackerExpiryQueue.first()) != null &&
@@ -548,7 +554,7 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
         try {
           Thread.sleep(RETIRE_JOB_CHECK_INTERVAL);
           List<JobInProgress> retiredJobs = new ArrayList<JobInProgress>();
-          long now = System.currentTimeMillis();
+          long now = clock.getTime();
           long retireBefore = now - RETIRE_JOB_INTERVAL;
 
           synchronized (jobs) {
@@ -636,9 +642,9 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
     private boolean isHealthy;
     private HashMap<ReasonForBlackListing, String>rfbMap;
     
-    FaultInfo() {
+    FaultInfo(long time) {
       numFaults = 0;
-      lastUpdated = System.currentTimeMillis();
+      lastUpdated = time;
       blacklisted = false;
       rfbMap = new  HashMap<ReasonForBlackListing, String>();
     }
@@ -730,7 +736,7 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
         int numFaults = fi.getFaultCount();
         ++numFaults;
         fi.setFaultCount(numFaults);
-        fi.setLastUpdated(System.currentTimeMillis());
+        fi.setLastUpdated(clock.getTime());
         if (exceedsFaults(fi)) {
           LOG.info("Adding " + hostName + " to the blacklist"
               + " across all jobs");
@@ -814,7 +820,7 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
         boolean createIfNeccessary) {
       FaultInfo fi = potentiallyFaultyTrackers.get(hostName);
       if (fi == null && createIfNeccessary) {
-        fi = new FaultInfo();
+        fi = new FaultInfo(clock.getTime());
         potentiallyFaultyTrackers.put(hostName, fi);
       }
       return fi;
@@ -1362,7 +1368,7 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
       TaskTrackerStatus ttStatus = 
         new TaskTrackerStatus(trackerName, trackerHostName, port, ttStatusList, 
                               0 , 0, 0);
-      ttStatus.setLastSeen(System.currentTimeMillis());
+      ttStatus.setLastSeen(clock.getTime());
 
       synchronized (JobTracker.this) {
         synchronized (taskTrackers) {
@@ -1688,7 +1694,7 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
         }
       }
 
-      long recoveryStartTime = System.currentTimeMillis();
+      long recoveryStartTime = clock.getTime();
 
       // II. Recover each job
       idIter = jobsToRecover.iterator();
@@ -1745,14 +1751,14 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
         }
       }
 
-      recoveryDuration = System.currentTimeMillis() - recoveryStartTime;
+      recoveryDuration = clock.getTime() - recoveryStartTime;
       hasRecovered = true;
 
       // III. Finalize the recovery
       synchronized (trackerExpiryQueue) {
         // Make sure that the tracker statuses in the expiry-tracker queue
         // are updated
-        long now = System.currentTimeMillis();
+        long now = clock.getTime();
         int size = trackerExpiryQueue.size();
         for (int i = 0; i < size ; ++i) {
           // Get the first tasktracker
@@ -1947,6 +1953,12 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
   
   JobTracker(final JobConf conf, String identifier) 
   throws IOException, InterruptedException {   
+    this(conf, identifier, new Clock());
+  }
+  
+  JobTracker(final JobConf conf, String identifier, Clock clock) 
+  throws IOException, InterruptedException { 
+    this.clock = clock;
     // find the owner of the process
     // get the desired principal to load
     String keytabFilename = conf.get(JT_KEYTAB_FILE);
@@ -2064,7 +2076,7 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
     InetSocketAddress infoSocAddr = NetUtils.createSocketAddr(infoAddr);
     String infoBindAddress = infoSocAddr.getHostName();
     int tmpInfoPort = infoSocAddr.getPort();
-    this.startTime = System.currentTimeMillis();
+    this.startTime = clock.getTime();
     infoServer = new HttpServer("job", infoBindAddress, tmpInfoPort, 
         tmpInfoPort == 0, conf);
     infoServer.setAttribute("job.tracker", this);
@@ -2598,7 +2610,7 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
     final JobTrackerInstrumentation metrics = getInstrumentation();
     metrics.finalizeJob(conf, id);
     
-    long now = System.currentTimeMillis();
+    long now = clock.getTime();
     
     // mark the job for cleanup at all the trackers
     addJobForCleanup(id);
@@ -2979,7 +2991,7 @@ public class JobTracker implements MRConstants, InterTrackerProtocol,
 
     // First check if the last heartbeat response got through
     String trackerName = status.getTrackerName();
-    long now = System.currentTimeMillis();
+    long now = clock.getTime();
     boolean isBlacklisted = false;
     if (restarted) {
       faultyTrackers.markTrackerHealthy(status.getHost());

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

@@ -226,7 +226,7 @@ class TaskInProgress {
    * Initialization common to Map and Reduce
    */
   void init(JobID jobId) {
-    this.startTime = System.currentTimeMillis();
+    this.startTime = jobtracker.getClock().getTime();
     this.id = new TaskID(jobId, isMapTask(), partition);
     this.skipping = startSkipping();
   }
@@ -635,7 +635,7 @@ class TaskInProgress {
 
       // tasktracker went down and failed time was not reported. 
       if (0 == status.getFinishTime()){
-        status.setFinishTime(System.currentTimeMillis());
+        status.setFinishTime(jobtracker.getClock().getTime());
       }
     }
 
@@ -740,7 +740,7 @@ class TaskInProgress {
     //
 
     this.completes++;
-    this.execFinishTime = System.currentTimeMillis();
+    this.execFinishTime = jobtracker.getClock().getTime();
     recomputeProgress();
     
   }
@@ -779,7 +779,7 @@ class TaskInProgress {
     }
     this.failed = true;
     killed = true;
-    this.execFinishTime = System.currentTimeMillis();
+    this.execFinishTime = jobtracker.getClock().getTime();
     recomputeProgress();
   }
 
@@ -903,7 +903,7 @@ class TaskInProgress {
   public Task getTaskToRun(String taskTracker) throws IOException {
     if (0 == execStartTime){
       // assume task starts running now
-      execStartTime = System.currentTimeMillis();
+      execStartTime = jobtracker.getClock().getTime();
     }
 
     // Create the 'taskid'; do not count the 'killed' tasks against the job!

src/test/findbugsExcludeFile.xml

@@ -89,4 +89,16 @@
        <Class name="org.apache.hadoop.mapred.FileOutputCommitter" />
        <Bug pattern="NM_WRONG_PACKAGE_INTENTIONAL" />
      </Match>
+    <!--
+       JobTracker's static variables should be ignored
+     -->
+     <Match>
+       <Class name="org.apache.hadoop.mapred.JobTracker" />
+       <Or>
+       <Field name="RETIRE_JOB_INTERVAL" />
+       <Field name="TASKTRACKER_EXPIRY_INTERVAL" />
+       <Field name="RETIRE_JOB_CHECK_INTERVAL" />
+       </Or>
+       <Bug pattern="ST_WRITE_TO_STATIC_FROM_INSTANCE_METHOD" />
+     </Match>
 </FindBugsFilter>

src/test/org/apache/hadoop/mapred/TestResourceEstimation.java

@@ -48,7 +48,7 @@ public class TestResourceEstimation extends TestCase {
       JobSplit.TaskSplitMetaInfo split =
           new JobSplit.TaskSplitMetaInfo(new String[0], 0, 0);
       TaskInProgress tip = 
-        new TaskInProgress(jid, "", split, null, jc, jip, 0, 1);
+        new TaskInProgress(jid, "", split, jip.jobtracker, jc, jip, 0, 1);
       re.updateWithCompletedTask(ts, tip);
     }
     assertEquals(2* singleMapOutputSize, re.getEstimatedMapOutputSize());
@@ -86,7 +86,7 @@ public class TestResourceEstimation extends TestCase {
               new JobSplit.TaskSplitMetaInfo(new String[0], 0,
                                            singleMapInputSize);
       TaskInProgress tip = 
-        new TaskInProgress(jid, "", split, null, jc, jip, 0, 1);
+        new TaskInProgress(jid, "", split, jip.jobtracker, jc, jip, 0, 1);
       re.updateWithCompletedTask(ts, tip);
     }
     
@@ -99,7 +99,7 @@ public class TestResourceEstimation extends TestCase {
     JobSplit.TaskSplitMetaInfo split =
         new JobSplit.TaskSplitMetaInfo(new String[0], 0, 0);
     TaskInProgress tip = 
-      new TaskInProgress(jid, "", split, null, jc, jip, 0, 1);
+      new TaskInProgress(jid, "", split, jip.jobtracker, jc, jip, 0, 1);
     re.updateWithCompletedTask(ts, tip);
     
     long expectedTotalMapOutSize = (singleMapOutputSize*11) * 

src/test/org/apache/hadoop/mapred/UtilsForTests.java

@@ -30,7 +30,6 @@ import java.util.Properties;
 import org.apache.commons.logging.LogFactory;
 import org.apache.commons.logging.Log;
 import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.examples.RandomWriter;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.hdfs.DFSTestUtil;
@@ -678,6 +677,19 @@ public class UtilsForTests {
     return job;
   }
 
+  static class FakeClock extends Clock {
+    long time = 0;
+    
+    public void advance(long millis) {
+      time += millis;
+    }
+     
+      @Override
+      long getTime() {
+        return time;
+      }
+    }
+  
   // Mapper that fails
   static class FailMapper extends MapReduceBase implements
       Mapper<WritableComparable, Writable, WritableComparable, Writable> {