package com.twitter.hadoop.isolated.example;

import static java.lang.Thread.sleep;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.util.Map.Entry;
import java.util.Properties;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import com.twitter.hadoop.isolated.IsolatedInputFormat;

public class IsolatedJob {

  public static void main(String[] args) throws Exception {
    Properties properties = new Properties();
    properties.load(new FileInputStream(new File(args[0])));
    Configuration conf = new Configuration();

    for (Entry<Object, Object> entry : properties.entrySet()) {
      conf.set((String)entry.getKey(), (String)entry.getValue());
    }

    Job job = new Job(conf, "example");

    job.setInputFormatClass(IsolatedInputFormat.class);
    job.setNumReduceTasks(0);
    job.setOutputFormatClass(TextOutputFormat.class);
    job.setMapperClass(Mapper.class);
    job.submit();
    System.out.println("job id = " + job.getJobID());
    System.out.println("URL: " + job.getTrackingURL());
    waitForJob(job);
  }


  private static void waitForJob(Job job) throws InterruptedException, IOException {
    while (!job.isComplete()) {
      System.out.println("waiting for job " + job.getJobName() + " " + (int)(job.mapProgress() * 100) + "%");
      sleep(100);
    }
    System.out.println("status for job " + job.getJobName() + ": " + (job.isSuccessful() ? "SUCCESS" : "FAILURE"));
    if (!job.isSuccessful()) {
      throw new RuntimeException("job failed " + job.getJobName());
    }
  }

}

@@ -705,24 +705,65 @@
 uration();%0A%0A
+%0A    Job job = new Job(conf, %22example%22);%0A
     for (Ent
@@ -820,20 +820,38 @@
 %7B%0A      
-conf
+job.getConfiguration()
 .set((St
@@ -909,49 +909,8 @@
  %7D%0A%0A
-    Job job = new Job(conf, %22example%22);%0A%0A
     
@@ -1075,32 +1075,74 @@
 (Mapper.class);%0A
+    job.setJarByClass(IsolatedJob.class);%0A
     job.submit()

fix example job