diff --git a/spring-cloud-task-integration-tests/src/test/java/configuration/JobConfiguration.java b/spring-cloud-task-integration-tests/src/test/java/configuration/JobConfiguration.java index 30def48e..d09c9d55 100644 --- a/spring-cloud-task-integration-tests/src/test/java/configuration/JobConfiguration.java +++ b/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").chunk(3) .reader(new ListItemReader<>(Arrays.asList("1", "2", "3", "4", "5", "6"))) - .processor(new ItemProcessor() { + .processor(new ItemProcessor() { @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() { + .writer(new ItemWriter() { @Override - public void write(List items) throws Exception { + public void write(List items) throws Exception { for (Object item : items) { System.out.println(">> " + item); } diff --git a/spring-cloud-task-integration-tests/src/test/java/configuration/JobSkipConfiguration.java b/spring-cloud-task-integration-tests/src/test/java/configuration/JobSkipConfiguration.java index e1ba64be..0b33f572 100644 --- a/spring-cloud-task-integration-tests/src/test/java/configuration/JobSkipConfiguration.java +++ b/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; diff --git a/spring-cloud-task-integration-tests/src/test/java/configuration/SkipItemWriter.java b/spring-cloud-task-integration-tests/src/test/java/configuration/SkipItemWriter.java index cae5ce4f..c7205b5f 100644 --- a/spring-cloud-task-integration-tests/src/test/java/configuration/SkipItemWriter.java +++ b/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"); diff --git a/spring-cloud-task-integration-tests/src/test/java/org/springframework/cloud/task/listener/BatchExecutionEventTests.java b/spring-cloud-task-integration-tests/src/test/java/org/springframework/cloud/task/listener/BatchExecutionEventTests.java index 87b891ed..9ab2c14c 100644 --- a/spring-cloud-task-integration-tests/src/test/java/org/springframework/cloud/task/listener/BatchExecutionEventTests.java +++ b/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; diff --git a/spring-cloud-task-samples/batch-events/pom.xml b/spring-cloud-task-samples/batch-events/pom.xml index 5b0eea71..b52965e9 100644 --- a/spring-cloud-task-samples/batch-events/pom.xml +++ b/spring-cloud-task-samples/batch-events/pom.xml @@ -34,12 +34,6 @@ spring-cloud-task-stream 1.0.0.BUILD-SNAPSHOT - - org.springframework.cloud - spring-cloud-stream-binder-test - 1.0.0.BUILD-SNAPSHOT - test - org.springframework.boot spring-boot-starter-batch diff --git a/spring-cloud-task-samples/batch-events/src/test/java/io/spring/cloud/BatchEventsApplicationTests.java b/spring-cloud-task-samples/batch-events/src/test/java/io/spring/cloud/BatchEventsApplicationTests.java index df7c61c3..3ae91cb9 100644 --- a/spring-cloud-task-samples/batch-events/src/test/java/io/spring/cloud/BatchEventsApplicationTests.java +++ b/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(); diff --git a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/BatchEventAutoConfiguration.java b/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/BatchEventAutoConfiguration.java index 099b24b2..b442c8e8 100644 --- a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/BatchEventAutoConfiguration.java +++ b/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: + * + *
    + *
  • {@link EventEmittingJobExecutionListener} - job-execution-events
    • + *
  • {@link EventEmittingStepExecutionListener} - step-execution-events
    • + *
  • {@link ChunkListener} - chunk-events
    • + *
  • {@link EventEmittingItemReadListener} - item-read-events
    • + *
  • {@link EventEmittingItemProcessListener} - item-process-events
    • + *
  • {@link EventEmittingItemWriteListener} - item-write-events
    • + *
  • {@link EventEmittingSkipListener} - skip-events
    • + *
+ * * @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 diff --git a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingItemProcessListener.java b/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingItemProcessListener.java index 414cb784..de5c737e 100644 --- a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingItemProcessListener.java +++ b/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 .equals), 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 messagePublisher; + private MessagePublisher 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 diff --git a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingItemReadEventsListener.java b/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingItemReadListener.java similarity index 59% rename from spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingItemReadEventsListener.java rename to spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingItemReadListener.java index 49865d3a..afb351e8 100644 --- a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingItemReadEventsListener.java +++ b/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 messagePublisher; + private MessagePublisher 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()); } } diff --git a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingItemWriteEventsListener.java b/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingItemWriteListener.java similarity index 67% rename from spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingItemWriteEventsListener.java rename to spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingItemWriteListener.java index 61e21b7f..d072ba68 100644 --- a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingItemWriteEventsListener.java +++ b/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 messagePublisher; + private MessagePublisher 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 diff --git a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingJobExecutionListener.java b/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingJobExecutionListener.java index 819a27a3..ae1c19bb 100644 --- a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingJobExecutionListener.java +++ b/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 messagePublisher; public EventEmittingJobExecutionListener(MessageChannel output) { Assert.notNull(output, "An output channel is required"); - this.output = output; this.messagePublisher = new MessagePublisher<>(output); } diff --git a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingSkipListener.java b/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingSkipListener.java index 4afc8976..a8921186 100644 --- a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingSkipListener.java +++ b/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 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 diff --git a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingStepExecutionListener.java b/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingStepExecutionListener.java index d1af72e9..42854e06 100644 --- a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/EventEmittingStepExecutionListener.java +++ b/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 messagePublisher; public EventEmittingStepExecutionListener(MessageChannel output) { Assert.notNull(output, "An output channel is required"); - this.output = output; this.messagePublisher = new MessagePublisher<>(output); } diff --git a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/ExitStatus.java b/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/ExitStatus.java index 54bf0a85..ce90b531 100644 --- a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/ExitStatus.java +++ b/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() { diff --git a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/JobExecutionEvent.java b/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/JobExecutionEvent.java index 0665cafd..f8c4b764 100644 --- a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/JobExecutionEvent.java +++ b/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 stepExecutions = new CopyOnWriteArraySet(); + private Collection stepExecutions = new CopyOnWriteArraySet(); - 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 failureExceptions = new CopyOnWriteArrayList(); + private List failureExceptions = new CopyOnWriteArrayList(); 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 getStepExecutions() { - return Collections.unmodifiableList(new ArrayList(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 getFailureExceptions() { - return failureExceptions; + return this.failureExceptions; } /** @@ -292,12 +262,12 @@ public class JobExecutionEvent extends Entity { */ public synchronized List getAllFailureExceptions() { - Set allExceptions = new HashSet(failureExceptions); - for (StepExecutionEvent stepExecution : stepExecutions) { + Set allExceptions = new HashSet<>(this.failureExceptions); + for (StepExecutionEvent stepExecution : this.stepExecutions) { allExceptions.addAll(stepExecution.getFailureExceptions()); } - return new ArrayList(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(); + 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 stepExecutions) { - if (stepExecutions!=null) { - this.stepExecutions.removeAll(stepExecutions); - this.stepExecutions.addAll(stepExecutions); - } - } - - } diff --git a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/JobInstanceEvent.java b/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/JobInstanceEvent.java index a53d1d1f..8e47366d 100644 --- a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/JobInstanceEvent.java +++ b/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(); } diff --git a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/JobParameterEvent.java b/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/JobParameterEvent.java index efc67c1b..87a33617 100644 --- a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/JobParameterEvent.java +++ b/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"); + } + } } } diff --git a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/JobParametersEvent.java b/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/JobParametersEvent.java index 623c083a..59590e99 100644 --- a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/JobParametersEvent.java +++ b/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 parameters; public JobParametersEvent() { - this.parameters = new LinkedHashMap(); + this.parameters = new LinkedHashMap<>(); } public JobParametersEvent(Map jobParameters) { - this.parameters = new LinkedHashMap(); + this.parameters = new LinkedHashMap<>(); for(Map.Entry 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())); } } } diff --git a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/StepExecutionEvent.java b/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/StepExecutionEvent.java index 09594684..40a95fc9 100644 --- a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/StepExecutionEvent.java +++ b/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 failureExceptions = new CopyOnWriteArrayList(); + private List failureExceptions = new CopyOnWriteArrayList(); 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 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(); + 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); } } diff --git a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/TaskBatchEventListenerBeanPostProcessor.java b/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/TaskBatchEventListenerBeanPostProcessor.java index fd744b89..77b5fd0a 100644 --- a/spring-cloud-task-stream/src/main/java/org/springframework/cloud/task/batch/listener/support/TaskBatchEventListenerBeanPostProcessor.java +++ b/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; diff --git a/spring-cloud-task-stream/src/test/java/org/springframework/cloud/task/batch/listener/EventJobExecutionTests.java b/spring-cloud-task-stream/src/test/java/org/springframework/cloud/task/batch/listener/EventJobExecutionTests.java index 30ec0c61..d6049165 100644 --- a/spring-cloud-task-stream/src/test/java/org/springframework/cloud/task/batch/listener/EventJobExecutionTests.java +++ b/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(); + Map jobParamMap = new LinkedHashMap<>(); for (int paramCount = 0; paramCount < JOB_PARAM_KEYS.length; paramCount++) { jobParamMap.put(JOB_PARAM_KEYS[paramCount], PARAMETERS[paramCount]); }