diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/StepExecutionListener.java b/spring-batch-core/src/main/java/org/springframework/batch/core/StepExecutionListener.java
index a542f9925..443f3c64e 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/StepExecutionListener.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/StepExecutionListener.java
@@ -48,6 +48,9 @@ public interface StepExecutionListener extends StepListener {
 	 * returned will be combined with the normal exit status using
 	 * {@link ExitStatus#and(ExitStatus)}.
 	 * 
+	 * Called after successful execution of step's processing logic. Throwing
+	 * exception in this method will cause step to fail.
+	 * 
 	 * @return an {@link ExitStatus} to combine with the normal value. Return
 	 * null to leave the old value unchanged.
 	 */
diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java
index f426622e2..d4c5f0e9e 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java
@@ -137,12 +137,7 @@ public class TaskletStep extends AbstractStep implements Step, InitializingBean,
 
 			listener.beforeStep(stepExecution);
 			exitStatus = tasklet.execute();
-			try {
-				exitStatus = exitStatus.and(listener.afterStep(stepExecution));
-			}
-			catch (Exception e) {
-				logger.error("Encountered an error on listener close.", e);
-			}
+			exitStatus = exitStatus.and(listener.afterStep(stepExecution));
 
 			try {
 				jobRepository.saveOrUpdateExecutionContext(stepExecution);
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/TaskletStepTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/TaskletStepTests.java
index 1b47edc8a..879268df7 100644
--- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/TaskletStepTests.java
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/TaskletStepTests.java
@@ -168,6 +168,36 @@ public class TaskletStepTests extends TestCase {
 			assertEquals("Job interrupted while executing tasklet", expected.getMessage());
 		}
 	}
+	
+	/**
+	 * Exception in {@link StepExecutionListener#afterStep(StepExecution)}
+	 * causes step to fail.
+	 * @throws JobInterruptedException
+	 */
+	public void testStepFailureInAfterStepCallback() throws JobInterruptedException {
+		TaskletStep step = new TaskletStep(new Tasklet() {
+			public ExitStatus execute() throws Exception {
+				return ExitStatus.FINISHED;
+			}
+		}, new JobRepositorySupport());
+		
+		StepExecutionListener listener = new StepExecutionListenerSupport() {
+			public ExitStatus afterStep(StepExecution stepExecution) {
+				throw new RuntimeException("exception thrown in afterStep to signal failure");
+			}
+		};
+		step.setStepListeners(new StepExecutionListener[] { listener });
+		try {
+			step.execute(stepExecution);
+			fail();
+		}
+		catch (RuntimeException expected) {
+			assertEquals("exception thrown in afterStep to signal failure", expected.getMessage());
+		}
+		
+		assertEquals(BatchStatus.FAILED, stepExecution.getStatus());
+		
+	}
 
 	private class StubTasklet extends StepExecutionListenerSupport implements Tasklet {