BATCH-63: added additional attributes to <process-task>
This commit is contained in:
@@ -229,6 +229,12 @@ public class StepParser {
|
||||
bd.getPropertyValues().addPropertyValue("itemWriter", writerRef);
|
||||
}
|
||||
|
||||
String taskExecutorBeanId = element.getAttribute("task-executor");
|
||||
if (StringUtils.hasText(taskExecutorBeanId)) {
|
||||
RuntimeBeanReference taskExecutorRef = new RuntimeBeanReference(taskExecutorBeanId);
|
||||
bd.getPropertyValues().addPropertyValue("taskExecutor", taskExecutorRef);
|
||||
}
|
||||
|
||||
String jobRepository = element.getAttribute("job-repository");
|
||||
RuntimeBeanReference jobRepositoryRef = new RuntimeBeanReference(jobRepository);
|
||||
bd.getPropertyValues().addPropertyValue("jobRepository", jobRepositoryRef);
|
||||
@@ -245,7 +251,7 @@ public class StepParser {
|
||||
String skipLimit = element.getAttribute("skip-limit");
|
||||
if (StringUtils.hasText(skipLimit)) {
|
||||
if (!isFaultTolerant) {
|
||||
throw new BeanCreationException("skip-limit can only be specified if fault-tolerant is set to 'true'");
|
||||
throw new BeanCreationException("skip-limit can only be specified if fault-tolerant is set to \"true\"");
|
||||
}
|
||||
bd.getPropertyValues().addPropertyValue("skipLimit", skipLimit);
|
||||
}
|
||||
@@ -253,11 +259,35 @@ public class StepParser {
|
||||
String retryLimit = element.getAttribute("retry-limit");
|
||||
if (StringUtils.hasText(retryLimit)) {
|
||||
if (!isFaultTolerant) {
|
||||
throw new BeanCreationException("retry-limit can only be specified if fault-tolerant is set to 'true'");
|
||||
throw new BeanCreationException("retry-limit can only be specified if fault-tolerant is set to \"true\"");
|
||||
}
|
||||
bd.getPropertyValues().addPropertyValue("retryLimit", retryLimit);
|
||||
}
|
||||
|
||||
String cacheCapacity = element.getAttribute("cache-capacity");
|
||||
if (StringUtils.hasText(cacheCapacity)) {
|
||||
if (!isFaultTolerant) {
|
||||
throw new BeanCreationException("cache-capacity can only be specified if fault-tolerant is set to \"true\"");
|
||||
}
|
||||
bd.getPropertyValues().addPropertyValue("cacheCapacity", cacheCapacity);
|
||||
}
|
||||
|
||||
String transactionAttribute = element.getAttribute("transaction-attribute");
|
||||
if (StringUtils.hasText(transactionAttribute)) {
|
||||
handleTransactionAttributesElement(element, bd);
|
||||
bd.getPropertyValues().addPropertyValue("transactionAttribute", transactionAttribute);
|
||||
}
|
||||
|
||||
String isReaderTransactionalQueue = element.getAttribute("is-reader-transactional-queue");
|
||||
if (StringUtils.hasText(isReaderTransactionalQueue)) {
|
||||
if (!isFaultTolerant && "true".equals(isReaderTransactionalQueue)) {
|
||||
throw new BeanCreationException("is-reader-transactional-queue=\"true\" can only be specified if fault-tolerant is set to \"true\"");
|
||||
}
|
||||
if (isFaultTolerant) {
|
||||
bd.getPropertyValues().addPropertyValue("isReaderTransactionalQueue", isReaderTransactionalQueue);
|
||||
}
|
||||
}
|
||||
|
||||
handleExceptionElement(element, bd, "skippable-exception-classes", "skippableExceptionClasses", isFaultTolerant);
|
||||
|
||||
handleExceptionElement(element, bd, "retryable-exception-classes", "retryableExceptionClasses",isFaultTolerant);
|
||||
@@ -279,13 +309,16 @@ public class StepParser {
|
||||
|
||||
}
|
||||
|
||||
private void handleTransactionAttributesElement(Element element, RootBeanDefinition bd) {
|
||||
}
|
||||
|
||||
private void handleExceptionElement(Element element, RootBeanDefinition bd,
|
||||
String attributeName, String propertyName, boolean isFaultTolerant) {
|
||||
String exceptions =
|
||||
DomUtils.getChildElementValueByTagName(element, attributeName);
|
||||
if (StringUtils.hasLength(exceptions)) {
|
||||
if (!isFaultTolerant) {
|
||||
throw new BeanCreationException(attributeName + " can only be specified if fault-tolerant is set to 'true'");
|
||||
throw new BeanCreationException(attributeName + " can only be specified if fault-tolerant is set to \"true\"");
|
||||
}
|
||||
String[] exceptionArray = StringUtils.tokenizeToStringArray(
|
||||
StringUtils.delete(exceptions, ","), "\n");
|
||||
|
||||
@@ -281,18 +281,6 @@
|
||||
</xsd:sequence>
|
||||
</xsd:complexType>
|
||||
</xsd:element>
|
||||
<xsd:element name="transaction-attributes" minOccurs="0" maxOccurs="1">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation><![CDATA[
|
||||
The transaction attributes to be used for the transaction used during the execution
|
||||
of the task within the step.
|
||||
]]>
|
||||
</xsd:documentation>
|
||||
</xsd:annotation>
|
||||
<xsd:simpleType>
|
||||
<xsd:restriction base="xsd:string"/>
|
||||
</xsd:simpleType>
|
||||
</xsd:element>
|
||||
<xsd:element name="skippable-exception-classes" minOccurs="0" maxOccurs="1">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation><![CDATA[
|
||||
@@ -384,6 +372,14 @@
|
||||
]]></xsd:documentation>
|
||||
</xsd:annotation>
|
||||
</xsd:attribute>
|
||||
<xsd:attribute name="transaction-attribute" type="xsd:string" use="optional">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation><![CDATA[
|
||||
The transaction attributes to be used for the transaction used during the execution
|
||||
of the task within the step.
|
||||
]]></xsd:documentation>
|
||||
</xsd:annotation>
|
||||
</xsd:attribute>
|
||||
<xsd:attribute name="is-reader-transactional-queue" type="xsd:boolean" default="false" use="optional">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation><![CDATA[
|
||||
|
||||
@@ -34,9 +34,13 @@ import org.springframework.batch.item.ItemStream;
|
||||
import org.springframework.batch.retry.RetryListener;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.beans.factory.annotation.Qualifier;
|
||||
import org.springframework.scheduling.concurrent.ConcurrentTaskExecutor;
|
||||
import org.springframework.test.context.ContextConfiguration;
|
||||
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
|
||||
import org.springframework.test.util.ReflectionTestUtils;
|
||||
import org.springframework.transaction.TransactionDefinition;
|
||||
import org.springframework.transaction.interceptor.RollbackRuleAttribute;
|
||||
import org.springframework.transaction.interceptor.RuleBasedTransactionAttribute;
|
||||
|
||||
|
||||
/**
|
||||
@@ -87,6 +91,23 @@ public class StepWithFaultTolerantProcessTaskJobParserTests {
|
||||
assertEquals("wrong skip-limit:", 20, sl);
|
||||
Object rl = ReflectionTestUtils.getField(factory, "retryLimit");
|
||||
assertEquals("wrong retry-limit:", 3, rl);
|
||||
Object cc = ReflectionTestUtils.getField(factory, "cacheCapacity");
|
||||
assertEquals("wrong cache-capacity:", 100, cc);
|
||||
Object txa = ReflectionTestUtils.getField(factory, "transactionAttribute");
|
||||
assertEquals("wrong transaction-attribute:", TransactionDefinition.PROPAGATION_REQUIRED,
|
||||
((RuleBasedTransactionAttribute)txa).getPropagationBehavior());
|
||||
assertEquals("wrong transaction-attribute:", TransactionDefinition.ISOLATION_DEFAULT,
|
||||
((RuleBasedTransactionAttribute)txa).getIsolationLevel());
|
||||
assertEquals("wrong transaction-attribute:", 10,
|
||||
((RuleBasedTransactionAttribute)txa).getTimeout());
|
||||
RollbackRuleAttribute rra =
|
||||
(RollbackRuleAttribute) ((RuleBasedTransactionAttribute)txa).getRollbackRules().get(0);
|
||||
assertEquals("wrong transaction-attribute:",
|
||||
"org.springframework.dao.DataIntegrityViolationException", rra.getExceptionName());
|
||||
Object txq = ReflectionTestUtils.getField(factory, "isReaderTransactionalQueue");
|
||||
assertEquals("wrong is-reader-transactional-queue:", true, txq);
|
||||
Object te = ReflectionTestUtils.getField(factory, "taskExecutor");
|
||||
assertEquals("wrong task-executor:", ConcurrentTaskExecutor.class, te.getClass());
|
||||
Object listeners = ReflectionTestUtils.getField(factory, "listeners");
|
||||
assertEquals("wrong number of listeners:", 2, ((StepListener[])listeners).length);
|
||||
Object retryListeners = ReflectionTestUtils.getField(factory, "retryListeners");
|
||||
|
||||
@@ -15,7 +15,6 @@ public class TestRetryListener extends AbstractTestComponent implements RetryLis
|
||||
}
|
||||
|
||||
public <T> boolean open(RetryContext context, RetryCallback<T> callback) {
|
||||
System.out.println("RETRY RETRY RETRY RETRY");
|
||||
executed = true;
|
||||
return true;
|
||||
}
|
||||
|
||||
@@ -7,9 +7,13 @@
|
||||
<beans:import resource="common-context.xml" />
|
||||
|
||||
<job id="job">
|
||||
<step name="ft-step">
|
||||
<step name="step">
|
||||
<process-task reader="reader" processor="processor" writer="writer"
|
||||
fault-tolerant="true" commit-interval="10" skip-limit="20" retry-limit="3">
|
||||
fault-tolerant="true" commit-interval="10" skip-limit="20"
|
||||
retry-limit="3" cache-capacity="100"
|
||||
transaction-attribute="PROPAGATION_REQUIRED,ISOLATION_DEFAULT,timeout_10,-org.springframework.dao.DataIntegrityViolationException"
|
||||
is-reader-transactional-queue="true"
|
||||
task-executor="taskExecutor">
|
||||
<listeners>
|
||||
<listener class="org.springframework.batch.core.configuration.xml.TestListener"/>
|
||||
<listener ref="listener"/>
|
||||
@@ -31,5 +35,7 @@
|
||||
<beans:bean id="writer" class="org.springframework.batch.core.configuration.xml.TestWriter"/>
|
||||
|
||||
<beans:bean id="listener" class="org.springframework.batch.core.configuration.xml.TestListener"/>
|
||||
|
||||
<beans:bean id="taskExecutor" class="org.springframework.scheduling.concurrent.ConcurrentTaskExecutor"/>
|
||||
|
||||
</beans:beans>
|
||||
Reference in New Issue
Block a user