From 19618f48f9d9e2132e57c66bbffecd3ec2d5ab17 Mon Sep 17 00:00:00 2001 From: Michael Minella Date: Thu, 10 Apr 2014 17:17:57 -0500 Subject: [PATCH] BATCH-2204: Updated StepParserStepFactoryBean to set the buffered flag correctly when building a fault tolerant step --- .../xml/StepParserStepFactoryBean.java | 44 ++++++++++--------- .../xml/StepParserStepFactoryBeanTests.java | 14 +++--- 2 files changed, 31 insertions(+), 27 deletions(-) diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBean.java index 7953432c1..d4d9b7fda 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBean.java @@ -16,25 +16,6 @@ package org.springframework.batch.core.configuration.xml; -import java.io.Serializable; -import java.util.Collection; -import java.util.HashMap; -import java.util.HashSet; -import java.util.LinkedHashSet; -import java.util.Map; -import java.util.Queue; -import java.util.Set; -import java.util.concurrent.locks.ReentrantLock; - -import javax.batch.api.chunk.listener.RetryProcessListener; -import javax.batch.api.chunk.listener.RetryReadListener; -import javax.batch.api.chunk.listener.RetryWriteListener; -import javax.batch.api.chunk.listener.SkipProcessListener; -import javax.batch.api.chunk.listener.SkipReadListener; -import javax.batch.api.chunk.listener.SkipWriteListener; -import javax.batch.api.partition.PartitionAnalyzer; -import javax.batch.api.partition.PartitionCollector; - import org.springframework.batch.core.ChunkListener; import org.springframework.batch.core.ItemProcessListener; import org.springframework.batch.core.ItemReadListener; @@ -98,6 +79,23 @@ import org.springframework.transaction.annotation.Propagation; import org.springframework.transaction.interceptor.DefaultTransactionAttribute; import org.springframework.util.Assert; +import javax.batch.api.chunk.listener.RetryProcessListener; +import javax.batch.api.chunk.listener.RetryReadListener; +import javax.batch.api.chunk.listener.RetryWriteListener; +import javax.batch.api.chunk.listener.SkipProcessListener; +import javax.batch.api.chunk.listener.SkipReadListener; +import javax.batch.api.chunk.listener.SkipWriteListener; +import javax.batch.api.partition.PartitionCollector; +import java.io.Serializable; +import java.util.Collection; +import java.util.HashMap; +import java.util.HashSet; +import java.util.LinkedHashSet; +import java.util.Map; +import java.util.Queue; +import java.util.Set; +import java.util.concurrent.locks.ReentrantLock; + /** * This {@link FactoryBean} is used by the batch namespace parser to create {@link Step} objects. Stores all of the * properties that are configurable on the <step/> (and its inner <tasklet/>). Based on which properties are @@ -244,14 +242,14 @@ public class StepParserStepFactoryBean implements FactoryBean, BeanN private StepExecutionAggregator stepExecutionAggregator; /** - * @param queue The {@link Queue} that is used for communication between {@link PartitionCollector} and {@link PartitionAnalyzer} + * @param queue The {@link Queue} that is used for communication between {@link javax.batch.api.partition.PartitionCollector} and {@link javax.batch.api.partition.PartitionAnalyzer} */ public void setPartitionQueue(Queue queue) { this.partitionQueue = queue; } /** - * Used to coordinate access to the partition queue between the {@link PartitionCollector} and {@link PartitionAnalyzer} + * Used to coordinate access to the partition queue between the {@link javax.batch.api.partition.PartitionCollector} and {@link javax.batch.api.partition.AbstractPartitionAnalyzer} * * @param lock a lock that will be locked around accessing the partition queue */ @@ -363,6 +361,10 @@ public class StepParserStepFactoryBean implements FactoryBean, BeanN builder.processorNonTransactional(); } + if (readerTransactionalQueue!=null && readerTransactionalQueue==true) { + builder.readerIsTransactionalQueue(); + } + for (SkipListener listener : skipListeners) { builder.listener(listener); } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBeanTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBeanTests.java index 8c02fe0bb..c04023398 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBeanTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepParserStepFactoryBeanTests.java @@ -16,12 +16,6 @@ package org.springframework.batch.core.configuration.xml; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertTrue; - -import java.util.HashMap; -import java.util.Map; - import org.junit.Test; import org.springframework.aop.framework.Advised; import org.springframework.aop.framework.ProxyFactory; @@ -50,6 +44,13 @@ import org.springframework.test.util.ReflectionTestUtils; import org.springframework.transaction.annotation.Isolation; import org.springframework.transaction.annotation.Propagation; +import java.util.HashMap; +import java.util.Map; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.assertFalse; + /** * @author Dan Garrette * @since 2.0 @@ -248,6 +249,7 @@ public class StepParserStepFactoryBeanTests { assertEquals(new Integer(10), throttleLimit); Object tasklet = ReflectionTestUtils.getField(step, "tasklet"); assertTrue(tasklet instanceof ChunkOrientedTasklet); + assertFalse((Boolean) ReflectionTestUtils.getField(tasklet, "buffering")); Object chunkProvider = ReflectionTestUtils.getField(tasklet, "chunkProvider"); Object repeatOperations = ReflectionTestUtils.getField(chunkProvider, "repeatOperations"); Object completionPolicy = ReflectionTestUtils.getField(repeatOperations, "completionPolicy");