Code review cleanup

spring-cloud-task-integration-tests/src/test/java/configuration/JobConfiguration.java

@@ -16,7 +16,8 @@
 
 package configuration;
 
-import java.util.*;
+import java.util.Arrays;
+import java.util.List;
 
 import org.springframework.batch.core.Job;
 import org.springframework.batch.core.Step;
@@ -68,17 +69,17 @@ public class JobConfiguration {
 
 	@Bean
 	public Step step2() {
-		return stepBuilderFactory.get("step2").chunk(3)
+		return stepBuilderFactory.get("step2").<String, String>chunk(3)
 				.reader(new ListItemReader<>(Arrays.asList("1", "2", "3", "4", "5", "6")))
-				.processor(new ItemProcessor<Object, Object>() {
+				.processor(new ItemProcessor<String, String>() {
 					@Override
-					public String process(Object item) throws Exception {
+					public String process(String item) throws Exception {
 						return String.valueOf(Integer.parseInt((String) item) * -1);
 					}
 				})
-				.writer(new ItemWriter<Object>() {
+				.writer(new ItemWriter<String>() {
 					@Override
-					public void write(List<? extends Object> items) throws Exception {
+					public void write(List<? extends String> items) throws Exception {
 						for (Object item : items) {
 							System.out.println(">> " + item);
 						}

spring-cloud-task-integration-tests/src/test/java/configuration/JobSkipConfiguration.java

@@ -16,8 +16,6 @@
 
 package configuration;
 
-import java.util.*;
-
 import org.springframework.batch.core.Job;
 import org.springframework.batch.core.Step;
 import org.springframework.batch.core.StepContribution;
@@ -27,7 +25,6 @@ import org.springframework.batch.core.configuration.annotation.StepBuilderFactor
 import org.springframework.batch.core.scope.context.ChunkContext;
 import org.springframework.batch.core.step.tasklet.Tasklet;
 import org.springframework.batch.item.ItemProcessor;
-import org.springframework.batch.item.ItemWriter;
 import org.springframework.batch.repeat.RepeatStatus;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.cloud.task.configuration.EnableTask;

spring-cloud-task-integration-tests/src/test/java/configuration/SkipItemWriter.java

@@ -16,13 +16,9 @@
 
 package configuration;
 
-import java.util.*;
+import java.util.List;
 
-import org.springframework.batch.item.ItemReader;
 import org.springframework.batch.item.ItemWriter;
-import org.springframework.batch.item.NonTransientResourceException;
-import org.springframework.batch.item.ParseException;
-import org.springframework.batch.item.UnexpectedInputException;
 
 /**
  * @author Glenn Renfro
@@ -30,11 +26,9 @@ import org.springframework.batch.item.UnexpectedInputException;
 public class SkipItemWriter implements ItemWriter {
 
 	int failCount = 0;
-	boolean finished = false;
 
 	@Override
 	public void write(List items) throws Exception {
-		String result = "1";
 		if(failCount < 2) {
 			failCount++;
 			throw new IllegalStateException("Writer FOOBAR");

spring-cloud-task-integration-tests/src/test/java/org/springframework/cloud/task/listener/BatchExecutionEventTests.java

@@ -17,7 +17,8 @@
 
 package org.springframework.cloud.task.listener;
 
-import java.util.concurrent.*;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 
 import configuration.JobConfiguration;
 import configuration.JobSkipConfiguration;

spring-cloud-task-samples/batch-events/pom.xml

@@ -34,12 +34,6 @@
 			<artifactId>spring-cloud-task-stream</artifactId>
 			<version>1.0.0.BUILD-SNAPSHOT</version>
 		</dependency>
-		<dependency>
-			<groupId>org.springframework.cloud</groupId>
-			<artifactId>spring-cloud-stream-binder-test</artifactId>
-			<version>1.0.0.BUILD-SNAPSHOT</version>
-			<scope>test</scope>
-		</dependency>
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-batch</artifactId>

spring-cloud-task-samples/batch-events/src/test/java/io/spring/cloud/BatchEventsApplicationTests.java

@@ -16,7 +16,8 @@
 
 package io.spring.cloud;
 
-import java.util.concurrent.*;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 
 import org.junit.Assert;
 import org.junit.ClassRule;
@@ -33,13 +34,8 @@ import org.springframework.cloud.stream.test.junit.redis.RedisTestSupport;
 import org.springframework.cloud.task.batch.listener.support.JobExecutionEvent;
 import org.springframework.context.annotation.PropertySource;
 
-import static junit.framework.Assert.assertEquals;
-import static junit.framework.Assert.assertTrue;
-
 public class BatchEventsApplicationTests {
 
-	private static final String ITEM_INDICATOR = ">> -";
-
 	@ClassRule
 	public static RedisTestSupport redisTestSupport = new RedisTestSupport();
 

spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/BatchEventAutoConfiguration.java

@@ -38,7 +38,18 @@ import org.springframework.integration.gateway.GatewayProxyFactoryBean;
 import org.springframework.messaging.MessageChannel;
 
 /**
- * Configures the listeners and channels that are required to emit job messages.
+ * Autoconfigures Spring Batch listeners designed to emit events on the following channels:
+ *
+ * <ul>
+ *     <li>{@link EventEmittingJobExecutionListener} - job-execution-events</li>
+ *     <li>{@link EventEmittingStepExecutionListener} - step-execution-events</li>
+ *     <li>{@link ChunkListener} - chunk-events</li>
+ *     <li>{@link EventEmittingItemReadListener} - item-read-events</li>
+ *     <li>{@link EventEmittingItemProcessListener} - item-process-events</li>
+ *     <li>{@link EventEmittingItemWriteListener} - item-write-events</li>
+ *     <li>{@link EventEmittingSkipListener} - skip-events</li>
+ * </ul>
+ *
  * @author Michael Minella
  * @author Glenn Renfro
  */
@@ -93,12 +104,12 @@ public class BatchEventAutoConfiguration {
 
 		@Bean
 		public ItemReadListener itemReadEventsListener() {
-			return new EventEmittingItemReadEventsListener(listenerChannels.itemReadEvents());
+			return new EventEmittingItemReadListener(listenerChannels.itemReadEvents());
 		}
 
 		@Bean
 		public ItemWriteListener itemWriteEventsListener() {
-			return new EventEmittingItemWriteEventsListener(listenerChannels.itemWriteEvents());
+			return new EventEmittingItemWriteListener(listenerChannels.itemWriteEvents());
 		}
 
 		@Bean

spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingItemProcessListener.java

@@ -16,25 +16,31 @@
 package org.springframework.cloud.task.batch.listener;
 
 import org.springframework.batch.core.ItemProcessListener;
+import org.springframework.cloud.task.batch.listener.support.BatchJobHeaders;
 import org.springframework.cloud.task.batch.listener.support.MessagePublisher;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.util.Assert;
 
 /**
- * Setups up the ItemProcessListener to emit events to the spring cloud stream output channel.
+ * Provides informational messages around the {@link ItemProcessListener} of a batch job.
+ *
+ * The {@link ItemProcessListener#beforeProcess(Object)} of this listener is a no-op.
+ * {@link ItemProcessListener#afterProcess(Object, Object)} returns a message if an item
+ * was filtered ({@link ItemProcessor} returned null), if the result of the processor was
+ * equal to the input (via <code>.equals</code>), or if they were not equal.
+ * {@link ItemProcessListener#onProcessError(Object, Exception)} provides the exception
+ * via the {@link BatchJobHeaders.BATCH_EXCEPTION} message header.
  *
  * @author Michael Minella
  * @author Glenn Renfro
  */
 public class EventEmittingItemProcessListener implements ItemProcessListener {
 
-	private MessageChannel output;
-	private MessagePublisher<Object> messagePublisher;
+	private MessagePublisher<String> messagePublisher;
 
 	public EventEmittingItemProcessListener(MessageChannel output) {
 		Assert.notNull(output, "An output channel is required");
-		this.output = output;
-		this.messagePublisher = new MessagePublisher(output);
+		this.messagePublisher = new MessagePublisher<>(output);
 	}
 
 	@Override

spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingItemReadListener.java

@@ -16,28 +16,33 @@
 
 package org.springframework.cloud.task.batch.listener;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
 import org.springframework.batch.core.ItemReadListener;
+import org.springframework.cloud.task.batch.listener.support.BatchJobHeaders;
 import org.springframework.cloud.task.batch.listener.support.MessagePublisher;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.util.Assert;
 
 /**
- *  Setups up the ItemReadEventsListener to emit events to the spring cloud stream output channel.
+ *  Provides informational messages around the {@link ItemReader} of a batch job.
+ *
+ *  The {@link ItemReadListener#beforeRead()} and
+ *  {@link ItemReadListener#afterRead(Object)} are both no-ops in this implementation.
+ *  {@link ItemReadListener#onReadError(Exception)} provides the exception
+ * via the {@link BatchJobHeaders.BATCH_EXCEPTION} message header.
  *
  * @author Glenn Renfro
  */
-public class EventEmittingItemReadEventsListener implements ItemReadListener {
+public class EventEmittingItemReadListener implements ItemReadListener {
 
-	private static final Logger logger = LoggerFactory.getLogger(EventEmittingItemReadEventsListener.class);
+	private static final Log logger = LogFactory.getLog(EventEmittingItemReadListener.class);
 
-	private MessageChannel output;
-	private MessagePublisher<Object> messagePublisher;
+	private MessagePublisher<String> messagePublisher;
 
-	public EventEmittingItemReadEventsListener(MessageChannel output) {
+	public EventEmittingItemReadListener(MessageChannel output) {
 		Assert.notNull(output, "An output channel is required");
-		this.output = output;
 		this.messagePublisher = new MessagePublisher(output);
 	}
 
@@ -56,6 +61,7 @@ public class EventEmittingItemReadEventsListener implements ItemReadListener {
 		if (logger.isDebugEnabled()) {
 			logger.debug("Executing onReadError: " + ex.getMessage(), ex);
 		}
-		this.messagePublisher.publish(ex.getMessage());
+
+		messagePublisher.publishWithThrowableHeader("Exception while item was being read", ex.getMessage());
 	}
 }

spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingItemWriteListener.java

@@ -16,11 +16,13 @@
 
 package org.springframework.cloud.task.batch.listener;
 
-import java.util.*;
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.springframework.batch.core.ItemWriteListener;
+import org.springframework.cloud.task.batch.listener.support.BatchJobHeaders;
 import org.springframework.cloud.task.batch.listener.support.MessagePublisher;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.util.Assert;
@@ -28,24 +30,26 @@ import org.springframework.util.Assert;
 /**
  *  Setups up the ItemWriteEventsListener to emit events to the spring cloud stream output channel.
  *
+ *  Each method provides an informational message.
+ *  {@link ItemWriteListener#onWriteError(Exception, List)} provides a message as well as
+ *  the exception's message via the {@link BatchJobHeaders.BATCH_EXCEPTION} message header.
+ *
  * @author Glenn Renfro
  */
-public class EventEmittingItemWriteEventsListener implements ItemWriteListener{
+public class EventEmittingItemWriteListener implements ItemWriteListener{
 
-	private static final Logger logger = LoggerFactory.getLogger(EventEmittingItemWriteEventsListener.class);
+	private static final Log logger = LogFactory.getLog(EventEmittingItemWriteListener.class);
 
-	private MessageChannel output;
-	private MessagePublisher<Object> messagePublisher;
+	private MessagePublisher<String> messagePublisher;
 
-	public EventEmittingItemWriteEventsListener(MessageChannel output) {
+	public EventEmittingItemWriteListener(MessageChannel output) {
 		Assert.notNull(output, "An output channel is required");
-		this.output = output;
-		this.messagePublisher = new MessagePublisher(output);
+		this.messagePublisher = new MessagePublisher<>(output);
 	}
 
 	@Override
 	public void beforeWrite(List items) {
-		messagePublisher.publish(items.size() + " items to be written.");
+		this.messagePublisher.publish(items.size() + " items to be written.");
 	}
 
 	@Override

spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingJobExecutionListener.java

@@ -23,20 +23,17 @@ import org.springframework.messaging.MessageChannel;
 import org.springframework.util.Assert;
 
 /**
- * Setups up the StepExecutionListener to emit events to the spring cloud stream output channel.
+ * Provides {@link JobExecutionEvent} at both the start and end of the job's execution.
  *
  * @author Michael Minella
  * @author Glenn Renfro
  */
 public class EventEmittingJobExecutionListener implements JobExecutionListener {
 
-	private MessageChannel output;
-
 	private MessagePublisher<JobExecutionEvent> messagePublisher;
 
 	public EventEmittingJobExecutionListener(MessageChannel output) {
 		Assert.notNull(output, "An output channel is required");
-		this.output = output;
 		this.messagePublisher = new MessagePublisher<>(output);
 	}
 

spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingSkipListener.java

@@ -16,33 +16,35 @@
 
 package org.springframework.cloud.task.batch.listener;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
 import org.springframework.batch.core.SkipListener;
 import org.springframework.cloud.task.batch.listener.support.BatchJobHeaders;
 import org.springframework.cloud.task.batch.listener.support.MessagePublisher;
-import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
-import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.util.Assert;
 
 /**
  * Setups up the SkipProcessListener to emit events to the spring cloud stream output channel.
  *
+ * This listener emits the exception's message via the
+ * {@link BatchJobHeaders.BATCH_EXCEPTION} message header for each method.  For
+ * {@link SkipListener#onSkipInProcess(Object, Throwable)} and
+ * {@link SkipListener#onSkipInWrite(Object, Throwable)} the body of the message consists
+ * of the item that caused the error.
+ *
  * @author Glenn Renfro
  */
 public class EventEmittingSkipListener implements SkipListener {
 
-	private static final Logger logger = LoggerFactory.getLogger(EventEmittingSkipListener.class);
-
-	private MessageChannel output;
+	private static final Log logger = LogFactory.getLog(EventEmittingSkipListener.class);
 
 	private MessagePublisher<Object> messagePublisher;
 
 	public EventEmittingSkipListener(MessageChannel output) {
 		Assert.notNull(output, "An output channel is required");
-		this.output = output;
-		this.messagePublisher = new MessagePublisher(output);
+		this.messagePublisher = new MessagePublisher<>(output);
 	}
 
 	@Override
@@ -51,7 +53,6 @@ public class EventEmittingSkipListener implements SkipListener {
 			logger.debug("Executing onSkipInRead: " + t.getMessage(), t);
 		}
 		messagePublisher.publishWithThrowableHeader("Skipped when reading.", t.getMessage());
-
 	}
 
 	@Override

spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingStepExecutionListener.java

@@ -24,20 +24,19 @@ import org.springframework.messaging.MessageChannel;
 import org.springframework.util.Assert;
 
 /**
- * Setups up the StepExecutionListener to emit events to the spring cloud stream output channel.
+ * Provides a {@link StepExecutionEvent} at the start and end of each step indicating the
+ * step's status.  The {@link StepExecutionListener#afterStep(StepExecution)} returns the
+ * {@link ExitStatus} of the inputted {@link StepExecution}.
  *
  * @author Michael Minella
  * @author Glenn Renfro
  */
 public class EventEmittingStepExecutionListener implements StepExecutionListener {
 
-	private MessageChannel output;
-
 	private MessagePublisher<StepExecutionEvent> messagePublisher;
 
 	public EventEmittingStepExecutionListener(MessageChannel output) {
 		Assert.notNull(output, "An output channel is required");
-		this.output = output;
 		this.messagePublisher = new MessagePublisher<>(output);
 	}
 

spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/ExitStatus.java

@@ -16,6 +16,8 @@
 
 package org.springframework.cloud.task.batch.listener.support;
 
+import org.springframework.util.Assert;
+
 /**
  * ExitStatus DTO created so that {@link org.springframework.batch.core.ExitStatus} can be serialized into Json without
  * having to add mixins to an ObjectMapper
@@ -28,12 +30,13 @@ public class ExitStatus {
 	private String exitDescription;
 
 	public ExitStatus(){
-
 	}
 
-	public ExitStatus(String exitCode, String exitDescription) {
-		this.exitCode = exitCode;
-		this.exitDescription = exitDescription;
+	public ExitStatus(org.springframework.batch.core.ExitStatus exitStatus) {
+		Assert.notNull(exitStatus);
+
+		this.exitCode = exitStatus.getExitCode();
+		this.exitDescription = exitStatus.getExitDescription();
 	}
 
 	public String getExitCode() {

spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/JobExecutionEvent.java

@@ -16,11 +16,22 @@
 
 package org.springframework.cloud.task.batch.listener.support;
 
-import java.io.*;
-import java.util.*;
-import java.util.concurrent.*;
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Date;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CopyOnWriteArraySet;
 
-import org.springframework.batch.core.*;
+import org.springframework.batch.core.BatchStatus;
+import org.springframework.batch.core.Entity;
+import org.springframework.batch.core.JobExecution;
+import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.item.ExecutionContext;
 
 /**
@@ -34,23 +45,23 @@ public class JobExecutionEvent extends Entity {
 
 	private JobInstanceEvent jobInstance;
 
-	private volatile Collection<StepExecutionEvent> stepExecutions = new CopyOnWriteArraySet<StepExecutionEvent>();
+	private Collection<StepExecutionEvent> stepExecutions = new CopyOnWriteArraySet<StepExecutionEvent>();
 
-	private volatile BatchStatus status = BatchStatus.STARTING;
+	private BatchStatus status = BatchStatus.STARTING;
 
-	private volatile Date startTime = null;
+	private Date startTime = null;
 
-	private volatile Date createTime = new Date(System.currentTimeMillis());
+	private Date createTime = new Date(System.currentTimeMillis());
 
-	private volatile Date endTime = null;
+	private Date endTime = null;
 
-	private volatile Date lastUpdated = null;
+	private Date lastUpdated = null;
 
-	private volatile ExitStatus exitStatus = new ExitStatus("UNKNOWN", null);
+	private ExitStatus exitStatus = new ExitStatus(new org.springframework.batch.core.ExitStatus("UNKNOWN", null));
 
-	private volatile ExecutionContext executionContext = new ExecutionContext();
+	private ExecutionContext executionContext = new ExecutionContext();
 
-	private transient volatile List<Throwable> failureExceptions = new CopyOnWriteArrayList<Throwable>();
+	private List<Throwable> failureExceptions = new CopyOnWriteArrayList<Throwable>();
 
 	private String jobConfigurationName;
 
@@ -74,8 +85,7 @@ public class JobExecutionEvent extends Entity {
 		this.createTime = original.getCreateTime();
 		this.endTime = original.getEndTime();
 		this.lastUpdated = original.getLastUpdated();
-		this.exitStatus = new ExitStatus(original.getExitStatus().getExitCode(),
-				original.getExitStatus().getExitDescription());
+		this.exitStatus = new ExitStatus(original.getExitStatus());
 		this.executionContext = original.getExecutionContext();
 		this.failureExceptions = original.getFailureExceptions();
 		this.jobConfigurationName = original.getJobConfigurationName();
@@ -88,7 +98,7 @@ public class JobExecutionEvent extends Entity {
 	}
 
 	public Date getEndTime() {
-		return endTime;
+		return this.endTime;
 	}
 
 	public void setJobInstance(JobInstanceEvent jobInstance) {
@@ -108,7 +118,7 @@ public class JobExecutionEvent extends Entity {
 	}
 
 	public BatchStatus getStatus() {
-		return status;
+		return this.status;
 	}
 
 	/**
@@ -155,14 +165,14 @@ public class JobExecutionEvent extends Entity {
 	 * @return the exitCode
 	 */
 	public ExitStatus getExitStatus() {
-		return exitStatus;
+		return this.exitStatus;
 	}
 
 	/**
 	 * @return the Job that is executing.
 	 */
 	public JobInstanceEvent getJobInstance() {
-		return jobInstance;
+		return this.jobInstance;
 	}
 
 	/**
@@ -171,38 +181,7 @@ public class JobExecutionEvent extends Entity {
 	 * @return the step executions that were registered
 	 */
 	public Collection<StepExecutionEvent> getStepExecutions() {
-		return Collections.unmodifiableList(new ArrayList<StepExecutionEvent>(stepExecutions));
-	}
-
-	/**
-	 * Test if this {@link JobExecution} indicates that it is running. It should
-	 * be noted that this does not necessarily mean that it has been persisted
-	 * as such yet.
-	 * @return true if the end time is null
-	 */
-	public boolean isRunning() {
-		return endTime == null;
-	}
-
-	/**
-	 * Test if this {@link JobExecution} indicates that it has been signalled to
-	 * stop.
-	 * @return true if the status is {@link BatchStatus#STOPPING}
-	 */
-	public boolean isStopping() {
-		return status == BatchStatus.STOPPING;
-	}
-
-	/**
-	 * Signal the {@link JobExecution} to stop. Iterates through the associated
-	 * {@link StepExecution}s, calling {@link StepExecution#setTerminateOnly()}.
-	 *
-	 */
-	public void stop() {
-		for (StepExecutionEvent stepExecution : stepExecutions) {
-			stepExecution.setTerminateOnly();
-		}
-		status = BatchStatus.STOPPING;
+		return Collections.unmodifiableList(new ArrayList<>(this.stepExecutions));
 	}
 
 	/**
@@ -221,14 +200,14 @@ public class JobExecutionEvent extends Entity {
 	 * @return the context
 	 */
 	public ExecutionContext getExecutionContext() {
-		return executionContext;
+		return this.executionContext;
 	}
 
 	/**
 	 * @return the time when this execution was created.
 	 */
 	public Date getCreateTime() {
-		return createTime;
+		return this.createTime;
 	}
 
 	/**
@@ -242,15 +221,6 @@ public class JobExecutionEvent extends Entity {
 		return this.jobConfigurationName;
 	}
 
-	/**
-	 * Package private method for re-constituting the step executions from
-	 * existing instances.
-	 * @param stepExecution
-	 */
-	void addStepExecution(StepExecutionEvent stepExecution) {
-		stepExecutions.add(stepExecution);
-	}
-
 	/**
 	 * Get the date representing the last time this JobExecution was updated in
 	 * the JobRepository.
@@ -258,7 +228,7 @@ public class JobExecutionEvent extends Entity {
 	 * @return Date representing the last time this JobExecution was updated.
 	 */
 	public Date getLastUpdated() {
-		return lastUpdated;
+		return this.lastUpdated;
 	}
 
 	/**
@@ -271,7 +241,7 @@ public class JobExecutionEvent extends Entity {
 	}
 
 	public List<Throwable> getFailureExceptions() {
-		return failureExceptions;
+		return this.failureExceptions;
 	}
 
 	/**
@@ -292,12 +262,12 @@ public class JobExecutionEvent extends Entity {
 	 */
 	public synchronized List<Throwable> getAllFailureExceptions() {
 
-		Set<Throwable> allExceptions = new HashSet<Throwable>(failureExceptions);
-		for (StepExecutionEvent stepExecution : stepExecutions) {
+		Set<Throwable> allExceptions = new HashSet<>(this.failureExceptions);
+		for (StepExecutionEvent stepExecution : this.stepExecutions) {
 			allExceptions.addAll(stepExecution.getFailureExceptions());
 		}
 
-		return new ArrayList<Throwable>(allExceptions);
+		return new ArrayList<>(allExceptions);
 	}
 
 	/**
@@ -306,7 +276,7 @@ public class JobExecutionEvent extends Entity {
 	 */
 	private void readObject(ObjectInputStream stream) throws IOException, ClassNotFoundException {
 		stream.defaultReadObject();
-		failureExceptions = new ArrayList<Throwable>();
+		this.failureExceptions = new ArrayList<>();
 	}
 
 	/*
@@ -320,17 +290,4 @@ public class JobExecutionEvent extends Entity {
 				+ String.format(", startTime=%s, endTime=%s, lastUpdated=%s, status=%s, exitStatus=%s, job=[%s], jobParameters=[%s]",
 				startTime, endTime, lastUpdated, status, exitStatus, jobInstance, jobParameters);
 	}
-
-	/**
-	 * Add some step executions.  For internal use only.
-	 * @param stepExecutions step executions to add to the current list
-	 */
-	public void addStepExecutions(List<StepExecutionEvent> stepExecutions) {
-		if (stepExecutions!=null) {
-			this.stepExecutions.removeAll(stepExecutions);
-			this.stepExecutions.addAll(stepExecutions);
-		}
-	}
-
-
 }

spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/JobInstanceEvent.java

@@ -26,7 +26,7 @@ import org.springframework.util.Assert;
  * @author Glenn Renfro
  */
 
-public class JobInstanceEvent extends Entity implements javax.batch.runtime.JobInstance {
+public class JobInstanceEvent extends Entity {
 
 	private String jobName;
 
@@ -43,17 +43,14 @@ public class JobInstanceEvent extends Entity implements javax.batch.runtime.JobI
 	/**
 	 * @return the job name. (Equivalent to getJob().getName())
 	 */
-	@Override
 	public String getJobName() {
 		return jobName;
 	}
 
-	@Override
 	public String toString() {
 		return super.toString() + ", Job=[" + jobName + "]";
 	}
 
-	@Override
 	public long getInstanceId() {
 		return super.getId();
 	}

spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/JobParameterEvent.java

@@ -16,8 +16,9 @@
 
 package org.springframework.cloud.task.batch.listener.support;
 
-import java.util.*;
+import java.util.Date;
 
+import org.springframework.batch.core.JobParameter;
 
 /**
  * This is a JobParameter DTO created so that a {@link org.springframework.batch.core.JobParameter} can be serialized
@@ -32,92 +33,12 @@ public class JobParameterEvent {
 	private boolean identifying;
 
 	public JobParameterEvent() {
-
 	}
 
-	/**
-	 * Construct a new JobParameter as a String.
-	 */
-	public JobParameterEvent(String parameter, boolean identifying) {
-		this.parameter = parameter;
-		parameterType = JobParameterEvent.ParameterType.STRING;
-		this.identifying = identifying;
-	}
-
-	/**
-	 * Construct a new JobParameter as a Long.
-	 *
-	 * @param parameter
-	 */
-	public JobParameterEvent(Long parameter, boolean identifying) {
-		this.parameter = parameter;
-		parameterType = JobParameterEvent.ParameterType.LONG;
-		this.identifying = identifying;
-	}
-
-	/**
-	 * Construct a new JobParameter as a Date.
-	 *
-	 * @param parameter
-	 */
-	public JobParameterEvent(Date parameter, boolean identifying) {
-		this.parameter = parameter;
-		parameterType = JobParameterEvent.ParameterType.DATE;
-		this.identifying = identifying;
-	}
-
-	/**
-	 * Construct a new JobParameter as a Double.
-	 *
-	 * @param parameter
-	 */
-	public JobParameterEvent(Double parameter, boolean identifying) {
-		this.parameter = parameter;
-		parameterType = JobParameterEvent.ParameterType.DOUBLE;
-		this.identifying = identifying;
-	}
-
-
-	/**
-	 * Construct a new JobParameter as a String.
-	 */
-	public JobParameterEvent(String parameter) {
-		this.parameter = parameter;
-		parameterType = JobParameterEvent.ParameterType.STRING;
-		this.identifying = true;
-	}
-
-	/**
-	 * Construct a new JobParameter as a Long.
-	 *
-	 * @param parameter
-	 */
-	public JobParameterEvent(Long parameter) {
-		this.parameter = parameter;
-		parameterType = JobParameterEvent.ParameterType.LONG;
-		this.identifying = true;
-	}
-
-	/**
-	 * Construct a new JobParameter as a Date.
-	 *
-	 * @param parameter
-	 */
-	public JobParameterEvent(Date parameter) {
-		this.parameter = parameter;
-		parameterType = JobParameterEvent.ParameterType.DATE;
-		this.identifying = true;
-	}
-
-	/**
-	 * Construct a new JobParameter as a Double.
-	 *
-	 * @param parameter
-	 */
-	public JobParameterEvent(Double parameter) {
-		this.parameter = parameter;
-		parameterType = JobParameterEvent.ParameterType.DOUBLE;
-		this.identifying = true;
+	public JobParameterEvent(JobParameter jobParameter) {
+		this.parameter = jobParameter.getValue();
+		this.parameterType = ParameterType.convert(jobParameter.getType());
+		this.identifying = jobParameter.isIdentifying();
 	}
 
 	public boolean isIdentifying() {
@@ -146,7 +67,7 @@ public class JobParameterEvent {
 
 	@Override
 	public boolean equals(Object obj) {
-		if (obj instanceof JobParameterEvent == false) {
+		if (!(obj instanceof JobParameterEvent)) {
 			return false;
 		}
 
@@ -173,7 +94,24 @@ public class JobParameterEvent {
 	 * Enumeration representing the type of a JobParameter.
 	 */
 	public enum ParameterType {
-
 		STRING, DATE, LONG, DOUBLE;
+
+		public static ParameterType convert(JobParameter.ParameterType type) {
+			if(JobParameter.ParameterType.DATE.equals(type)) {
+				return DATE;
+			}
+			else if(JobParameter.ParameterType.DOUBLE.equals(type)) {
+				return DOUBLE;
+			}
+			else if(JobParameter.ParameterType.LONG.equals(type)) {
+				return LONG;
+			}
+			else if(JobParameter.ParameterType.STRING.equals(type)) {
+				return STRING;
+			}
+			else {
+				throw new IllegalArgumentException("Unable to convert type");
+			}
+		}
 	}
 }

spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/JobParametersEvent.java

@@ -16,7 +16,10 @@
 
 package org.springframework.cloud.task.batch.listener.support;
 
-import java.util.*;
+import java.util.Date;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Properties;
 
 import org.springframework.batch.core.JobParameter;
 
@@ -30,23 +33,23 @@ public class JobParametersEvent {
 	private final Map<String,JobParameterEvent> parameters;
 
 	public JobParametersEvent() {
-		this.parameters = new LinkedHashMap<String, JobParameterEvent>();
+		this.parameters = new LinkedHashMap<>();
 	}
 
 	public JobParametersEvent(Map<String,JobParameter> jobParameters) {
-		this.parameters = new LinkedHashMap<String,JobParameterEvent>();
+		this.parameters = new LinkedHashMap<>();
 		for(Map.Entry<String, JobParameter> entry: jobParameters.entrySet()){
 			if(entry.getValue().getValue() instanceof String){
-				parameters.put(entry.getKey(), new JobParameterEvent(((String) entry.getValue().getValue()), entry.getValue().isIdentifying()));
+				parameters.put(entry.getKey(), new JobParameterEvent(entry.getValue()));
 			}
 			else if(entry.getValue().getValue() instanceof Long){
-				parameters.put(entry.getKey(), new JobParameterEvent(((Long) entry.getValue().getValue()), entry.getValue().isIdentifying()));
+				parameters.put(entry.getKey(), new JobParameterEvent(entry.getValue()));
 			}
 			else if(entry.getValue().getValue() instanceof Date){
-				parameters.put(entry.getKey(), new JobParameterEvent(((Date) entry.getValue().getValue()), entry.getValue().isIdentifying()));
+				parameters.put(entry.getKey(), new JobParameterEvent(entry.getValue()));
 			}
 			else if(entry.getValue().getValue() instanceof Double){
-				parameters.put(entry.getKey(), new JobParameterEvent(((Double) entry.getValue().getValue()), entry.getValue().isIdentifying()));
+				parameters.put(entry.getKey(), new JobParameterEvent(entry.getValue()));
 			}
 		}
 	}

spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/StepExecutionEvent.java

@@ -16,13 +16,15 @@
 
 package org.springframework.cloud.task.batch.listener.support;
 
-import java.io.*;
-import java.util.*;
-import java.util.concurrent.*;
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
 
 import org.springframework.batch.core.BatchStatus;
 import org.springframework.batch.core.Entity;
-
 import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.item.ExecutionContext;
 import org.springframework.util.Assert;
@@ -34,41 +36,41 @@ import org.springframework.util.Assert;
  */
 public class StepExecutionEvent extends Entity {
 
-	private volatile long jobExecutionId;
+	private long jobExecutionId;
 
-	private volatile String stepName;
+	private String stepName;
 
-	private volatile BatchStatus status = BatchStatus.STARTING;
+	private BatchStatus status = BatchStatus.STARTING;
 
-	private volatile int readCount = 0;
+	private int readCount = 0;
 
-	private volatile int writeCount = 0;
+	private int writeCount = 0;
 
-	private volatile int commitCount = 0;
+	private int commitCount = 0;
 
-	private volatile int rollbackCount = 0;
+	private int rollbackCount = 0;
 
-	private volatile int readSkipCount = 0;
+	private int readSkipCount = 0;
 
-	private volatile int processSkipCount = 0;
+	private int processSkipCount = 0;
 
-	private volatile int writeSkipCount = 0;
+	private int writeSkipCount = 0;
 
-	private volatile Date startTime = new Date(System.currentTimeMillis());
+	private Date startTime = new Date(System.currentTimeMillis());
 
-	private volatile Date endTime = null;
+	private Date endTime = null;
 
-	private volatile Date lastUpdated = null;
+	private Date lastUpdated = null;
 
-	private volatile ExecutionContext executionContext = new ExecutionContext();
+	private ExecutionContext executionContext = new ExecutionContext();
 
-	private volatile ExitStatus exitStatus = new ExitStatus("EXECUTING", null);
+	private ExitStatus exitStatus = new ExitStatus(org.springframework.batch.core.ExitStatus.EXECUTING);
 
-	private volatile boolean terminateOnly;
+	private boolean terminateOnly;
 
-	private volatile int filterCount;
+	private int filterCount;
 
-	private transient volatile List<Throwable> failureExceptions = new CopyOnWriteArrayList<Throwable>();
+	private List<Throwable> failureExceptions = new CopyOnWriteArrayList<Throwable>();
 
 
 	public StepExecutionEvent() {
@@ -89,8 +91,7 @@ public class StepExecutionEvent extends Entity {
 		this.stepName = stepExecution.getStepName();
 
 		this.status = stepExecution.getStatus();
-		this.exitStatus = new ExitStatus(stepExecution.getExitStatus().getExitCode(),
-				stepExecution.getExitStatus().getExitDescription());
+		this.exitStatus = new ExitStatus(stepExecution.getExitStatus());
 		this.executionContext = stepExecution.getExecutionContext();
 		for (Throwable throwable : stepExecution.getFailureExceptions()){
 			this.failureExceptions.add(throwable);
@@ -118,7 +119,7 @@ public class StepExecutionEvent extends Entity {
 	 * @return the attributes
 	 */
 	public ExecutionContext getExecutionContext() {
-		return executionContext;
+		return this.executionContext;
 	}
 
 	/**
@@ -136,7 +137,7 @@ public class StepExecutionEvent extends Entity {
 	 * @return the current number of commits
 	 */
 	public int getCommitCount() {
-		return commitCount;
+		return this.commitCount;
 	}
 
 	/**
@@ -154,7 +155,7 @@ public class StepExecutionEvent extends Entity {
 	 * @return the time that this execution ended
 	 */
 	public Date getEndTime() {
-		return endTime;
+		return this.endTime;
 	}
 
 	/**
@@ -172,7 +173,7 @@ public class StepExecutionEvent extends Entity {
 	 * @return the current number of items read for this execution
 	 */
 	public int getReadCount() {
-		return readCount;
+		return this.readCount;
 	}
 
 	/**
@@ -190,7 +191,7 @@ public class StepExecutionEvent extends Entity {
 	 * @return the current number of items written for this execution
 	 */
 	public int getWriteCount() {
-		return writeCount;
+		return this.writeCount;
 	}
 
 	/**
@@ -208,7 +209,7 @@ public class StepExecutionEvent extends Entity {
 	 * @return the current number of rollbacks for this execution
 	 */
 	public int getRollbackCount() {
-		return rollbackCount;
+		return this.rollbackCount;
 	}
 
 	/**
@@ -217,7 +218,7 @@ public class StepExecutionEvent extends Entity {
 	 * @return the current number of items filtered out of this execution
 	 */
 	public int getFilterCount() {
-		return filterCount;
+		return this.filterCount;
 	}
 
 	/**
@@ -242,7 +243,7 @@ public class StepExecutionEvent extends Entity {
 	 * @return the time this execution started
 	 */
 	public Date getStartTime() {
-		return startTime;
+		return this.startTime;
 	}
 
 	/**
@@ -260,7 +261,7 @@ public class StepExecutionEvent extends Entity {
 	 * @return the current status of this step
 	 */
 	public BatchStatus getStatus() {
-		return status;
+		return this.status;
 	}
 
 	/**
@@ -272,17 +273,6 @@ public class StepExecutionEvent extends Entity {
 		this.status = status;
 	}
 
-	/**
-	 * Upgrade the status field if the provided value is greater than the
-	 * existing one. Clients using this method to set the status can be sure
-	 * that they don't overwrite a failed status with an successful one.
-	 *
-	 * @param status the new status value
-	 */
-	public void upgradeStatus(BatchStatus status) {
-		this.status = this.status.upgradeTo(status);
-	}
-
 	public void setStepName(String stepName) {
 		this.stepName = stepName;
 	}
@@ -290,7 +280,7 @@ public class StepExecutionEvent extends Entity {
 	 * @return the name of the step
 	 */
 	public String getStepName() {
-		return stepName;
+		return this.stepName;
 	}
 
 	/**
@@ -304,14 +294,7 @@ public class StepExecutionEvent extends Entity {
 	 * @return the exitCode
 	 */
 	public ExitStatus getExitStatus() {
-		return exitStatus;
-	}
-
-	/**
-	 * On unsuccessful execution after a chunk has rolled back.
-	 */
-	public synchronized void incrementRollbackCount() {
-		rollbackCount++;
+		return this.exitStatus;
 	}
 
 	/**
@@ -333,28 +316,28 @@ public class StepExecutionEvent extends Entity {
 	 * @return the total number of items skipped.
 	 */
 	public int getSkipCount() {
-		return readSkipCount + processSkipCount + writeSkipCount;
+		return this.readSkipCount + this.processSkipCount + this.writeSkipCount;
 	}
 
 	/**
 	 * Increment the number of commits
 	 */
 	public void incrementCommitCount() {
-		commitCount++;
+		this.commitCount++;
 	}
 
 	/**
 	 * @return the number of records skipped on read
 	 */
 	public int getReadSkipCount() {
-		return readSkipCount;
+		return this.readSkipCount;
 	}
 
 	/**
 	 * @return the number of records skipped on write
 	 */
 	public int getWriteSkipCount() {
-		return writeSkipCount;
+		return this.writeSkipCount;
 	}
 
 	/**
@@ -379,7 +362,7 @@ public class StepExecutionEvent extends Entity {
 	 * @return the number of records skipped during processing
 	 */
 	public int getProcessSkipCount() {
-		return processSkipCount;
+		return this.processSkipCount;
 	}
 
 	/**
@@ -395,7 +378,7 @@ public class StepExecutionEvent extends Entity {
 	 * @return the Date representing the last time this execution was persisted.
 	 */
 	public Date getLastUpdated() {
-		return lastUpdated;
+		return this.lastUpdated;
 	}
 
 	/**
@@ -408,15 +391,11 @@ public class StepExecutionEvent extends Entity {
 	}
 
 	public List<Throwable> getFailureExceptions() {
-		return failureExceptions;
-	}
-
-	public void addFailureException(Throwable throwable) {
-		this.failureExceptions.add(throwable);
+		return this.failureExceptions;
 	}
 
 	public long getJobExecutionId() {
-		return jobExecutionId;
+		return this.jobExecutionId;
 	}
 
 	/*
@@ -434,7 +413,7 @@ public class StepExecutionEvent extends Entity {
 		}
 		StepExecution other = (StepExecution) obj;
 
-		return stepName.equals(other.getStepName()) && (jobExecutionId == other.getJobExecutionId())
+		return this.stepName.equals(other.getStepName()) && (this.jobExecutionId == other.getJobExecutionId())
 				&& getId().equals(other.getId());
 	}
 
@@ -444,7 +423,7 @@ public class StepExecutionEvent extends Entity {
 	 */
 	private void readObject(ObjectInputStream stream) throws IOException, ClassNotFoundException {
 		stream.defaultReadObject();
-		failureExceptions = new ArrayList<Throwable>();
+		this.failureExceptions = new ArrayList<>();
 	}
 
 	/*
@@ -456,21 +435,21 @@ public class StepExecutionEvent extends Entity {
 	public int hashCode() {
 		Object jobExecutionId = getJobExecutionId();
 		Long id = getId();
-		return super.hashCode() + 31 * (stepName != null ? stepName.hashCode() : 0) + 91
+		return super.hashCode() + 31 * (this.stepName != null ? this.stepName.hashCode() : 0) + 91
 				* (jobExecutionId != null ? jobExecutionId.hashCode() : 0) + 59 * (id != null ? id.hashCode() : 0);
 	}
 
 	@Override
 	public String toString() {
-		return String.format(getSummary() + ", exitDescription=%s", exitStatus.getExitDescription());
+		return String.format(getSummary() + ", exitDescription=%s", this.exitStatus.getExitDescription());
 	}
 
 	public String getSummary() {
 		return super.toString()
 				+ String.format(
 				", name=%s, status=%s, exitStatus=%s, readCount=%d, filterCount=%d, writeCount=%d readSkipCount=%d, writeSkipCount=%d"
-						+ ", processSkipCount=%d, commitCount=%d, rollbackCount=%d", stepName, status,
-				exitStatus.getExitCode(), readCount, filterCount, writeCount, readSkipCount, writeSkipCount,
-				processSkipCount, commitCount, rollbackCount);
+						+ ", processSkipCount=%d, commitCount=%d, rollbackCount=%d", this.stepName, this.status,
+				this.exitStatus.getExitCode(), this.readCount, this.filterCount, this.writeCount, this.readSkipCount, this.writeSkipCount,
+				this.processSkipCount, this.commitCount, this.rollbackCount);
 	}
 }

spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/TaskBatchEventListenerBeanPostProcessor.java

@@ -15,7 +15,7 @@
  */
 package org.springframework.cloud.task.batch.listener.support;
 
-import java.lang.reflect.*;
+import java.lang.reflect.Field;
 
 import org.springframework.batch.core.ChunkListener;
 import org.springframework.batch.core.ItemProcessListener;

spring-cloud-task-stream/src/test/java/org/springframework/cloud/task/batch/listener/EventJobExecutionTests.java

@@ -16,10 +16,16 @@
 
 package org.springframework.cloud.task.batch.listener;
 
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
 
 import org.junit.Before;
 import org.junit.Test;
+
 import org.springframework.batch.core.JobExecution;
 import org.springframework.batch.core.JobInstance;
 import org.springframework.batch.core.JobParameter;
@@ -72,7 +78,7 @@ public class EventJobExecutionTests {
 		JobParameter[] PARAMETERS = { new JobParameter("FOO", true), new JobParameter(1L, true),
 				new JobParameter(1D, true), new JobParameter(testDate, false) };
 
-		Map jobParamMap = new LinkedHashMap<String, JobParameter>();
+		Map jobParamMap = new LinkedHashMap<>();
 		for (int paramCount = 0; paramCount < JOB_PARAM_KEYS.length; paramCount++) {
 			jobParamMap.put(JOB_PARAM_KEYS[paramCount], PARAMETERS[paramCount]);
 		}