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 to...@apache.org on 2007/04/18 22:27:10 UTC
svn commit: r530146 - in /lucene/hadoop/trunk: ./ conf/
src/java/org/apache/hadoop/mapred/ src/test/org/apache/hadoop/mapred/
Author: tomwhite
Date: Wed Apr 18 13:27:09 2007
New Revision: 530146
URL: http://svn.apache.org/viewvc?view=rev&rev=530146
Log:
HADOOP-1111. Add support for client notification of job completion. Contributed by Alejandro Abdelnur.
Added:
lucene/hadoop/trunk/src/java/org/apache/hadoop/mapred/JobEndNotifier.java
lucene/hadoop/trunk/src/test/org/apache/hadoop/mapred/HadoopTestCase.java
lucene/hadoop/trunk/src/test/org/apache/hadoop/mapred/NotificationTestCase.java
lucene/hadoop/trunk/src/test/org/apache/hadoop/mapred/TestClusterMRNotification.java
lucene/hadoop/trunk/src/test/org/apache/hadoop/mapred/TestLocalMRNotification.java
Modified:
lucene/hadoop/trunk/CHANGES.txt
lucene/hadoop/trunk/conf/hadoop-default.xml
lucene/hadoop/trunk/src/java/org/apache/hadoop/mapred/JobInProgress.java
lucene/hadoop/trunk/src/java/org/apache/hadoop/mapred/JobTracker.java
lucene/hadoop/trunk/src/java/org/apache/hadoop/mapred/LocalJobRunner.java
Modified: lucene/hadoop/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/CHANGES.txt?view=diff&rev=530146&r1=530145&r2=530146
==============================================================================
--- lucene/hadoop/trunk/CHANGES.txt (original)
+++ lucene/hadoop/trunk/CHANGES.txt Wed Apr 18 13:27:09 2007
@@ -225,6 +225,12 @@
68. HADOOP-968. Move shuffle and sort to run in reduce's child JVM,
rather than in TaskTracker. (Devaraj Das via cutting)
+69. HADOOP-1111. Add support for client notification of job
+ completion. If the job configuration has a job.end.notification.url
+ property it will make a HTTP GET request to the specified URL.
+ The number of retries and the interval between retries is also
+ configurable. (Alejandro Abdelnur via tomwhite)
+
Release 0.12.3 - 2007-04-06
Modified: lucene/hadoop/trunk/conf/hadoop-default.xml
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/conf/hadoop-default.xml?view=diff&rev=530146&r1=530145&r2=530146
==============================================================================
--- lucene/hadoop/trunk/conf/hadoop-default.xml (original)
+++ lucene/hadoop/trunk/conf/hadoop-default.xml Wed Apr 18 13:27:09 2007
@@ -857,4 +857,33 @@
</description>
</property>
+<!-- Job Notification Configuration -->
+
+<!--
+<property>
+ <name>job.end.notification.url</name>
+ <value>http://localhost:8080/jobstatus.php?jobId=$jobId&jobStatus=$jobStatus</value>
+ <description>Indicates url which will be called on completion of job to inform
+ end status of job.
+ User can give at most 2 variables with URI : $jobId and $jobStatus.
+ If they are present in URI, then they will be replaced by their
+ respective values.
+</description>
+</property>
+-->
+
+<property>
+ <name>job.end.retry.attempt</name>
+ <value>0</value>
+ <description>Indicates how many times hadoop should attempt to contact the
+ notification URL </description>
+</property>
+
+<property>
+ <name>job.end.retry.interval</name>
+ <value>30000</value>
+ <description>Indicates time in milliseconds between notification URL retry
+ calls</description>
+</property>
+
</configuration>
Added: lucene/hadoop/trunk/src/java/org/apache/hadoop/mapred/JobEndNotifier.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/java/org/apache/hadoop/mapred/JobEndNotifier.java?view=auto&rev=530146
==============================================================================
--- lucene/hadoop/trunk/src/java/org/apache/hadoop/mapred/JobEndNotifier.java (added)
+++ lucene/hadoop/trunk/src/java/org/apache/hadoop/mapred/JobEndNotifier.java Wed Apr 18 13:27:09 2007
@@ -0,0 +1,207 @@
+package org.apache.hadoop.mapred;
+
+import java.io.IOException;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.DelayQueue;
+import java.util.concurrent.Delayed;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.commons.httpclient.HttpClient;
+import org.apache.commons.httpclient.HttpMethod;
+import org.apache.commons.httpclient.URI;
+import org.apache.commons.httpclient.methods.GetMethod;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+public class JobEndNotifier {
+ private static final Log LOG =
+ LogFactory.getLog(JobEndNotifier.class.getName());
+
+ private static Thread thread;
+ private static volatile boolean running;
+ private static BlockingQueue<JobEndStatusInfo> queue =
+ new DelayQueue<JobEndStatusInfo>();
+
+ public static void startNotifier() {
+ running = true;
+ thread = new Thread(
+ new Runnable() {
+ public void run() {
+ try {
+ while (running) {
+ sendNotification(queue.take());
+ }
+ }
+ catch (InterruptedException irex) {
+ if (running) {
+ LOG.error("Thread has ended unexpectedly", irex);
+ }
+ }
+ }
+
+ private void sendNotification(JobEndStatusInfo notification) {
+ try {
+ int code = httpNotification(notification.getUri());
+ if (code != 200) {
+ throw new IOException("Invalid response status code: " + code);
+ }
+ }
+ catch (IOException ioex) {
+ LOG.error("Notification failure [" + notification + "]", ioex);
+ if (notification.configureForRetry()) {
+ try {
+ queue.put(notification);
+ }
+ catch (InterruptedException iex) {
+ LOG.error("Notification queuing error [" + notification + "]",
+ iex);
+ }
+ }
+ }
+ catch (Exception ex) {
+ LOG.error("Notification failure [" + notification + "]", ex);
+ }
+ }
+
+ }
+
+ );
+ thread.start();
+ }
+
+ public static void stopNotifier() {
+ running = false;
+ thread.interrupt();
+ }
+
+ private static JobEndStatusInfo createNotification(JobConf conf,
+ JobStatus status) {
+ JobEndStatusInfo notification = null;
+ String uri = conf.get("job.end.notification.url");
+ if (uri != null) {
+ // +1 to make logic for first notification identical to a retry
+ int retryAttempts = conf.getInt("job.end.retry.attempts", 0) + 1;
+ long retryInterval = conf.getInt("job.end.retry.interval", 30000);
+ if (uri.contains("$jobId")) {
+ uri = uri.replace("$jobId", status.getJobId());
+ }
+ if (uri.contains("$jobStatus")) {
+ String statusStr =
+ (status.getRunState() == JobStatus.SUCCEEDED) ? "SUCCEEDED" : "FAILED";
+ uri = uri.replace("$jobStatus", statusStr);
+ }
+ notification = new JobEndStatusInfo(uri, retryAttempts, retryInterval);
+ }
+ return notification;
+ }
+
+ public static void registerNotification(JobConf jobConf, JobStatus status) {
+ JobEndStatusInfo notification = createNotification(jobConf, status);
+ if (notification != null) {
+ try {
+ queue.put(notification);
+ }
+ catch (InterruptedException iex) {
+ LOG.error("Notification queuing failure [" + notification + "]", iex);
+ }
+ }
+ }
+
+ private static int httpNotification(String uri) throws IOException {
+ URI url = new URI(uri, false);
+ HttpClient m_client = new HttpClient();
+ HttpMethod method = new GetMethod(url.getEscapedURI());
+ method.setRequestHeader("Accept", "*/*");
+ return m_client.executeMethod(method);
+ }
+
+ // for use by the LocalJobRunner, without using a thread&queue,
+ // simple synchronous way
+ public static void localRunnerNotification(JobConf conf, JobStatus status) {
+ JobEndStatusInfo notification = createNotification(conf, status);
+ if (notification != null) {
+ while (notification.configureForRetry()) {
+ try {
+ int code = httpNotification(notification.getUri());
+ if (code != 200) {
+ throw new IOException("Invalid response status code: " + code);
+ }
+ else {
+ break;
+ }
+ }
+ catch (IOException ioex) {
+ LOG.error("Notification error [" + notification.getUri() + "]", ioex);
+ }
+ catch (Exception ex) {
+ LOG.error("Notification error [" + notification.getUri() + "]", ex);
+ }
+ try {
+ synchronized (Thread.currentThread()) {
+ Thread.currentThread().sleep(notification.getRetryInterval());
+ }
+ }
+ catch (InterruptedException iex) {
+ LOG.error("Notification retry error [" + notification + "]", iex);
+ }
+ }
+ }
+ }
+
+ private static class JobEndStatusInfo implements Delayed {
+ private String uri;
+ private int retryAttempts;
+ private long retryInterval;
+ private long delayTime;
+
+ JobEndStatusInfo(String uri, int retryAttempts, long retryInterval) {
+ this.uri = uri;
+ this.retryAttempts = retryAttempts;
+ this.retryInterval = retryInterval;
+ this.delayTime = System.currentTimeMillis();
+ }
+
+ public String getUri() {
+ return uri;
+ }
+
+ public int getRetryAttempts() {
+ return retryAttempts;
+ }
+
+ public long getRetryInterval() {
+ return retryInterval;
+ }
+
+ public long getDelayTime() {
+ return delayTime;
+ }
+
+ public boolean configureForRetry() {
+ boolean retry = false;
+ if (getRetryAttempts() > 0) {
+ retry = true;
+ delayTime = System.currentTimeMillis() + retryInterval;
+ }
+ retryAttempts--;
+ return retry;
+ }
+
+ public long getDelay(TimeUnit unit) {
+ long n = this.delayTime - System.currentTimeMillis();
+ return unit.convert(n, TimeUnit.MILLISECONDS);
+ }
+
+ public int compareTo(Delayed d) {
+ return (int)(delayTime - ((JobEndStatusInfo)d).delayTime);
+ }
+
+
+ public String toString() {
+ return "URL: " + uri + " remaining retries: " + retryAttempts +
+ " interval: " + retryInterval;
+ }
+
+ }
+
+}
Modified: lucene/hadoop/trunk/src/java/org/apache/hadoop/mapred/JobInProgress.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/java/org/apache/hadoop/mapred/JobInProgress.java?view=diff&rev=530146&r1=530145&r2=530146
==============================================================================
--- lucene/hadoop/trunk/src/java/org/apache/hadoop/mapred/JobInProgress.java (original)
+++ lucene/hadoop/trunk/src/java/org/apache/hadoop/mapred/JobInProgress.java Wed Apr 18 13:27:09 2007
@@ -198,9 +198,13 @@
this.finishTime = System.currentTimeMillis();
this.status = new JobStatus(status.getJobId(), 1.0f, 1.0f, JobStatus.SUCCEEDED);
tasksInited = true;
+
+ // Special case because the Job is not queued
+ JobEndNotifier.registerNotification(this.getJobConf(), this.getStatus());
+
return;
}
-
+
//
// Create reduce tasks
//
Modified: lucene/hadoop/trunk/src/java/org/apache/hadoop/mapred/JobTracker.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/java/org/apache/hadoop/mapred/JobTracker.java?view=diff&rev=530146&r1=530145&r2=530146
==============================================================================
--- lucene/hadoop/trunk/src/java/org/apache/hadoop/mapred/JobTracker.java (original)
+++ lucene/hadoop/trunk/src/java/org/apache/hadoop/mapred/JobTracker.java Wed Apr 18 13:27:09 2007
@@ -115,7 +115,10 @@
} catch (InterruptedException e) {
}
}
- if (runTracker) { tracker.offerService(); }
+ if (runTracker) {
+ JobEndNotifier.startNotifier();
+ tracker.offerService();
+ }
}
public static JobTracker getTracker() {
@@ -125,6 +128,7 @@
public static void stopTracker() throws IOException {
runTracker = false;
if (tracker != null) {
+ JobEndNotifier.stopNotifier();
tracker.close();
tracker = null;
}
@@ -873,7 +877,9 @@
synchronized void finalizeJob(JobInProgress job) {
// Mark the 'non-running' tasks for pruning
markCompletedJob(job);
-
+
+ JobEndNotifier.registerNotification(job.getJobConf(), job.getStatus());
+
// Purge oldest jobs and keep at-most MAX_COMPLETE_USER_JOBS_IN_MEMORY jobs of a given user
// in memory; information about the purged jobs is available via
// JobHistory.
Modified: lucene/hadoop/trunk/src/java/org/apache/hadoop/mapred/LocalJobRunner.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/java/org/apache/hadoop/mapred/LocalJobRunner.java?view=diff&rev=530146&r1=530145&r2=530146
==============================================================================
--- lucene/hadoop/trunk/src/java/org/apache/hadoop/mapred/LocalJobRunner.java (original)
+++ lucene/hadoop/trunk/src/java/org/apache/hadoop/mapred/LocalJobRunner.java Wed Apr 18 13:27:09 2007
@@ -161,9 +161,13 @@
this.status.setRunState(JobStatus.SUCCEEDED);
+ JobEndNotifier.localRunnerNotification(job, status);
+
} catch (Throwable t) {
this.status.setRunState(JobStatus.FAILED);
LOG.warn(id, t);
+
+ JobEndNotifier.localRunnerNotification(job, status);
} finally {
try {
Added: lucene/hadoop/trunk/src/test/org/apache/hadoop/mapred/HadoopTestCase.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/test/org/apache/hadoop/mapred/HadoopTestCase.java?view=auto&rev=530146
==============================================================================
--- lucene/hadoop/trunk/src/test/org/apache/hadoop/mapred/HadoopTestCase.java (added)
+++ lucene/hadoop/trunk/src/test/org/apache/hadoop/mapred/HadoopTestCase.java Wed Apr 18 13:27:09 2007
@@ -0,0 +1,190 @@
+package org.apache.hadoop.mapred;
+
+import junit.framework.TestCase;
+import org.apache.hadoop.dfs.MiniDFSCluster;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+import java.io.File;
+import java.io.IOException;
+
+/**
+ * Abstract Test case class to run MR in local or cluster mode and in local FS
+ * or DFS.
+ *
+ * The Hadoop instance is started and stopped on each test method.
+ *
+ * If using DFS the filesystem is reformated at each start (test method).
+ *
+ * Job Configurations should be created using a configuration returned by the
+ * 'createJobConf()' method.
+ *
+ * @author Alejandro Abdelnur
+ */
+public abstract class HadoopTestCase extends TestCase {
+ public static final int LOCAL_MR = 1;
+ public static final int CLUSTER_MR = 2;
+ public static final int LOCAL_FS = 4;
+ public static final int DFS_FS = 8;
+
+ private boolean localMR;
+ private boolean localFS;
+
+ private int taskTrackers;
+ private int dataNodes;
+
+ /**
+ * Creates a testcase for local or cluster MR using DFS.
+ *
+ * The DFS will be formatted regardless if there was one or not before in the
+ * given location.
+ *
+ * @param mrMode indicates if the MR should be local (LOCAL_MR) or cluster
+ * (CLUSTER_MR)
+ * @param fsMode indicates if the FS should be local (LOCAL_FS) or DFS (DFS_FS)
+ *
+ * local FS when using relative PATHs)
+ *
+ * @param taskTrackers number of task trackers to start when using cluster
+ *
+ * @param dataNodes number of data nodes to start when using DFS
+ *
+ * @throws IOException thrown if the base directory cannot be set.
+ */
+ public HadoopTestCase(int mrMode, int fsMode, int taskTrackers, int dataNodes)
+ throws IOException {
+ if (mrMode != LOCAL_MR && mrMode != CLUSTER_MR) {
+ throw new IllegalArgumentException(
+ "Invalid MapRed mode, must be LOCAL_MR or CLUSTER_MR");
+ }
+ if (fsMode != LOCAL_FS && fsMode != DFS_FS) {
+ throw new IllegalArgumentException(
+ "Invalid FileSystem mode, must be LOCAL_FS or DFS_FS");
+ }
+ if (taskTrackers < 1) {
+ throw new IllegalArgumentException(
+ "Invalid taskTrackers value, must be greater than 0");
+ }
+ if (dataNodes < 1) {
+ throw new IllegalArgumentException(
+ "Invalid dataNodes value, must be greater than 0");
+ }
+ localMR = (mrMode == LOCAL_MR);
+ localFS = (fsMode == LOCAL_FS);
+/*
+ JobConf conf = new JobConf();
+ fsRoot = conf.get("hadoop.tmp.dir");
+
+ if (fsRoot == null) {
+ throw new IllegalArgumentException(
+ "hadoop.tmp.dir is not defined");
+ }
+
+ fsRoot = fsRoot.replace(' ', '+') + "/fs";
+
+ File file = new File(fsRoot);
+ if (!file.exists()) {
+ if (!file.mkdirs()) {
+ throw new RuntimeException("Could not create FS base path: " + file);
+ }
+ }
+*/
+ this.taskTrackers = taskTrackers;
+ this.dataNodes = dataNodes;
+ }
+
+ /**
+ * Indicates if the MR is running in local or cluster mode.
+ *
+ * @return returns TRUE if the MR is running locally, FALSE if running in
+ * cluster mode.
+ */
+ public boolean isLocalMR() {
+ return localMR;
+ }
+
+ /**
+ * Indicates if the filesystem is local or DFS.
+ *
+ * @return returns TRUE if the filesystem is local, FALSE if it is DFS.
+ */
+ public boolean isLocalFS() {
+ return localFS;
+ }
+
+
+ private MiniDFSCluster dfsCluster = null;
+ private MiniMRCluster mrCluster = null;
+ private FileSystem fileSystem = null;
+
+ /**
+ * Creates Hadoop instance based on constructor configuration before
+ * a test case is run.
+ *
+ * @throws Exception
+ */
+ protected void setUp() throws Exception {
+ super.setUp();
+ if (localFS) {
+ fileSystem = FileSystem.getLocal(new JobConf());
+ }
+ else {
+ dfsCluster = new MiniDFSCluster(new JobConf(), dataNodes, true, null);
+ fileSystem = dfsCluster.getFileSystem();
+ }
+ if (localMR) {
+ }
+ else {
+ //noinspection deprecation
+ mrCluster = new MiniMRCluster(taskTrackers, fileSystem.getName(), 1);
+ }
+ }
+
+ /**
+ * Destroys Hadoop instance based on constructor configuration after
+ * a test case is run.
+ *
+ * @throws Exception
+ */
+ protected void tearDown() throws Exception {
+ try {
+ if (mrCluster != null) {
+ mrCluster.shutdown();
+ }
+ }
+ catch (Exception ex) {
+ System.out.println(ex);
+ }
+ try {
+ if (dfsCluster != null) {
+ dfsCluster.shutdown();
+ }
+ }
+ catch (Exception ex) {
+ System.out.println(ex);
+ }
+ super.tearDown();
+ }
+
+ /**
+ * Returns the Filesystem in use.
+ *
+ * TestCases should use this Filesystem as it
+ * is properly configured with the workingDir for relative PATHs.
+ *
+ * @return the filesystem used by Hadoop.
+ */
+ protected FileSystem getFileSystem() {
+ return fileSystem;
+ }
+
+ /**
+ * Returns a job configuration preconfigured to run against the Hadoop
+ * managed by the testcase.
+ * @return configuration that works on the testcase Hadoop instance
+ */
+ protected JobConf createJobConf() {
+ return (localMR) ? new JobConf() : mrCluster.createJobConf();
+ }
+
+}
Added: lucene/hadoop/trunk/src/test/org/apache/hadoop/mapred/NotificationTestCase.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/test/org/apache/hadoop/mapred/NotificationTestCase.java?view=auto&rev=530146
==============================================================================
--- lucene/hadoop/trunk/src/test/org/apache/hadoop/mapred/NotificationTestCase.java (added)
+++ lucene/hadoop/trunk/src/test/org/apache/hadoop/mapred/NotificationTestCase.java Wed Apr 18 13:27:09 2007
@@ -0,0 +1,200 @@
+package org.apache.hadoop.mapred;
+
+import org.mortbay.http.HttpServer;
+import org.mortbay.http.SocketListener;
+import org.mortbay.http.HttpContext;
+import org.mortbay.jetty.servlet.ServletHandler;
+import org.apache.log4j.net.SocketServer;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.LocalFileSystem;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.examples.WordCount;
+
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+import javax.servlet.http.HttpServlet;
+import javax.servlet.ServletException;
+import java.io.IOException;
+import java.io.DataOutputStream;
+import java.util.Date;
+import java.net.Socket;
+import java.net.InetSocketAddress;
+
+/**
+ * Base class to test Job end notification in local and cluster mode.
+ *
+ * Starts up hadoop on Local or Cluster mode (by extending of the
+ * HadoopTestCase class) and it starts a servlet engine that hosts
+ * a servlet that will receive the notification of job finalization.
+ *
+ * The notification servlet returns a HTTP 400 the first time is called
+ * and a HTTP 200 the second time, thus testing retry.
+ *
+ * In both cases local file system is used (this is irrelevant for
+ * the tested functionality)
+ *
+ * @author Alejandro Abdelnur
+ *
+ */
+public abstract class NotificationTestCase extends HadoopTestCase {
+
+ private static void stdPrintln(String s) {
+ //System.out.println(s);
+ }
+
+ protected NotificationTestCase(int mode) throws IOException {
+ super(mode, HadoopTestCase.LOCAL_FS, 1, 1);
+ }
+
+ private int port;
+ private String contextPath = "/notification";
+ private Class servletClass = NotificationServlet.class;
+ private String servletPath = "/mapred";
+ private HttpServer server;
+
+ private void startHttpServer() throws Exception {
+
+ // Create the server
+ if (server != null) {
+ server.stop();
+ server = null;
+ }
+ server = new HttpServer();
+
+ // Create a port listener
+ SocketListener listener = new SocketListener();
+ listener.setPort(0); // letting OS to pickup the PORT
+ server.addListener(listener);
+
+ // create context
+ HttpContext context = new HttpContext();
+ context.setContextPath(contextPath + "/*");
+
+ // create servlet handler
+ ServletHandler handler = new ServletHandler();
+ handler.addServlet(servletClass.getName(), servletPath,
+ servletClass.getName());
+
+ // bind servlet handler to context
+ context.addHandler(handler);
+
+ // bind context to servlet engine
+ server.addContext(context);
+
+ // Start server
+ server.start();
+ port = listener.getPort();
+
+ }
+
+ private void stopHttpServer() throws Exception {
+ if (server != null) {
+ server.stop();
+ server.destroy();
+ server = null;
+ }
+ }
+
+ public static class NotificationServlet extends HttpServlet {
+ public static int counter = 0;
+
+ protected void doGet(HttpServletRequest req, HttpServletResponse res)
+ throws ServletException, IOException {
+ if (counter == 0) {
+ stdPrintln((new Date()).toString() +
+ "Receiving First notification for [" + req.getQueryString() +
+ "], returning error");
+ res.sendError(HttpServletResponse.SC_BAD_REQUEST, "forcing error");
+ }
+ else {
+ stdPrintln((new Date()).toString() +
+ "Receiving Second notification for [" + req.getQueryString() +
+ "], returning OK");
+ res.setStatus(HttpServletResponse.SC_OK);
+ }
+ counter++;
+ }
+ }
+
+ private String getNotificationUrlTemplate() {
+ return "http://localhost:" + port + contextPath + servletPath +
+ "?jobId=$jobId&jobStatus=$jobStatus";
+ }
+
+ protected JobConf createJobConf() {
+ JobConf conf = super.createJobConf();
+ conf.set("job.end.notification.url", getNotificationUrlTemplate());
+ conf.set("job.end.retry.attempts", 3);
+ conf.set("job.end.retry.interval", 200);
+ return conf;
+ }
+
+
+ protected void setUp() throws Exception {
+ super.setUp();
+ startHttpServer();
+ }
+
+ protected void tearDown() throws Exception {
+ stopHttpServer();
+ super.tearDown();
+ }
+
+ public void testMR() throws Exception {
+ System.out.println(launchWordCount(this.createJobConf(),
+ "a b c d e f g h", 1, 1));
+ synchronized(Thread.currentThread()) {
+ stdPrintln("Sleeping for 2 seconds to give time for retry");
+ Thread.currentThread().sleep(2000);
+ }
+ assertEquals(2, NotificationServlet.counter);
+ }
+
+ private String launchWordCount(JobConf conf,
+ String input,
+ int numMaps,
+ int numReduces) throws IOException {
+ Path inDir = new Path("testing/wc/input");
+ Path outDir = new Path("testing/wc/output");
+
+ // Hack for local FS that does not have the concept of a 'mounting point'
+ if (isLocalFS()) {
+ String localPathRoot = System.getProperty("test.build.data","/tmp")
+ .toString().replace(' ', '+');;
+ inDir = new Path(localPathRoot, inDir);
+ outDir = new Path(localPathRoot, outDir);
+ }
+
+ FileSystem fs = FileSystem.get(conf);
+ fs.delete(outDir);
+ if (!fs.mkdirs(inDir)) {
+ throw new IOException("Mkdirs failed to create " + inDir.toString());
+ }
+ {
+ DataOutputStream file = fs.create(new Path(inDir, "part-0"));
+ file.writeBytes(input);
+ file.close();
+ }
+ conf.setJobName("wordcount");
+ conf.setInputFormat(TextInputFormat.class);
+
+ // the keys are words (strings)
+ conf.setOutputKeyClass(Text.class);
+ // the values are counts (ints)
+ conf.setOutputValueClass(IntWritable.class);
+
+ conf.setMapperClass(WordCount.MapClass.class);
+ conf.setCombinerClass(WordCount.Reduce.class);
+ conf.setReducerClass(WordCount.Reduce.class);
+
+ conf.setInputPath(inDir);
+ conf.setOutputPath(outDir);
+ conf.setNumMapTasks(numMaps);
+ conf.setNumReduceTasks(numReduces);
+ JobClient.runJob(conf);
+ return TestMiniMRWithDFS.readOutput(outDir, conf);
+ }
+
+}
Added: lucene/hadoop/trunk/src/test/org/apache/hadoop/mapred/TestClusterMRNotification.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/test/org/apache/hadoop/mapred/TestClusterMRNotification.java?view=auto&rev=530146
==============================================================================
--- lucene/hadoop/trunk/src/test/org/apache/hadoop/mapred/TestClusterMRNotification.java (added)
+++ lucene/hadoop/trunk/src/test/org/apache/hadoop/mapred/TestClusterMRNotification.java Wed Apr 18 13:27:09 2007
@@ -0,0 +1,16 @@
+package org.apache.hadoop.mapred;
+
+import java.io.IOException;
+
+/**
+ * Tests Job end notification in cluster mode.
+ *
+ * @author Alejandro Abdelnur
+ */
+public class TestClusterMRNotification extends NotificationTestCase {
+
+ public TestClusterMRNotification() throws IOException {
+ super(HadoopTestCase.CLUSTER_MR);
+ }
+
+}
Added: lucene/hadoop/trunk/src/test/org/apache/hadoop/mapred/TestLocalMRNotification.java
URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/test/org/apache/hadoop/mapred/TestLocalMRNotification.java?view=auto&rev=530146
==============================================================================
--- lucene/hadoop/trunk/src/test/org/apache/hadoop/mapred/TestLocalMRNotification.java (added)
+++ lucene/hadoop/trunk/src/test/org/apache/hadoop/mapred/TestLocalMRNotification.java Wed Apr 18 13:27:09 2007
@@ -0,0 +1,17 @@
+package org.apache.hadoop.mapred;
+
+import java.io.IOException;
+
+
+/**
+ * Tests Job end notification in local mode.
+ *
+ * @author Alejandro Abdelnur
+ */
+public class TestLocalMRNotification extends NotificationTestCase {
+
+ public TestLocalMRNotification() throws IOException {
+ super(HadoopTestCase.LOCAL_MR);
+ }
+
+}