Fix TCK test JobOperatorTests.testJobOperatorRestartJobAlreadyAbandoned

* On restart if a previous JobExecution has a status of ABANDONED, throw a JobRestartException
* Ensure any previous JobParameters are merged with new parameters when creating a new job execution on restart

spring-batch-core/src/main/java/org/springframework/batch/core/jsr/launch/JsrJobOperator.java

@@ -414,6 +414,10 @@ public class JsrJobOperator implements JobOperator {
 			if(jobExecution.getCreateTime().compareTo(previousJobExecution.getCreateTime()) > 0) {
 				throw new JobExecutionNotMostRecentException("The requested JobExecution to restart was not the most recently run");
 			}
+
+			if(jobExecution.getStatus().equals(BatchStatus.ABANDONED)) {
+				throw new JobRestartException("JobExecution ID: " + jobExecution.getId() + " is abandoned and attempted to be restarted.");
+			}
 		}
 
 		String jobName = previousJobExecution.getJobInstance().getJobName();
@@ -445,7 +449,8 @@ public class JsrJobOperator implements JobOperator {
 		final org.springframework.batch.core.JobExecution jobExecution;
 
 		try {
-			JobParameters jobParameters = jobParametersConverter.getJobParameters(params);
+			Properties jobRestartProperties = getJobRestartProperties(params, previousJobExecution);
+			JobParameters jobParameters = jobParametersConverter.getJobParameters(jobRestartProperties);
 			jobExecution = jobRepository.createJobExecution(previousJobExecution.getJobInstance(), jobParameters, previousJobExecution.getJobConfigurationName());
 		} catch (Exception e) {
 			throw new JobRestartException(e);
@@ -485,6 +490,24 @@ public class JsrJobOperator implements JobOperator {
 		return jobExecution.getId();
 	}
 
+	protected Properties getJobRestartProperties(Properties params, org.springframework.batch.core.JobExecution previousJobExecution) {
+		Properties jobRestartProperties = new Properties();
+
+		if (params != null && !params.isEmpty()) {
+			jobRestartProperties.putAll(params);
+		}
+
+		if (previousJobExecution != null) {
+			JobParameters previousJobParameters = previousJobExecution.getJobParameters();
+
+			if (previousJobParameters != null && !previousJobParameters.isEmpty()) {
+				jobRestartProperties.putAll(previousJobParameters.toProperties());
+			}
+		}
+
+		return jobRestartProperties;
+	}
+
 	/**
 	 * Creates a child {@link ApplicationContext} for the job being requested based upon
 	 * the /META-INF/batch.xml (if exists) and the /META-INF/batch-jobs/<jobName>.xml

spring-batch-core/src/test/java/org/springframework/batch/core/jsr/launch/JsrJobOperatorTests.java

@@ -17,6 +17,7 @@ import java.util.Set;
 
 import javax.batch.operations.JobExecutionIsRunningException;
 import javax.batch.operations.JobOperator;
+import javax.batch.operations.JobRestartException;
 import javax.batch.operations.NoSuchJobException;
 import javax.batch.operations.NoSuchJobExecutionException;
 import javax.batch.operations.NoSuchJobInstanceException;
@@ -374,4 +375,40 @@ public class JsrJobOperatorTests {
 
 		assertEquals(BatchStatus.COMPLETED, jsrJobOperator.getJobExecution(finalExecutionId).getBatchStatus());
 	}
+
+	@Test(expected = JobRestartException.class)
+	public void testRestartAbandoned() {
+		jsrJobOperator = BatchRuntime.getJobOperator();
+
+		long executionId = jsrJobOperator.start("jsrJobOperatorTestRestartAbandonJob", null);
+
+		assertEquals(BatchStatus.FAILED, jsrJobOperator.getJobExecution(executionId).getBatchStatus());
+
+		jsrJobOperator.abandon(executionId);
+		jsrJobOperator.restart(executionId, null);
+	}
+
+	@Test
+	public void testGetNoRestartJobParameters() {
+		JsrJobOperator jobOperator = (JsrJobOperator) jsrJobOperator;
+		Properties properties = jobOperator.getJobRestartProperties(null, null);
+		assertTrue(properties.isEmpty());
+	}
+
+	@Test
+	public void testGetRestartJobParameters() {
+		JsrJobOperator jobOperator = (JsrJobOperator) jsrJobOperator;
+
+		JobExecution jobExecution = new JobExecution(1L,
+			new JobParametersBuilder().addString("prevKey1", "prevVal1").toJobParameters());
+
+		Properties userProperties = new Properties();
+		userProperties.put("userKey1", "userVal1");
+
+		Properties properties = jobOperator.getJobRestartProperties(userProperties, jobExecution);
+
+		assertTrue(properties.size() == 2);
+		assertTrue(properties.getProperty("prevKey1").equals("prevVal1"));
+		assertTrue(properties.getProperty("userKey1").equals("userVal1"));
+	}
 }

spring-batch-core/src/test/java/org/springframework/batch/core/jsr/step/batchlet/FailingBatchlet.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2013 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.core.jsr.step.batchlet;
+
+import javax.batch.api.Batchlet;
+
+/**
+ * <p>
+ * Test batchlet that always fails..
+ * </p>
+ *
+ * @author Chris Schaefer
+ */
+public class FailingBatchlet implements Batchlet {
+	@Override
+	public String process() throws Exception {
+		throw new RuntimeException("process failed");
+	}
+
+	@Override
+	public void stop() throws Exception {
+	}
+}

spring-batch-core/src/test/resources/META-INF/batch-jobs/jsrJobOperatorTestRestartAbandonJob.xml

@@ -0,0 +1,7 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<job id="failingJob" xmlns="http://xmlns.jcp.org/xml/ns/javaee" version="1.0">
+    <step id="step1" >
+        <batchlet ref="failingBatchlet" />
+    </step>
+</job>

spring-batch-core/src/test/resources/META-INF/batch.xml

@@ -1,4 +1,5 @@
 <batch-artifacts xmlns="http://xmlns.jcp.org/xml/ns/javaee">
 	<ref id="testBatchlet" class="org.springframework.batch.core.jsr.step.batchlet.BatchletSupport" />
-	<ref id="restartBatchlet" class="org.springframework.batch.core.jsr.step.batchlet.RestartBatchlet" />
+    <ref id="restartBatchlet" class="org.springframework.batch.core.jsr.step.batchlet.RestartBatchlet" />
+    <ref id="failingBatchlet" class="org.springframework.batch.core.jsr.step.batchlet.FailingBatchlet" />
 </batch-artifacts>