From bcaaf4e2b09dcc1f61afd108955884d1f6fb82f6 Mon Sep 17 00:00:00 2001 From: dhgarrette Date: Sat, 14 Mar 2009 08:34:04 +0000 Subject: [PATCH] BATCH-1154: - Created StepFactoryBean, as a "master" factory that holds all the properties configurable on a step and decides which type of step to create. - the StepFactoryBean is used to generate a tasklet step as well as chunk-oriented step because if the 'tasklet=' attribute is specified on at one level and the element is specified at another, the factory bean will be able to throw a useful error, whereas the parser would not know. - removed hacks from TaskletStep and SimpleStepFactoryBean --- .../configuration/xml/AbstractStepParser.java | 14 +- .../configuration/xml/InlineStepParser.java | 4 - .../xml/TaskletElementParser.java | 74 +-- .../core/step/item/SimpleStepFactoryBean.java | 25 +- .../batch/core/step/item/StepFactoryBean.java | 588 ++++++++++++++++++ .../batch/core/step/tasklet/TaskletStep.java | 33 - .../configuration/xml/StepParserTests.java | 7 +- ...tepWithBasicProcessTaskJobParserTests.java | 4 +- ...aultTolerantProcessTaskJobParserTests.java | 4 +- .../xml/TaskletElementParserTests.java | 2 +- .../core/step/item/StepFactoryBeanTests.java | 195 ++++++ 11 files changed, 822 insertions(+), 128 deletions(-) create mode 100644 spring-batch-core/src/main/java/org/springframework/batch/core/step/item/StepFactoryBean.java create mode 100644 spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StepFactoryBeanTests.java diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/AbstractStepParser.java b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/AbstractStepParser.java index 165b2461a..622e55aab 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/AbstractStepParser.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/AbstractStepParser.java @@ -18,7 +18,7 @@ package org.springframework.batch.core.configuration.xml; import java.util.Arrays; import java.util.List; -import org.springframework.batch.core.step.tasklet.TaskletStep; +import org.springframework.batch.core.step.item.StepFactoryBean; import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.config.RuntimeBeanReference; import org.springframework.beans.factory.parsing.CompositeComponentDefinition; @@ -61,6 +61,7 @@ public abstract class AbstractStepParser { @SuppressWarnings("unchecked") List taskletElements = (List) DomUtils.getChildElementsByTagName(stepElement, "tasklet"); boolean taskletElementExists = taskletElements.size() > 0; + boolean stepUnderspecified = stepUnderspecified(stepElement); AbstractBeanDefinition bd = null; if (StringUtils.hasText(taskletRef)) { if (taskletElementExists) { @@ -73,12 +74,17 @@ public abstract class AbstractStepParser { } else if (taskletElementExists) { Element taskElement = taskletElements.get(0); - bd = taskletElementParser.parse(taskElement, parserContext, stepUnderspecified(stepElement)); + bd = taskletElementParser.parse(taskElement, parserContext, stepUnderspecified); } if (bd != null) { setUpBeanDefinition(stepElement, bd, parserContext, jobRepositoryRef); } + else if (!stepUnderspecified) { + parserContext.getReaderContext().error( + "Step [" + stepElement.getAttribute("id") + + "] has neither a element nor a 'tasklet' attribute.", stepElement); + } return bd; @@ -105,7 +111,7 @@ public abstract class AbstractStepParser { String jobRepositoryRef) { GenericBeanDefinition bd = new GenericBeanDefinition(); - bd.setBeanClass(TaskletStep.class); + bd.setBeanClass(StepFactoryBean.class); if (StringUtils.hasText(taskletRef)) { RuntimeBeanReference taskletBeanRef = new RuntimeBeanReference(taskletRef); @@ -121,7 +127,7 @@ public abstract class AbstractStepParser { String jobRepositoryRef) { checkStepAttributes(stepElement, bd); - bd.setAbstract(stepElement.hasAttribute("abstract") && Boolean.valueOf(stepElement.getAttribute("abstract"))); + bd.setAbstract(Boolean.valueOf(stepElement.getAttribute("abstract"))); RuntimeBeanReference jobRepositoryBeanRef = new RuntimeBeanReference(jobRepositoryRef); bd.getPropertyValues().addPropertyValue("jobRepository", jobRepositoryBeanRef); diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/InlineStepParser.java b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/InlineStepParser.java index 26bb0ca9b..584e1ed2b 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/InlineStepParser.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/InlineStepParser.java @@ -104,10 +104,6 @@ public class InlineStepParser extends AbstractStepParser { parserContext.registerBeanComponent(new BeanComponentDefinition(bd, stepId)); stateBuilder.addConstructorArgReference(stepId); } - else { - parserContext.getReaderContext().error( - "Incomplete configuration detected while creating step with name " + stepRef, element); - } } return FlowParser.getNextElements(parserContext, stepId, stateBuilder.getBeanDefinition(), element); diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/TaskletElementParser.java b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/TaskletElementParser.java index f79891241..ab2a1130b 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/TaskletElementParser.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/TaskletElementParser.java @@ -18,8 +18,7 @@ package org.springframework.batch.core.configuration.xml; import java.util.Arrays; import java.util.List; -import org.springframework.batch.core.step.item.FaultTolerantStepFactoryBean; -import org.springframework.batch.core.step.item.SimpleStepFactoryBean; +import org.springframework.batch.core.step.item.StepFactoryBean; import org.springframework.beans.MutablePropertyValues; import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.config.BeanReference; @@ -50,27 +49,13 @@ public class TaskletElementParser { */ protected AbstractBeanDefinition parse(Element element, ParserContext parserContext, boolean underspecified) { - String skipLimit = element.getAttribute("skip-limit"); - String retryLimit = element.getAttribute("retry-limit"); - String cacheCapacity = element.getAttribute("cache-capacity"); - String isReaderTransactionalQueue = element.getAttribute("is-reader-transactional-queue"); - - boolean useFaultTolerant = underspecified - || (StringUtils.hasText(isReaderTransactionalQueue) && Boolean.valueOf(isReaderTransactionalQueue)) - || isPositive(skipLimit) || isPositive(retryLimit) || isPositive(cacheCapacity) - || hasElement(element, "skippable-exception-classes") - || hasElement(element, "retryable-exception-classes") || hasElement(element, "fatal-exception-classes"); - GenericBeanDefinition bd = new GenericBeanDefinition(); - if (useFaultTolerant) { - bd.setBeanClass(FaultTolerantStepFactoryBean.class); - } - else { - bd.setBeanClass(SimpleStepFactoryBean.class); - } + bd.setBeanClass(StepFactoryBean.class); MutablePropertyValues propertyValues = bd.getPropertyValues(); + propertyValues.addPropertyValue("hasTaskletElement", Boolean.TRUE); + String readerBeanId = element.getAttribute("reader"); if (StringUtils.hasText(readerBeanId)) { RuntimeBeanReference readerRef = new RuntimeBeanReference(readerBeanId); @@ -113,32 +98,31 @@ public class TaskletElementParser { + "or 'chunk-completion-policy', but not both.", element); } + String skipLimit = element.getAttribute("skip-limit"); if (StringUtils.hasText(skipLimit)) { propertyValues.addPropertyValue("skipLimit", skipLimit); } + String retryLimit = element.getAttribute("retry-limit"); if (StringUtils.hasText(retryLimit)) { propertyValues.addPropertyValue("retryLimit", retryLimit); } + String cacheCapacity = element.getAttribute("cache-capacity"); if (StringUtils.hasText(cacheCapacity)) { propertyValues.addPropertyValue("cacheCapacity", cacheCapacity); } + String isReaderTransactionalQueue = element.getAttribute("is-reader-transactional-queue"); if (StringUtils.hasText(isReaderTransactionalQueue)) { - if (useFaultTolerant) { - propertyValues.addPropertyValue("isReaderTransactionalQueue", isReaderTransactionalQueue); - } + propertyValues.addPropertyValue("isReaderTransactionalQueue", isReaderTransactionalQueue); } - handleExceptionElement(element, parserContext, bd, "skippable-exception-classes", "skippableExceptionClasses", - useFaultTolerant, underspecified); + handleExceptionElement(element, parserContext, bd, "skippable-exception-classes", "skippableExceptionClasses"); - handleExceptionElement(element, parserContext, bd, "retryable-exception-classes", "retryableExceptionClasses", - useFaultTolerant, underspecified); + handleExceptionElement(element, parserContext, bd, "retryable-exception-classes", "retryableExceptionClasses"); - handleExceptionElement(element, parserContext, bd, "fatal-exception-classes", "fatalExceptionClasses", - useFaultTolerant, underspecified); + handleExceptionElement(element, parserContext, bd, "fatal-exception-classes", "fatalExceptionClasses"); handleRetryListenersElement(element, bd, parserContext); @@ -148,39 +132,19 @@ public class TaskletElementParser { } - private boolean isPositive(String stringValue) { - if (StringUtils.hasText(stringValue)) { - if (Integer.valueOf(stringValue) > 0) { - return true; - } - } - return false; - } - - private boolean hasElement(Element element, String subElementName) { - return StringUtils.hasLength(DomUtils.getChildElementValueByTagName(element, subElementName)); - } - @SuppressWarnings("unchecked") private void handleExceptionElement(Element element, ParserContext parserContext, BeanDefinition bd, - String subElementName, String propertyName, boolean isFaultTolerant, boolean isAbstract) { + String subElementName, String propertyName) { Element child = DomUtils.getChildElementByTagName(element, subElementName); if (child != null) { String exceptions = DomUtils.getTextValue(child); if (StringUtils.hasLength(exceptions)) { - if (isFaultTolerant || isAbstract) { - String[] exceptionArray = StringUtils.tokenizeToStringArray(exceptions, ",\n"); - if (exceptionArray.length > 0) { - ManagedList managedList = new ManagedList(); - managedList.setMergeEnabled(Boolean.valueOf(child.getAttribute("merge"))); - managedList.addAll(Arrays.asList(exceptionArray)); - bd.getPropertyValues().addPropertyValue(propertyName, managedList); - } - } - else { - parserContext.getReaderContext().error( - subElementName + " can only be specified for fault-tolerant " - + "configurations providing skip-limit, retry-limit or cache-capacity", element); + String[] exceptionArray = StringUtils.tokenizeToStringArray(exceptions, ",\n"); + if (exceptionArray.length > 0) { + ManagedList managedList = new ManagedList(); + managedList.setMergeEnabled(Boolean.valueOf(child.getAttribute("merge"))); + managedList.addAll(Arrays.asList(exceptionArray)); + bd.getPropertyValues().addPropertyValue(propertyName, managedList); } } } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java index 4521449dd..d87466406 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleStepFactoryBean.java @@ -15,9 +15,6 @@ */ package org.springframework.batch.core.step.item; -import java.beans.PropertyEditor; -import java.util.List; - import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.batch.core.ChunkListener; @@ -48,9 +45,7 @@ import org.springframework.core.task.TaskExecutor; import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.interceptor.DefaultTransactionAttribute; import org.springframework.transaction.interceptor.TransactionAttribute; -import org.springframework.transaction.interceptor.TransactionAttributeEditor; import org.springframework.util.Assert; -import org.springframework.util.StringUtils; /** * Most common configuration options for simple steps should be found here. Use @@ -143,10 +138,8 @@ public class SimpleStepFactoryBean implements FactoryBean, BeanNameAware { } /** - * public void setIsReaderTransactionalQueue(boolean - * isReaderTransactionalQueue) { this.isReaderTransactionalQueue = - * isReaderTransactionalQueue; } Set the bean name property, which will - * become the name of the {@link Step} when it is created. + * Set the bean name property, which will become the name of the + * {@link Step} when it is created. * * @see org.springframework.beans.factory.BeanNameAware#setBeanName(java.lang.String) */ @@ -274,20 +267,6 @@ public class SimpleStepFactoryBean implements FactoryBean, BeanNameAware { this.transactionManager = transactionManager; } - /** - * Public setter for the {@link TransactionAttribute}. - * - * @param transactionAttributeList A list of all the transaction attributes - * to set - */ - public void setTransactionAttributeList(List transactionAttributeList) { - String[] stringArray = transactionAttributeList.toArray(new String[0]); - String attributeString = StringUtils.arrayToCommaDelimitedString(stringArray); - PropertyEditor editor = new TransactionAttributeEditor(); - editor.setAsText(attributeString); - this.setTransactionAttribute((TransactionAttribute) editor.getValue()); - } - /** * Public setter for the {@link TransactionAttribute}. * @param transactionAttribute the {@link TransactionAttribute} to set diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/StepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/StepFactoryBean.java new file mode 100644 index 000000000..7c7ed02ba --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/StepFactoryBean.java @@ -0,0 +1,588 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.batch.core.step.item; + +import java.beans.PropertyEditor; +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; + +import org.springframework.batch.core.Step; +import org.springframework.batch.core.StepExecutionListener; +import org.springframework.batch.core.StepListener; +import org.springframework.batch.core.repository.JobRepository; +import org.springframework.batch.core.step.tasklet.Tasklet; +import org.springframework.batch.core.step.tasklet.TaskletStep; +import org.springframework.batch.item.ItemProcessor; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ItemStream; +import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.repeat.CompletionPolicy; +import org.springframework.batch.repeat.policy.SimpleCompletionPolicy; +import org.springframework.batch.retry.RetryListener; +import org.springframework.batch.retry.policy.MapRetryContextCache; +import org.springframework.beans.factory.BeanNameAware; +import org.springframework.beans.factory.FactoryBean; +import org.springframework.core.task.TaskExecutor; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.interceptor.TransactionAttribute; +import org.springframework.transaction.interceptor.TransactionAttributeEditor; +import org.springframework.util.Assert; +import org.springframework.util.StringUtils; + +/** + * This {@link FactoryBean} is used by the batch namespace parser to create + * {@link Step} objects. + * + * @author Dan Garrette + * @since 2.0 + * @see SimpleStepFactoryBean + * @see FaultTolerantStepFactoryBean + */ +public class StepFactoryBean implements FactoryBean, BeanNameAware { + + // + // Step Attributes + // + private String name; + private Boolean allowStartIfComplete; + private JobRepository jobRepository; + private Integer startLimit; + private Tasklet tasklet; + private PlatformTransactionManager transactionManager; + + // + // Step Elements + // + private StepListener[] listeners; + private TransactionAttribute transactionAttribute; + + // + // Tasklet Attributes + // + private Integer cacheCapacity; + private CompletionPolicy chunkCompletionPolicy; + private Integer commitInterval; + private Boolean isReaderTransactionalQueue; + private Integer retryLimit; + private Integer skipLimit; + private TaskExecutor taskExecutor; + private ItemReader itemReader; + private ItemProcessor itemProcessor; + private ItemWriter itemWriter; + + // + // Tasklet Elements + // + private RetryListener[] retryListeners; + private Collection> skippableExceptionClasses; + private Collection> retryableExceptionClasses; + private Collection> fatalExceptionClasses; + private ItemStream[] streams; + + // + // Additional + // + private boolean hasTaskletElement = false; + + /** + * Create a {@link Step} from the configuration provided. + * + * @see FactoryBean#getObject() + */ + public final Object getObject() throws Exception { + if (hasTaskletElement) { + Assert.isNull(tasklet, "Step [" + name + "] has both a element and a 'tasklet' attribute."); + + if (isFaultTolerant()) { + FaultTolerantStepFactoryBean fb = new FaultTolerantStepFactoryBean(); + configureSimple(fb); + configureFaultTolerant(fb); + return fb.getObject(); + } + else { + validateSimpleStep(); + SimpleStepFactoryBean fb = new SimpleStepFactoryBean(); + configureSimple(fb); + return fb.getObject(); + } + } + else if (tasklet != null) { + validateTaskletStep(); + TaskletStep ts = new TaskletStep(); + configureTaskletStep(ts); + return ts; + } + else { + throw new IllegalStateException("Step [" + name + + "] has neither a element nor a 'tasklet' attribute."); + } + } + + private void configureSimple(SimpleStepFactoryBean fb) { + if (name != null) { + fb.setBeanName(name); + } + if (allowStartIfComplete != null) { + fb.setAllowStartIfComplete(allowStartIfComplete); + } + if (jobRepository != null) { + fb.setJobRepository(jobRepository); + } + if (startLimit != null) { + fb.setStartLimit(startLimit); + } + if (transactionManager != null) { + fb.setTransactionManager(transactionManager); + } + if (listeners != null) { + fb.setListeners(listeners); + } + if (transactionAttribute != null) { + fb.setTransactionAttribute(transactionAttribute); + } + + if (chunkCompletionPolicy != null) { + fb.setChunkCompletionPolicy(chunkCompletionPolicy); + } + if (commitInterval != null) { + fb.setCommitInterval(commitInterval); + } + if (taskExecutor != null) { + fb.setTaskExecutor(taskExecutor); + } + if (itemReader != null) { + fb.setItemReader(itemReader); + } + if (itemProcessor != null) { + fb.setItemProcessor(itemProcessor); + } + if (itemWriter != null) { + fb.setItemWriter(itemWriter); + } + + if (streams != null) { + fb.setStreams(streams); + } + } + + private void configureFaultTolerant(FaultTolerantStepFactoryBean fb) { + if (cacheCapacity != null) { + fb.setCacheCapacity(cacheCapacity); + } + if (isReaderTransactionalQueue != null) { + fb.setIsReaderTransactionalQueue(isReaderTransactionalQueue); + } + if (retryLimit != null) { + fb.setRetryLimit(retryLimit); + } + if (skipLimit != null) { + fb.setSkipLimit(skipLimit); + } + + if (retryListeners != null) { + fb.setRetryListeners(retryListeners); + } + if (skippableExceptionClasses != null) { + fb.setSkippableExceptionClasses(skippableExceptionClasses); + } + if (retryableExceptionClasses != null) { + fb.setRetryableExceptionClasses(retryableExceptionClasses); + } + if (fatalExceptionClasses != null) { + fb.setFatalExceptionClasses(fatalExceptionClasses); + } + } + + private void configureTaskletStep(TaskletStep ts) { + if (name != null) { + ts.setName(name); + } + if (allowStartIfComplete != null) { + ts.setAllowStartIfComplete(allowStartIfComplete); + } + if (jobRepository != null) { + ts.setJobRepository(jobRepository); + } + if (startLimit != null) { + ts.setStartLimit(startLimit); + } + if (tasklet != null) { + ts.setTasklet(tasklet); + } + if (transactionManager != null) { + ts.setTransactionManager(transactionManager); + } + if (listeners != null) { + int i = 0; + StepExecutionListener[] newListeners = new StepExecutionListener[listeners.length]; + for (StepListener listener : listeners) { + newListeners[i++] = (StepExecutionListener) listener; + } + ts.setStepExecutionListeners((StepExecutionListener[]) newListeners); + } + if (transactionAttribute != null) { + ts.setTransactionAttribute(transactionAttribute); + } + } + + private void validateSimpleStep() { + PropertyNamePair[] notPermitted = new PropertyNamePair[] { + new PropertyNamePair(cacheCapacity, "cacheCapacity"), new PropertyNamePair(retryLimit, "retryLimit"), + new PropertyNamePair(skipLimit, "skipLimit"), new PropertyNamePair(retryListeners, "retryListeners"), + new PropertyNamePair(skippableExceptionClasses, "skippableExceptionClasses"), + new PropertyNamePair(retryableExceptionClasses, "retryableExceptionClasses"), + new PropertyNamePair(fatalExceptionClasses, "fatalExceptionClasses") }; + validateDisallowedFields("on the simple step [" + name + "]. They can only be specified for fault-tolerant " + + "configurations providing skip-limit, retry-limit, or cache-capacity", notPermitted); + } + + private void validateTaskletStep() { + PropertyNamePair[] notPermitted = new PropertyNamePair[] { + new PropertyNamePair(cacheCapacity, "cacheCapacity"), + new PropertyNamePair(chunkCompletionPolicy, "chunkCompletionPolicy"), + new PropertyNamePair(commitInterval, "commitInterval"), new PropertyNamePair(retryLimit, "retryLimit"), + new PropertyNamePair(skipLimit, "skipLimit"), new PropertyNamePair(taskExecutor, "taskExecutor"), + new PropertyNamePair(itemReader, "itemReader"), new PropertyNamePair(itemProcessor, "itemProcessor"), + new PropertyNamePair(itemWriter, "itemWriter"), new PropertyNamePair(retryListeners, "retryListeners"), + new PropertyNamePair(skippableExceptionClasses, "skippableExceptionClasses"), + new PropertyNamePair(retryableExceptionClasses, "retryableExceptionClasses"), + new PropertyNamePair(fatalExceptionClasses, "fatalExceptionClasses"), + new PropertyNamePair(streams, "streams") }; + validateDisallowedFields("when a 'tasklet' attribute is specified, as it is on [" + name + "]", notPermitted); + } + + private void validateDisallowedFields(String msg, PropertyNamePair... notPermitted) { + List wrong = new ArrayList(); + for (PropertyNamePair field : notPermitted) { + if (field.getProperty() != null) { + wrong.add(field.getName()); + } + } + if (!wrong.isEmpty()) { + throw new IllegalArgumentException("The field" + (wrong.size() > 1 ? "s " : " ") + wrong + + (wrong.size() == 1 ? " is" : " are") + " not permitted " + msg); + } + } + + private static class PropertyNamePair { + private Object property; + private String name; + + public PropertyNamePair(Object property, String name) { + super(); + this.property = property; + this.name = name; + } + + public Object getProperty() { + return property; + } + + public String getName() { + return name; + } + } + + private boolean isFaultTolerant() { + return isPositive(skipLimit) || isPositive(retryLimit) || isPositive(cacheCapacity) + || isTrue(isReaderTransactionalQueue); + } + + private boolean isTrue(Boolean b) { + return b != null && b.booleanValue(); + } + + private boolean isPositive(Integer n) { + return n != null && n > 0; + } + + public Class getObjectType() { + return Step.class; + } + + public boolean isSingleton() { + return false; + } + + // ========================================================= + // Step Attributes + // ========================================================= + + /** + * Set the bean name property, which will become the name of the + * {@link Step} when it is created. + * + * @see org.springframework.beans.factory.BeanNameAware#setBeanName(java.lang.String) + */ + public void setBeanName(String name) { + if (this.name == null) { + this.name = name; + } + } + + /** + * Public setter for the flag to indicate that the step should be replayed + * on a restart, even if successful the first time. + * + * @param allowStartIfComplete the shouldAllowStartIfComplete to set + */ + public void setAllowStartIfComplete(boolean allowStartIfComplete) { + this.allowStartIfComplete = allowStartIfComplete; + + } + + /** + * Public setter for {@link JobRepository}. + * + * @param jobRepository + */ + public void setJobRepository(JobRepository jobRepository) { + this.jobRepository = jobRepository; + } + + /** + * The number of times that the step should be allowed to start + * + * @param startLimit + */ + public void setStartLimit(int startLimit) { + this.startLimit = startLimit; + } + + /** + * A preconfigured {@link Tasklet} to use. + * + * @param tasklet + */ + public void setTasklet(Tasklet tasklet) { + this.tasklet = tasklet; + } + + /** + * @param transactionManager the transaction manager to set + */ + public void setTransactionManager(PlatformTransactionManager transactionManager) { + this.transactionManager = transactionManager; + } + + // ========================================================= + // Step Elements + // ========================================================= + + /** + * The listeners to inject into the {@link Step}. Any instance of + * {@link StepListener} can be used, and will then receive callbacks at the + * appropriate stage in the step. + * + * @param listeners an array of listeners + */ + public void setListeners(StepListener[] listeners) { + this.listeners = listeners; + } + + /** + * Set the transaction attribute with a list of all the individual + * attributes. + * + * @param transactionAttributeList + */ + public void setTransactionAttributeList(List transactionAttributeList) { + String[] stringArray = transactionAttributeList.toArray(new String[0]); + String attributeString = StringUtils.arrayToCommaDelimitedString(stringArray); + PropertyEditor editor = new TransactionAttributeEditor(); + editor.setAsText(attributeString); + this.setTransactionAttribute((TransactionAttribute) editor.getValue()); + } + + /** + * @param transactionAttribute the {@link TransactionAttribute} to set + */ + public void setTransactionAttribute(TransactionAttribute transactionAttribute) { + this.transactionAttribute = transactionAttribute; + } + + // ========================================================= + // Tasklet Attributes + // ========================================================= + + /** + * Public setter for the capacity of the cache in the retry policy. If more + * items than this fail without being skipped or recovered an exception will + * be thrown. This is to guard against inadvertent infinite loops generated + * by item identity problems.
+ * + * The default value should be high enough and more for most purposes. To + * breach the limit in a single-threaded step typically you have to have + * this many failures in a single transaction. Defaults to the value in the + * {@link MapRetryContextCache}.
+ * + * @param cacheCapacity the cache capacity to set (greater than 0 else + * ignored) + */ + public void setCacheCapacity(int cacheCapacity) { + this.cacheCapacity = cacheCapacity; + } + + /** + * Public setter for the {@link CompletionPolicy} applying to the chunk + * level. A transaction will be committed when this policy decides to + * complete. Defaults to a {@link SimpleCompletionPolicy} with chunk size + * equal to the commitInterval property. + * + * @param chunkCompletionPolicy the chunkCompletionPolicy to set + */ + public void setChunkCompletionPolicy(CompletionPolicy chunkCompletionPolicy) { + this.chunkCompletionPolicy = chunkCompletionPolicy; + } + + /** + * Set the commit interval. Either set this or the chunkCompletionPolicy but + * not both. + * + * @param commitInterval 1 by default + */ + public void setCommitInterval(int commitInterval) { + this.commitInterval = commitInterval; + } + + /** + * Flag to signal that the reader is transactional (usually a JMS consumer) + * so that items are re-presented after a rollback. The default is false and + * readers are assumed to be forward-only. + * + * @param isReaderTransactionalQueue the value of the flag + */ + public void setIsReaderTransactionalQueue(boolean isReaderTransactionalQueue) { + this.isReaderTransactionalQueue = isReaderTransactionalQueue; + } + + /** + * Public setter for the retry limit. Each item can be retried up to this + * limit. Note this limit includes the initial attempt to process the item, + * therefore retryLimit == 1 by default. + * + * @param retryLimit the retry limit to set, must be greater or equal to 1. + */ + public void setRetryLimit(int retryLimit) { + this.retryLimit = retryLimit; + } + + /** + * Public setter for a limit that determines skip policy. If this value is + * positive then an exception in chunk processing will cause the item to be + * skipped and no exception propagated until the limit is reached. If it is + * zero then all exceptions will be propagated from the chunk and cause the + * step to abort. + * + * @param skipLimit the value to set. Default is 0 (never skip). + */ + public void setSkipLimit(int skipLimit) { + this.skipLimit = skipLimit; + } + + /** + * Public setter for the {@link TaskExecutor}. If this is set, then it will + * be used to execute the chunk processing inside the {@link Step}. + * + * @param taskExecutor the taskExecutor to set + */ + public void setTaskExecutor(TaskExecutor taskExecutor) { + this.taskExecutor = taskExecutor; + } + + /** + * @param itemReader the {@link ItemReader} to set + */ + public void setItemReader(ItemReader itemReader) { + this.itemReader = itemReader; + } + + /** + * @param itemProcessor the {@link ItemProcessor} to set + */ + public void setItemProcessor(ItemProcessor itemProcessor) { + this.itemProcessor = itemProcessor; + } + + /** + * @param itemWriter the {@link ItemWriter} to set + */ + public void setItemWriter(ItemWriter itemWriter) { + this.itemWriter = itemWriter; + } + + // ========================================================= + // Tasklet Elements + // ========================================================= + + /** + * Public setter for the {@link RetryListener}s. + * + * @param retryListeners the {@link RetryListener}s to set + */ + public void setRetryListeners(RetryListener... retryListeners) { + this.retryListeners = retryListeners; + } + + /** + * Public setter for exception classes that when raised won't crash the job + * but will result in transaction rollback and the item which handling + * caused the exception will be skipped. + * + * @param exceptionClasses + */ + public void setSkippableExceptionClasses(Collection> exceptionClasses) { + this.skippableExceptionClasses = exceptionClasses; + } + + /** + * Public setter for exception classes that will retry the item when raised. + * + * @param retryableExceptionClasses the retryableExceptionClasses to set + */ + public void setRetryableExceptionClasses(Collection> retryableExceptionClasses) { + this.retryableExceptionClasses = retryableExceptionClasses; + } + + /** + * Public setter for exception classes that should cause immediate failure. + * + * @param fatalExceptionClasses + */ + public void setFatalExceptionClasses(Collection> fatalExceptionClasses) { + this.fatalExceptionClasses = fatalExceptionClasses; + } + + /** + * The streams to inject into the {@link Step}. Any instance of + * {@link ItemStream} can be used, and will then receive callbacks at the + * appropriate stage in the step. + * + * @param streams an array of listeners + */ + public void setStreams(ItemStream[] streams) { + this.streams = streams; + } + + /** + * @param hasTaskletElement + */ + public void setHasTaskletElement(boolean hasTaskletElement) { + this.hasTaskletElement = hasTaskletElement; + } +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java index 50d0c42bd..6083be1df 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java @@ -15,8 +15,6 @@ */ package org.springframework.batch.core.step.tasklet; -import java.beans.PropertyEditor; -import java.util.List; import java.util.concurrent.Semaphore; import org.apache.commons.logging.Log; @@ -34,7 +32,6 @@ import org.springframework.batch.core.scope.context.StepContextRepeatCallback; import org.springframework.batch.core.step.AbstractStep; import org.springframework.batch.core.step.StepInterruptionPolicy; import org.springframework.batch.core.step.ThreadStepInterruptionPolicy; -import org.springframework.batch.core.step.item.SimpleStepFactoryBean; import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemStream; @@ -48,9 +45,7 @@ import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.TransactionStatus; import org.springframework.transaction.interceptor.DefaultTransactionAttribute; import org.springframework.transaction.interceptor.TransactionAttribute; -import org.springframework.transaction.interceptor.TransactionAttributeEditor; import org.springframework.util.Assert; -import org.springframework.util.StringUtils; /** * Simple implementation of executing the step as a call to a {@link Tasklet}, @@ -130,20 +125,6 @@ public class TaskletStep extends AbstractStep { this.transactionManager = transactionManager; } - /** - * Public setter for the {@link TransactionAttribute}. - * - * @param transactionAttributeList A list of all the transaction attributes - * to set - */ - public void setTransactionAttributeList(List transactionAttributeList) { - String[] stringArray = transactionAttributeList.toArray(new String[0]); - String attributeString = StringUtils.arrayToCommaDelimitedString(stringArray); - PropertyEditor editor = new TransactionAttributeEditor(); - editor.setAsText(attributeString); - this.setTransactionAttribute((TransactionAttribute) editor.getValue()); - } - /** * Public setter for the {@link TransactionAttribute}. * @@ -235,20 +216,6 @@ public class TaskletStep extends AbstractStep { this.interruptionPolicy = interruptionPolicy; } - /** - * Variation on - * {@link AbstractStep#setStepExecutionListeners(StepExecutionListener[])}. - * This method exists because the parser requires a "listeners" property - * setter to match the one on {@link SimpleStepFactoryBean}. - * - * @param listeners - * @see AbstractStep#setStepExecutionListeners(StepExecutionListener[]) - * @see SimpleStepFactoryBean#setListeners(org.springframework.batch.core.StepListener[]) - */ - public void setListeners(StepExecutionListener[] listeners) { - this.setStepExecutionListeners(listeners); - } - /** * Process the step and update its context so that progress can be monitored * by the caller. The step is broken down into chunks, each one executing in diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepParserTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepParserTests.java index 88b0d7e75..c315b0864 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepParserTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepParserTests.java @@ -27,7 +27,7 @@ import org.springframework.aop.framework.Advised; import org.springframework.batch.core.Step; import org.springframework.batch.core.StepExecutionListener; import org.springframework.batch.core.listener.StepExecutionListenerSupport; -import org.springframework.batch.core.step.item.FaultTolerantStepFactoryBean; +import org.springframework.batch.core.step.item.StepFactoryBean; import org.springframework.batch.core.step.tasklet.TaskletStep; import org.springframework.batch.repeat.CompletionPolicy; import org.springframework.batch.repeat.policy.SimpleCompletionPolicy; @@ -50,10 +50,9 @@ public class StepParserTests { public void testTaskletStepAttributes() throws Exception { ConfigurableApplicationContext ctx = new ClassPathXmlApplicationContext( "org/springframework/batch/core/configuration/xml/StepParserTaskletAttributesTests-context.xml"); - Map beans = ctx.getBeansOfType(FaultTolerantStepFactoryBean.class); + Map beans = ctx.getBeansOfType(StepFactoryBean.class); String factoryName = (String) beans.keySet().toArray()[0]; - FaultTolerantStepFactoryBean factory = (FaultTolerantStepFactoryBean) beans - .get(factoryName); + StepFactoryBean factory = (StepFactoryBean) beans.get(factoryName); TaskletStep bean = (TaskletStep) factory.getObject(); assertEquals("wrong start-limit:", 25, bean.getStartLimit()); } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepWithBasicProcessTaskJobParserTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepWithBasicProcessTaskJobParserTests.java index ee392879e..119a6d84a 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepWithBasicProcessTaskJobParserTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepWithBasicProcessTaskJobParserTests.java @@ -29,7 +29,7 @@ import org.springframework.batch.core.JobParameters; import org.springframework.batch.core.StepListener; import org.springframework.batch.core.repository.JobRepository; import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean; -import org.springframework.batch.core.step.item.SimpleStepFactoryBean; +import org.springframework.batch.core.step.item.StepFactoryBean; import org.springframework.batch.item.ItemStream; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; @@ -67,7 +67,7 @@ public class StepWithBasicProcessTaskJobParserTests { @SuppressWarnings("unchecked") @Autowired - private SimpleStepFactoryBean factory; + private StepFactoryBean factory; @Before public void setUp() { diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepWithFaultTolerantProcessTaskJobParserTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepWithFaultTolerantProcessTaskJobParserTests.java index c806e6768..174e26182 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepWithFaultTolerantProcessTaskJobParserTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepWithFaultTolerantProcessTaskJobParserTests.java @@ -29,7 +29,7 @@ import org.springframework.batch.core.JobParameters; import org.springframework.batch.core.StepListener; import org.springframework.batch.core.repository.JobRepository; import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean; -import org.springframework.batch.core.step.item.FaultTolerantStepFactoryBean; +import org.springframework.batch.core.step.item.StepFactoryBean; import org.springframework.batch.item.ItemStream; import org.springframework.batch.retry.RetryListener; import org.springframework.beans.factory.annotation.Autowired; @@ -75,7 +75,7 @@ public class StepWithFaultTolerantProcessTaskJobParserTests { @SuppressWarnings("unchecked") @Autowired - private FaultTolerantStepFactoryBean factory; + private StepFactoryBean factory; @Before public void setUp() { diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/TaskletElementParserTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/TaskletElementParserTests.java index a6b099ad4..41ec5a977 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/TaskletElementParserTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/TaskletElementParserTests.java @@ -43,7 +43,7 @@ import org.springframework.test.util.ReflectionTestUtils; */ public class TaskletElementParserTests { - ConfigurableApplicationContext taskletElementParentAttributeParserTestsContext = new ClassPathXmlApplicationContext( + private ConfigurableApplicationContext taskletElementParentAttributeParserTestsContext = new ClassPathXmlApplicationContext( "org/springframework/batch/core/configuration/xml/TaskletElementParentAttributeParserTests-context.xml"); @Test diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StepFactoryBeanTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StepFactoryBeanTests.java new file mode 100644 index 000000000..f9845b89b --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/StepFactoryBeanTests.java @@ -0,0 +1,195 @@ +/* + * Copyright 2006-2007 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.batch.core.step.item; + +import static org.junit.Assert.assertTrue; + +import java.util.ArrayList; + +import org.junit.Test; +import org.springframework.batch.core.StepListener; +import org.springframework.batch.core.configuration.xml.DummyCompletionPolicy; +import org.springframework.batch.core.configuration.xml.DummyItemReader; +import org.springframework.batch.core.configuration.xml.DummyItemWriter; +import org.springframework.batch.core.configuration.xml.DummyTasklet; +import org.springframework.batch.core.listener.StepExecutionListenerSupport; +import org.springframework.batch.core.step.JobRepositorySupport; +import org.springframework.batch.core.step.tasklet.TaskletStep; +import org.springframework.batch.item.ItemStream; +import org.springframework.batch.item.file.FlatFileItemReader; +import org.springframework.batch.item.support.PassThroughItemProcessor; +import org.springframework.batch.retry.listener.RetryListenerSupport; +import org.springframework.batch.support.transaction.ResourcelessTransactionManager; +import org.springframework.core.task.SyncTaskExecutor; +import org.springframework.test.util.ReflectionTestUtils; + +/** + * @author Dan Garrette + * @since 2.0 + */ +public class StepFactoryBeanTests { + + @Test(expected = IllegalStateException.class) + public void testNothingSet() throws Exception { + StepFactoryBean fb = new StepFactoryBean(); + fb.getObject(); + } + + @Test + public void testOnlyTaskletSet() throws Exception { + StepFactoryBean fb = new StepFactoryBean(); + fb.setTasklet(new DummyTasklet()); + Object step = fb.getObject(); + assertTrue(step instanceof TaskletStep); + Object tasklet = ReflectionTestUtils.getField(step, "tasklet"); + assertTrue(tasklet instanceof DummyTasklet); + } + + @Test(expected = IllegalStateException.class) + public void testSkipLimitSet() throws Exception { + StepFactoryBean fb = new StepFactoryBean(); + fb.setSkipLimit(5); + fb.getObject(); + } + + @Test + public void testTaskletStep_All() throws Exception { + StepFactoryBean fb = new StepFactoryBean(); + fb.setBeanName("step1"); + fb.setAllowStartIfComplete(true); + fb.setJobRepository(new JobRepositorySupport()); + fb.setStartLimit(5); + fb.setTasklet(new DummyTasklet()); + fb.setTransactionManager(new ResourcelessTransactionManager()); + fb.setListeners(new StepExecutionListenerSupport[] { new StepExecutionListenerSupport() }); + fb.setTransactionAttributeList(new ArrayList()); + Object step = fb.getObject(); + assertTrue(step instanceof TaskletStep); + Object tasklet = ReflectionTestUtils.getField(step, "tasklet"); + assertTrue(tasklet instanceof DummyTasklet); + } + + @Test(expected = IllegalStateException.class) + public void testSimpleStep_All() throws Exception { + StepFactoryBean fb = new StepFactoryBean(); + fb.setBeanName("step1"); + fb.setAllowStartIfComplete(true); + fb.setJobRepository(new JobRepositorySupport()); + fb.setStartLimit(5); + fb.setTransactionManager(new ResourcelessTransactionManager()); + fb.setListeners(new StepListener[] { new StepExecutionListenerSupport() }); + fb.setTransactionAttributeList(new ArrayList()); + fb.setChunkCompletionPolicy(new DummyCompletionPolicy()); + fb.setCommitInterval(5); + fb.setTaskExecutor(new SyncTaskExecutor()); + fb.setItemReader(new DummyItemReader()); + fb.setItemWriter(new DummyItemWriter()); + fb.setStreams(new ItemStream[] { new FlatFileItemReader() }); + + Object step = fb.getObject(); + assertTrue(step instanceof TaskletStep); + Object tasklet = ReflectionTestUtils.getField(step, "tasklet"); + assertTrue(tasklet instanceof ChunkOrientedTasklet); + } + + @Test(expected = IllegalStateException.class) + public void testFaultTolerantStep_All() throws Exception { + StepFactoryBean fb = new StepFactoryBean(); + fb.setBeanName("step1"); + fb.setAllowStartIfComplete(true); + fb.setJobRepository(new JobRepositorySupport()); + fb.setStartLimit(5); + fb.setTransactionManager(new ResourcelessTransactionManager()); + fb.setListeners(new StepListener[] { new StepExecutionListenerSupport() }); + fb.setTransactionAttributeList(new ArrayList()); + fb.setChunkCompletionPolicy(new DummyCompletionPolicy()); + fb.setCommitInterval(5); + fb.setTaskExecutor(new SyncTaskExecutor()); + fb.setItemReader(new DummyItemReader()); + fb.setItemWriter(new DummyItemWriter()); + fb.setStreams(new ItemStream[] { new FlatFileItemReader() }); + fb.setCacheCapacity(5); + fb.setIsReaderTransactionalQueue(true); + fb.setRetryLimit(5); + fb.setSkipLimit(100); + fb.setRetryListeners(new RetryListenerSupport()); + fb.setSkippableExceptionClasses(new ArrayList>()); + fb.setRetryableExceptionClasses(new ArrayList>()); + fb.setFatalExceptionClasses(new ArrayList>()); + + Object step = fb.getObject(); + assertTrue(step instanceof TaskletStep); + Object tasklet = ReflectionTestUtils.getField(step, "tasklet"); + assertTrue(tasklet instanceof ChunkOrientedTasklet); + } + + @Test + public void testSimpleStep() throws Exception { + StepFactoryBean fb = new StepFactoryBean(); + fb.setHasTaskletElement(true); + fb.setBeanName("step1"); + fb.setAllowStartIfComplete(true); + fb.setJobRepository(new JobRepositorySupport()); + fb.setStartLimit(5); + fb.setTransactionManager(new ResourcelessTransactionManager()); + fb.setListeners(new StepListener[] { new StepExecutionListenerSupport() }); + fb.setTransactionAttributeList(new ArrayList()); + fb.setChunkCompletionPolicy(new DummyCompletionPolicy()); + fb.setTaskExecutor(new SyncTaskExecutor()); + fb.setItemReader(new DummyItemReader()); + fb.setItemProcessor(new PassThroughItemProcessor()); + fb.setItemWriter(new DummyItemWriter()); + fb.setStreams(new ItemStream[] { new FlatFileItemReader() }); + + Object step = fb.getObject(); + assertTrue(step instanceof TaskletStep); + Object tasklet = ReflectionTestUtils.getField(step, "tasklet"); + assertTrue(tasklet instanceof ChunkOrientedTasklet); + } + + @Test + public void testFaultTolerantStep() throws Exception { + StepFactoryBean fb = new StepFactoryBean(); + fb.setHasTaskletElement(true); + fb.setBeanName("step1"); + fb.setAllowStartIfComplete(true); + fb.setJobRepository(new JobRepositorySupport()); + fb.setStartLimit(5); + fb.setTransactionManager(new ResourcelessTransactionManager()); + fb.setListeners(new StepListener[] { new StepExecutionListenerSupport() }); + fb.setTransactionAttributeList(new ArrayList()); + fb.setChunkCompletionPolicy(new DummyCompletionPolicy()); + fb.setTaskExecutor(new SyncTaskExecutor()); + fb.setItemReader(new DummyItemReader()); + fb.setItemProcessor(new PassThroughItemProcessor()); + fb.setItemWriter(new DummyItemWriter()); + fb.setStreams(new ItemStream[] { new FlatFileItemReader() }); + fb.setCacheCapacity(5); + fb.setIsReaderTransactionalQueue(true); + fb.setRetryLimit(5); + fb.setSkipLimit(100); + fb.setRetryListeners(new RetryListenerSupport()); + fb.setSkippableExceptionClasses(new ArrayList>()); + fb.setRetryableExceptionClasses(new ArrayList>()); + fb.setFatalExceptionClasses(new ArrayList>()); + + Object step = fb.getObject(); + assertTrue(step instanceof TaskletStep); + Object tasklet = ReflectionTestUtils.getField(step, "tasklet"); + assertTrue(tasklet instanceof ChunkOrientedTasklet); + } +}