diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowJob.java b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowJob.java
index 00c7e8ca5..2845f32c7 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowJob.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/job/flow/FlowJob.java
@@ -69,13 +69,19 @@ public class FlowJob extends AbstractJob {
 	public Step getStep(String stepName) {
 		State state = this.flow.getState(stepName);
 		if (state instanceof StepHolder) {
-			return ((StepHolder) state).getStep();
+			Step step = ((StepHolder) state).getStep();
+			if (stepName.equals(step.getName())) {
+				return step;
+			}
 		}
 		// The state names can be prefixed with the job name for
 		// uniqueness...
-		state = this.flow.getState(getName()+"."+stepName);
+		state = this.flow.getState(getName() + "." + stepName);
 		if (state instanceof StepHolder) {
-			return ((StepHolder) state).getStep();
+			Step step = ((StepHolder) state).getStep();
+			if (stepName.equals(step.getName())) {
+				return step;
+			}
 		}
 		return null;
 	}
@@ -88,7 +94,8 @@ public class FlowJob extends AbstractJob {
 		Collection<String> steps = new HashSet<String>();
 		for (State state : flow.getStates()) {
 			if (state instanceof StepHolder) {
-				steps.add(state.getName());
+				String name = ((StepHolder) state).getStep().getName();
+				steps.add(name);
 			}
 		}
 		return steps;
@@ -100,8 +107,8 @@ public class FlowJob extends AbstractJob {
 	@Override
 	protected void doExecute(final JobExecution execution) throws JobExecutionException {
 		try {
-			JobFlowExecutor executor = new JobFlowExecutor(getJobRepository(), new SimpleStepHandler(getJobRepository()),
-					execution);
+			JobFlowExecutor executor = new JobFlowExecutor(getJobRepository(),
+					new SimpleStepHandler(getJobRepository()), execution);
 			executor.updateJobExecutionStatus(flow.start(executor).getStatus());
 		}
 		catch (FlowExecutionException e) {
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/FlowJobTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/FlowJobTests.java
index 706be10d6..834665621 100644
--- a/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/FlowJobTests.java
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/job/flow/FlowJobTests.java
@@ -535,6 +535,21 @@ public class FlowJobTests {
 		assertEquals("step", step.getName());
 	}
 
+	@Test
+	public void testGetStepNamesWithPrefix() throws Exception {
+		SimpleFlow flow = new SimpleFlow("job");
+		List<StateTransition> transitions = new ArrayList<StateTransition>();
+		transitions.add(StateTransition.createStateTransition(new StepState("job.step", new StubStep("step")), "end0"));
+		transitions.add(StateTransition.createEndStateTransition(new EndState(FlowExecutionStatus.COMPLETED, "end0")));
+		flow.setStateTransitions(transitions);
+		flow.afterPropertiesSet();
+		job.setFlow(flow);
+		job.setName(flow.getName());
+		job.afterPropertiesSet();
+
+		assertEquals("[step]", job.getStepNames().toString());
+	}
+
 	@Test
 	public void testGetStepNotExists() throws Exception {
 		SimpleFlow flow = new SimpleFlow("job");