diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/JobListener.java b/spring-batch-core/src/main/java/org/springframework/batch/core/JobListener.java
index c54004ea2..a3bf00673 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/JobListener.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/JobListener.java
@@ -37,4 +37,9 @@ public interface JobListener {
 	 * Callback after successful completion of a job.
 	 */
 	void afterJob(JobExecution jobExecution);
+	
+	/**
+	 * Callback on job failure.
+	 */
+	void onError(JobExecution jobExecution, Throwable e);
 }
diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/SimpleJob.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/SimpleJob.java
index 48f03b82a..f68f86180 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/job/SimpleJob.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/SimpleJob.java
@@ -133,6 +133,7 @@ public class SimpleJob extends AbstractJob {
 		}
 		catch (Throwable t) {
 			execution.setStatus(BatchStatus.FAILED);
+			listener.onError(execution, t);
 			rethrow(t);
 		}
 		finally {
diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/listener/CompositeJobListener.java b/spring-batch-core/src/main/java/org/springframework/batch/core/listener/CompositeJobListener.java
index b7737f640..a743a71fc 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/listener/CompositeJobListener.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/listener/CompositeJobListener.java
@@ -51,24 +51,26 @@ public class CompositeJobListener implements JobListener {
 		}
 	}
 
-	/* (non-Javadoc)
-	 * @see org.springframework.batch.core.domain.StepListener#close()
-	 */
 	public void afterJob(JobExecution jobExecution) {
-		for (Iterator iterator = listeners.iterator(); iterator.hasNext();) {
+		for (Iterator iterator = listeners.listIterator(); iterator.hasNext();) {
 			JobListener listener = (JobListener) iterator.next();
 			listener.afterJob(jobExecution);
 		}
 	}
 
-	/* (non-Javadoc)
-	 * @see org.springframework.batch.core.domain.StepListener#open(org.springframework.batch.core.domain.JobParameters)
-	 */
 	public void beforeJob(JobExecution jobExecution) {
-		for (Iterator iterator = listeners.iterator(); iterator.hasNext();) {
+		for (Iterator iterator = listeners.listIterator(); iterator.hasNext();) {
 			JobListener listener = (JobListener) iterator.next();
 			listener.beforeJob(jobExecution);
 		}
 	}
 
+	public void onError(JobExecution jobExecution, Throwable e) {
+		for (Iterator iterator = listeners.listIterator(); iterator.hasNext();) {
+			JobListener listener = (JobListener) iterator.next();
+			listener.onError(jobExecution, e);
+		}
+		
+	}
+
 }
diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/listener/JobListenerSupport.java b/spring-batch-core/src/main/java/org/springframework/batch/core/listener/JobListenerSupport.java
index 5832c6c55..7a621bbd2 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/listener/JobListenerSupport.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/listener/JobListenerSupport.java
@@ -36,4 +36,9 @@ public class JobListenerSupport implements JobListener {
 	public void beforeJob(JobExecution jobExecution) {
 	}
 
+	public void onError(JobExecution jobExecution, Throwable e) {
+		// TODO Auto-generated method stub
+		
+	}
+
 }
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/job/SimpleJobTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/job/SimpleJobTests.java
index a54aa0224..7b427cf03 100644
--- a/spring-batch-core/src/test/java/org/springframework/batch/core/job/SimpleJobTests.java
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/job/SimpleJobTests.java
@@ -250,6 +250,25 @@ public class SimpleJobTests extends TestCase {
 		assertEquals(0, list.size());
 		checkRepository(BatchStatus.FAILED, ExitStatus.FAILED);
 	}
+	
+	public void testFailedWithListener() throws Exception {
+		job.setJobListeners(new JobListenerSupport[] { new JobListenerSupport() {
+			public void onError(JobExecution jobExecution, Throwable t) {
+				list.add(t);
+			}
+		} });
+		final RuntimeException exception = new RuntimeException("Foo!");
+		stepConfiguration1.setProcessException(exception);
+		
+		try {
+			job.execute(jobExecution);
+		} catch (RuntimeException e) {
+			assertEquals(exception, e);
+		}
+		assertEquals(1, list.size());
+		assertSame(exception, list.get(0));
+		checkRepository(BatchStatus.FAILED, ExitStatus.FAILED);
+	}
 
 	public void testFailedWithError() throws Exception {
 		stepConfiguration1.setStartLimit(5);