From a6479602e2121126c50eec2fc204c45a7e9244ca Mon Sep 17 00:00:00 2001 From: dsyer Date: Tue, 30 Dec 2008 16:32:29 +0000 Subject: [PATCH] BATCH-919: reuse ChunkProcessor in integration --- .../batch/core/StepContribution.java | 4 +- .../batch/core/step/item/Chunk.java | 9 ++- .../core/step/item/SimpleChunkProcessor.java | 41 +++++++++- .../chunk/ChunkMessageChannelItemWriter.java | 12 +-- .../integration/chunk/ChunkProcessor.java | 32 -------- .../chunk/ChunkProcessorChunkHandler.java | 34 ++++---- .../batch/integration/chunk/ChunkRequest.java | 27 ++++--- .../integration/chunk/ChunkResponse.java | 26 +++---- .../chunk/SimpleChunkProcessor.java | 78 ------------------- ...hunkMessageItemWriterIntegrationTests.java | 13 +++- .../ChunkProcessorChunkHandlerTests.java | 20 +++-- ...tTransactionalPollingIntegrationTests.java | 12 +-- .../TransactionalPollingIntegrationTests.java | 4 +- ...sageItemWriterIntegrationTests-context.xml | 2 +- ...ctionalPollingIntegrationTests-context.xml | 2 +- ...ctionalPollingIntegrationTests-context.xml | 46 +++++++---- ...ctionalPollingIntegrationTests-context.xml | 2 +- ...ctionalPollingIntegrationTests-context.xml | 2 +- 18 files changed, 166 insertions(+), 200 deletions(-) delete mode 100644 spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkProcessor.java delete mode 100644 spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/SimpleChunkProcessor.java diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/StepContribution.java b/spring-batch-core/src/main/java/org/springframework/batch/core/StepContribution.java index 4318d9d49..01abf129c 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/StepContribution.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/StepContribution.java @@ -15,6 +15,8 @@ */ package org.springframework.batch.core; +import java.io.Serializable; + /** * Represents a contribution to a {@link StepExecution}, buffering changes until * they can be applied at a chunk boundary. @@ -22,7 +24,7 @@ package org.springframework.batch.core; * @author Dave Syer * */ -public class StepContribution { +public class StepContribution implements Serializable { private volatile int readCount = 0; diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/Chunk.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/Chunk.java index 444d529e7..a521d688a 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/Chunk.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/Chunk.java @@ -1,6 +1,7 @@ package org.springframework.batch.core.step.item; import java.util.ArrayList; +import java.util.Collection; import java.util.Collections; import java.util.Iterator; import java.util.List; @@ -15,7 +16,7 @@ import java.util.List; * @author Dave Syer * */ -class Chunk implements Iterable { +public class Chunk implements Iterable { private List items = new ArrayList(); @@ -30,8 +31,12 @@ class Chunk implements Iterable { public Chunk() { this(null,null); } + + public Chunk(Collection items) { + this(items,null); + } - public Chunk(List items, List> skips) { + public Chunk(Collection items, List> skips) { super(); if (items!=null) { this.items = new ArrayList(items); diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkProcessor.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkProcessor.java index 4fb972595..eef3134da 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkProcessor.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkProcessor.java @@ -8,20 +8,55 @@ import org.springframework.batch.core.listener.MulticasterBatchListener; import org.springframework.batch.core.step.skip.SkipListenerFailedException; import org.springframework.batch.item.ItemProcessor; import org.springframework.batch.item.ItemWriter; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.util.Assert; -public class SimpleChunkProcessor implements ChunkProcessor { +public class SimpleChunkProcessor implements ChunkProcessor, InitializingBean { - private final ItemProcessor itemProcessor; + private ItemProcessor itemProcessor; - private final ItemWriter itemWriter; + private ItemWriter itemWriter; private final MulticasterBatchListener listener = new MulticasterBatchListener(); + /** + * Default constructor for ease of configuration (both itemWriter and + * itemProcessor are mandatory). + */ + @SuppressWarnings("unused") + private SimpleChunkProcessor() { + this(null, null); + } + public SimpleChunkProcessor(ItemProcessor itemProcessor, ItemWriter itemWriter) { this.itemProcessor = itemProcessor; this.itemWriter = itemWriter; } + /** + * @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; + } + + /** + * Check mandatory properties. + * + * @see InitializingBean#afterPropertiesSet() + */ + public void afterPropertiesSet() throws Exception { + Assert.notNull(itemWriter, "ItemWriter must be set"); + Assert.notNull(itemProcessor, "ItemProcessor must be set"); + } + /** * Register some {@link StepListener}s with the handler. Each will get the * callbacks in the order specified at the correct stage. diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkMessageChannelItemWriter.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkMessageChannelItemWriter.java index 093d7fa50..5326e0ec3 100644 --- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkMessageChannelItemWriter.java +++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkMessageChannelItemWriter.java @@ -6,8 +6,10 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.batch.core.BatchStatus; import org.springframework.batch.core.ExitStatus; +import org.springframework.batch.core.StepContribution; import org.springframework.batch.core.StepExecution; import org.springframework.batch.core.listener.StepExecutionListenerSupport; +import org.springframework.batch.core.step.item.Chunk; import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.ItemStreamException; @@ -19,9 +21,9 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSuppo private static final Log logger = LogFactory.getLog(ChunkMessageChannelItemWriter.class); - static final String ACTUAL = "ACTUAL"; + static final String ACTUAL = ChunkMessageChannelItemWriter.class.getName()+".ACTUAL"; - static final String EXPECTED = "EXPECTED"; + static final String EXPECTED = ChunkMessageChannelItemWriter.class.getName()+".EXPECTED"; private static final long DEFAULT_THROTTLE_LIMIT = 6; @@ -54,7 +56,7 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSuppo if (!items.isEmpty()) { logger.debug("Dispatching chunk: " + items); - ChunkRequest request = new ChunkRequest(items, localState.getJobId(), localState.getSkipCount()); + ChunkRequest request = new ChunkRequest(new Chunk(items), localState.getJobId(), localState.createStepContribution()); messagingGateway.send(request); localState.expected++; @@ -156,8 +158,8 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSuppo return expected - actual; } - public int getSkipCount() { - return stepExecution.getSkipCount(); + public StepContribution createStepContribution() { + return stepExecution.createStepContribution(); } public Long getJobId() { diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkProcessor.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkProcessor.java deleted file mode 100644 index bd5bd1027..000000000 --- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkProcessor.java +++ /dev/null @@ -1,32 +0,0 @@ -/* - * 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.integration.chunk; - -import java.util.Collection; - -import org.springframework.transaction.annotation.Propagation; -import org.springframework.transaction.annotation.Transactional; - -/** - * @author Dave Syer - */ -public interface ChunkProcessor { - - // This is transactional with REQUIRES_NEW because we need to force rollback - @Transactional(propagation = Propagation.REQUIRES_NEW) - int process(Collection items, int skipCount) throws Exception; - -} diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkProcessorChunkHandler.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkProcessorChunkHandler.java index ee5772c29..f1b722f33 100644 --- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkProcessorChunkHandler.java +++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkProcessorChunkHandler.java @@ -2,17 +2,17 @@ package org.springframework.batch.integration.chunk; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.springframework.batch.core.StepContribution; +import org.springframework.batch.core.step.item.ChunkProcessor; import org.springframework.beans.factory.InitializingBean; import org.springframework.integration.annotation.MessageEndpoint; import org.springframework.integration.annotation.ServiceActivator; import org.springframework.util.Assert; @MessageEndpoint -public class ChunkProcessorChunkHandler implements ChunkHandler, - InitializingBean { +public class ChunkProcessorChunkHandler implements ChunkHandler, InitializingBean { - private static final Log logger = LogFactory - .getLog(ChunkProcessorChunkHandler.class); + private static final Log logger = LogFactory.getLog(ChunkProcessorChunkHandler.class); private ChunkProcessor chunkProcessor; @@ -29,37 +29,33 @@ public class ChunkProcessorChunkHandler implements ChunkHandler, /** * Public setter for the {@link ChunkProcessor}. * - * @param chunkProcessor - * the chunkProcessor to set + * @param chunkProcessor the chunkProcessor to set */ public void setChunkProcessor(ChunkProcessor chunkProcessor) { this.chunkProcessor = chunkProcessor; } - /* - * (non-Javadoc) + /** * - * @see - * org.springframework.integration.batch.slave.ChunkHandler#handleChunk( - * java.util.Collection) + * @see ChunkHandler#handleChunk(ChunkRequest) */ @ServiceActivator public ChunkResponse handleChunk(ChunkRequest chunkRequest) { logger.debug("Handling chunk: " + chunkRequest); - int skipCount = 0; + StepContribution stepContribution = chunkRequest.getStepContribution(); try { - skipCount = chunkProcessor.process(chunkRequest.getItems(), - chunkRequest.getSkipCount()); - } catch (Exception e) { + chunkProcessor.process(stepContribution, chunkRequest.getChunk()); + } + catch (Exception e) { logger.debug("Failed chunk", e); - return new ChunkResponse(false, chunkRequest.getJobId(), skipCount, - e.getClass().getName() + ": " + e.getMessage()); + return new ChunkResponse(false, chunkRequest.getJobId(), stepContribution, e.getClass().getName() + ": " + + e.getMessage()); } - logger.debug("Completed chunk handling with " + skipCount + " skips"); - return new ChunkResponse(true, chunkRequest.getJobId(), skipCount); + logger.debug("Completed chunk handling with " + stepContribution); + return new ChunkResponse(true, chunkRequest.getJobId(), stepContribution); } } diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkRequest.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkRequest.java index d0a7493cb..a6e553a93 100644 --- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkRequest.java +++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkRequest.java @@ -1,38 +1,43 @@ package org.springframework.batch.integration.chunk; import java.io.Serializable; -import java.util.Collection; + +import org.springframework.batch.core.StepContribution; +import org.springframework.batch.core.step.item.Chunk; public class ChunkRequest implements Serializable { - private final int skipCount; private final Long jobId; - private final Collection items; + private final Chunk items; + private final StepContribution stepContribution; - public ChunkRequest(Collection items, Long jobId, int skipCount) { + public ChunkRequest(Chunk items, Long jobId, StepContribution stepContribution) { this.items = items; this.jobId = jobId; - this.skipCount = skipCount; - } - - public int getSkipCount() { - return skipCount; + this.stepContribution = stepContribution; } public Long getJobId() { return jobId; } - public Collection getItems() { + public Chunk getChunk() { return items; } + /** + * @return the {@link StepContribution} for this chunk + */ + public StepContribution getStepContribution() { + return stepContribution; + } + /** * @see java.lang.Object#toString() */ @Override public String toString() { - return getClass().getSimpleName()+": jobId="+jobId+", skipCount="+skipCount+", item count="+items.size(); + return getClass().getSimpleName()+": jobId="+jobId+", contribution="+stepContribution+", item count="+items.size(); } } diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkResponse.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkResponse.java index f6729fbfc..2d358e38b 100644 --- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkResponse.java +++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkResponse.java @@ -2,34 +2,32 @@ package org.springframework.batch.integration.chunk; import java.io.Serializable; +import org.springframework.batch.core.StepContribution; + public class ChunkResponse implements Serializable { - private final int skipCount; + private final StepContribution stepContribution; private final Long jobId; private final boolean status; private final String message; - public ChunkResponse(Long jobId) { - this(true, jobId, 0, null); + public ChunkResponse(Long jobId, StepContribution stepContribution) { + this(true, jobId, stepContribution, null); } - public ChunkResponse(Long jobId, int skipCount) { - this(true, jobId, skipCount, null); + public ChunkResponse(boolean status, Long jobId, StepContribution stepContribution) { + this(status, jobId, stepContribution, null); } - public ChunkResponse(boolean status, Long jobId, int skipCount) { - this(status, jobId, skipCount, null); - } - - public ChunkResponse(boolean status, Long jobId, int skipCount, String message) { + public ChunkResponse(boolean status, Long jobId, StepContribution stepContribution, String message) { this.status = status; this.jobId = jobId; - this.skipCount = skipCount; + this.stepContribution = stepContribution; this.message = message; } - public int getSkipCount() { - return skipCount; + public StepContribution getStepContribution() { + return stepContribution; } public Long getJobId() { @@ -49,7 +47,7 @@ public class ChunkResponse implements Serializable { */ @Override public String toString() { - return getClass().getSimpleName()+": jobId="+jobId+", skipCount="+skipCount+", successful="+status; + return getClass().getSimpleName()+": jobId="+jobId+", stepContribution="+stepContribution+", successful="+status; } } diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/SimpleChunkProcessor.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/SimpleChunkProcessor.java deleted file mode 100644 index 4ca82260c..000000000 --- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/SimpleChunkProcessor.java +++ /dev/null @@ -1,78 +0,0 @@ -/* - * 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.integration.chunk; - -import java.util.ArrayList; -import java.util.Collection; -import java.util.List; - -import org.springframework.batch.item.ItemProcessor; -import org.springframework.batch.item.ItemWriter; -import org.springframework.beans.factory.InitializingBean; -import org.springframework.util.Assert; - -/** - * @author Dave Syer - * - */ -public class SimpleChunkProcessor implements ChunkProcessor, InitializingBean { - - private ItemProcessor itemProcessor; - - private ItemWriter itemWriter; - - /* - * (non-Javadoc) - * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet() - */ - public void afterPropertiesSet() throws Exception { - Assert.notNull(itemProcessor, "An ItemProcessor must be provided"); - Assert.notNull(itemWriter, "An ItemWriter must be provided"); - } - - /** - * @param itemWriter - */ - public void setItemWriter(ItemWriter itemWriter) { - this.itemWriter = itemWriter; - } - - /** - * Public setter for the {@link ItemProcessor}. - * @param itemProcessor the {@link ItemProcessor} to set - */ - public void setItemProcessor(ItemProcessor itemProcessor) { - this.itemProcessor = itemProcessor; - } - - /* - * (non-Javadoc) - * @see org.springframework.batch.integration.chunk.ChunkProcessor#process(java.util.Collection, - * int) - */ - public int process(Collection items, int parentSkipCount) throws Exception { - - List processed = new ArrayList(); - for (S item : items) { - processed.add(itemProcessor.process(item)); - } - itemWriter.write(processed); - - return 0; - - } - -} diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkMessageItemWriterIntegrationTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkMessageItemWriterIntegrationTests.java index 316d36a27..72a010e5f 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkMessageItemWriterIntegrationTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkMessageItemWriterIntegrationTests.java @@ -12,8 +12,11 @@ import org.junit.runner.RunWith; import org.springframework.batch.core.BatchStatus; import org.springframework.batch.core.ExitStatus; import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobInstance; +import org.springframework.batch.core.JobParameters; import org.springframework.batch.core.JobParametersBuilder; import org.springframework.batch.core.Step; +import org.springframework.batch.core.StepContribution; import org.springframework.batch.core.StepExecution; import org.springframework.batch.core.job.SimpleJob; import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException; @@ -24,6 +27,7 @@ import org.springframework.batch.core.repository.dao.MapJobExecutionDao; import org.springframework.batch.core.repository.dao.MapJobInstanceDao; import org.springframework.batch.core.repository.dao.MapStepExecutionDao; import org.springframework.batch.core.repository.support.SimpleJobRepository; +import org.springframework.batch.core.step.item.Chunk; import org.springframework.batch.core.step.item.SimpleStepFactoryBean; import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.item.support.ListItemReader; @@ -69,7 +73,7 @@ public class ChunkMessageItemWriterIntegrationTests { factory.setBeanName("step"); factory.setItemWriter(writer); factory.setCommitInterval(4); - + SimpleMessagingGateway gateway = new SimpleMessagingGateway(); writer.setMessagingGateway(gateway); @@ -85,7 +89,7 @@ public class ChunkMessageItemWriterIntegrationTests { System.err.println(message); message = replies.receive(10); } - + } @After @@ -186,7 +190,10 @@ public class ChunkMessageItemWriterIntegrationTests { */ @SuppressWarnings("unchecked") private GenericMessage getSimpleMessage(String string, Long jobId) { - ChunkRequest chunk = new ChunkRequest(StringUtils.commaDelimitedListToSet(string), jobId, 0); + StepContribution stepContribution = new JobExecution(new JobInstance(0L, new JobParameters(), "job"), 1L) + .createStepExecution("step").createStepContribution(); + ChunkRequest chunk = new ChunkRequest(new Chunk(StringUtils.commaDelimitedListToSet(string)), jobId, + stepContribution); GenericMessage message = new GenericMessage(chunk); return message; } diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkProcessorChunkHandlerTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkProcessorChunkHandlerTests.java index 39a2694fb..40b784959 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkProcessorChunkHandlerTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkProcessorChunkHandlerTests.java @@ -3,9 +3,13 @@ package org.springframework.batch.integration.chunk; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; -import java.util.Collection; - import org.junit.Test; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobInstance; +import org.springframework.batch.core.JobParameters; +import org.springframework.batch.core.StepContribution; +import org.springframework.batch.core.step.item.Chunk; +import org.springframework.batch.core.step.item.ChunkProcessor; import org.springframework.util.StringUtils; public class ChunkProcessorChunkHandlerTests { @@ -17,15 +21,15 @@ public class ChunkProcessorChunkHandlerTests { @Test public void testVanillaHandleChunk() { handler.setChunkProcessor(new ChunkProcessor() { - public int process(Collection items, int skipCount) throws Exception { - count += items.size(); - return 0; + public void process(StepContribution contribution, Chunk chunk) throws Exception { + count += chunk.size(); } }); + StepContribution stepContribution = new JobExecution(new JobInstance(0L, new JobParameters(), "job"), 1L).createStepExecution("step").createStepContribution(); @SuppressWarnings("unchecked") - ChunkResponse response = handler.handleChunk(new ChunkRequest(StringUtils - .commaDelimitedListToSet("foo,bar"), 12L, 10)); - assertEquals(0, response.getSkipCount()); + ChunkResponse response = handler.handleChunk(new ChunkRequest(new Chunk(StringUtils + .commaDelimitedListToSet("foo,bar")), 12L, stepContribution)); + assertEquals(stepContribution, response.getStepContribution()); assertEquals(12, response.getJobId().longValue()); assertTrue(response.isSuccessful()); assertEquals(2, count); diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/RetryRepeatTransactionalPollingIntegrationTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/RetryRepeatTransactionalPollingIntegrationTests.java index f4084dee4..dacac8386 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/RetryRepeatTransactionalPollingIntegrationTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/RetryRepeatTransactionalPollingIntegrationTests.java @@ -30,7 +30,7 @@ public class RetryRepeatTransactionalPollingIntegrationTests implements Applicat private Log logger = LogFactory.getLog(getClass()); - private static List list = new ArrayList(); + private volatile static List list = new ArrayList(); @Autowired private SimpleRecoverer recoverer; @@ -38,10 +38,10 @@ public class RetryRepeatTransactionalPollingIntegrationTests implements Applicat @Autowired private SimpleService service; - private Lifecycle bus; + private Lifecycle lifecycle; public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { - bus = (Lifecycle) applicationContext; + lifecycle = (Lifecycle) applicationContext; } private static volatile int count = 0; @@ -73,7 +73,7 @@ public class RetryRepeatTransactionalPollingIntegrationTests implements Applicat List expected = TransactionAwareProxyFactory.createTransactionalList(Arrays.asList(StringUtils .commaDelimitedListToStringArray("a,b,c,d"))); service.setExpected(expected); - waitForResults(bus, expected.size(), 60); + waitForResults(lifecycle, expected.size(), 60); assertEquals(4,service.getProcessed().size()); // a,b,c,d assertEquals(expected, service.getProcessed()); } @@ -86,8 +86,8 @@ public class RetryRepeatTransactionalPollingIntegrationTests implements Applicat List expected = TransactionAwareProxyFactory.createTransactionalList(Arrays.asList(StringUtils .commaDelimitedListToStringArray("a,b,fail,fail,d,e,f"))); service.setExpected(expected); - waitForResults(bus, expected.size(), 60); - waitForResults(bus, 6, 100); // (a,b), (fail), (fail), ([fail],d), (e,f) + waitForResults(lifecycle, expected.size(), 60); // (a,b), (fail), (fail), ([fail],d), (e,f) + System.err.println(service.getProcessed()); assertEquals(7,service.getProcessed().size()); // a,b,fail,fail,d,e,f assertEquals(1,recoverer.getRecovered().size()); // fail assertEquals(expected, service.getProcessed()); diff --git a/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/TransactionalPollingIntegrationTests.java b/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/TransactionalPollingIntegrationTests.java index 4de098b94..ffc005039 100644 --- a/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/TransactionalPollingIntegrationTests.java +++ b/spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/TransactionalPollingIntegrationTests.java @@ -69,7 +69,9 @@ public class TransactionalPollingIntegrationTests implements ApplicationContextA } public void output(String message) { - handled.add(message); + if (count < expected.size()) { + handled.add(message); + } logger.debug("Handled: " + message); } diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/ChunkMessageItemWriterIntegrationTests-context.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/ChunkMessageItemWriterIntegrationTests-context.xml index 37381fe45..28c3f78d8 100644 --- a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/ChunkMessageItemWriterIntegrationTests-context.xml +++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/ChunkMessageItemWriterIntegrationTests-context.xml @@ -19,7 +19,7 @@ - + diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/retry/RepeatTransactionalPollingIntegrationTests-context.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/retry/RepeatTransactionalPollingIntegrationTests-context.xml index c0ba9486a..7f6d9e742 100644 --- a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/retry/RepeatTransactionalPollingIntegrationTests-context.xml +++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/retry/RepeatTransactionalPollingIntegrationTests-context.xml @@ -20,7 +20,7 @@ - + diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/retry/RetryRepeatTransactionalPollingIntegrationTests-context.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/retry/RetryRepeatTransactionalPollingIntegrationTests-context.xml index b350cde8b..b8bff6c29 100644 --- a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/retry/RetryRepeatTransactionalPollingIntegrationTests-context.xml +++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/retry/RetryRepeatTransactionalPollingIntegrationTests-context.xml @@ -1,7 +1,9 @@ - - + + - - + + @@ -39,32 +47,44 @@ - + + - + - - - + + + + + + - + + - + + \ No newline at end of file diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/retry/RetryTransactionalPollingIntegrationTests-context.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/retry/RetryTransactionalPollingIntegrationTests-context.xml index d199962a1..da7ca017d 100644 --- a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/retry/RetryTransactionalPollingIntegrationTests-context.xml +++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/retry/RetryTransactionalPollingIntegrationTests-context.xml @@ -18,7 +18,7 @@ - + diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/retry/TransactionalPollingIntegrationTests-context.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/retry/TransactionalPollingIntegrationTests-context.xml index 96bf5da2f..84256151e 100644 --- a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/retry/TransactionalPollingIntegrationTests-context.xml +++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/retry/TransactionalPollingIntegrationTests-context.xml @@ -20,7 +20,7 @@ - +