diff --git a/spring-batch-integration/.settings/com.springsource.sts.config.flow.prefs b/spring-batch-integration/.settings/com.springsource.sts.config.flow.prefs index fb9097303..dfd0e28d0 100644 --- a/spring-batch-integration/.settings/com.springsource.sts.config.flow.prefs +++ b/spring-batch-integration/.settings/com.springsource.sts.config.flow.prefs @@ -1,9 +1,9 @@ -#Fri Apr 09 18:01:26 BST 2010 -//com.springsource.sts.config.flow.coordinates\:http\://www.springframework.org/schema/batch\:/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/ChunkStepIntegrationTests-context.xml=\n\n\n\n\n\n\n\n\n\n -//com.springsource.sts.config.flow.coordinates\:http\://www.springframework.org/schema/batch\:/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/RemoteChunkFaultTolerantStepIntegrationTests-context.xml=\n\n\n\n\n\n -//com.springsource.sts.config.flow.coordinates\:http\://www.springframework.org/schema/batch\:/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/RemoteChunkFaultTolerantStepJmsIntegrationTests-context.xml=\n\n\n\n\n\n -//com.springsource.sts.config.flow.coordinates\:http\://www.springframework.org/schema/batch\:/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/RemoteChunkStepIntegrationTests-context.xml=\n\n\n\n\n\n -//com.springsource.sts.config.flow.coordinates\:http\://www.springframework.org/schema/batch\:/spring-batch-integration/src/test/resources/org/springframework/batch/integration/file/FileToMessagesJobIntegrationTests-context.xml=\r\n\r\n\r\n\r\n\r\n\r\n -//com.springsource.sts.config.flow.coordinates\:http\://www.springframework.org/schema/batch\:/spring-batch-integration/src/test/resources/org/springframework/batch/integration/step/StepGatewayIntegrationTests-context.xml=\n\n\n\n\n\n\n\n\n\n -//com.springsource.sts.config.flow.coordinates\:http\://www.springframework.org/schema/batch\:/spring-batch-integration/src/test/resources/org/springframework/batch/integration/tasklet/StepGatewayIntegrationTests-context.xml=\n\n\n\n\n\n\n\n\n\n -eclipse.preferences.version=1 +#Mon Apr 19 14:39:29 BST 2010 +//com.springsource.sts.config.flow.coordinates\:http\://www.springframework.org/schema/batch\:/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/ChunkStepIntegrationTests-context.xml=\n\n\n\n\n\n\n\n\n\n +//com.springsource.sts.config.flow.coordinates\:http\://www.springframework.org/schema/batch\:/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/RemoteChunkFaultTolerantStepIntegrationTests-context.xml=\n\n\n\n\n\n +//com.springsource.sts.config.flow.coordinates\:http\://www.springframework.org/schema/batch\:/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/RemoteChunkFaultTolerantStepJmsIntegrationTests-context.xml=\r\n\r\n\r\n\r\n\r\n\r\n +//com.springsource.sts.config.flow.coordinates\:http\://www.springframework.org/schema/batch\:/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/RemoteChunkStepIntegrationTests-context.xml=\n\n\n\n\n\n +//com.springsource.sts.config.flow.coordinates\:http\://www.springframework.org/schema/batch\:/spring-batch-integration/src/test/resources/org/springframework/batch/integration/file/FileToMessagesJobIntegrationTests-context.xml=\r\n\r\n\r\n\r\n\r\n\r\n +//com.springsource.sts.config.flow.coordinates\:http\://www.springframework.org/schema/batch\:/spring-batch-integration/src/test/resources/org/springframework/batch/integration/step/StepGatewayIntegrationTests-context.xml=\n\n\n\n\n\n\n\n\n\n +//com.springsource.sts.config.flow.coordinates\:http\://www.springframework.org/schema/batch\:/spring-batch-integration/src/test/resources/org/springframework/batch/integration/tasklet/StepGatewayIntegrationTests-context.xml=\n\n\n\n\n\n\n\n\n\n +eclipse.preferences.version=1 diff --git a/spring-batch-integration/.settings/org.eclipse.wst.common.component b/spring-batch-integration/.settings/org.eclipse.wst.common.component index 849ec8679..62ba7949e 100644 --- a/spring-batch-integration/.settings/org.eclipse.wst.common.component +++ b/spring-batch-integration/.settings/org.eclipse.wst.common.component @@ -1,8 +1,7 @@ - - - - - - - - + + + + + + + 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 c184a24be..aee497eb4 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 @@ -95,7 +95,7 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSuppo ChunkRequest request = new ChunkRequest(items, localState.getJobId(), localState .createStepContribution()); messagingGateway.send(request); - localState.expected.incrementAndGet(); + localState.incrementExpected(); } @@ -201,8 +201,17 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSuppo Assert.state(jobInstanceId != null, "Message did not contain job instance id."); Assert.state(jobInstanceId.equals(localState.getJobId()), "Message contained wrong job instance id [" + jobInstanceId + "] should have been [" + localState.getJobId() + "]."); - localState.actual.incrementAndGet(); + if (payload.isRedelivered()) { + logger + .warn("Redelivered result detected, which may indicate stale state. In the best case, we just picked up a timed out message " + + "from a previous failed execution. In the worst case (and if this is not a restart), " + + "the step may now timeout. In that case if you believe that all messages " + + "from workers have been sent, the business state " + + "is probably inconsistent, and the step will fail."); + localState.incrementRedelivered(); + } localState.pushStepContribution(payload.getStepContribution()); + localState.incrementActual(); if (!payload.isSuccessful()) { throw new AsynchronousFailureException("Failure or interrupt detected in handler: " + payload.getMessage()); @@ -232,6 +241,8 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSuppo private AtomicInteger expected = new AtomicInteger(); + private AtomicInteger redelivered = new AtomicInteger(); + private StepExecution stepExecution; private Queue contributions = new LinkedBlockingQueue(); @@ -263,6 +274,18 @@ public class ChunkMessageChannelItemWriter extends StepExecutionListenerSuppo } } + public void incrementRedelivered() { + redelivered.incrementAndGet(); + } + + public void incrementActual() { + actual.incrementAndGet(); + } + + public void incrementExpected() { + expected.incrementAndGet(); + } + public StepContribution createStepContribution() { return stepExecution.createStepContribution(); } 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 3a32b51c9..46a21dbf5 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 @@ -35,6 +35,8 @@ public class ChunkResponse implements Serializable { private final boolean status; private final String message; + + private final boolean redelivered; public ChunkResponse(Long jobId, StepContribution stepContribution) { this(true, jobId, stepContribution, null); @@ -45,10 +47,19 @@ public class ChunkResponse implements Serializable { } public ChunkResponse(boolean status, Long jobId, StepContribution stepContribution, String message) { + this(status, jobId, stepContribution, message, false); + } + + public ChunkResponse(ChunkResponse input, boolean redelivered) { + this(input.status, input.jobId, input.stepContribution, input.message, redelivered); + } + + public ChunkResponse(boolean status, Long jobId, StepContribution stepContribution, String message, boolean redelivered) { this.status = status; this.jobId = jobId; this.stepContribution = stepContribution; this.message = message; + this.redelivered = redelivered; } public StepContribution getStepContribution() { @@ -62,6 +73,10 @@ public class ChunkResponse implements Serializable { public boolean isSuccessful() { return status; } + + public boolean isRedelivered() { + return redelivered; + } public String getMessage() { return message; diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/JmsRedeliveredExtractor.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/JmsRedeliveredExtractor.java new file mode 100644 index 000000000..c2bc0ee29 --- /dev/null +++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/JmsRedeliveredExtractor.java @@ -0,0 +1,35 @@ +/* + * Copyright 2009-2010 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 org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.integration.annotation.Header; + +/** + * @author Dave Syer + * + */ +public class JmsRedeliveredExtractor { + + private static final Log logger = LogFactory.getLog(JmsRedeliveredExtractor.class); + + public ChunkResponse extract(ChunkResponse input, @Header("springintegration_jms_redelivered") boolean redelivered) { + logger.debug("Extracted redelivered flag for response, value="+redelivered); + return new ChunkResponse(input, redelivered); + } + +} diff --git a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/MessageSourcePollerInterceptor.java b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/MessageSourcePollerInterceptor.java index b4f6eeff7..2b6671713 100644 --- a/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/MessageSourcePollerInterceptor.java +++ b/spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/MessageSourcePollerInterceptor.java @@ -26,6 +26,8 @@ public class MessageSourcePollerInterceptor extends ChannelInterceptorAdapter im private MessageSource source; + private MessageChannel channel; + /** * Convenient default constructor for configuration purposes. */ @@ -39,6 +41,16 @@ public class MessageSourcePollerInterceptor extends ChannelInterceptorAdapter im this.source = source; } + /** + * Optional MessageChannel for injecting the message receieved from the source (defaults to the channel + * intercepted in {@link #preReceive(MessageChannel)}). + * + * @param channel the channel to set + */ + public void setChannel(MessageChannel channel) { + this.channel = channel; + } + /** * Asserts that mandatory properties are set. * @see InitializingBean#afterPropertiesSet() @@ -64,6 +76,9 @@ public class MessageSourcePollerInterceptor extends ChannelInterceptorAdapter im public boolean preReceive(MessageChannel channel) { Message message = source.receive(); if (message != null) { + if (this.channel!=null) { + channel = this.channel; + } channel.send(message); if (logger.isDebugEnabled()) { logger.debug("Sent " + message + " to channel " + channel.getName()); diff --git a/spring-batch-integration/src/site/apt/chunking.apt b/spring-batch-integration/src/site/apt/chunking.apt index 20aec1eff..4173f2814 100644 --- a/spring-batch-integration/src/site/apt/chunking.apt +++ b/spring-batch-integration/src/site/apt/chunking.apt @@ -17,15 +17,52 @@ Remote Chunking Implementation [[1]] Worker thread replies to response channel - [[1]] Step picks up reply and aggregates the counts + [[1]] Step picks up reply if there is one and aggregates the counts - [[1]] Step blocks until all the requests are satisfied + [[1]] Step reapeats until no more input data + + [[1]] Step blocks until all the outstanding requests are satisfied * Implementation - A ChunkProcessor acts as an asynchronous Messaging Gateway. - Asynchronous Messaging Gateway isn't a pattern that is supported out - of the box with Spring Integration. There is a - SimpleMessagingGateway that provides programmatic access to send and - receive payloads (instead of messages), so the pattern can be - manually implemented. + A ChunkProcessor acts as a kind of Throttling Asynchronous Messaging + Gateway, which isn't a pattern that is supported out of the box with + Spring Integration. There is a SimpleMessagingGateway that provides + programmatic access to send and receive payloads (instead of + messages), so the pattern can be manually implemented in the + ChunkProcessor. + + The current implementation is in the form of an ItemWriter + (ChunkMessageChannelItemWriter) which is a StepExecutionListener + (blocks and waits for the outstanding responses in the afterStep). + The ChunkProcessor can then simply be a vanilla implementation from + Spring Batch. + + The ChunkMessageChannelItemWriter implements the Throttling part of + the pattern by keeping track of the number of outstanding requests + (which it has to do anyway) and blocking until a response arrives if + the number is above a configurable limit. It wouldn't be necessary + to do this manually in the writer if the messages were only going + over local MessageChannels: the requests would either be processed + serially in a single thread, or else there would be a thread pool + with limited size controlling the workers. But since the messages + are going to JMS we need to either explicitly throttle in the writer + (or else rely on vendor features for producer flow control), + otherwise the JMS Queue could easily be overwhelmed and start + barfing (which happened in one of the early prototypes on an + Accenture project). + + Throttling MessageChannel.send() might be something Spring + Integration could do, but it only makes sense really in the context + of this gateway pattern (because you need something to react against + to decide when to release another send). + + The gateway is used to send requests to the workers, and then to + receive responses in the same thread, but only waiting for a + response when the step is complete. To implement this with JMS + backed channels we need a Spring Integration inbound adapter that + translates PollableChannel.receive() into + JmsTemplate.receiveAndConvert(). In the unlikely event of a problem + in the receiver the JMS message should roll back. + JmsDestinationPollingAdapter actually almost does what we need but + there is no support for configuring it without a scheduled poller. diff --git a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/RemoteChunkFaultTolerantStepJmsIntegrationTests-context.xml b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/RemoteChunkFaultTolerantStepJmsIntegrationTests-context.xml index a5df7b030..233291203 100644 --- a/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/RemoteChunkFaultTolerantStepJmsIntegrationTests-context.xml +++ b/spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/RemoteChunkFaultTolerantStepJmsIntegrationTests-context.xml @@ -56,6 +56,12 @@ + + + + + + @@ -70,6 +76,7 @@ +