svn commit: r1163100 - in /pig/trunk: CHANGES.txt src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceLauncher.java

[da...@apache.org]

Author: daijy
Date: Tue Aug 30 06:31:11 2011
New Revision: 1163100

URL: http://svn.apache.org/viewvc?rev=1163100&view=rev
Log:
PIG-604: Kill the Pig job should kill all associated Hadoop Jobs

Modified:
    pig/trunk/CHANGES.txt
    pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceLauncher.java

Modified: pig/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/pig/trunk/CHANGES.txt?rev=1163100&r1=1163099&r2=1163100&view=diff
==============================================================================
--- pig/trunk/CHANGES.txt (original)
+++ pig/trunk/CHANGES.txt Tue Aug 30 06:31:11 2011
@@ -23,6 +23,9 @@ Trunk (unreleased changes)
 INCOMPATIBLE CHANGES
 
 IMPROVEMENTS
+
+PIG-604: Kill the Pig job should kill all associated Hadoop Jobs (daijy)
+
 PIG-2096: End to end tests for new Macro feature (gates)
 
 PIG-2242: Allow the delimiter to be specified when calling TOKENIZE (markroddy via hashutosh)

Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceLauncher.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceLauncher.java?rev=1163100&r1=1163099&r2=1163100&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceLauncher.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceLauncher.java Tue Aug 30 06:31:11 2011
@@ -89,6 +89,33 @@ public class MapReduceLauncher extends L
     public static final String SUCCESSFUL_JOB_OUTPUT_DIR_MARKER =
         "mapreduce.fileoutputcommitter.marksuccessfuljobs";
     
+    private JobControl jc=null;
+    
+    private class HangingJobKiller extends Thread {
+        public HangingJobKiller() {
+        }
+        @Override
+        public void run() {
+            try {
+                log.debug("Receive kill signal");
+                if (jc!=null) {
+                    for (Job job : jc.getRunningJobs()) {
+                        RunningJob runningJob = job.getJobClient().getJob(job.getAssignedJobID());
+                        if (runningJob!=null)
+                            runningJob.killJob();
+                        log.info("Job " + job.getJobID() + " killed");
+                    }
+                }
+            } catch (Exception e) {
+                log.warn("Encounter exception on cleanup:" + e);
+            }
+        }
+    }
+
+    public MapReduceLauncher() {
+        Runtime.getRuntime().addShutdownHook(new HangingJobKiller());
+    }
+    
     /**
      * Get the exception that caused a failure on the backend for a
      * store location (if any).
@@ -132,7 +159,6 @@ public class MapReduceLauncher extends L
         List<NativeMapReduceOper> failedNativeMR = new LinkedList<NativeMapReduceOper>();
         List<Job> completeFailedJobsInThisRun = new LinkedList<Job>();
         List<Job> succJobs = new LinkedList<Job>();
-        JobControl jc;
         int totalMRJobs = mrp.size();
         int numMRJobsCompl = 0;
         double lastProg = -1;