You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by vi...@apache.org on 2012/11/10 03:35:38 UTC
svn commit: r1407717 - in /hadoop/common/branches/branch-1: CHANGES.txt
src/mapred/org/apache/hadoop/mapred/TaskTracker.java
src/test/org/apache/hadoop/mapred/TestTaskTrackerActionCleanup.java
Author: vinodkv
Date: Sat Nov 10 02:35:36 2012
New Revision: 1407717
URL: http://svn.apache.org/viewvc?rev=1407717&view=rev
Log:
MAPREDUCE-4749. Fixed a bug in TaskTracker because of which kill-actions get delayed progressively. Contributed by Arpit Gupta.
Added:
hadoop/common/branches/branch-1/src/test/org/apache/hadoop/mapred/TestTaskTrackerActionCleanup.java
Modified:
hadoop/common/branches/branch-1/CHANGES.txt
hadoop/common/branches/branch-1/src/mapred/org/apache/hadoop/mapred/TaskTracker.java
Modified: hadoop/common/branches/branch-1/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1/CHANGES.txt?rev=1407717&r1=1407716&r2=1407717&view=diff
==============================================================================
--- hadoop/common/branches/branch-1/CHANGES.txt (original)
+++ hadoop/common/branches/branch-1/CHANGES.txt Sat Nov 10 02:35:36 2012
@@ -342,6 +342,9 @@ Release 1.1.1 - Unreleased
MAPREDUCE-4782. NLineInputFormat skips first line of last InputSplit
(Mark Fuhs via bobby)
+ MAPREDUCE-4749. Fixed a bug in TaskTracker because of which kill-actions get
+ delayed progressively. (Arpit Gupta via vinodkv)
+
Release 1.1.0 - 2012.09.28
INCOMPATIBLE CHANGES
Modified: hadoop/common/branches/branch-1/src/mapred/org/apache/hadoop/mapred/TaskTracker.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1/src/mapred/org/apache/hadoop/mapred/TaskTracker.java?rev=1407717&r1=1407716&r2=1407717&view=diff
==============================================================================
--- hadoop/common/branches/branch-1/src/mapred/org/apache/hadoop/mapred/TaskTracker.java (original)
+++ hadoop/common/branches/branch-1/src/mapred/org/apache/hadoop/mapred/TaskTracker.java Sat Nov 10 02:35:36 2012
@@ -44,6 +44,7 @@ import java.util.TreeMap;
import java.util.Vector;
import java.util.Map.Entry;
import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.regex.Pattern;
@@ -440,50 +441,105 @@ public class TaskTracker implements MRCo
}
/**
- * A list of tips that should be cleaned up.
+ * A list of clean-up actions
*/
- private BlockingQueue<TaskTrackerAction> tasksToCleanup =
- new LinkedBlockingQueue<TaskTrackerAction>();
-
+ // No ConcurrentHashSet :(
+ ConcurrentHashMap<String, String> allCleanupActions =
+ new ConcurrentHashMap<String, String>();
+ BlockingQueue<TaskTrackerAction> activeCleanupActions =
+ new LinkedBlockingQueue<TaskTrackerAction>();
+ List<TaskTrackerAction> inactiveCleanupActions =
+ new ArrayList<TaskTrackerAction>();
+
+ /*
+ * Add action to clean up queue. Check to make sure action is not already
+ * present before adding it to the queue.
+ */
+ void addActionToCleanup(TaskTrackerAction action) throws InterruptedException {
+
+ String actionId = getIdForCleanUpAction(action);
+
+ // add the action to the queue only if its not added in the first place
+ String previousActionId = allCleanupActions.putIfAbsent(actionId, actionId);
+ if (previousActionId != null) {
+ return;
+ } else {
+ activeCleanupActions.put(action);
+ }
+ }
+
+ /*
+ * Get the id for a given clean up action. It is expected to be either a
+ * KillJobAction or KillTaskAction
+ */
+ private String getIdForCleanUpAction(TaskTrackerAction action) {
+ String actionId = null;
+ // get the id of the action
+ if (action instanceof KillJobAction) {
+ actionId = ((KillJobAction) action).getJobID().toString();
+ } else if (action instanceof KillTaskAction) {
+ actionId = ((KillTaskAction) action).getTaskID().toString();
+ }
+ // Assuming actionId is not null as all clean-up actions are either KillJob
+ // or KillTask
+ return actionId;
+ }
+
/**
* A daemon-thread that pulls tips off the list of things to cleanup.
*/
- private Thread taskCleanupThread =
- new Thread(new Runnable() {
- public void run() {
- while (true) {
- try {
- TaskTrackerAction action = tasksToCleanup.take();
- if( !checkJobStatusAndWait(action) ) {
- //If job is still localizing, put it back into the queue and
- //pick another one in the next iteration
- StringBuffer sb = new StringBuffer("Cleanup for job ");
- if (action instanceof KillJobAction) {
- sb.append(((KillJobAction)action).getJobID());
- } else if (action instanceof KillTaskAction) {
- sb.append(((KillTaskAction)action).getTaskID().getJobID() +
- ", task " + ((KillTaskAction)action).getTaskID());
- }
- sb.append(" was postponed after waiting for 5 seconds.");
- LOG.info(sb);
- tasksToCleanup.put(action);
- continue;
- }
- if (action instanceof KillJobAction) {
- purgeJob((KillJobAction) action);
- } else if (action instanceof KillTaskAction) {
- processKillTaskAction((KillTaskAction) action);
- } else {
- LOG.error("Non-delete action given to cleanup thread: "
- + action);
- }
- } catch (Throwable except) {
- LOG.warn(StringUtils.stringifyException(except));
- }
- }
+ private Thread taskCleanupThread = new Thread(new Runnable() {
+ public void run() {
+ while (true) {
+ try {
+ taskCleanUp();
+ } catch (Throwable except) {
+ LOG.warn(StringUtils.stringifyException(except));
}
- }, "taskCleanup");
+ }
+ }
+ }, "taskCleanup");
+
+ void taskCleanUp() throws InterruptedException, IOException {
+ // process all the localizing tasks and move to the clean up queue
+ // if the job is not localizing.
+ Iterator<TaskTrackerAction> itr = inactiveCleanupActions.iterator();
+ while (itr.hasNext()) {
+ TaskTrackerAction action = itr.next();
+ if (!isJobLocalizing(action)) {
+ activeCleanupActions.put(action);
+ itr.remove();
+ }
+ }
+
+ // if the tasks to clean is empty better sleep for 2 seconds and
+ // re process
+ if (activeCleanupActions.isEmpty()) {
+ Thread.sleep(2000);
+ return;
+ }
+ TaskTrackerAction action = activeCleanupActions.take();
+ String actionId = getIdForCleanUpAction(action);
+ if (isJobLocalizing(action)) {
+ // If job is still localizing, put it back into the queue and
+ // pick another one in the next iteration
+ LOG.info("Cleanup for id " + actionId + " skipped as its localizing.");
+ inactiveCleanupActions.add(action);
+ return;
+ } else {
+ // remove the action from the hash as its being processed.
+ allCleanupActions.remove(actionId);
+ }
+ if (action instanceof KillJobAction) {
+ purgeJob((KillJobAction) action);
+ } else if (action instanceof KillTaskAction) {
+ processKillTaskAction((KillTaskAction) action);
+ } else {
+ LOG.error("Non-delete action given to cleanup thread: " + action);
+ }
+ }
+
void processKillTaskAction(KillTaskAction killAction) throws IOException {
TaskInProgress tip;
synchronized (TaskTracker.this) {
@@ -494,21 +550,19 @@ public class TaskTracker implements MRCo
}
/**
- * Wait until the job has completed localizing if it was doing so.
+ * Check if the job for the given action is localizing or not.
* @param action The command received from the JobTracker
- * @return true if the wait was successful and the task is not localizing
- * anymore. false if after 5 seconds, the job was still localizing
- * @throws InterruptedException
+ * @return true if job is localizing and false otherwise.
*/
- private boolean checkJobStatusAndWait(TaskTrackerAction action)
- throws InterruptedException {
+ private boolean isJobLocalizing(TaskTrackerAction action)
+ {
JobID jobId = null;
if (action instanceof KillJobAction) {
jobId = ((KillJobAction)action).getJobID();
} else if (action instanceof KillTaskAction) {
jobId = ((KillTaskAction)action).getTaskID().getJobID();
} else {
- return true;
+ return false;
}
RunningJob rjob = null;
synchronized (runningJobs) {
@@ -516,11 +570,10 @@ public class TaskTracker implements MRCo
}
if (rjob != null) {
synchronized (rjob) {
- rjob.wait(5000);
- return !rjob.localizing;
+ return rjob.localizing;
}
}
- return true;
+ return false;
}
public TaskController getTaskController() {
@@ -1819,7 +1872,7 @@ public class TaskTracker implements MRCo
commitResponses.add(commitAction.getTaskID());
}
} else {
- tasksToCleanup.put(action);
+ addActionToCleanup(action);
}
}
}
@@ -3841,7 +3894,7 @@ public class TaskTracker implements MRCo
* @return has this task tracker finished and cleaned up all of its tasks?
*/
public synchronized boolean isIdleAndClean() {
- return tasks.isEmpty() && tasksToCleanup.isEmpty();
+ return tasks.isEmpty() && allCleanupActions.isEmpty();
}
/**
Added: hadoop/common/branches/branch-1/src/test/org/apache/hadoop/mapred/TestTaskTrackerActionCleanup.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/branch-1/src/test/org/apache/hadoop/mapred/TestTaskTrackerActionCleanup.java?rev=1407717&view=auto
==============================================================================
--- hadoop/common/branches/branch-1/src/test/org/apache/hadoop/mapred/TestTaskTrackerActionCleanup.java (added)
+++ hadoop/common/branches/branch-1/src/test/org/apache/hadoop/mapred/TestTaskTrackerActionCleanup.java Sat Nov 10 02:35:36 2012
@@ -0,0 +1,388 @@
+/**
+ * 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;
+
+import java.io.IOException;
+
+import org.apache.hadoop.mapred.TaskTracker.RunningJob;
+import org.junit.Before;
+import org.junit.Test;
+
+import junit.framework.TestCase;
+
+public class TestTaskTrackerActionCleanup extends TestCase {
+ String jtIdentifier = "201210122331";
+ TestTaskTracker tt = null;
+
+ @Before
+ public void setUp() {
+ tt = new TestTaskTracker();
+ }
+
+ @Test
+ // check duplicate entries do not make it to the queue
+ public void testDuplicateEntries() throws InterruptedException {
+
+ KillJobAction action = new KillJobAction();
+ // add the action twice
+ tt.addActionToCleanup(action);
+ tt.addActionToCleanup(action);
+
+ checkItemCountInQueue(tt, 1, 1, 0);
+ }
+
+ @Test
+ // test to make sure all tasks with localizing jobs are added to another queue
+ public void testLocalizingJobActions() throws InterruptedException,
+ IOException {
+
+ // job and attempt ids
+ JobID jobId1 = new JobID(jtIdentifier, 1);
+ JobID jobId2 = new JobID(jtIdentifier, 2);
+
+ TaskAttemptID taskAttemptId1 = new TaskAttemptID(jtIdentifier, 3, true, 1,
+ 1);
+ TaskAttemptID taskAttemptId2 = new TaskAttemptID(jtIdentifier, 4, true, 1,
+ 1);
+
+ // job actions which is localizing
+ KillJobAction jAction1 = new KillJobAction(jobId1);
+ RunningJob rjob1 = new RunningJob(jAction1.getJobID());
+ rjob1.localizing = true;
+ // add the job to the runningJobs tree
+ tt.runningJobs.put(jobId1, rjob1);
+ tt.addActionToCleanup(jAction1);
+
+ KillJobAction jAction2 = new KillJobAction(jobId2);
+ RunningJob rjob2 = new RunningJob(jAction2.getJobID());
+ rjob2.localizing = true;
+ // add the job to the runningJobs tree
+ tt.runningJobs.put(jobId2, rjob2);
+ tt.addActionToCleanup(jAction2);
+
+ // task action which is localizing
+ KillTaskAction tAction1 = new KillTaskAction(taskAttemptId1);
+ RunningJob rjob3 = new RunningJob(tAction1.getTaskID().getJobID());
+ // add the job to the runningJobs tree
+ tt.runningJobs.put(rjob3.getJobID(), rjob3);
+ rjob3.localizing = true;
+ tt.addActionToCleanup(tAction1);
+
+ KillTaskAction tAction2 = new KillTaskAction(taskAttemptId2);
+ RunningJob rjob4 = new RunningJob(tAction2.getTaskID().getJobID());
+ rjob4.localizing = true;
+ // add the job to the runningJobs tree
+ tt.runningJobs.put(rjob4.getJobID(), rjob4);
+ tt.addActionToCleanup(tAction2);
+
+ // before the task clean up test the queue
+ checkItemCountInQueue(tt, 4, 4, 0);
+
+ // now run the task clean up
+ tt.taskCleanUp();
+
+ // after the one round the first task gets moved to localizing list
+ checkItemCountInQueue(tt, 4, 3, 1);
+ // now run the task clean up
+ tt.taskCleanUp();
+
+ // after the another round the 2nd task gets moved to localizing list
+ checkItemCountInQueue(tt, 4, 2, 2);
+
+ // now run the task clean up
+ tt.taskCleanUp();
+
+ // after the another round the 3rd task gets moved to localizing list
+ checkItemCountInQueue(tt, 4, 1, 3);
+
+ // now run the task clean up
+ tt.taskCleanUp();
+
+ // after the another round the 4th task gets moved to localizing list
+ checkItemCountInQueue(tt, 4, 0, 4);
+
+ // now run the task clean up
+ tt.taskCleanUp();
+
+ // this is a no-op there is a sleep here but cant really test it
+ checkItemCountInQueue(tt, 4, 0, 4);
+ }
+
+ @Test
+ // test when all active items in the queue
+ public void testAllActiveJobActions() throws InterruptedException,
+ IOException {
+ // job and attempt ids
+ JobID jobId1 = new JobID(jtIdentifier, 1);
+ JobID jobId2 = new JobID(jtIdentifier, 2);
+
+ TaskAttemptID taskAttemptId1 = new TaskAttemptID(jtIdentifier, 3, true, 1,
+ 1);
+ TaskAttemptID taskAttemptId2 = new TaskAttemptID(jtIdentifier, 4, true, 1,
+ 1);
+
+ // job actions which is localizing
+ KillJobAction jAction1 = new KillJobAction(jobId1);
+ RunningJob rjob1 = new RunningJob(jAction1.getJobID());
+ rjob1.localizing = false;
+ // add the job to the runningJobs tree
+ tt.runningJobs.put(jobId1, rjob1);
+ tt.addActionToCleanup(jAction1);
+
+ KillJobAction jAction2 = new KillJobAction(jobId2);
+ RunningJob rjob2 = new RunningJob(jAction2.getJobID());
+ rjob2.localizing = false;
+ // add the job to the runningJobs tree
+ tt.runningJobs.put(jobId2, rjob2);
+ tt.addActionToCleanup(jAction2);
+
+ // task action which is localizing
+ KillTaskAction tAction1 = new KillTaskAction(taskAttemptId1);
+ RunningJob rjob3 = new RunningJob(tAction1.getTaskID().getJobID());
+ // add the job to the runningJobs tree
+ tt.runningJobs.put(rjob3.getJobID(), rjob3);
+ rjob3.localizing = false;
+ tt.addActionToCleanup(tAction1);
+
+ KillTaskAction tAction2 = new KillTaskAction(taskAttemptId2);
+ RunningJob rjob4 = new RunningJob(tAction2.getTaskID().getJobID());
+ rjob4.localizing = false;
+ // add the job to the runningJobs tree
+ tt.runningJobs.put(rjob4.getJobID(), rjob4);
+ tt.addActionToCleanup(tAction2);
+
+ // before the task clean up test the queue
+ checkItemCountInQueue(tt, 4, 4, 0);
+
+ // now run the task clean up
+ tt.taskCleanUp();
+
+ // after the one round the first task gets killed
+ checkItemCountInQueue(tt, 3, 3, 0);
+ // now run the task clean up
+ tt.taskCleanUp();
+
+ // after the another round the 2nd task gets killed
+ checkItemCountInQueue(tt, 2, 2, 0);
+
+ // now run the task clean up
+ tt.taskCleanUp();
+
+ // after the another round the 3rd task gets killed
+ checkItemCountInQueue(tt, 1, 1, 0);
+
+ // now run the task clean up
+ tt.taskCleanUp();
+
+ // after the another round the 4th task gets killed
+ checkItemCountInQueue(tt, 0, 0, 0);
+
+ // now run the task clean up
+ tt.taskCleanUp();
+
+ // this is a no-op there is a sleep here but cant really test it
+ checkItemCountInQueue(tt, 0, 0, 0);
+ }
+
+ @Test
+ // test when some items are localizing and some are not in the queue
+ public void testMixedJobActions() throws InterruptedException, IOException {
+ // job and attempt ids
+ JobID jobId1 = new JobID(jtIdentifier, 1);
+ JobID jobId2 = new JobID(jtIdentifier, 2);
+
+ TaskAttemptID taskAttemptId1 = new TaskAttemptID(jtIdentifier, 3, true, 1,
+ 1);
+ TaskAttemptID taskAttemptId2 = new TaskAttemptID(jtIdentifier, 4, true, 1,
+ 1);
+
+ // job actions which is localizing
+ KillJobAction jAction1 = new KillJobAction(jobId1);
+ RunningJob rjob1 = new RunningJob(jAction1.getJobID());
+ rjob1.localizing = true;
+ // add the job to the runningJobs tree
+ tt.runningJobs.put(jobId1, rjob1);
+ tt.addActionToCleanup(jAction1);
+
+ KillJobAction jAction2 = new KillJobAction(jobId2);
+ RunningJob rjob2 = new RunningJob(jAction2.getJobID());
+ rjob2.localizing = false;
+ // add the job to the runningJobs tree
+ tt.runningJobs.put(jobId2, rjob2);
+ tt.addActionToCleanup(jAction2);
+
+ // task action which is localizing
+ KillTaskAction tAction1 = new KillTaskAction(taskAttemptId1);
+ RunningJob rjob3 = new RunningJob(tAction1.getTaskID().getJobID());
+ // add the job to the runningJobs tree
+ tt.runningJobs.put(rjob3.getJobID(), rjob3);
+ rjob3.localizing = true;
+ tt.addActionToCleanup(tAction1);
+
+ KillTaskAction tAction2 = new KillTaskAction(taskAttemptId2);
+ RunningJob rjob4 = new RunningJob(tAction2.getTaskID().getJobID());
+ rjob4.localizing = false;
+ // add the job to the runningJobs tree
+ tt.runningJobs.put(rjob4.getJobID(), rjob4);
+ tt.addActionToCleanup(tAction2);
+
+ // before the task clean up test the queue
+ checkItemCountInQueue(tt, 4, 4, 0);
+
+ // now run the task clean up
+ tt.taskCleanUp();
+
+ // since the first attempt is localizing it will move to the
+ // localizing queue
+ checkItemCountInQueue(tt, 4, 3, 1);
+
+ // now run the task clean up
+ tt.taskCleanUp();
+
+ // after the 2nd round the tip is a job that can be cleaned up
+ checkItemCountInQueue(tt, 3, 2, 1);
+
+ // now run the task clean up
+ tt.taskCleanUp();
+
+ // after the 3rd round the 3rd task is getting localized
+ checkItemCountInQueue(tt, 3, 1, 2);
+
+ // now run the task clean up
+ tt.taskCleanUp();
+
+ // after the another round the 4th task gets killed
+ checkItemCountInQueue(tt, 2, 0, 2);
+
+ // now run the task clean up
+ tt.taskCleanUp();
+
+ // nothing should change as the tasks are being localized
+ checkItemCountInQueue(tt, 2, 0, 2);
+
+ }
+
+ // test when some items are localizing and some are not in the queue
+ // and we update the status of the actions between clean up jobs
+ @Test
+ public void testMixedJobActionsAndUpdateActions()
+ throws InterruptedException, IOException {
+ // job and attempt ids
+ JobID jobId1 = new JobID(jtIdentifier, 1);
+ JobID jobId2 = new JobID(jtIdentifier, 2);
+
+ TaskAttemptID taskAttemptId1 = new TaskAttemptID(jtIdentifier, 3, true, 1,
+ 1);
+ TaskAttemptID taskAttemptId2 = new TaskAttemptID(jtIdentifier, 4, true, 1,
+ 1);
+
+ // job actions which is localizing
+ KillJobAction jAction1 = new KillJobAction(jobId1);
+ RunningJob rjob1 = new RunningJob(jAction1.getJobID());
+ rjob1.localizing = true;
+ // add the job to the runningJobs tree
+ tt.runningJobs.put(jobId1, rjob1);
+ tt.addActionToCleanup(jAction1);
+
+ KillJobAction jAction2 = new KillJobAction(jobId2);
+ RunningJob rjob2 = new RunningJob(jAction2.getJobID());
+ rjob2.localizing = false;
+ // add the job to the runningJobs tree
+ tt.runningJobs.put(jobId2, rjob2);
+ tt.addActionToCleanup(jAction2);
+
+ // task action which is localizing
+ KillTaskAction tAction1 = new KillTaskAction(taskAttemptId1);
+ RunningJob rjob3 = new RunningJob(tAction1.getTaskID().getJobID());
+ // add the job to the runningJobs tree
+ tt.runningJobs.put(rjob3.getJobID(), rjob3);
+ rjob3.localizing = true;
+ tt.addActionToCleanup(tAction1);
+
+ KillTaskAction tAction2 = new KillTaskAction(taskAttemptId2);
+ RunningJob rjob4 = new RunningJob(tAction2.getTaskID().getJobID());
+ rjob4.localizing = false;
+ // add the job to the runningJobs tree
+ tt.runningJobs.put(rjob4.getJobID(), rjob4);
+ tt.addActionToCleanup(tAction2);
+
+ // before the task clean up test the queue
+ checkItemCountInQueue(tt, 4, 4, 0);
+
+ // now run the task clean up
+ tt.taskCleanUp();
+
+ // since the first attempt is localizing it will move to the
+ // localizing queue
+ checkItemCountInQueue(tt, 4, 3, 1);
+
+ // before running clean up again change the status of the first attempt to
+ // not be localizing.
+ rjob1.localizing = false;
+
+ // now run the task clean up
+ tt.taskCleanUp();
+
+ // after the 2nd round the tip is a job that can be cleaned up so it is
+ // cleaned and the localizing job is no longer being localized so it moves
+ // to the active queue
+ checkItemCountInQueue(tt, 3, 3, 0);
+
+ // now run the task clean up
+ tt.taskCleanUp();
+
+ // after the 3rd round the 3rd task is getting localized
+ // so gets moved to the localized queue.
+ checkItemCountInQueue(tt, 3, 2, 1);
+
+ // now run the task clean up
+ tt.taskCleanUp();
+
+ // after the another round the 4th task gets killed
+ // and leaves 2 tasks one which switched from localizing to clean
+ // and one that is being localized.
+ checkItemCountInQueue(tt, 2, 1, 1);
+ }
+
+ /*
+ * Method to test the number of items in the various datastructures holding
+ * clean up actions
+ */
+ private void checkItemCountInQueue(TaskTracker tt, int allSize,
+ int activeSize, int inactiveSize) {
+ // check the size and the content of the hashset and queue
+ assertEquals("Size of allCleanUpActions is not " + allSize, allSize,
+ tt.allCleanupActions.size());
+ assertEquals("Size of activeCleanUpActions is not " + activeSize,
+ activeSize, tt.activeCleanupActions.size());
+ assertEquals("Size of inactiveCleanUpActions is not " + inactiveSize,
+ inactiveSize, tt.inactiveCleanupActions.size());
+ }
+
+ class TestTaskTracker extends TaskTracker {
+ // override the method so its a no-op
+ synchronized void purgeJob(KillJobAction action) throws IOException {
+ LOG.info("Received 'KillJobAction' for job: " + action.getJobID());
+ }
+
+ // override the method so its a no-op
+ void processKillTaskAction(KillTaskAction killAction) throws IOException {
+ LOG.info("Received KillTaskAction for task: " + killAction.getTaskID());
+ }
+ }
+}
\ No newline at end of file