Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Apply spring-javaformat style for consistency with other projects

Browse Source 
Resolves #4118
This commit is contained in:
Mahmoud Ben Hassine
2022-05-25 17:47:13 +02:00
parent b3fe088879
commit c4ad90b9b1
1626 changed files with 44478 additions and 45612 deletions
Download Patch File Download Diff File Expand all files Collapse all files

252
spring-batch-integration/src/main/java/org/springframework/batch/integration/async/AsyncItemProcessor.java
View File

@@ -1,129 +1,123 @@
/*
* Copyright 2006-2019 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
*
* https://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.async;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.scope.context.StepContext;
import org.springframework.batch.core.scope.context.StepSynchronizationManager;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemWriter;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.task.SyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
/**
* An {@link ItemProcessor} that delegates to a nested processor and in the
* background. To allow for background processing the return value from the
* processor is a {@link Future} which needs to be unpacked before the item can
* be used by a client.
*
* Because the {@link Future} is typically unwrapped in the {@link ItemWriter},
* there are lifecycle and stats limitations (since the framework doesn't know
* what the result of the processor is). While not an exhaustive list, things like
* {@link StepExecution#filterCount} will not reflect the number of filtered items
* and {@link org.springframework.batch.core.ItemProcessListener#onProcessError(Object, Exception)}
* will not be called.
*
* @author Dave Syer
*
* @param <I> the input object type
* @param <O> the output object type (will be wrapped in a Future)
* @see AsyncItemWriter
*/
public class AsyncItemProcessor<I, O> implements ItemProcessor<I, Future<O>>, InitializingBean {
private ItemProcessor<I, O> delegate;
private TaskExecutor taskExecutor = new SyncTaskExecutor();
/**
* Check mandatory properties (the {@link #setDelegate(ItemProcessor)}).
*
* @see InitializingBean#afterPropertiesSet()
*/
public void afterPropertiesSet() throws Exception {
Assert.notNull(delegate, "The delegate must be set.");
}
/**
* The {@link ItemProcessor} to use to delegate processing to in a
* background thread.
*
* @param delegate the {@link ItemProcessor} to use as a delegate
*/
public void setDelegate(ItemProcessor<I, O> delegate) {
this.delegate = delegate;
}
/**
* The {@link TaskExecutor} to use to allow the item processing to proceed
* in the background. Defaults to a {@link SyncTaskExecutor} so no threads
* are created unless this is overridden.
*
* @param taskExecutor a {@link TaskExecutor}
*/
public void setTaskExecutor(TaskExecutor taskExecutor) {
this.taskExecutor = taskExecutor;
}
/**
* Transform the input by delegating to the provided item processor. The
* return value is wrapped in a {@link Future} so that clients can unpack it
* later.
*
* @see ItemProcessor#process(Object)
*/
@Nullable
public Future<O> process(final I item) throws Exception {
final StepExecution stepExecution = getStepExecution();
FutureTask<O> task = new FutureTask<>(new Callable<O>() {
public O call() throws Exception {
if (stepExecution != null) {
StepSynchronizationManager.register(stepExecution);
}
try {
return delegate.process(item);
}
finally {
if (stepExecution != null) {
StepSynchronizationManager.close();
}
}
}
});
taskExecutor.execute(task);
return task;
}
/**
* @return the current step execution if there is one
*/
private StepExecution getStepExecution() {
StepContext context = StepSynchronizationManager.getContext();
if (context==null) {
return null;
}
StepExecution stepExecution = context.getStepExecution();
return stepExecution;
}
}
/*
* Copyright 2006-2019 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
*
* https://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.async;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.scope.context.StepContext;
import org.springframework.batch.core.scope.context.StepSynchronizationManager;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemWriter;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.task.SyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
/**
* An {@link ItemProcessor} that delegates to a nested processor and in the background. To
* allow for background processing the return value from the processor is a {@link Future}
* which needs to be unpacked before the item can be used by a client.
*
* Because the {@link Future} is typically unwrapped in the {@link ItemWriter}, there are
* lifecycle and stats limitations (since the framework doesn't know what the result of
* the processor is). While not an exhaustive list, things like
* {@link StepExecution#filterCount} will not reflect the number of filtered items and
* {@link org.springframework.batch.core.ItemProcessListener#onProcessError(Object, Exception)}
* will not be called.
*
* @author Dave Syer
* @param <I> the input object type
* @param <O> the output object type (will be wrapped in a Future)
* @see AsyncItemWriter
*/
public class AsyncItemProcessor<I, O> implements ItemProcessor<I, Future<O>>, InitializingBean {
private ItemProcessor<I, O> delegate;
private TaskExecutor taskExecutor = new SyncTaskExecutor();
/**
* Check mandatory properties (the {@link #setDelegate(ItemProcessor)}).
*
* @see InitializingBean#afterPropertiesSet()
*/
public void afterPropertiesSet() throws Exception {
Assert.notNull(delegate, "The delegate must be set.");
}
/**
* The {@link ItemProcessor} to use to delegate processing to in a background thread.
* @param delegate the {@link ItemProcessor} to use as a delegate
*/
public void setDelegate(ItemProcessor<I, O> delegate) {
this.delegate = delegate;
}
/**
* The {@link TaskExecutor} to use to allow the item processing to proceed in the
* background. Defaults to a {@link SyncTaskExecutor} so no threads are created unless
* this is overridden.
* @param taskExecutor a {@link TaskExecutor}
*/
public void setTaskExecutor(TaskExecutor taskExecutor) {
this.taskExecutor = taskExecutor;
}
/**
* Transform the input by delegating to the provided item processor. The return value
* is wrapped in a {@link Future} so that clients can unpack it later.
*
* @see ItemProcessor#process(Object)
*/
@Nullable
public Future<O> process(final I item) throws Exception {
final StepExecution stepExecution = getStepExecution();
FutureTask<O> task = new FutureTask<>(new Callable<O>() {
public O call() throws Exception {
if (stepExecution != null) {
StepSynchronizationManager.register(stepExecution);
}
try {
return delegate.process(item);
}
finally {
if (stepExecution != null) {
StepSynchronizationManager.close();
}
}
}
});
taskExecutor.execute(task);
return task;
}
/**
* @return the current step execution if there is one
*/
private StepExecution getStepExecution() {
StepContext context = StepSynchronizationManager.getContext();
if (context == null) {
return null;
}
StepExecution stepExecution = context.getStepExecution();
return stepExecution;
}
}

218
spring-batch-integration/src/main/java/org/springframework/batch/integration/async/AsyncItemWriter.java
View File

@@ -1,108 +1,110 @@
/*
* Copyright 2006-2018 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
*
* https://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.async;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.ItemStreamException;
import org.springframework.batch.item.ItemStreamWriter;
import org.springframework.batch.item.ItemWriter;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;
public class AsyncItemWriter<T> implements ItemStreamWriter<Future<T>>, InitializingBean {
private static final Log logger = LogFactory.getLog(AsyncItemWriter.class);
private ItemWriter<T> delegate;
public void afterPropertiesSet() throws Exception {
Assert.notNull(delegate, "A delegate ItemWriter must be provided.");
}
/**
* @param delegate ItemWriter that does the actual writing of the Future results
*/
public void setDelegate(ItemWriter<T> delegate) {
this.delegate = delegate;
}
/**
* In the processing of the {@link java.util.concurrent.Future}s passed, nulls are <em>not</em> passed to the
* delegate since they are considered filtered out by the {@link org.springframework.batch.integration.async.AsyncItemProcessor}'s
* delegated {@link org.springframework.batch.item.ItemProcessor}. If the unwrapping
* of the {@link Future} results in an {@link ExecutionException}, that will be
* unwrapped and the cause will be thrown.
*
* @param items {@link java.util.concurrent.Future}s to be unwrapped and passed to the delegate
* @throws Exception The exception returned by the Future if one was thrown
*/
public void write(List<? extends Future<T>> items) throws Exception {
List<T> list = new ArrayList<>();
for (Future<T> future : items) {
try {
T item = future.get();
if(item != null) {
list.add(future.get());
}
}
catch (ExecutionException e) {
Throwable cause = e.getCause();
if(cause != null && cause instanceof Exception) {
logger.debug("An exception was thrown while processing an item", e);
throw (Exception) cause;
}
else {
throw e;
}
}
}
delegate.write(list);
}
@Override
public void open(ExecutionContext executionContext) throws ItemStreamException {
if (delegate instanceof ItemStream) {
((ItemStream) delegate).open(executionContext);
}
}
@Override
public void update(ExecutionContext executionContext) throws ItemStreamException {
if (delegate instanceof ItemStream) {
((ItemStream) delegate).update(executionContext);
}
}
@Override
public void close() throws ItemStreamException {
if (delegate instanceof ItemStream) {
((ItemStream) delegate).close();
}
}
}
/*
* Copyright 2006-2018 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
*
* https://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.async;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.ItemStreamException;
import org.springframework.batch.item.ItemStreamWriter;
import org.springframework.batch.item.ItemWriter;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;
public class AsyncItemWriter<T> implements ItemStreamWriter<Future<T>>, InitializingBean {
private static final Log logger = LogFactory.getLog(AsyncItemWriter.class);
private ItemWriter<T> delegate;
public void afterPropertiesSet() throws Exception {
Assert.notNull(delegate, "A delegate ItemWriter must be provided.");
}
/**
* @param delegate ItemWriter that does the actual writing of the Future results
*/
public void setDelegate(ItemWriter<T> delegate) {
this.delegate = delegate;
}
/**
* In the processing of the {@link java.util.concurrent.Future}s passed, nulls are
* <em>not</em> passed to the delegate since they are considered filtered out by the
* {@link org.springframework.batch.integration.async.AsyncItemProcessor}'s delegated
* {@link org.springframework.batch.item.ItemProcessor}. If the unwrapping of the
* {@link Future} results in an {@link ExecutionException}, that will be unwrapped and
* the cause will be thrown.
* @param items {@link java.util.concurrent.Future}s to be unwrapped and passed to the
* delegate
* @throws Exception The exception returned by the Future if one was thrown
*/
public void write(List<? extends Future<T>> items) throws Exception {
List<T> list = new ArrayList<>();
for (Future<T> future : items) {
try {
T item = future.get();
if (item != null) {
list.add(future.get());
}
}
catch (ExecutionException e) {
Throwable cause = e.getCause();
if (cause != null && cause instanceof Exception) {
logger.debug("An exception was thrown while processing an item", e);
throw (Exception) cause;
}
else {
throw e;
}
}
}
delegate.write(list);
}
@Override
public void open(ExecutionContext executionContext) throws ItemStreamException {
if (delegate instanceof ItemStream) {
((ItemStream) delegate).open(executionContext);
}
}
@Override
public void update(ExecutionContext executionContext) throws ItemStreamException {
if (delegate instanceof ItemStream) {
((ItemStream) delegate).update(executionContext);
}
}
@Override
public void close() throws ItemStreamException {
if (delegate instanceof ItemStream) {
((ItemStream) delegate).close();
}
}
}

9
spring-batch-integration/src/main/java/org/springframework/batch/integration/async/StepExecutionInterceptor.java
View File

@@ -24,11 +24,10 @@ import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.ChannelInterceptor;
/**
* A {@link ChannelInterceptor} that adds the current {@link StepExecution} (if
* there is one) as a header to the message. Downstream asynchronous handlers
* can then take advantage of the step context without needing to be step
* scoped, which is a problem for handlers executing in another thread because
* the scope context is not available.
* A {@link ChannelInterceptor} that adds the current {@link StepExecution} (if there is
* one) as a header to the message. Downstream asynchronous handlers can then take
* advantage of the step context without needing to be step scoped, which is a problem for
* handlers executing in another thread because the scope context is not available.
*
* @author Dave Syer
*

3
spring-batch-integration/src/main/java/org/springframework/batch/integration/async/package-info.java
View File

@@ -1,5 +1,6 @@
/**
* Components for executing item processing asynchronously and writing the results when processing is complete.
* Components for executing item processing asynchronously and writing the results when
* processing is complete.
*
* @author Michael Minella
* @author Mahmoud Ben Hassine

23
spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/AsynchronousFailureException.java
View File

@@ -19,24 +19,21 @@ package org.springframework.batch.integration.chunk;
import org.springframework.batch.item.ItemWriterException;
/**
* Exception indicating that a failure or early completion condition was
* detected in a remote worker.
*
* Exception indicating that a failure or early completion condition was detected in a
* remote worker.
*
* @author Dave Syer
*
*
*/
public class AsynchronousFailureException extends ItemWriterException {
private static final long serialVersionUID = 1L;
/**
* Create a new {@link AsynchronousFailureException} based on a message and
* another exception.
*
* @param message
* the message for this exception
* @param cause
* the other exception
* Create a new {@link AsynchronousFailureException} based on a message and another
* exception.
* @param message the message for this exception
* @param cause the other exception
*/
public AsynchronousFailureException(String message, Throwable cause) {
super(message, cause);
@@ -44,9 +41,7 @@ public class AsynchronousFailureException extends ItemWriterException {
/**
* Create a new {@link AsynchronousFailureException} based on a message.
*
* @param message
* the message for this exception
* @param message the message for this exception
*/
public AsynchronousFailureException(String message) {
super(message);

27
spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkHandler.java
View File

@@ -16,28 +16,27 @@
package org.springframework.batch.integration.chunk;
/**
* Interface for a remote worker in the Remote Chunking pattern. A request comes from a manager process containing some
* items to be processed. Once the items are done with a response needs to be generated containing a summary of the
* result.
*
* Interface for a remote worker in the Remote Chunking pattern. A request comes from a
* manager process containing some items to be processed. Once the items are done with a
* response needs to be generated containing a summary of the result.
*
* @author Dave Syer
*
* @param <T> the type of the items to be processed (it is recommended to use a Memento like a primary key)
* @param <T> the type of the items to be processed (it is recommended to use a Memento
* like a primary key)
*/
public interface ChunkHandler<T> {
/**
* Handle the chunk, processing all the items and returning a response summarising the result. If the result is a
* failure then the response should say so. The handler only throws an exception if it needs to roll back a
* transaction and knows that the request will be re-delivered (if not to the same handler then to one processing
* the same Step).
*
* Handle the chunk, processing all the items and returning a response summarising the
* result. If the result is a failure then the response should say so. The handler
* only throws an exception if it needs to roll back a transaction and knows that the
* request will be re-delivered (if not to the same handler then to one processing the
* same Step).
* @param chunk a request containing the chunk to process
* @return a response summarising the result
*
* @throws Exception if the handler needs to roll back a transaction and have the chunk re-delivered
* @throws Exception if the handler needs to roll back a transaction and have the
* chunk re-delivered
*/
ChunkResponse handleChunk(ChunkRequest<T> chunk) throws Exception;

688
spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkMessageChannelItemWriter.java
View File

@@ -1,344 +1,344 @@
/*
* Copyright 2006-2021 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
*
* https://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 java.util.Queue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
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.StepExecutionListener;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.ItemStreamException;
import org.springframework.batch.item.ItemWriter;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.util.Assert;
public class ChunkMessageChannelItemWriter<T> implements StepExecutionListener, ItemWriter<T>,
ItemStream, StepContributionSource {
private static final Log logger = LogFactory.getLog(ChunkMessageChannelItemWriter.class);
static final String ACTUAL = ChunkMessageChannelItemWriter.class.getName() + ".ACTUAL";
static final String EXPECTED = ChunkMessageChannelItemWriter.class.getName() + ".EXPECTED";
private static final long DEFAULT_THROTTLE_LIMIT = 6;
private MessagingTemplate messagingGateway;
private final LocalState localState = new LocalState();
private long throttleLimit = DEFAULT_THROTTLE_LIMIT;
private final int DEFAULT_MAX_WAIT_TIMEOUTS = 40;
private int maxWaitTimeouts = DEFAULT_MAX_WAIT_TIMEOUTS;
private PollableChannel replyChannel;
/**
* The maximum number of times to wait at the end of a step for a non-null result from the remote workers. This is a
* multiplier on the receive timeout set separately on the gateway. The ideal value is a compromise between allowing
* slow workers time to finish, and responsiveness if there is a dead worker. Defaults to 40.
*
* @param maxWaitTimeouts the maximum number of wait timeouts
*/
public void setMaxWaitTimeouts(int maxWaitTimeouts) {
this.maxWaitTimeouts = maxWaitTimeouts;
}
/**
* Public setter for the throttle limit. This limits the number of pending requests for chunk processing to avoid
* overwhelming the receivers.
* @param throttleLimit the throttle limit to set
*/
public void setThrottleLimit(long throttleLimit) {
this.throttleLimit = throttleLimit;
}
public void setMessagingOperations(MessagingTemplate messagingGateway) {
this.messagingGateway = messagingGateway;
}
public void setReplyChannel(PollableChannel replyChannel) {
this.replyChannel = replyChannel;
}
public void write(List<? extends T> items) throws Exception {
// Block until expecting <= throttle limit
while (localState.getExpecting() > throttleLimit) {
getNextResult();
}
if (!items.isEmpty()) {
ChunkRequest<T> request = localState.getRequest(items);
if (logger.isDebugEnabled()) {
logger.debug("Dispatching chunk: " + request);
}
messagingGateway.send(new GenericMessage<>(request));
localState.incrementExpected();
}
}
@Override
public void beforeStep(StepExecution stepExecution) {
localState.setStepExecution(stepExecution);
}
@Nullable
@Override
public ExitStatus afterStep(StepExecution stepExecution) {
if (!(stepExecution.getStatus() == BatchStatus.COMPLETED)) {
return ExitStatus.EXECUTING;
}
long expecting = localState.getExpecting();
boolean timedOut;
try {
logger.debug("Waiting for results in step listener...");
timedOut = !waitForResults();
logger.debug("Finished waiting for results in step listener.");
}
catch (RuntimeException e) {
logger.debug("Detected failure waiting for results in step listener.", e);
stepExecution.setStatus(BatchStatus.FAILED);
return ExitStatus.FAILED.addExitDescription(e.getClass().getName() + ": " + e.getMessage());
}
finally {
if (logger.isDebugEnabled()) {
logger.debug("Finished waiting for results in step listener. Still expecting: "
+ localState.getExpecting());
}
for (StepContribution contribution : getStepContributions()) {
stepExecution.apply(contribution);
}
}
if (timedOut) {
stepExecution.setStatus(BatchStatus.FAILED);
return ExitStatus.FAILED.addExitDescription("Timed out waiting for " + localState.getExpecting()
+ " backlog at end of step");
}
return ExitStatus.COMPLETED.addExitDescription("Waited for " + expecting + " results.");
}
public void close() throws ItemStreamException {
localState.reset();
}
public void open(ExecutionContext executionContext) throws ItemStreamException {
if (executionContext.containsKey(EXPECTED)) {
localState.open(executionContext.getInt(EXPECTED), executionContext.getInt(ACTUAL));
if (!waitForResults()) {
throw new ItemStreamException("Timed out waiting for back log on open");
}
}
}
public void update(ExecutionContext executionContext) throws ItemStreamException {
executionContext.putInt(EXPECTED, localState.expected.intValue());
executionContext.putInt(ACTUAL, localState.actual.intValue());
}
public Collection<StepContribution> getStepContributions() {
List<StepContribution> contributions = new ArrayList<>();
for (ChunkResponse response : localState.pollChunkResponses()) {
StepContribution contribution = response.getStepContribution();
if (logger.isDebugEnabled()) {
logger.debug("Applying: " + response);
}
contributions.add(contribution);
}
return contributions;
}
/**
* Wait until all the results that are in the pipeline come back to the reply channel.
*
* @return true if successfully received a result, false if timed out
*/
private boolean waitForResults() throws AsynchronousFailureException {
int count = 0;
int maxCount = maxWaitTimeouts;
Throwable failure = null;
if (logger.isInfoEnabled()) {
logger.info("Waiting for " + localState.getExpecting() + " results");
}
while (localState.getExpecting() > 0 && count++ < maxCount) {
try {
getNextResult();
}
catch (Throwable t) {
logger.error("Detected error in remote result. Trying to recover " + localState.getExpecting()
+ " outstanding results before completing.", t);
failure = t;
}
}
if (failure != null) {
throw wrapIfNecessary(failure);
}
return count < maxCount;
}
/**
* Get the next result if it is available (within the timeout specified in the gateway), otherwise do nothing.
*
* @throws AsynchronousFailureException If there is a response and it contains a failed chunk response.
*
* @throws IllegalStateException if the result contains the wrong job instance id (maybe we are sharing a channel
* and we shouldn't be)
*/
@SuppressWarnings("unchecked")
private void getNextResult() throws AsynchronousFailureException {
Message<ChunkResponse> message = (Message<ChunkResponse>) messagingGateway.receive(replyChannel);
if (message != null) {
ChunkResponse payload = message.getPayload();
if (logger.isDebugEnabled()) {
logger.debug("Found result: " + payload);
}
Long jobInstanceId = payload.getJobId();
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() + "].");
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.pushResponse(payload);
localState.incrementActual();
if (!payload.isSuccessful()) {
throw new AsynchronousFailureException("Failure or interrupt detected in handler: "
+ payload.getMessage());
}
}
}
/**
* Re-throws the original throwable if it is unchecked, wraps checked exceptions into
* {@link AsynchronousFailureException}.
*/
private static AsynchronousFailureException wrapIfNecessary(Throwable throwable) {
if (throwable instanceof Error) {
throw (Error) throwable;
}
else if (throwable instanceof AsynchronousFailureException) {
return (AsynchronousFailureException) throwable;
}
else {
return new AsynchronousFailureException("Exception in remote process", throwable);
}
}
private static class LocalState {
private final AtomicInteger current = new AtomicInteger(-1);
private final AtomicInteger actual = new AtomicInteger();
private final AtomicInteger expected = new AtomicInteger();
private final AtomicInteger redelivered = new AtomicInteger();
private StepExecution stepExecution;
private final Queue<ChunkResponse> contributions = new LinkedBlockingQueue<>();
public int getExpecting() {
return expected.get() - actual.get();
}
public <T> ChunkRequest<T> getRequest(List<? extends T> items) {
return new ChunkRequest<>(current.incrementAndGet(), items, getJobId(), createStepContribution());
}
public void open(int expectedValue, int actualValue) {
actual.set(actualValue);
expected.set(expectedValue);
}
public Collection<ChunkResponse> pollChunkResponses() {
Collection<ChunkResponse> set = new ArrayList<>();
synchronized (contributions) {
ChunkResponse item = contributions.poll();
while (item != null) {
set.add(item);
item = contributions.poll();
}
}
return set;
}
public void pushResponse(ChunkResponse stepContribution) {
synchronized (contributions) {
contributions.add(stepContribution);
}
}
public void incrementRedelivered() {
redelivered.incrementAndGet();
}
public void incrementActual() {
actual.incrementAndGet();
}
public void incrementExpected() {
expected.incrementAndGet();
}
public StepContribution createStepContribution() {
return stepExecution.createStepContribution();
}
public Long getJobId() {
return stepExecution.getJobExecution().getJobId();
}
public void setStepExecution(StepExecution stepExecution) {
this.stepExecution = stepExecution;
}
public void reset() {
expected.set(0);
actual.set(0);
}
}
}
/*
* Copyright 2006-2021 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
*
* https://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 java.util.Queue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
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.StepExecutionListener;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.ItemStreamException;
import org.springframework.batch.item.ItemWriter;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.util.Assert;
public class ChunkMessageChannelItemWriter<T>
implements StepExecutionListener, ItemWriter<T>, ItemStream, StepContributionSource {
private static final Log logger = LogFactory.getLog(ChunkMessageChannelItemWriter.class);
static final String ACTUAL = ChunkMessageChannelItemWriter.class.getName() + ".ACTUAL";
static final String EXPECTED = ChunkMessageChannelItemWriter.class.getName() + ".EXPECTED";
private static final long DEFAULT_THROTTLE_LIMIT = 6;
private MessagingTemplate messagingGateway;
private final LocalState localState = new LocalState();
private long throttleLimit = DEFAULT_THROTTLE_LIMIT;
private final int DEFAULT_MAX_WAIT_TIMEOUTS = 40;
private int maxWaitTimeouts = DEFAULT_MAX_WAIT_TIMEOUTS;
private PollableChannel replyChannel;
/**
* The maximum number of times to wait at the end of a step for a non-null result from
* the remote workers. This is a multiplier on the receive timeout set separately on
* the gateway. The ideal value is a compromise between allowing slow workers time to
* finish, and responsiveness if there is a dead worker. Defaults to 40.
* @param maxWaitTimeouts the maximum number of wait timeouts
*/
public void setMaxWaitTimeouts(int maxWaitTimeouts) {
this.maxWaitTimeouts = maxWaitTimeouts;
}
/**
* Public setter for the throttle limit. This limits the number of pending requests
* for chunk processing to avoid overwhelming the receivers.
* @param throttleLimit the throttle limit to set
*/
public void setThrottleLimit(long throttleLimit) {
this.throttleLimit = throttleLimit;
}
public void setMessagingOperations(MessagingTemplate messagingGateway) {
this.messagingGateway = messagingGateway;
}
public void setReplyChannel(PollableChannel replyChannel) {
this.replyChannel = replyChannel;
}
public void write(List<? extends T> items) throws Exception {
// Block until expecting <= throttle limit
while (localState.getExpecting() > throttleLimit) {
getNextResult();
}
if (!items.isEmpty()) {
ChunkRequest<T> request = localState.getRequest(items);
if (logger.isDebugEnabled()) {
logger.debug("Dispatching chunk: " + request);
}
messagingGateway.send(new GenericMessage<>(request));
localState.incrementExpected();
}
}
@Override
public void beforeStep(StepExecution stepExecution) {
localState.setStepExecution(stepExecution);
}
@Nullable
@Override
public ExitStatus afterStep(StepExecution stepExecution) {
if (!(stepExecution.getStatus() == BatchStatus.COMPLETED)) {
return ExitStatus.EXECUTING;
}
long expecting = localState.getExpecting();
boolean timedOut;
try {
logger.debug("Waiting for results in step listener...");
timedOut = !waitForResults();
logger.debug("Finished waiting for results in step listener.");
}
catch (RuntimeException e) {
logger.debug("Detected failure waiting for results in step listener.", e);
stepExecution.setStatus(BatchStatus.FAILED);
return ExitStatus.FAILED.addExitDescription(e.getClass().getName() + ": " + e.getMessage());
}
finally {
if (logger.isDebugEnabled()) {
logger.debug("Finished waiting for results in step listener. Still expecting: "
+ localState.getExpecting());
}
for (StepContribution contribution : getStepContributions()) {
stepExecution.apply(contribution);
}
}
if (timedOut) {
stepExecution.setStatus(BatchStatus.FAILED);
return ExitStatus.FAILED.addExitDescription(
"Timed out waiting for " + localState.getExpecting() + " backlog at end of step");
}
return ExitStatus.COMPLETED.addExitDescription("Waited for " + expecting + " results.");
}
public void close() throws ItemStreamException {
localState.reset();
}
public void open(ExecutionContext executionContext) throws ItemStreamException {
if (executionContext.containsKey(EXPECTED)) {
localState.open(executionContext.getInt(EXPECTED), executionContext.getInt(ACTUAL));
if (!waitForResults()) {
throw new ItemStreamException("Timed out waiting for back log on open");
}
}
}
public void update(ExecutionContext executionContext) throws ItemStreamException {
executionContext.putInt(EXPECTED, localState.expected.intValue());
executionContext.putInt(ACTUAL, localState.actual.intValue());
}
public Collection<StepContribution> getStepContributions() {
List<StepContribution> contributions = new ArrayList<>();
for (ChunkResponse response : localState.pollChunkResponses()) {
StepContribution contribution = response.getStepContribution();
if (logger.isDebugEnabled()) {
logger.debug("Applying: " + response);
}
contributions.add(contribution);
}
return contributions;
}
/**
* Wait until all the results that are in the pipeline come back to the reply channel.
* @return true if successfully received a result, false if timed out
*/
private boolean waitForResults() throws AsynchronousFailureException {
int count = 0;
int maxCount = maxWaitTimeouts;
Throwable failure = null;
if (logger.isInfoEnabled()) {
logger.info("Waiting for " + localState.getExpecting() + " results");
}
while (localState.getExpecting() > 0 && count++ < maxCount) {
try {
getNextResult();
}
catch (Throwable t) {
logger.error("Detected error in remote result. Trying to recover " + localState.getExpecting()
+ " outstanding results before completing.", t);
failure = t;
}
}
if (failure != null) {
throw wrapIfNecessary(failure);
}
return count < maxCount;
}
/**
* Get the next result if it is available (within the timeout specified in the
* gateway), otherwise do nothing.
* @throws AsynchronousFailureException If there is a response and it contains a
* failed chunk response.
* @throws IllegalStateException if the result contains the wrong job instance id
* (maybe we are sharing a channel and we shouldn't be)
*/
@SuppressWarnings("unchecked")
private void getNextResult() throws AsynchronousFailureException {
Message<ChunkResponse> message = (Message<ChunkResponse>) messagingGateway.receive(replyChannel);
if (message != null) {
ChunkResponse payload = message.getPayload();
if (logger.isDebugEnabled()) {
logger.debug("Found result: " + payload);
}
Long jobInstanceId = payload.getJobId();
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() + "].");
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.pushResponse(payload);
localState.incrementActual();
if (!payload.isSuccessful()) {
throw new AsynchronousFailureException(
"Failure or interrupt detected in handler: " + payload.getMessage());
}
}
}
/**
* Re-throws the original throwable if it is unchecked, wraps checked exceptions into
* {@link AsynchronousFailureException}.
*/
private static AsynchronousFailureException wrapIfNecessary(Throwable throwable) {
if (throwable instanceof Error) {
throw (Error) throwable;
}
else if (throwable instanceof AsynchronousFailureException) {
return (AsynchronousFailureException) throwable;
}
else {
return new AsynchronousFailureException("Exception in remote process", throwable);
}
}
private static class LocalState {
private final AtomicInteger current = new AtomicInteger(-1);
private final AtomicInteger actual = new AtomicInteger();
private final AtomicInteger expected = new AtomicInteger();
private final AtomicInteger redelivered = new AtomicInteger();
private StepExecution stepExecution;
private final Queue<ChunkResponse> contributions = new LinkedBlockingQueue<>();
public int getExpecting() {
return expected.get() - actual.get();
}
public <T> ChunkRequest<T> getRequest(List<? extends T> items) {
return new ChunkRequest<>(current.incrementAndGet(), items, getJobId(), createStepContribution());
}
public void open(int expectedValue, int actualValue) {
actual.set(actualValue);
expected.set(expectedValue);
}
public Collection<ChunkResponse> pollChunkResponses() {
Collection<ChunkResponse> set = new ArrayList<>();
synchronized (contributions) {
ChunkResponse item = contributions.poll();
while (item != null) {
set.add(item);
item = contributions.poll();
}
}
return set;
}
public void pushResponse(ChunkResponse stepContribution) {
synchronized (contributions) {
contributions.add(stepContribution);
}
}
public void incrementRedelivered() {
redelivered.incrementAndGet();
}
public void incrementActual() {
actual.incrementAndGet();
}
public void incrementExpected() {
expected.incrementAndGet();
}
public StepContribution createStepContribution() {
return stepExecution.createStepContribution();
}
public Long getJobId() {
return stepExecution.getJobExecution().getJobId();
}
public void setStepExecution(StepExecution stepExecution) {
this.stepExecution = stepExecution;
}
public void reset() {
expected.set(0);
actual.set(0);
}
}
}

13
spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkProcessorChunkHandler.java
View File

@@ -33,13 +33,13 @@ import org.springframework.retry.RetryException;
import org.springframework.util.Assert;
/**
* A {@link ChunkHandler} based on a {@link ChunkProcessor}. Knows how to distinguish between a processor that is fault
* tolerant, and one that is not. If the processor is fault tolerant then exceptions can be propagated on the assumption
* that there will be a roll back and the request will be re-delivered.
* A {@link ChunkHandler} based on a {@link ChunkProcessor}. Knows how to distinguish
* between a processor that is fault tolerant, and one that is not. If the processor is
* fault tolerant then exceptions can be propagated on the assumption that there will be a
* roll back and the request will be re-delivered.
*
* @author Dave Syer
* @author Michael Minella
*
* @param <S> the type of the items in the chunk to be handled
*/
@MessageEndpoint
@@ -60,7 +60,6 @@ public class ChunkProcessorChunkHandler<S> implements ChunkHandler<S>, Initializ
/**
* Public setter for the {@link ChunkProcessor}.
*
* @param chunkProcessor the chunkProcessor to set
*/
public void setChunkProcessor(ChunkProcessor<S> chunkProcessor) {
@@ -83,8 +82,8 @@ public class ChunkProcessorChunkHandler<S> implements ChunkHandler<S>, Initializ
Throwable failure = process(chunkRequest, stepContribution);
if (failure != null) {
logger.debug("Failed chunk", failure);
return new ChunkResponse(false, chunkRequest.getSequence(), chunkRequest.getJobId(), stepContribution, failure.getClass().getName()
+ ": " + failure.getMessage());
return new ChunkResponse(false, chunkRequest.getSequence(), chunkRequest.getJobId(), stepContribution,
failure.getClass().getName() + ": " + failure.getMessage());
}
if (logger.isDebugEnabled()) {

6
spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkRequest.java
View File

@@ -22,11 +22,9 @@ import java.util.Collection;
import org.springframework.batch.core.StepContribution;
/**
* Encapsulation of a chunk of items to be processed remotely as part of a step
* execution.
*
* Encapsulation of a chunk of items to be processed remotely as part of a step execution.
*
* @author Dave Syer
*
* @param <T> the type of the items to process
*/
public class ChunkRequest<T> implements Serializable {

23
spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkResponse.java
View File

@@ -22,11 +22,12 @@ import org.springframework.batch.core.StepContribution;
import org.springframework.lang.Nullable;
/**
* Encapsulates a response to processing a chunk of items, summarising the result as a {@link StepContribution}.
*
* Encapsulates a response to processing a chunk of items, summarising the result as a
* {@link StepContribution}.
*
* @author Dave Syer
* @author Mahmoud Ben Hassine
*
*
*/
public class ChunkResponse implements Serializable {
@@ -39,7 +40,7 @@ public class ChunkResponse implements Serializable {
private final boolean status;
private final String message;
private final boolean redelivered;
private final int sequence;
@@ -52,7 +53,8 @@ public class ChunkResponse implements Serializable {
this(status, sequence, jobId, stepContribution, null);
}
public ChunkResponse(boolean status, int sequence, Long jobId, StepContribution stepContribution, @Nullable String message) {
public ChunkResponse(boolean status, int sequence, Long jobId, StepContribution stepContribution,
@Nullable String message) {
this(status, sequence, jobId, stepContribution, message, false);
}
@@ -60,7 +62,8 @@ public class ChunkResponse implements Serializable {
this(input.status, input.sequence, input.jobId, input.stepContribution, input.message, redelivered);
}
public ChunkResponse(boolean status, int sequence, Long jobId, StepContribution stepContribution, @Nullable String message, boolean redelivered) {
public ChunkResponse(boolean status, int sequence, Long jobId, StepContribution stepContribution,
@Nullable String message, boolean redelivered) {
this.status = status;
this.sequence = sequence;
this.jobId = jobId;
@@ -76,7 +79,7 @@ public class ChunkResponse implements Serializable {
public Long getJobId() {
return jobId;
}
public int getSequence() {
return sequence;
}
@@ -84,7 +87,7 @@ public class ChunkResponse implements Serializable {
public boolean isSuccessful() {
return status;
}
public boolean isRedelivered() {
return redelivered;
}
@@ -98,8 +101,8 @@ public class ChunkResponse implements Serializable {
*/
@Override
public String toString() {
return getClass().getSimpleName() + ": jobId=" + jobId + ", sequence=" + sequence + ", stepContribution=" + stepContribution
+ ", successful=" + status;
return getClass().getSimpleName() + ": jobId=" + jobId + ", sequence=" + sequence + ", stepContribution="
+ stepContribution + ", successful=" + status;
}
}

76
spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/JmsRedeliveredExtractor.java
View File

@@ -1,38 +1,38 @@
/*
* 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
*
* https://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.messaging.handler.annotation.Header;
import org.springframework.jms.support.JmsHeaders;
/**
* @author Dave Syer
*
*/
public class JmsRedeliveredExtractor {
private static final Log logger = LogFactory.getLog(JmsRedeliveredExtractor.class);
public ChunkResponse extract(ChunkResponse input, @Header(JmsHeaders.REDELIVERED) boolean redelivered) {
if (logger.isDebugEnabled()) {
logger.debug("Extracted redelivered flag for response, value="+redelivered);
}
return new ChunkResponse(input, redelivered);
}
}
/*
* 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
*
* https://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.messaging.handler.annotation.Header;
import org.springframework.jms.support.JmsHeaders;
/**
* @author Dave Syer
*
*/
public class JmsRedeliveredExtractor {
private static final Log logger = LogFactory.getLog(JmsRedeliveredExtractor.class);
public ChunkResponse extract(ChunkResponse input, @Header(JmsHeaders.REDELIVERED) boolean redelivered) {
if (logger.isDebugEnabled()) {
logger.debug("Extracted redelivered flag for response, value=" + redelivered);
}
return new ChunkResponse(input, redelivered);
}
}

19
spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/MessageSourcePollerInterceptor.java
View File

@@ -9,12 +9,12 @@ import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.util.Assert;
/**
* A {@link ChannelInterceptor} that turns a pollable channel into a "pass-thru channel": if a client calls
* <code>receive()</code> on the channel it will delegate to a {@link MessageSource} to pull the message directly from
* an external source. This is particularly useful in combination with a message channel in thread scope, in which case
* the <code>receive()</code> can join a transaction which was started by the caller.
* A {@link ChannelInterceptor} that turns a pollable channel into a "pass-thru channel":
* if a client calls <code>receive()</code> on the channel it will delegate to a
* {@link MessageSource} to pull the message directly from an external source. This is
* particularly useful in combination with a message channel in thread scope, in which
* case the <code>receive()</code> can join a transaction which was started by the caller.
*
* @author Dave Syer
*
@@ -41,9 +41,8 @@ public class MessageSourcePollerInterceptor implements ChannelInterceptor, Initi
}
/**
* Optional MessageChannel for injecting the message received from the source (defaults to the channel intercepted
* in {@link #preReceive(MessageChannel)}).
*
* Optional MessageChannel for injecting the message received from the source
* (defaults to the channel intercepted in {@link #preReceive(MessageChannel)}).
* @param channel the channel to set
*/
public void setChannel(MessageChannel channel) {
@@ -66,8 +65,8 @@ public class MessageSourcePollerInterceptor implements ChannelInterceptor, Initi
}
/**
* Receive from the {@link MessageSource} and send immediately to the input channel, so that the call that we are
* intercepting always a message to receive.
* Receive from the {@link MessageSource} and send immediately to the input channel,
* so that the call that we are intercepting always a message to receive.
*
* @see ChannelInterceptor#preReceive(MessageChannel)
*/

83
spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/RemoteChunkHandlerFactoryBean.java
View File

@@ -36,16 +36,18 @@ import org.springframework.util.Assert;
import org.springframework.util.ReflectionUtils;
/**
* Convenient factory bean for a chunk handler that also converts an existing chunk-oriented step into a remote chunk
* manager. The idea is to lift the existing chunk processor out of a Step that works locally, and replace it with a one
* that writes chunks into a message channel. The existing step hands its business chunk processing responsibility over
* to the handler produced by the factory, which then needs to be set up as a worker on the other end of the channel the
* chunks are being sent to. Once this chunk handler is installed the application is playing the role of both the manager
* and the worker listeners in the Remote Chunking pattern for the Step in question.
*
* Convenient factory bean for a chunk handler that also converts an existing
* chunk-oriented step into a remote chunk manager. The idea is to lift the existing chunk
* processor out of a Step that works locally, and replace it with a one that writes
* chunks into a message channel. The existing step hands its business chunk processing
* responsibility over to the handler produced by the factory, which then needs to be set
* up as a worker on the other end of the channel the chunks are being sent to. Once this
* chunk handler is installed the application is playing the role of both the manager and
* the worker listeners in the Remote Chunking pattern for the Step in question.
*
* @author Dave Syer
* @author Mahmoud Ben Hassine
*
*
*/
public class RemoteChunkHandlerFactoryBean<T> implements FactoryBean<ChunkHandler<T>> {
@@ -59,7 +61,6 @@ public class RemoteChunkHandlerFactoryBean<T> implements FactoryBean<ChunkHandle
/**
* The local step that is to be converted to a remote chunk manager.
*
* @param step the step to set
*/
public void setStep(TaskletStep step) {
@@ -67,9 +68,9 @@ public class RemoteChunkHandlerFactoryBean<T> implements FactoryBean<ChunkHandle
}
/**
* The item writer to be injected into the step. Its responsibility is to send chunks of items to remote workers.
* Usually in practice it will be a {@link ChunkMessageChannelItemWriter}.
*
* The item writer to be injected into the step. Its responsibility is to send chunks
* of items to remote workers. Usually in practice it will be a
* {@link ChunkMessageChannelItemWriter}.
* @param chunkWriter the chunk writer to set
*/
public void setChunkWriter(ItemWriter<T> chunkWriter) {
@@ -78,8 +79,8 @@ public class RemoteChunkHandlerFactoryBean<T> implements FactoryBean<ChunkHandle
/**
* A source of {@link StepContribution} instances coming back from remote workers.
*
* @param stepContributionSource the step contribution source to set (defaults to the chunk writer)
* @param stepContributionSource the step contribution source to set (defaults to the
* chunk writer)
*/
public void setStepContributionSource(StepContributionSource stepContributionSource) {
this.stepContributionSource = stepContributionSource;
@@ -87,7 +88,7 @@ public class RemoteChunkHandlerFactoryBean<T> implements FactoryBean<ChunkHandle
/**
* The type of object created by this factory. Returns {@link ChunkHandler} class.
*
*
* @see FactoryBean#getObjectType()
*/
public Class<?> getObjectType() {
@@ -96,7 +97,7 @@ public class RemoteChunkHandlerFactoryBean<T> implements FactoryBean<ChunkHandle
/**
* Optimization for the bean factory (always returns true).
*
*
* @see FactoryBean#isSingleton()
*/
public boolean isSingleton() {
@@ -104,10 +105,10 @@ public class RemoteChunkHandlerFactoryBean<T> implements FactoryBean<ChunkHandle
}
/**
* Builds a {@link ChunkHandler} from the {@link ChunkProcessor} extracted from the {@link #setStep(TaskletStep)
* step} provided. Also modifies the step to send chunks to the chunk handler via the
* {@link #setChunkWriter(ItemWriter) chunk writer}.
*
* Builds a {@link ChunkHandler} from the {@link ChunkProcessor} extracted from the
* {@link #setStep(TaskletStep) step} provided. Also modifies the step to send chunks
* to the chunk handler via the {@link #setChunkWriter(ItemWriter) chunk writer}.
*
* @see FactoryBean#getObject()
*/
public ChunkHandler<T> getObject() throws Exception {
@@ -124,8 +125,8 @@ public class RemoteChunkHandlerFactoryBean<T> implements FactoryBean<ChunkHandle
}
Tasklet tasklet = getTasklet(step);
Assert.state(tasklet instanceof ChunkOrientedTasklet<?>, "Tasklet must be ChunkOrientedTasklet in step="
+ step.getName());
Assert.state(tasklet instanceof ChunkOrientedTasklet<?>,
"Tasklet must be ChunkOrientedTasklet in step=" + step.getName());
ChunkProcessor<T> chunkProcessor = getChunkProcessor((ChunkOrientedTasklet<?>) tasklet);
Assert.state(chunkProcessor != null, "ChunkProcessor must be accessible in Tasklet in step=" + step.getName());
@@ -161,35 +162,37 @@ public class RemoteChunkHandlerFactoryBean<T> implements FactoryBean<ChunkHandle
}
/**
* Replace the chunk processor in the tasklet provided with one that can act as a manager in the Remote Chunking
* pattern.
*
* Replace the chunk processor in the tasklet provided with one that can act as a
* manager in the Remote Chunking pattern.
* @param tasklet a ChunkOrientedTasklet
* @param chunkWriter an ItemWriter that can send the chunks to remote workers
* @param stepContributionSource a StepContributionSource used to gather results from the workers
* @param stepContributionSource a StepContributionSource used to gather results from
* the workers
*/
private void replaceChunkProcessor(ChunkOrientedTasklet<?> tasklet, ItemWriter<T> chunkWriter,
final StepContributionSource stepContributionSource) {
setField(tasklet, "chunkProcessor", new SimpleChunkProcessor<T, T>(new PassThroughItemProcessor<>(),
chunkWriter) {
@Override
protected void write(StepContribution contribution, Chunk<T> inputs, Chunk<T> outputs) throws Exception {
doWrite(outputs.getItems());
// Do not update the step contribution until the chunks are
// actually processed
updateStepContribution(contribution, stepContributionSource);
}
});
setField(tasklet, "chunkProcessor",
new SimpleChunkProcessor<T, T>(new PassThroughItemProcessor<>(), chunkWriter) {
@Override
protected void write(StepContribution contribution, Chunk<T> inputs, Chunk<T> outputs)
throws Exception {
doWrite(outputs.getItems());
// Do not update the step contribution until the chunks are
// actually processed
updateStepContribution(contribution, stepContributionSource);
}
});
}
/**
* Update a StepContribution with all the data from a StepContributionSource. The filter and write counts plus the
* exit status will be updated to reflect the data in the source.
*
* Update a StepContribution with all the data from a StepContributionSource. The
* filter and write counts plus the exit status will be updated to reflect the data in
* the source.
* @param contribution the current contribution
* @param stepContributionSource a source of StepContributions
*/
protected void updateStepContribution(StepContribution contribution, StepContributionSource stepContributionSource) {
protected void updateStepContribution(StepContribution contribution,
StepContributionSource stepContributionSource) {
for (StepContribution result : stepContributionSource.getStepContributions()) {
contribution.incrementFilterCount(result.getFilterCount());
contribution.incrementWriteCount(result.getWriteCount());

104
spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/RemoteChunkingManagerStepBuilder.java
View File

@@ -44,38 +44,46 @@ import org.springframework.transaction.interceptor.TransactionAttribute;
import org.springframework.util.Assert;
/**
* Builder for a manager step in a remote chunking setup. This builder creates and
* sets a {@link ChunkMessageChannelItemWriter} on the manager step.
* Builder for a manager step in a remote chunking setup. This builder creates and sets a
* {@link ChunkMessageChannelItemWriter} on the manager step.
*
* <p>If no {@code messagingTemplate} is provided through
* {@link RemoteChunkingManagerStepBuilder#messagingTemplate(MessagingTemplate)},
* this builder will create one and set its default channel to the {@code outputChannel}
* provided through {@link RemoteChunkingManagerStepBuilder#outputChannel(MessageChannel)}.</p>
* <p>
* If no {@code messagingTemplate} is provided through
* {@link RemoteChunkingManagerStepBuilder#messagingTemplate(MessagingTemplate)}, this
* builder will create one and set its default channel to the {@code outputChannel}
* provided through
* {@link RemoteChunkingManagerStepBuilder#outputChannel(MessageChannel)}.
* </p>
*
* <p>If a {@code messagingTemplate} is provided, it is assumed that it is fully configured
* <p>
* If a {@code messagingTemplate} is provided, it is assumed that it is fully configured
* and that its default channel is set to an output channel on which requests to workers
* will be sent.</p>
* will be sent.
* </p>
*
* @param <I> type of input items
* @param <O> type of output items
*
* @since 4.2
* @author Mahmoud Ben Hassine
*/
public class RemoteChunkingManagerStepBuilder<I, O> extends FaultTolerantStepBuilder<I, O> {
private MessagingTemplate messagingTemplate;
private PollableChannel inputChannel;
private MessageChannel outputChannel;
private final int DEFAULT_MAX_WAIT_TIMEOUTS = 40;
private static final long DEFAULT_THROTTLE_LIMIT = 6;
private int maxWaitTimeouts = DEFAULT_MAX_WAIT_TIMEOUTS;
private long throttleLimit = DEFAULT_THROTTLE_LIMIT;
/**
* Create a new {@link RemoteChunkingManagerStepBuilder}.
*
* @param stepName name of the manager step
*/
public RemoteChunkingManagerStepBuilder(String stepName) {
@@ -83,10 +91,9 @@ public class RemoteChunkingManagerStepBuilder<I, O> extends FaultTolerantStepBui
}
/**
* Set the input channel on which replies from workers will be received.
* The provided input channel will be set as a reply channel on the
* Set the input channel on which replies from workers will be received. The provided
* input channel will be set as a reply channel on the
* {@link ChunkMessageChannelItemWriter} created by this builder.
*
* @param inputChannel the input channel
* @return this builder instance for fluent chaining
*
@@ -99,12 +106,14 @@ public class RemoteChunkingManagerStepBuilder<I, O> extends FaultTolerantStepBui
}
/**
* Set the output channel on which requests to workers will be sent. By using
* this setter, a default messaging template will be created and the output
* channel will be set as its default channel.
* <p>Use either this setter or {@link RemoteChunkingManagerStepBuilder#messagingTemplate(MessagingTemplate)}
* to provide a fully configured messaging template.</p>
*
* Set the output channel on which requests to workers will be sent. By using this
* setter, a default messaging template will be created and the output channel will be
* set as its default channel.
* <p>
* Use either this setter or
* {@link RemoteChunkingManagerStepBuilder#messagingTemplate(MessagingTemplate)} to
* provide a fully configured messaging template.
* </p>
* @param outputChannel the output channel.
* @return this builder instance for fluent chaining
*
@@ -117,12 +126,14 @@ public class RemoteChunkingManagerStepBuilder<I, O> extends FaultTolerantStepBui
}
/**
* Set the {@link MessagingTemplate} to use to send data to workers.
* <strong>The default channel of the messaging template must be set</strong>.
* <p>Use either this setter to provide a fully configured messaging template or
* provide an output channel through {@link RemoteChunkingManagerStepBuilder#outputChannel(MessageChannel)}
* and a default messaging template will be created.</p>
*
* Set the {@link MessagingTemplate} to use to send data to workers. <strong>The
* default channel of the messaging template must be set</strong>.
* <p>
* Use either this setter to provide a fully configured messaging template or provide
* an output channel through
* {@link RemoteChunkingManagerStepBuilder#outputChannel(MessageChannel)} and a
* default messaging template will be created.
* </p>
* @param messagingTemplate the messaging template to use
* @return this builder instance for fluent chaining
* @see RemoteChunkingManagerStepBuilder#outputChannel(MessageChannel)
@@ -134,10 +145,10 @@ public class RemoteChunkingManagerStepBuilder<I, O> extends FaultTolerantStepBui
}
/**
* The maximum number of times to wait at the end of a step for a non-null result from the remote workers. This is a
* multiplier on the receive timeout set separately on the gateway. The ideal value is a compromise between allowing
* slow workers time to finish, and responsiveness if there is a dead worker. Defaults to 40.
*
* The maximum number of times to wait at the end of a step for a non-null result from
* the remote workers. This is a multiplier on the receive timeout set separately on
* the gateway. The ideal value is a compromise between allowing slow workers time to
* finish, and responsiveness if there is a dead worker. Defaults to 40.
* @param maxWaitTimeouts the maximum number of wait timeouts
* @return this builder instance for fluent chaining
* @see ChunkMessageChannelItemWriter#setMaxWaitTimeouts(int)
@@ -149,9 +160,8 @@ public class RemoteChunkingManagerStepBuilder<I, O> extends FaultTolerantStepBui
}
/**
* Public setter for the throttle limit. This limits the number of pending requests for chunk processing to avoid
* overwhelming the receivers.
*
* Public setter for the throttle limit. This limits the number of pending requests
* for chunk processing to avoid overwhelming the receivers.
* @param throttleLimit the throttle limit to set
* @return this builder instance for fluent chaining
* @see ChunkMessageChannelItemWriter#setThrottleLimit(long)
@@ -164,7 +174,6 @@ public class RemoteChunkingManagerStepBuilder<I, O> extends FaultTolerantStepBui
/**
* Build a manager {@link TaskletStep}.
*
* @return the configured manager step
* @see RemoteChunkHandlerFactoryBean
*/
@@ -194,10 +203,9 @@ public class RemoteChunkingManagerStepBuilder<I, O> extends FaultTolerantStepBui
}
/*
* The following methods override those from parent builders and return
* the current builder type.
* FIXME: Change parent builders to be generic and return current builder
* type in each method.
* The following methods override those from parent builders and return the current
* builder type. FIXME: Change parent builders to be generic and return current
* builder type in each method.
*/
@Override
@@ -339,23 +347,22 @@ public class RemoteChunkingManagerStepBuilder<I, O> extends FaultTolerantStepBui
}
/**
* This method will throw a {@link UnsupportedOperationException} since
* the item writer of the manager step in a remote chunking setup will be
* automatically set to an instance of {@link ChunkMessageChannelItemWriter}.
*
* When building a manager step for remote chunking, no item writer must be
* provided.
* This method will throw a {@link UnsupportedOperationException} since the item
* writer of the manager step in a remote chunking setup will be automatically set to
* an instance of {@link ChunkMessageChannelItemWriter}.
*
* When building a manager step for remote chunking, no item writer must be provided.
* @throws UnsupportedOperationException if an item writer is provided
* @see ChunkMessageChannelItemWriter
* @see RemoteChunkHandlerFactoryBean#setChunkWriter(ItemWriter)
*/
@Override
public RemoteChunkingManagerStepBuilder<I, O> writer(ItemWriter<? super O> writer) throws UnsupportedOperationException {
throw new UnsupportedOperationException("When configuring a manager step " +
"for remote chunking, the item writer will be automatically set " +
"to an instance of ChunkMessageChannelItemWriter. The item writer " +
"must not be provided in this case.");
public RemoteChunkingManagerStepBuilder<I, O> writer(ItemWriter<? super O> writer)
throws UnsupportedOperationException {
throw new UnsupportedOperationException(
"When configuring a manager step " + "for remote chunking, the item writer will be automatically set "
+ "to an instance of ChunkMessageChannelItemWriter. The item writer "
+ "must not be provided in this case.");
}
@Override
@@ -417,4 +424,5 @@ public class RemoteChunkingManagerStepBuilder<I, O> extends FaultTolerantStepBui
super.processor(itemProcessor);
return this;
}
}

12
spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/RemoteChunkingManagerStepBuilderFactory.java
View File

@@ -19,8 +19,8 @@ import org.springframework.batch.core.repository.JobRepository;
import org.springframework.transaction.PlatformTransactionManager;
/**
* Convenient factory for a {@link RemoteChunkingManagerStepBuilder} which sets
* the {@link JobRepository} and {@link PlatformTransactionManager} automatically.
* Convenient factory for a {@link RemoteChunkingManagerStepBuilder} which sets the
* {@link JobRepository} and {@link PlatformTransactionManager} automatically.
*
* @since 4.2
* @author Mahmoud Ben Hassine
@@ -33,12 +33,10 @@ public class RemoteChunkingManagerStepBuilderFactory {
/**
* Create a new {@link RemoteChunkingManagerStepBuilderFactory}.
*
* @param jobRepository the job repository to use
* @param transactionManager the transaction manager to use
*/
public RemoteChunkingManagerStepBuilderFactory(
JobRepository jobRepository,
public RemoteChunkingManagerStepBuilderFactory(JobRepository jobRepository,
PlatformTransactionManager transactionManager) {
this.jobRepository = jobRepository;
@@ -48,15 +46,13 @@ public class RemoteChunkingManagerStepBuilderFactory {
/**
* Creates a {@link RemoteChunkingManagerStepBuilder} and initializes its job
* repository and transaction manager.
*
* @param name the name of the step
* @param <I> type of input items
* @param <O> type of output items
* @return a {@link RemoteChunkingManagerStepBuilder}
*/
public <I, O> RemoteChunkingManagerStepBuilder<I, O> get(String name) {
return new RemoteChunkingManagerStepBuilder<I, O>(name)
.repository(this.jobRepository)
return new RemoteChunkingManagerStepBuilder<I, O>(name).repository(this.jobRepository)
.transactionManager(this.transactionManager);
}

40
spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/RemoteChunkingWorkerBuilder.java
View File

@@ -28,18 +28,16 @@ import org.springframework.util.Assert;
* Builder for a worker in a remote chunking setup. This builder:
*
* <ul>
* <li>creates a {@link ChunkProcessorChunkHandler} with the provided
* item processor and writer. If no item processor is provided, a
* {@link PassThroughItemProcessor} will be used</li>
* <li>creates an {@link IntegrationFlow} with the
* {@link ChunkProcessorChunkHandler} as a service activator which listens
* to incoming requests on <code>inputChannel</code> and sends replies
* on <code>outputChannel</code></li>
* <li>creates a {@link ChunkProcessorChunkHandler} with the provided item processor and
* writer. If no item processor is provided, a {@link PassThroughItemProcessor} will be
* used</li>
* <li>creates an {@link IntegrationFlow} with the {@link ChunkProcessorChunkHandler} as a
* service activator which listens to incoming requests on <code>inputChannel</code> and
* sends replies on <code>outputChannel</code></li>
* </ul>
*
* @param <I> type of input items
* @param <O> type of output items
*
* @since 4.1
* @author Mahmoud Ben Hassine
*/
@@ -48,14 +46,15 @@ public class RemoteChunkingWorkerBuilder<I, O> {
private static final String SERVICE_ACTIVATOR_METHOD_NAME = "handleChunk";
private ItemProcessor<I, O> itemProcessor;
private ItemWriter<O> itemWriter;
private MessageChannel inputChannel;
private MessageChannel outputChannel;
/**
* Set the {@link ItemProcessor} to use to process items sent by the manager
* step.
*
* Set the {@link ItemProcessor} to use to process items sent by the manager step.
* @param itemProcessor to use
* @return this builder instance for fluent chaining
*/
@@ -67,7 +66,6 @@ public class RemoteChunkingWorkerBuilder<I, O> {
/**
* Set the {@link ItemWriter} to use to write items sent by the manager step.
*
* @param itemWriter to use
* @return this builder instance for fluent chaining
*/
@@ -79,7 +77,6 @@ public class RemoteChunkingWorkerBuilder<I, O> {
/**
* Set the input channel on which items sent by the manager are received.
*
* @param inputChannel the input channel
* @return this builder instance for fluent chaining
*/
@@ -91,7 +88,6 @@ public class RemoteChunkingWorkerBuilder<I, O> {
/**
* Set the output channel on which replies will be sent to the manager step.
*
* @param outputChannel the output channel
* @return this builder instance for fluent chaining
*/
@@ -103,18 +99,17 @@ public class RemoteChunkingWorkerBuilder<I, O> {
/**
* Create an {@link IntegrationFlow} with a {@link ChunkProcessorChunkHandler}
* configured as a service activator listening to the input channel and replying
* on the output channel.
*
* configured as a service activator listening to the input channel and replying on
* the output channel.
* @return the integration flow
*/
@SuppressWarnings({"unchecked", "rawtypes"})
@SuppressWarnings({ "unchecked", "rawtypes" })
public IntegrationFlow build() {
Assert.notNull(this.itemWriter, "An ItemWriter must be provided");
Assert.notNull(this.inputChannel, "An InputChannel must be provided");
Assert.notNull(this.outputChannel, "An OutputChannel must be provided");
if(this.itemProcessor == null) {
if (this.itemProcessor == null) {
this.itemProcessor = new PassThroughItemProcessor();
}
SimpleChunkProcessor<I, O> chunkProcessor = new SimpleChunkProcessor<>(this.itemProcessor, this.itemWriter);
@@ -122,11 +117,8 @@ public class RemoteChunkingWorkerBuilder<I, O> {
ChunkProcessorChunkHandler<I> chunkProcessorChunkHandler = new ChunkProcessorChunkHandler<>();
chunkProcessorChunkHandler.setChunkProcessor(chunkProcessor);
return IntegrationFlows
.from(this.inputChannel)
.handle(chunkProcessorChunkHandler, SERVICE_ACTIVATOR_METHOD_NAME)
.channel(this.outputChannel)
.get();
return IntegrationFlows.from(this.inputChannel)
.handle(chunkProcessorChunkHandler, SERVICE_ACTIVATOR_METHOD_NAME).channel(this.outputChannel).get();
}
}

13
spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/StepContributionSource.java
View File

@@ -22,18 +22,17 @@ import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.StepExecution;
/**
* A source of {@link StepContribution} instances that can be aggregated and used to update an ongoing
* {@link StepExecution}.
*
* A source of {@link StepContribution} instances that can be aggregated and used to
* update an ongoing {@link StepExecution}.
*
* @author Dave Syer
*
*
*/
public interface StepContributionSource {
/**
* Get the currently available contributions and drain the source. The next call would return an empty collection,
* unless new contributions have arrived.
*
* Get the currently available contributions and drain the source. The next call would
* return an empty collection, unless new contributions have arrived.
* @return a collection of {@link StepContribution} instances
*/
Collection<StepContribution> getStepContributions();

15
spring-batch-integration/src/main/java/org/springframework/batch/integration/config/annotation/BatchIntegrationConfiguration.java
View File

@@ -51,9 +51,7 @@ public class BatchIntegrationConfiguration implements InitializingBean {
private RemotePartitioningWorkerStepBuilderFactory remotePartitioningWorkerStepBuilderFactory;
@Autowired
public BatchIntegrationConfiguration(
JobRepository jobRepository,
JobExplorer jobExplorer,
public BatchIntegrationConfiguration(JobRepository jobRepository, JobExplorer jobExplorer,
PlatformTransactionManager transactionManager) {
this.jobRepository = jobRepository;
@@ -67,7 +65,7 @@ public class BatchIntegrationConfiguration implements InitializingBean {
}
@Bean
public <I,O> RemoteChunkingWorkerBuilder<I, O> remoteChunkingWorkerBuilder() {
public <I, O> RemoteChunkingWorkerBuilder<I, O> remoteChunkingWorkerBuilder() {
return remoteChunkingWorkerBuilder;
}
@@ -86,9 +84,10 @@ public class BatchIntegrationConfiguration implements InitializingBean {
this.remoteChunkingManagerStepBuilderFactory = new RemoteChunkingManagerStepBuilderFactory(this.jobRepository,
this.transactionManager);
this.remoteChunkingWorkerBuilder = new RemoteChunkingWorkerBuilder<>();
this.remotePartitioningManagerStepBuilderFactory = new RemotePartitioningManagerStepBuilderFactory(this.jobRepository,
this.jobExplorer, this.transactionManager);
this.remotePartitioningWorkerStepBuilderFactory = new RemotePartitioningWorkerStepBuilderFactory(this.jobRepository,
this.jobExplorer, this.transactionManager);
this.remotePartitioningManagerStepBuilderFactory = new RemotePartitioningManagerStepBuilderFactory(
this.jobRepository, this.jobExplorer, this.transactionManager);
this.remotePartitioningWorkerStepBuilderFactory = new RemotePartitioningWorkerStepBuilderFactory(
this.jobRepository, this.jobExplorer, this.transactionManager);
}
}

32
spring-batch-integration/src/main/java/org/springframework/batch/integration/config/annotation/EnableBatchIntegration.java
View File

@@ -29,24 +29,25 @@ import org.springframework.context.annotation.Import;
import org.springframework.integration.config.EnableIntegration;
/**
* Enable Spring Batch Integration features and provide a base configuration for
* setting up remote chunking or partitioning infrastructure beans.
* Enable Spring Batch Integration features and provide a base configuration for setting
* up remote chunking or partitioning infrastructure beans.
*
* By adding this annotation on a {@link org.springframework.context.annotation.Configuration}
* class, it will be possible to autowire the following beans:
* By adding this annotation on a
* {@link org.springframework.context.annotation.Configuration} class, it will be possible
* to autowire the following beans:
*
* <ul>
* <li>{@link RemoteChunkingManagerStepBuilderFactory}:
* used to create a manager step of a remote chunking setup by automatically
* setting the job repository and transaction manager.</li>
* <li>{@link RemoteChunkingWorkerBuilder}: used to create the integration
* flow on the worker side of a remote chunking setup.</li>
* <li>{@link RemotePartitioningManagerStepBuilderFactory}: used to create
* a manager step of a remote partitioning setup by automatically setting
* the job repository, job explorer, bean factory and transaction manager.</li>
* <li>{@link RemotePartitioningWorkerStepBuilderFactory}: used to create
* a worker step of a remote partitioning setup by automatically setting
* the job repository, job explorer, bean factory and transaction manager.</li>
* <li>{@link RemoteChunkingManagerStepBuilderFactory}: used to create a manager step of a
* remote chunking setup by automatically setting the job repository and transaction
* manager.</li>
* <li>{@link RemoteChunkingWorkerBuilder}: used to create the integration flow on the
* worker side of a remote chunking setup.</li>
* <li>{@link RemotePartitioningManagerStepBuilderFactory}: used to create a manager step
* of a remote partitioning setup by automatically setting the job repository, job
* explorer, bean factory and transaction manager.</li>
* <li>{@link RemotePartitioningWorkerStepBuilderFactory}: used to create a worker step of
* a remote partitioning setup by automatically setting the job repository, job explorer,
* bean factory and transaction manager.</li>
* </ul>
*
* For remote chunking, an example of a configuration class would be:
@@ -140,4 +141,5 @@ import org.springframework.integration.config.EnableIntegration;
@EnableIntegration
@Import(BatchIntegrationConfiguration.class)
public @interface EnableBatchIntegration {
}

8
spring-batch-integration/src/main/java/org/springframework/batch/integration/config/xml/BatchIntegrationNamespaceHandler.java
View File

@@ -26,14 +26,18 @@ import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHa
* @since 1.3
*/
public class BatchIntegrationNamespaceHandler extends AbstractIntegrationNamespaceHandler {
/* (non-Javadoc)
/*
* (non-Javadoc)
*
* @see org.springframework.beans.factory.xml.NamespaceHandler#init()
*/
public void init() {
this.registerBeanDefinitionParser("job-launching-gateway", new JobLaunchingGatewayParser());
this.registerBeanDefinitionParser("job-launching-gateway", new JobLaunchingGatewayParser());
RemoteChunkingManagerParser remoteChunkingManagerParser = new RemoteChunkingManagerParser();
this.registerBeanDefinitionParser("remote-chunking-manager", remoteChunkingManagerParser);
RemoteChunkingWorkerParser remoteChunkingWorkerParser = new RemoteChunkingWorkerParser();
this.registerBeanDefinitionParser("remote-chunking-worker", remoteChunkingWorkerParser);
}
}

12
spring-batch-integration/src/main/java/org/springframework/batch/integration/config/xml/JobLaunchingGatewayParser.java
View File

@@ -28,9 +28,8 @@ import org.w3c.dom.Element;
/**
* The parser for the Job-Launching Gateway, which will instantiate a
* {@link JobLaunchingGatewayParser}. If no {@link JobLauncher} reference has
* been provided, this parse will use the use the globally registered bean
* 'jobLauncher'.
* {@link JobLaunchingGatewayParser}. If no {@link JobLauncher} reference has been
* provided, this parse will use the use the globally registered bean 'jobLauncher'.
*
* @author Gunnar Hillert
* @since 1.3
@@ -48,8 +47,8 @@ public class JobLaunchingGatewayParser extends AbstractConsumerEndpointParser {
@Override
protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) {
final BeanDefinitionBuilder jobLaunchingGatewayBuilder =
BeanDefinitionBuilder.genericBeanDefinition(JobLaunchingGateway.class);
final BeanDefinitionBuilder jobLaunchingGatewayBuilder = BeanDefinitionBuilder
.genericBeanDefinition(JobLaunchingGateway.class);
final String jobLauncher = element.getAttribute("job-launcher");
@@ -63,7 +62,8 @@ public class JobLaunchingGatewayParser extends AbstractConsumerEndpointParser {
jobLaunchingGatewayBuilder.addConstructorArgReference("jobLauncher");
}
IntegrationNamespaceUtils.setValueIfAttributeDefined(jobLaunchingGatewayBuilder, element, "reply-timeout", "sendTimeout");
IntegrationNamespaceUtils.setValueIfAttributeDefined(jobLaunchingGatewayBuilder, element, "reply-timeout",
"sendTimeout");
final String replyChannel = element.getAttribute("reply-channel");

32
spring-batch-integration/src/main/java/org/springframework/batch/integration/config/xml/RemoteChunkingManagerParser.java
View File

@@ -36,13 +36,21 @@ import org.w3c.dom.Element;
* @since 3.1
*/
public class RemoteChunkingManagerParser extends AbstractBeanDefinitionParser {
private static final String MESSAGE_TEMPLATE_ATTRIBUTE = "message-template";
private static final String STEP_ATTRIBUTE = "step";
private static final String REPLY_CHANNEL_ATTRIBUTE = "reply-channel";
private static final String MESSAGING_OPERATIONS_PROPERTY = "messagingOperations";
private static final String REPLY_CHANNEL_PROPERTY = "replyChannel";
private static final String CHUNK_WRITER_PROPERTY = "chunkWriter";
private static final String STEP_PROPERTY = "step";
private static final String CHUNK_HANDLER_BEAN_NAME_PREFIX = "remoteChunkHandlerFactoryBean_";
@Override
@@ -61,24 +69,22 @@ public class RemoteChunkingManagerParser extends AbstractBeanDefinitionParser {
BeanDefinitionRegistry beanDefinitionRegistry = parserContext.getRegistry();
BeanDefinition chunkMessageChannelItemWriter =
BeanDefinitionBuilder
.genericBeanDefinition(ChunkMessageChannelItemWriter.class)
.addPropertyReference(MESSAGING_OPERATIONS_PROPERTY, messageTemplate)
.addPropertyReference(REPLY_CHANNEL_PROPERTY, replyChannel)
.getBeanDefinition();
BeanDefinition chunkMessageChannelItemWriter = BeanDefinitionBuilder
.genericBeanDefinition(ChunkMessageChannelItemWriter.class)
.addPropertyReference(MESSAGING_OPERATIONS_PROPERTY, messageTemplate)
.addPropertyReference(REPLY_CHANNEL_PROPERTY, replyChannel).getBeanDefinition();
beanDefinitionRegistry.registerBeanDefinition(id, chunkMessageChannelItemWriter);
BeanDefinition remoteChunkHandlerFactoryBean =
BeanDefinitionBuilder
.genericBeanDefinition(RemoteChunkHandlerFactoryBean.class)
.addPropertyValue(CHUNK_WRITER_PROPERTY, chunkMessageChannelItemWriter)
.addPropertyValue(STEP_PROPERTY, step)
.getBeanDefinition();
BeanDefinition remoteChunkHandlerFactoryBean = BeanDefinitionBuilder
.genericBeanDefinition(RemoteChunkHandlerFactoryBean.class)
.addPropertyValue(CHUNK_WRITER_PROPERTY, chunkMessageChannelItemWriter)
.addPropertyValue(STEP_PROPERTY, step).getBeanDefinition();
beanDefinitionRegistry.registerBeanDefinition(CHUNK_HANDLER_BEAN_NAME_PREFIX + step, remoteChunkHandlerFactoryBean);
beanDefinitionRegistry.registerBeanDefinition(CHUNK_HANDLER_BEAN_NAME_PREFIX + step,
remoteChunkHandlerFactoryBean);
return null;
}
}

44
spring-batch-integration/src/main/java/org/springframework/batch/integration/config/xml/RemoteChunkingWorkerParser.java
View File

@@ -45,13 +45,21 @@ import org.springframework.util.StringUtils;
* @since 3.1
*/
public class RemoteChunkingWorkerParser extends AbstractBeanDefinitionParser {
private static final String INPUT_CHANNEL_ATTRIBUTE = "input-channel";
private static final String OUTPUT_CHANNEL_ATTRIBUTE = "output-channel";
private static final String ITEM_PROCESSOR_ATTRIBUTE = "item-processor";
private static final String ITEM_WRITER_ATTRIBUTE = "item-writer";
private static final String ITEM_PROCESSOR_PROPERTY_NAME = "itemProcessor";
private static final String ITEM_WRITER_PROPERTY_NAME = "itemWriter";
private static final String CHUNK_PROCESSOR_PROPERTY_NAME = "chunkProcessor";
private static final String CHUNK_PROCESSOR_CHUNK_HANDLER_BEAN_NAME_PREFIX = "chunkProcessorChunkHandler_";
@Override
@@ -72,24 +80,24 @@ public class RemoteChunkingWorkerParser extends AbstractBeanDefinitionParser {
BeanDefinitionRegistry beanDefinitionRegistry = parserContext.getRegistry();
BeanDefinitionBuilder chunkProcessorBuilder =
BeanDefinitionBuilder
.genericBeanDefinition(SimpleChunkProcessor.class)
.addPropertyReference(ITEM_WRITER_PROPERTY_NAME, itemWriter);
BeanDefinitionBuilder chunkProcessorBuilder = BeanDefinitionBuilder
.genericBeanDefinition(SimpleChunkProcessor.class)
.addPropertyReference(ITEM_WRITER_PROPERTY_NAME, itemWriter);
if(StringUtils.hasText(itemProcessor)) {
if (StringUtils.hasText(itemProcessor)) {
chunkProcessorBuilder.addPropertyReference(ITEM_PROCESSOR_PROPERTY_NAME, itemProcessor);
} else {
}
else {
chunkProcessorBuilder.addPropertyValue(ITEM_PROCESSOR_PROPERTY_NAME, new PassThroughItemProcessor<>());
}
BeanDefinition chunkProcessorChunkHandler =
BeanDefinitionBuilder
.genericBeanDefinition(ChunkProcessorChunkHandler.class)
.addPropertyValue(CHUNK_PROCESSOR_PROPERTY_NAME, chunkProcessorBuilder.getBeanDefinition())
.getBeanDefinition();
BeanDefinition chunkProcessorChunkHandler = BeanDefinitionBuilder
.genericBeanDefinition(ChunkProcessorChunkHandler.class)
.addPropertyValue(CHUNK_PROCESSOR_PROPERTY_NAME, chunkProcessorBuilder.getBeanDefinition())
.getBeanDefinition();
beanDefinitionRegistry.registerBeanDefinition(CHUNK_PROCESSOR_CHUNK_HANDLER_BEAN_NAME_PREFIX + id, chunkProcessorChunkHandler);
beanDefinitionRegistry.registerBeanDefinition(CHUNK_PROCESSOR_CHUNK_HANDLER_BEAN_NAME_PREFIX + id,
chunkProcessorChunkHandler);
new ServiceActivatorParser(id).parse(element, parserContext);
@@ -97,9 +105,13 @@ public class RemoteChunkingWorkerParser extends AbstractBeanDefinitionParser {
}
private static class ServiceActivatorParser extends AbstractConsumerEndpointParser {
private static final String TARGET_METHOD_NAME_PROPERTY_NAME = "targetMethodName";
private static final String TARGET_OBJECT_PROPERTY_NAME = "targetObject";
private static final String HANDLE_CHUNK_METHOD_NAME = "handleChunk";
private static final String CHUNK_PROCESSOR_CHUNK_HANDLER_BEAN_NAME_PREFIX = "chunkProcessorChunkHandler_";
private String id;
@@ -110,10 +122,14 @@ public class RemoteChunkingWorkerParser extends AbstractBeanDefinitionParser {
@Override
protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(ServiceActivatorFactoryBean.class);
BeanDefinitionBuilder builder = BeanDefinitionBuilder
.genericBeanDefinition(ServiceActivatorFactoryBean.class);
builder.addPropertyValue(TARGET_METHOD_NAME_PROPERTY_NAME, HANDLE_CHUNK_METHOD_NAME);
builder.addPropertyValue(TARGET_OBJECT_PROPERTY_NAME, new RuntimeBeanReference(CHUNK_PROCESSOR_CHUNK_HANDLER_BEAN_NAME_PREFIX + id));
builder.addPropertyValue(TARGET_OBJECT_PROPERTY_NAME,
new RuntimeBeanReference(CHUNK_PROCESSOR_CHUNK_HANDLER_BEAN_NAME_PREFIX + id));
return builder;
}
}
}

7
spring-batch-integration/src/main/java/org/springframework/batch/integration/launch/JobLaunchRequest.java
View File

@@ -19,10 +19,11 @@ import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobParameters;
/**
* Encapsulation of a {@link Job} and its {@link JobParameters} forming a request for a job to be launched.
*
* Encapsulation of a {@link Job} and its {@link JobParameters} forming a request for a
* job to be launched.
*
* @author Dave Syer
*
*
*/
public class JobLaunchRequest {

1
spring-batch-integration/src/main/java/org/springframework/batch/integration/launch/JobLaunchRequestHandler.java
View File

@@ -21,6 +21,7 @@ import org.springframework.batch.core.JobExecutionException;
/**
* Interface for handling a {@link JobLaunchRequest} and returning a {@link JobExecution}.
*
* @author Dave Syer
*
*/

22
spring-batch-integration/src/main/java/org/springframework/batch/integration/launch/JobLaunchingGateway.java
View File

@@ -25,11 +25,10 @@ import org.springframework.messaging.MessageHandlingException;
import org.springframework.util.Assert;
/**
* The {@link JobLaunchingGateway} is used to launch Batch Jobs. Internally it
* delegates to a {@link JobLaunchingMessageHandler}.
* The {@link JobLaunchingGateway} is used to launch Batch Jobs. Internally it delegates
* to a {@link JobLaunchingMessageHandler}.
*
* @author Gunnar Hillert
*
* @since 1.3
*/
public class JobLaunchingGateway extends AbstractReplyProducingMessageHandler {
@@ -38,7 +37,6 @@ public class JobLaunchingGateway extends AbstractReplyProducingMessageHandler {
/**
* Constructor taking a {@link JobLauncher} as parameter.
*
* @param jobLauncher Must not be null.
*
*/
@@ -48,15 +46,12 @@ public class JobLaunchingGateway extends AbstractReplyProducingMessageHandler {
}
/**
* Launches a Batch Job using the provided request {@link Message}. The payload
* of the {@link Message} <em>must</em> be an instance of {@link JobLaunchRequest}.
*
* Launches a Batch Job using the provided request {@link Message}. The payload of the
* {@link Message} <em>must</em> be an instance of {@link JobLaunchRequest}.
* @param requestMessage must not be null.
* @return Generally a {@link JobExecution} will always be returned. An
* exception ({@link MessageHandlingException}) will only be thrown if there
* is a failure to start the job. The cause of the exception will be a
* {@link JobExecutionException}.
*
* @return Generally a {@link JobExecution} will always be returned. An exception
* ({@link MessageHandlingException}) will only be thrown if there is a failure to
* start the job. The cause of the exception will be a {@link JobExecutionException}.
* @throws MessageHandlingException when a job cannot be launched
*/
@Override
@@ -74,7 +69,8 @@ public class JobLaunchingGateway extends AbstractReplyProducingMessageHandler {
try {
jobExecution = this.jobLaunchingMessageHandler.launch(jobLaunchRequest);
} catch (JobExecutionException e) {
}
catch (JobExecutionException e) {
throw new MessageHandlingException(requestMessage, e);
}

7
spring-batch-integration/src/main/java/org/springframework/batch/integration/launch/JobLaunchingMessageHandler.java
View File

@@ -24,7 +24,9 @@ import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.integration.annotation.ServiceActivator;
/**
* Message handler which uses strategies to convert a Message into a job and a set of job parameters
* Message handler which uses strategies to convert a Message into a job and a set of job
* parameters
*
* @author Jonas Partner
* @author Dave Syer
* @author Gunnar Hillert
@@ -35,7 +37,8 @@ public class JobLaunchingMessageHandler implements JobLaunchRequestHandler {
private final JobLauncher jobLauncher;
/**
* @param jobLauncher {@link org.springframework.batch.core.launch.JobLauncher} used to execute Spring Batch jobs
* @param jobLauncher {@link org.springframework.batch.core.launch.JobLauncher} used
* to execute Spring Batch jobs
*/
public JobLaunchingMessageHandler(JobLauncher jobLauncher) {
super();

96
spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/BeanFactoryStepLocator.java
View File

@@ -1,48 +1,48 @@
package org.springframework.batch.integration.partition;
import java.util.Arrays;
import java.util.Collection;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.step.StepLocator;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.util.Assert;
/**
* A {@link StepLocator} implementation that just looks in its enclosing bean
* factory for components of type {@link Step}.
*
* @author Dave Syer
* @author Mahmoud Ben Hassine
*
*/
public class BeanFactoryStepLocator implements StepLocator, BeanFactoryAware {
private BeanFactory beanFactory;
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.beanFactory = beanFactory;
}
/**
* Look up a bean with the provided name of type {@link Step}.
* @see StepLocator#getStep(String)
*/
public Step getStep(String stepName) {
return beanFactory.getBean(stepName, Step.class);
}
/**
* Look in the bean factory for all beans of type {@link Step}.
* @throws IllegalStateException if the {@link BeanFactory} is not listable
* @see StepLocator#getStepNames()
*/
public Collection<String> getStepNames() {
Assert.state(beanFactory instanceof ListableBeanFactory, "BeanFactory is not listable.");
return Arrays.asList(((ListableBeanFactory) beanFactory).getBeanNamesForType(Step.class));
}
}
package org.springframework.batch.integration.partition;
import java.util.Arrays;
import java.util.Collection;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.step.StepLocator;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.util.Assert;
/**
* A {@link StepLocator} implementation that just looks in its enclosing bean factory for
* components of type {@link Step}.
*
* @author Dave Syer
* @author Mahmoud Ben Hassine
*
*/
public class BeanFactoryStepLocator implements StepLocator, BeanFactoryAware {
private BeanFactory beanFactory;
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.beanFactory = beanFactory;
}
/**
* Look up a bean with the provided name of type {@link Step}.
* @see StepLocator#getStep(String)
*/
public Step getStep(String stepName) {
return beanFactory.getBean(stepName, Step.class);
}
/**
* Look in the bean factory for all beans of type {@link Step}.
* @throws IllegalStateException if the {@link BeanFactory} is not listable
* @see StepLocator#getStepNames()
*/
public Collection<String> getStepNames() {
Assert.state(beanFactory instanceof ListableBeanFactory, "BeanFactory is not listable.");
return Arrays.asList(((ListableBeanFactory) beanFactory).getBeanNamesForType(Step.class));
}
}

654
spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/MessageChannelPartitionHandler.java
View File

@@ -1,319 +1,335 @@
/*
* Copyright 2009-2021 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
*
* https://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.partition;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import javax.sql.DataSource;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.explore.JobExplorer;
import org.springframework.batch.core.explore.support.JobExplorerFactoryBean;
import org.springframework.batch.core.partition.PartitionHandler;
import org.springframework.batch.core.partition.StepExecutionSplitter;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.poller.DirectPoller;
import org.springframework.batch.poller.Poller;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.MessageTimeoutException;
import org.springframework.integration.annotation.Aggregator;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.annotation.Payloads;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
/**
* A {@link PartitionHandler} that uses {@link MessageChannel} instances to send instructions to remote workers and
* receive their responses. The {@link MessageChannel} provides a nice abstraction so that the location of the workers
* and the transport used to communicate with them can be changed at run time. The communication with the remote workers
* does not need to be transactional or have guaranteed delivery, so a local thread pool based implementation works as
* well as a remote web service or JMS implementation. If a remote worker fails, the job will fail and can be restarted
* to pick up missing messages and processing. The remote workers need access to the Spring Batch {@link JobRepository}
* so that the shared state across those restarts can be managed centrally.
*
* While a {@link org.springframework.messaging.MessageChannel} is used for sending the requests to the workers, the
* worker's responses can be obtained in one of two ways:
* <ul>
* <li>A reply channel - Workers will respond with messages that will be aggregated via this component.</li>
* <li>Polling the job repository - Since the state of each worker is maintained independently within the job
* repository, we can poll the store to determine the state without the need of the workers to formally respond.</li>
* </ul>
*
* Note: The reply channel for this is instance based. Sharing this component across
* multiple step instances may result in the crossing of messages. It's recommended that
* this component be step or job scoped.
*
* @author Dave Syer
* @author Will Schipp
* @author Michael Minella
* @author Mahmoud Ben Hassine
*
*/
@MessageEndpoint
public class MessageChannelPartitionHandler implements PartitionHandler, InitializingBean {
private static Log logger = LogFactory.getLog(MessageChannelPartitionHandler.class);
private int gridSize = 1;
private MessagingTemplate messagingGateway;
private String stepName;
private long pollInterval = 10000;
private JobExplorer jobExplorer;
private boolean pollRepositoryForResults = false;
private long timeout = -1;
private DataSource dataSource;
/**
* pollable channel for the replies
*/
private PollableChannel replyChannel;
@Override
public void afterPropertiesSet() throws Exception {
Assert.notNull(stepName, "A step name must be provided for the remote workers.");
Assert.state(messagingGateway != null, "The MessagingOperations must be set");
pollRepositoryForResults = !(dataSource == null && jobExplorer == null);
if(pollRepositoryForResults) {
logger.debug("MessageChannelPartitionHandler is configured to poll the job repository for worker results");
}
if(dataSource != null && jobExplorer == null) {
JobExplorerFactoryBean jobExplorerFactoryBean = new JobExplorerFactoryBean();
jobExplorerFactoryBean.setDataSource(dataSource);
jobExplorerFactoryBean.afterPropertiesSet();
jobExplorer = jobExplorerFactoryBean.getObject();
}
if (!pollRepositoryForResults && replyChannel == null) {
replyChannel = new QueueChannel();
}//end if
}
/**
* When using job repository polling, the time limit to wait.
*
* @param timeout milliseconds to wait, defaults to -1 (no timeout).
*/
public void setTimeout(long timeout) {
this.timeout = timeout;
}
/**
* {@link org.springframework.batch.core.explore.JobExplorer} to use to query the job repository. Either this or
* a {@link javax.sql.DataSource} is required when using job repository polling.
*
* @param jobExplorer {@link org.springframework.batch.core.explore.JobExplorer} to use for lookups
*/
public void setJobExplorer(JobExplorer jobExplorer) {
this.jobExplorer = jobExplorer;
}
/**
* How often to poll the job repository for the status of the workers.
*
* @param pollInterval milliseconds between polls, defaults to 10000 (10 seconds).
*/
public void setPollInterval(long pollInterval) {
this.pollInterval = pollInterval;
}
/**
* {@link javax.sql.DataSource} pointing to the job repository
*
* @param dataSource {@link javax.sql.DataSource} that points to the job repository's store
*/
public void setDataSource(DataSource dataSource) {
this.dataSource = dataSource;
}
/**
* A pre-configured gateway for sending and receiving messages to the remote workers. Using this property allows a
* large degree of control over the timeouts and other properties of the send. It should have channels set up
* internally: <ul> <li>request channel capable of accepting {@link StepExecutionRequest} payloads</li> <li>reply
* channel that returns a list of {@link StepExecution} results</li> </ul> The timeout for the reply should be set
* sufficiently long that the remote steps have time to complete.
*
* @param messagingGateway the {@link org.springframework.integration.core.MessagingTemplate} to set
*/
public void setMessagingOperations(MessagingTemplate messagingGateway) {
this.messagingGateway = messagingGateway;
}
/**
* Passed to the {@link StepExecutionSplitter} in the {@link #handle(StepExecutionSplitter, StepExecution)} method,
* instructing it how many {@link StepExecution} instances are required, ideally. The {@link StepExecutionSplitter}
* is allowed to ignore the grid size in the case of a restart, since the input data partitions must be preserved.
*
* @param gridSize the number of step executions that will be created
*/
public void setGridSize(int gridSize) {
this.gridSize = gridSize;
}
/**
* The name of the {@link Step} that will be used to execute the partitioned {@link StepExecution}. This is a
* regular Spring Batch step, with all the business logic required to complete an execution based on the input
* parameters in its {@link StepExecution} context. The name will be translated into a {@link Step} instance by the
* remote worker.
*
* @param stepName the name of the {@link Step} instance to execute business logic
*/
public void setStepName(String stepName) {
this.stepName = stepName;
}
/**
* @param messages the messages to be aggregated
* @return the list as it was passed in
*/
@Aggregator(sendPartialResultsOnExpiry = "true")
public List<?> aggregate(@Payloads List<?> messages) {
return messages;
}
public void setReplyChannel(PollableChannel replyChannel) {
this.replyChannel = replyChannel;
}
/**
* Sends {@link StepExecutionRequest} objects to the request channel of the {@link MessagingTemplate}, and then
* receives the result back as a list of {@link StepExecution} on a reply channel. Use the {@link #aggregate(List)}
* method as an aggregator of the individual remote replies. The receive timeout needs to be set realistically in
* the {@link MessagingTemplate} <b>and</b> the aggregator, so that there is a good chance of all work being done.
*
* @see PartitionHandler#handle(StepExecutionSplitter, StepExecution)
*/
public Collection<StepExecution> handle(StepExecutionSplitter stepExecutionSplitter,
final StepExecution managerStepExecution) throws Exception {
final Set<StepExecution> split = stepExecutionSplitter.split(managerStepExecution, gridSize);
if(CollectionUtils.isEmpty(split)) {
return split;
}
int count = 0;
for (StepExecution stepExecution : split) {
Message<StepExecutionRequest> request = createMessage(count++, split.size(), new StepExecutionRequest(
stepName, stepExecution.getJobExecutionId(), stepExecution.getId()), replyChannel);
if (logger.isDebugEnabled()) {
logger.debug("Sending request: " + request);
}
messagingGateway.send(request);
}
if(!pollRepositoryForResults) {
return receiveReplies(replyChannel);
}
else {
return pollReplies(managerStepExecution, split);
}
}
private Collection<StepExecution> pollReplies(final StepExecution managerStepExecution, final Set<StepExecution> split) throws Exception {
final Collection<StepExecution> result = new ArrayList<>(split.size());
Callable<Collection<StepExecution>> callback = new Callable<Collection<StepExecution>>() {
@Override
public Collection<StepExecution> call() throws Exception {
for(Iterator<StepExecution> stepExecutionIterator = split.iterator(); stepExecutionIterator.hasNext(); ) {
StepExecution curStepExecution = stepExecutionIterator.next();
if(!result.contains(curStepExecution)) {
StepExecution partitionStepExecution =
jobExplorer.getStepExecution(managerStepExecution.getJobExecutionId(), curStepExecution.getId());
if(!partitionStepExecution.getStatus().isRunning()) {
result.add(partitionStepExecution);
}
}
}
if(logger.isDebugEnabled()) {
logger.debug(String.format("Currently waiting on %s partitions to finish", split.size()));
}
if(result.size() == split.size()) {
return result;
}
else {
return null;
}
}
};
Poller<Collection<StepExecution>> poller = new DirectPoller<>(pollInterval);
Future<Collection<StepExecution>> resultsFuture = poller.poll(callback);
if(timeout >= 0) {
return resultsFuture.get(timeout, TimeUnit.MILLISECONDS);
}
else {
return resultsFuture.get();
}
}
private Collection<StepExecution> receiveReplies(PollableChannel currentReplyChannel) {
@SuppressWarnings("unchecked")
Message<Collection<StepExecution>> message = (Message<Collection<StepExecution>>) messagingGateway.receive(currentReplyChannel);
if(message == null) {
throw new MessageTimeoutException("Timeout occurred before all partitions returned");
} else if (logger.isDebugEnabled()) {
logger.debug("Received replies: " + message);
}
return message.getPayload();
}
private Message<StepExecutionRequest> createMessage(int sequenceNumber, int sequenceSize,
StepExecutionRequest stepExecutionRequest, PollableChannel replyChannel) {
return MessageBuilder.withPayload(stepExecutionRequest).setSequenceNumber(sequenceNumber)
.setSequenceSize(sequenceSize)
.setCorrelationId(stepExecutionRequest.getJobExecutionId() + ":" + stepExecutionRequest.getStepName())
.setReplyChannel(replyChannel)
.build();
}
}
/*
* Copyright 2009-2021 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
*
* https://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.partition;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import javax.sql.DataSource;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.explore.JobExplorer;
import org.springframework.batch.core.explore.support.JobExplorerFactoryBean;
import org.springframework.batch.core.partition.PartitionHandler;
import org.springframework.batch.core.partition.StepExecutionSplitter;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.poller.DirectPoller;
import org.springframework.batch.poller.Poller;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.MessageTimeoutException;
import org.springframework.integration.annotation.Aggregator;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.annotation.Payloads;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
/**
* A {@link PartitionHandler} that uses {@link MessageChannel} instances to send
* instructions to remote workers and receive their responses. The {@link MessageChannel}
* provides a nice abstraction so that the location of the workers and the transport used
* to communicate with them can be changed at run time. The communication with the remote
* workers does not need to be transactional or have guaranteed delivery, so a local
* thread pool based implementation works as well as a remote web service or JMS
* implementation. If a remote worker fails, the job will fail and can be restarted to
* pick up missing messages and processing. The remote workers need access to the Spring
* Batch {@link JobRepository} so that the shared state across those restarts can be
* managed centrally.
*
* While a {@link org.springframework.messaging.MessageChannel} is used for sending the
* requests to the workers, the worker's responses can be obtained in one of two ways:
* <ul>
* <li>A reply channel - Workers will respond with messages that will be aggregated via
* this component.</li>
* <li>Polling the job repository - Since the state of each worker is maintained
* independently within the job repository, we can poll the store to determine the state
* without the need of the workers to formally respond.</li>
* </ul>
*
* Note: The reply channel for this is instance based. Sharing this component across
* multiple step instances may result in the crossing of messages. It's recommended that
* this component be step or job scoped.
*
* @author Dave Syer
* @author Will Schipp
* @author Michael Minella
* @author Mahmoud Ben Hassine
*
*/
@MessageEndpoint
public class MessageChannelPartitionHandler implements PartitionHandler, InitializingBean {
private static Log logger = LogFactory.getLog(MessageChannelPartitionHandler.class);
private int gridSize = 1;
private MessagingTemplate messagingGateway;
private String stepName;
private long pollInterval = 10000;
private JobExplorer jobExplorer;
private boolean pollRepositoryForResults = false;
private long timeout = -1;
private DataSource dataSource;
/**
* pollable channel for the replies
*/
private PollableChannel replyChannel;
@Override
public void afterPropertiesSet() throws Exception {
Assert.notNull(stepName, "A step name must be provided for the remote workers.");
Assert.state(messagingGateway != null, "The MessagingOperations must be set");
pollRepositoryForResults = !(dataSource == null && jobExplorer == null);
if (pollRepositoryForResults) {
logger.debug("MessageChannelPartitionHandler is configured to poll the job repository for worker results");
}
if (dataSource != null && jobExplorer == null) {
JobExplorerFactoryBean jobExplorerFactoryBean = new JobExplorerFactoryBean();
jobExplorerFactoryBean.setDataSource(dataSource);
jobExplorerFactoryBean.afterPropertiesSet();
jobExplorer = jobExplorerFactoryBean.getObject();
}
if (!pollRepositoryForResults && replyChannel == null) {
replyChannel = new QueueChannel();
} // end if
}
/**
* When using job repository polling, the time limit to wait.
* @param timeout milliseconds to wait, defaults to -1 (no timeout).
*/
public void setTimeout(long timeout) {
this.timeout = timeout;
}
/**
* {@link org.springframework.batch.core.explore.JobExplorer} to use to query the job
* repository. Either this or a {@link javax.sql.DataSource} is required when using
* job repository polling.
* @param jobExplorer {@link org.springframework.batch.core.explore.JobExplorer} to
* use for lookups
*/
public void setJobExplorer(JobExplorer jobExplorer) {
this.jobExplorer = jobExplorer;
}
/**
* How often to poll the job repository for the status of the workers.
* @param pollInterval milliseconds between polls, defaults to 10000 (10 seconds).
*/
public void setPollInterval(long pollInterval) {
this.pollInterval = pollInterval;
}
/**
* {@link javax.sql.DataSource} pointing to the job repository
* @param dataSource {@link javax.sql.DataSource} that points to the job repository's
* store
*/
public void setDataSource(DataSource dataSource) {
this.dataSource = dataSource;
}
/**
* A pre-configured gateway for sending and receiving messages to the remote workers.
* Using this property allows a large degree of control over the timeouts and other
* properties of the send. It should have channels set up internally:
* <ul>
* <li>request channel capable of accepting {@link StepExecutionRequest} payloads</li>
* <li>reply channel that returns a list of {@link StepExecution} results</li>
* </ul>
* The timeout for the reply should be set sufficiently long that the remote steps
* have time to complete.
* @param messagingGateway the
* {@link org.springframework.integration.core.MessagingTemplate} to set
*/
public void setMessagingOperations(MessagingTemplate messagingGateway) {
this.messagingGateway = messagingGateway;
}
/**
* Passed to the {@link StepExecutionSplitter} in the
* {@link #handle(StepExecutionSplitter, StepExecution)} method, instructing it how
* many {@link StepExecution} instances are required, ideally. The
* {@link StepExecutionSplitter} is allowed to ignore the grid size in the case of a
* restart, since the input data partitions must be preserved.
* @param gridSize the number of step executions that will be created
*/
public void setGridSize(int gridSize) {
this.gridSize = gridSize;
}
/**
* The name of the {@link Step} that will be used to execute the partitioned
* {@link StepExecution}. This is a regular Spring Batch step, with all the business
* logic required to complete an execution based on the input parameters in its
* {@link StepExecution} context. The name will be translated into a {@link Step}
* instance by the remote worker.
* @param stepName the name of the {@link Step} instance to execute business logic
*/
public void setStepName(String stepName) {
this.stepName = stepName;
}
/**
* @param messages the messages to be aggregated
* @return the list as it was passed in
*/
@Aggregator(sendPartialResultsOnExpiry = "true")
public List<?> aggregate(@Payloads List<?> messages) {
return messages;
}
public void setReplyChannel(PollableChannel replyChannel) {
this.replyChannel = replyChannel;
}
/**
* Sends {@link StepExecutionRequest} objects to the request channel of the
* {@link MessagingTemplate}, and then receives the result back as a list of
* {@link StepExecution} on a reply channel. Use the {@link #aggregate(List)} method
* as an aggregator of the individual remote replies. The receive timeout needs to be
* set realistically in the {@link MessagingTemplate} <b>and</b> the aggregator, so
* that there is a good chance of all work being done.
*
* @see PartitionHandler#handle(StepExecutionSplitter, StepExecution)
*/
public Collection<StepExecution> handle(StepExecutionSplitter stepExecutionSplitter,
final StepExecution managerStepExecution) throws Exception {
final Set<StepExecution> split = stepExecutionSplitter.split(managerStepExecution, gridSize);
if (CollectionUtils.isEmpty(split)) {
return split;
}
int count = 0;
for (StepExecution stepExecution : split) {
Message<StepExecutionRequest> request = createMessage(count++, split.size(),
new StepExecutionRequest(stepName, stepExecution.getJobExecutionId(), stepExecution.getId()),
replyChannel);
if (logger.isDebugEnabled()) {
logger.debug("Sending request: " + request);
}
messagingGateway.send(request);
}
if (!pollRepositoryForResults) {
return receiveReplies(replyChannel);
}
else {
return pollReplies(managerStepExecution, split);
}
}
private Collection<StepExecution> pollReplies(final StepExecution managerStepExecution,
final Set<StepExecution> split) throws Exception {
final Collection<StepExecution> result = new ArrayList<>(split.size());
Callable<Collection<StepExecution>> callback = new Callable<Collection<StepExecution>>() {
@Override
public Collection<StepExecution> call() throws Exception {
for (Iterator<StepExecution> stepExecutionIterator = split.iterator(); stepExecutionIterator
.hasNext();) {
StepExecution curStepExecution = stepExecutionIterator.next();
if (!result.contains(curStepExecution)) {
StepExecution partitionStepExecution = jobExplorer
.getStepExecution(managerStepExecution.getJobExecutionId(), curStepExecution.getId());
if (!partitionStepExecution.getStatus().isRunning()) {
result.add(partitionStepExecution);
}
}
}
if (logger.isDebugEnabled()) {
logger.debug(String.format("Currently waiting on %s partitions to finish", split.size()));
}
if (result.size() == split.size()) {
return result;
}
else {
return null;
}
}
};
Poller<Collection<StepExecution>> poller = new DirectPoller<>(pollInterval);
Future<Collection<StepExecution>> resultsFuture = poller.poll(callback);
if (timeout >= 0) {
return resultsFuture.get(timeout, TimeUnit.MILLISECONDS);
}
else {
return resultsFuture.get();
}
}
private Collection<StepExecution> receiveReplies(PollableChannel currentReplyChannel) {
@SuppressWarnings("unchecked")
Message<Collection<StepExecution>> message = (Message<Collection<StepExecution>>) messagingGateway
.receive(currentReplyChannel);
if (message == null) {
throw new MessageTimeoutException("Timeout occurred before all partitions returned");
}
else if (logger.isDebugEnabled()) {
logger.debug("Received replies: " + message);
}
return message.getPayload();
}
private Message<StepExecutionRequest> createMessage(int sequenceNumber, int sequenceSize,
StepExecutionRequest stepExecutionRequest, PollableChannel replyChannel) {
return MessageBuilder.withPayload(stepExecutionRequest).setSequenceNumber(sequenceNumber)
.setSequenceSize(sequenceSize)
.setCorrelationId(stepExecutionRequest.getJobExecutionId() + ":" + stepExecutionRequest.getStepName())
.setReplyChannel(replyChannel).build();
}
}

92
spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/RemotePartitioningManagerStepBuilder.java
View File

@@ -42,14 +42,19 @@ import org.springframework.util.Assert;
* Builder for a manager step in a remote partitioning setup. This builder creates and
* sets a {@link MessageChannelPartitionHandler} on the manager step.
*
* <p>If no {@code messagingTemplate} is provided through
* {@link RemotePartitioningManagerStepBuilder#messagingTemplate(MessagingTemplate)},
* this builder will create one and set its default channel to the {@code outputChannel}
* provided through {@link RemotePartitioningManagerStepBuilder#outputChannel(MessageChannel)}.</p>
* <p>
* If no {@code messagingTemplate} is provided through
* {@link RemotePartitioningManagerStepBuilder#messagingTemplate(MessagingTemplate)}, this
* builder will create one and set its default channel to the {@code outputChannel}
* provided through
* {@link RemotePartitioningManagerStepBuilder#outputChannel(MessageChannel)}.
* </p>
*
* <p>If a {@code messagingTemplate} is provided, it is assumed that it is fully configured
* <p>
* If a {@code messagingTemplate} is provided, it is assumed that it is fully configured
* and that its default channel is set to an output channel on which requests to workers
* will be sent.</p>
* will be sent.
* </p>
*
* @since 4.2
* @author Mahmoud Ben Hassine
@@ -57,14 +62,21 @@ import org.springframework.util.Assert;
public class RemotePartitioningManagerStepBuilder extends PartitionStepBuilder {
private static final long DEFAULT_POLL_INTERVAL = 10000L;
private static final long DEFAULT_TIMEOUT = -1L;
private MessagingTemplate messagingTemplate;
private MessageChannel inputChannel;
private MessageChannel outputChannel;
private JobExplorer jobExplorer;
private BeanFactory beanFactory;
private long pollInterval = DEFAULT_POLL_INTERVAL;
private long timeout = DEFAULT_TIMEOUT;
/**
@@ -87,12 +99,14 @@ public class RemotePartitioningManagerStepBuilder extends PartitionStepBuilder {
}
/**
* Set the output channel on which requests to workers will be sent. By using
* this setter, a default messaging template will be created and the output
* channel will be set as its default channel.
* <p>Use either this setter or {@link RemotePartitioningManagerStepBuilder#messagingTemplate(MessagingTemplate)}
* to provide a fully configured messaging template.</p>
*
* Set the output channel on which requests to workers will be sent. By using this
* setter, a default messaging template will be created and the output channel will be
* set as its default channel.
* <p>
* Use either this setter or
* {@link RemotePartitioningManagerStepBuilder#messagingTemplate(MessagingTemplate)}
* to provide a fully configured messaging template.
* </p>
* @param outputChannel the output channel.
* @return this builder instance for fluent chaining
* @see RemotePartitioningManagerStepBuilder#messagingTemplate(MessagingTemplate)
@@ -104,12 +118,14 @@ public class RemotePartitioningManagerStepBuilder extends PartitionStepBuilder {
}
/**
* Set the {@link MessagingTemplate} to use to send data to workers.
* <strong>The default channel of the messaging template must be set</strong>.
* <p>Use either this setter to provide a fully configured messaging template or
* provide an output channel through {@link RemotePartitioningManagerStepBuilder#outputChannel(MessageChannel)}
* and a default messaging template will be created.</p>
*
* Set the {@link MessagingTemplate} to use to send data to workers. <strong>The
* default channel of the messaging template must be set</strong>.
* <p>
* Use either this setter to provide a fully configured messaging template or provide
* an output channel through
* {@link RemotePartitioningManagerStepBuilder#outputChannel(MessageChannel)} and a
* default messaging template will be created.
* </p>
* @param messagingTemplate the messaging template to use
* @return this builder instance for fluent chaining
* @see RemotePartitioningManagerStepBuilder#outputChannel(MessageChannel)
@@ -132,7 +148,8 @@ public class RemotePartitioningManagerStepBuilder extends PartitionStepBuilder {
}
/**
* How often to poll the job repository for the status of the workers. Defaults to 10 seconds.
* How often to poll the job repository for the status of the workers. Defaults to 10
* seconds.
* @param pollInterval the poll interval value in milliseconds
* @return this builder instance for fluent chaining
*/
@@ -143,7 +160,8 @@ public class RemotePartitioningManagerStepBuilder extends PartitionStepBuilder {
}
/**
* When using job repository polling, the time limit to wait. Defaults to -1 (no timeout).
* When using job repository polling, the time limit to wait. Defaults to -1 (no
* timeout).
* @param timeout the timeout value in milliseconds
* @return this builder instance for fluent chaining
*/
@@ -189,15 +207,10 @@ public class RemotePartitioningManagerStepBuilder extends PartitionStepBuilder {
else {
PollableChannel replies = new QueueChannel();
partitionHandler.setReplyChannel(replies);
StandardIntegrationFlow standardIntegrationFlow = IntegrationFlows
.from(this.inputChannel)
.aggregate(aggregatorSpec -> aggregatorSpec.processor(partitionHandler))
.channel(replies)
.get();
StandardIntegrationFlow standardIntegrationFlow = IntegrationFlows.from(this.inputChannel)
.aggregate(aggregatorSpec -> aggregatorSpec.processor(partitionHandler)).channel(replies).get();
IntegrationFlowContext integrationFlowContext = this.beanFactory.getBean(IntegrationFlowContext.class);
integrationFlowContext.registration(standardIntegrationFlow)
.autoStartup(false)
.register();
integrationFlowContext.registration(standardIntegrationFlow).autoStartup(false).register();
}
try {
@@ -282,24 +295,23 @@ public class RemotePartitioningManagerStepBuilder extends PartitionStepBuilder {
}
/**
* This method will throw a {@link UnsupportedOperationException} since
* the partition handler of the manager step will be automatically set to an
* instance of {@link MessageChannelPartitionHandler}.
*
* When building a manager step for remote partitioning using this builder,
* no partition handler must be provided.
* This method will throw a {@link UnsupportedOperationException} since the partition
* handler of the manager step will be automatically set to an instance of
* {@link MessageChannelPartitionHandler}.
*
* When building a manager step for remote partitioning using this builder, no
* partition handler must be provided.
* @param partitionHandler a partition handler
* @return this builder instance for fluent chaining
* @throws UnsupportedOperationException if a partition handler is provided
*/
@Override
public RemotePartitioningManagerStepBuilder partitionHandler(PartitionHandler partitionHandler) throws UnsupportedOperationException {
throw new UnsupportedOperationException("When configuring a manager step " +
"for remote partitioning using the RemotePartitioningManagerStepBuilder, " +
"the partition handler will be automatically set to an instance " +
"of MessageChannelPartitionHandler. The partition handler must " +
"not be provided in this case.");
public RemotePartitioningManagerStepBuilder partitionHandler(PartitionHandler partitionHandler)
throws UnsupportedOperationException {
throw new UnsupportedOperationException("When configuring a manager step "
+ "for remote partitioning using the RemotePartitioningManagerStepBuilder, "
+ "the partition handler will be automatically set to an instance "
+ "of MessageChannelPartitionHandler. The partition handler must " + "not be provided in this case.");
}
}

22
spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/RemotePartitioningManagerStepBuilderFactory.java
View File

@@ -24,8 +24,8 @@ import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.transaction.PlatformTransactionManager;
/**
* Convenient factory for a {@link RemotePartitioningManagerStepBuilder} which sets
* the {@link JobRepository}, {@link JobExplorer}, {@link BeanFactory} and
* Convenient factory for a {@link RemotePartitioningManagerStepBuilder} which sets the
* {@link JobRepository}, {@link JobExplorer}, {@link BeanFactory} and
* {@link PlatformTransactionManager} automatically.
*
* @since 4.2
@@ -34,10 +34,12 @@ import org.springframework.transaction.PlatformTransactionManager;
public class RemotePartitioningManagerStepBuilderFactory implements BeanFactoryAware {
private BeanFactory beanFactory;
final private JobExplorer jobExplorer;
final private JobRepository jobRepository;
final private PlatformTransactionManager transactionManager;
final private JobExplorer jobExplorer;
final private JobRepository jobRepository;
final private PlatformTransactionManager transactionManager;
/**
* Create a new {@link RemotePartitioningManagerStepBuilderFactory}.
@@ -45,8 +47,8 @@ public class RemotePartitioningManagerStepBuilderFactory implements BeanFactoryA
* @param jobExplorer the job explorer to use
* @param transactionManager the transaction manager to use
*/
public RemotePartitioningManagerStepBuilderFactory(JobRepository jobRepository,
JobExplorer jobExplorer, PlatformTransactionManager transactionManager) {
public RemotePartitioningManagerStepBuilderFactory(JobRepository jobRepository, JobExplorer jobExplorer,
PlatformTransactionManager transactionManager) {
this.jobRepository = jobRepository;
this.jobExplorer = jobExplorer;
@@ -65,10 +67,8 @@ public class RemotePartitioningManagerStepBuilderFactory implements BeanFactoryA
* @return a {@link RemotePartitioningManagerStepBuilder}
*/
public RemotePartitioningManagerStepBuilder get(String name) {
return new RemotePartitioningManagerStepBuilder(name)
.repository(this.jobRepository)
.jobExplorer(this.jobExplorer)
.beanFactory(this.beanFactory)
return new RemotePartitioningManagerStepBuilder(name).repository(this.jobRepository)
.jobExplorer(this.jobExplorer).beanFactory(this.beanFactory)
.transactionManager(this.transactionManager);
}

52
spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/RemotePartitioningWorkerStepBuilder.java
View File

@@ -46,20 +46,20 @@ import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.util.Assert;
/**
* Builder for a worker step in a remote partitioning setup. This builder
* creates an {@link IntegrationFlow} that:
* Builder for a worker step in a remote partitioning setup. This builder creates an
* {@link IntegrationFlow} that:
*
* <ul>
* <li>listens to {@link StepExecutionRequest}s coming from the manager
* on the input channel</li>
* <li>invokes the {@link StepExecutionRequestHandler} to execute the worker
* step for each incoming request. The worker step is located using the provided
* {@link StepLocator}. If no {@link StepLocator} is provided, a {@link BeanFactoryStepLocator}
* configured with the current {@link BeanFactory} will be used
* <li>replies to the manager on the output channel (when the manager step is
* configured to aggregate replies from workers). If no output channel
* is provided, a {@link NullChannel} will be used (assuming the manager side
* is configured to poll the job repository for workers status)</li>
* <li>listens to {@link StepExecutionRequest}s coming from the manager on the input
* channel</li>
* <li>invokes the {@link StepExecutionRequestHandler} to execute the worker step for each
* incoming request. The worker step is located using the provided {@link StepLocator}. If
* no {@link StepLocator} is provided, a {@link BeanFactoryStepLocator} configured with
* the current {@link BeanFactory} will be used
* <li>replies to the manager on the output channel (when the manager step is configured
* to aggregate replies from workers). If no output channel is provided, a
* {@link NullChannel} will be used (assuming the manager side is configured to poll the
* job repository for workers status)</li>
* </ul>
*
* @since 4.1
@@ -68,12 +68,17 @@ import org.springframework.util.Assert;
public class RemotePartitioningWorkerStepBuilder extends StepBuilder {
private static final String SERVICE_ACTIVATOR_METHOD_NAME = "handle";
private static final Log logger = LogFactory.getLog(RemotePartitioningWorkerStepBuilder.class);
private MessageChannel inputChannel;
private MessageChannel outputChannel;
private JobExplorer jobExplorer;
private StepLocator stepLocator;
private BeanFactory beanFactory;
/**
@@ -85,8 +90,8 @@ public class RemotePartitioningWorkerStepBuilder extends StepBuilder {
}
/**
* Set the input channel on which step execution requests sent by the manager
* are received.
* Set the input channel on which step execution requests sent by the manager are
* received.
* @param inputChannel the input channel
* @return this builder instance for fluent chaining
*/
@@ -220,8 +225,8 @@ public class RemotePartitioningWorkerStepBuilder extends StepBuilder {
/**
* Create an {@link IntegrationFlow} with a {@link StepExecutionRequestHandler}
* configured as a service activator listening to the input channel and replying
* on the output channel.
* configured as a service activator listening to the input channel and replying on
* the output channel.
*/
private void configureWorkerIntegrationFlow() {
Assert.notNull(this.inputChannel, "An InputChannel must be provided");
@@ -234,8 +239,8 @@ public class RemotePartitioningWorkerStepBuilder extends StepBuilder {
}
if (this.outputChannel == null) {
if (logger.isDebugEnabled()) {
logger.debug("The output channel is set to a NullChannel. " +
"The manager step must poll the job repository for workers status.");
logger.debug("The output channel is set to a NullChannel. "
+ "The manager step must poll the job repository for workers status.");
}
this.outputChannel = new NullChannel();
}
@@ -244,15 +249,10 @@ public class RemotePartitioningWorkerStepBuilder extends StepBuilder {
stepExecutionRequestHandler.setJobExplorer(this.jobExplorer);
stepExecutionRequestHandler.setStepLocator(this.stepLocator);
StandardIntegrationFlow standardIntegrationFlow = IntegrationFlows
.from(this.inputChannel)
.handle(stepExecutionRequestHandler, SERVICE_ACTIVATOR_METHOD_NAME)
.channel(this.outputChannel)
.get();
StandardIntegrationFlow standardIntegrationFlow = IntegrationFlows.from(this.inputChannel)
.handle(stepExecutionRequestHandler, SERVICE_ACTIVATOR_METHOD_NAME).channel(this.outputChannel).get();
IntegrationFlowContext integrationFlowContext = this.beanFactory.getBean(IntegrationFlowContext.class);
integrationFlowContext.registration(standardIntegrationFlow)
.autoStartup(false)
.register();
integrationFlowContext.registration(standardIntegrationFlow).autoStartup(false).register();
}
}

23
spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/RemotePartitioningWorkerStepBuilderFactory.java
View File

@@ -24,8 +24,8 @@ import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.transaction.PlatformTransactionManager;
/**
* Convenient factory for a {@link RemotePartitioningWorkerStepBuilder} which sets
* the {@link JobRepository}, {@link JobExplorer}, {@link BeanFactory} and
* Convenient factory for a {@link RemotePartitioningWorkerStepBuilder} which sets the
* {@link JobRepository}, {@link JobExplorer}, {@link BeanFactory} and
* {@link PlatformTransactionManager} automatically.
*
* @since 4.1
@@ -34,10 +34,12 @@ import org.springframework.transaction.PlatformTransactionManager;
public class RemotePartitioningWorkerStepBuilderFactory implements BeanFactoryAware {
private BeanFactory beanFactory;
final private JobExplorer jobExplorer;
final private JobRepository jobRepository;
final private PlatformTransactionManager transactionManager;
final private JobExplorer jobExplorer;
final private JobRepository jobRepository;
final private PlatformTransactionManager transactionManager;
/**
* Create a new {@link RemotePartitioningWorkerStepBuilderFactory}.
@@ -45,9 +47,8 @@ public class RemotePartitioningWorkerStepBuilderFactory implements BeanFactoryAw
* @param jobExplorer the job explorer to use
* @param transactionManager the transaction manager to use
*/
public RemotePartitioningWorkerStepBuilderFactory(JobRepository jobRepository,
JobExplorer jobExplorer,
PlatformTransactionManager transactionManager) {
public RemotePartitioningWorkerStepBuilderFactory(JobRepository jobRepository, JobExplorer jobExplorer,
PlatformTransactionManager transactionManager) {
this.jobExplorer = jobExplorer;
this.jobRepository = jobRepository;
@@ -66,10 +67,8 @@ public class RemotePartitioningWorkerStepBuilderFactory implements BeanFactoryAw
* @return a {@link RemotePartitioningWorkerStepBuilder}
*/
public RemotePartitioningWorkerStepBuilder get(String name) {
return new RemotePartitioningWorkerStepBuilder(name)
.repository(this.jobRepository)
.jobExplorer(this.jobExplorer)
.beanFactory(this.beanFactory)
return new RemotePartitioningWorkerStepBuilder(name).repository(this.jobRepository)
.jobExplorer(this.jobExplorer).beanFactory(this.beanFactory)
.transactionManager(this.transactionManager);
}

142
spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/StepExecutionRequest.java
View File

@@ -1,71 +1,71 @@
/*
* Copyright 2009-2018 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
*
* https://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.partition;
import java.io.Serializable;
/**
* Class encapsulating information required to request a step execution in
* a remote partitioning setup.
*
* @author Dave Syer
* @author Mahmoud Ben Hassine
*/
public class StepExecutionRequest implements Serializable {
private static final long serialVersionUID = 1L;
private Long stepExecutionId;
private String stepName;
private Long jobExecutionId;
private StepExecutionRequest() {
//For Jackson deserialization
}
/**
* Create a new {@link StepExecutionRequest} instance.
* @param stepName the name of the step to execute
* @param jobExecutionId the id of the job execution
* @param stepExecutionId the id of the step execution
*/
public StepExecutionRequest(String stepName, Long jobExecutionId, Long stepExecutionId) {
this.stepName = stepName;
this.jobExecutionId = jobExecutionId;
this.stepExecutionId = stepExecutionId;
}
public Long getJobExecutionId() {
return jobExecutionId;
}
public Long getStepExecutionId() {
return stepExecutionId;
}
public String getStepName() {
return stepName;
}
@Override
public String toString() {
return String.format("StepExecutionRequest: [jobExecutionId=%d, stepExecutionId=%d, stepName=%s]",
jobExecutionId, stepExecutionId, stepName);
}
}
/*
* Copyright 2009-2018 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
*
* https://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.partition;
import java.io.Serializable;
/**
* Class encapsulating information required to request a step execution in a remote
* partitioning setup.
*
* @author Dave Syer
* @author Mahmoud Ben Hassine
*/
public class StepExecutionRequest implements Serializable {
private static final long serialVersionUID = 1L;
private Long stepExecutionId;
private String stepName;
private Long jobExecutionId;
private StepExecutionRequest() {
// For Jackson deserialization
}
/**
* Create a new {@link StepExecutionRequest} instance.
* @param stepName the name of the step to execute
* @param jobExecutionId the id of the job execution
* @param stepExecutionId the id of the step execution
*/
public StepExecutionRequest(String stepName, Long jobExecutionId, Long stepExecutionId) {
this.stepName = stepName;
this.jobExecutionId = jobExecutionId;
this.stepExecutionId = stepExecutionId;
}
public Long getJobExecutionId() {
return jobExecutionId;
}
public Long getStepExecutionId() {
return stepExecutionId;
}
public String getStepName() {
return stepName;
}
@Override
public String toString() {
return String.format("StepExecutionRequest: [jobExecutionId=%d, stepExecutionId=%d, stepName=%s]",
jobExecutionId, stepExecutionId, stepName);
}
}

158
spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/StepExecutionRequestHandler.java
View File

@@ -1,80 +1,78 @@
package org.springframework.batch.integration.partition;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.JobInterruptedException;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.explore.JobExplorer;
import org.springframework.batch.core.step.NoSuchStepException;
import org.springframework.batch.core.step.StepLocator;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.annotation.ServiceActivator;
/**
* A {@link MessageEndpoint} that can handle a {@link StepExecutionRequest} and
* return a {@link StepExecution} as the result. Typically these need to be
* aggregated into a response to a partition handler.
*
* @author Dave Syer
*
*/
@MessageEndpoint
public class StepExecutionRequestHandler {
private JobExplorer jobExplorer;
private StepLocator stepLocator;
/**
* Used to locate a {@link Step} to execute for each request.
* @param stepLocator a {@link StepLocator}
*/
public void setStepLocator(StepLocator stepLocator) {
this.stepLocator = stepLocator;
}
/**
* An explorer that should be used to check for {@link StepExecution}
* completion.
*
* @param jobExplorer a {@link JobExplorer} that is linked to the shared
* repository used by all remote workers.
*/
public void setJobExplorer(JobExplorer jobExplorer) {
this.jobExplorer = jobExplorer;
}
@ServiceActivator
public StepExecution handle(StepExecutionRequest request) {
Long jobExecutionId = request.getJobExecutionId();
Long stepExecutionId = request.getStepExecutionId();
StepExecution stepExecution = jobExplorer.getStepExecution(jobExecutionId, stepExecutionId);
if (stepExecution == null) {
throw new NoSuchStepException("No StepExecution could be located for this request: " + request);
}
String stepName = request.getStepName();
Step step = stepLocator.getStep(stepName);
if (step == null) {
throw new NoSuchStepException(String.format("No Step with name [%s] could be located.", stepName));
}
try {
step.execute(stepExecution);
}
catch (JobInterruptedException e) {
stepExecution.setStatus(BatchStatus.STOPPED);
// The receiver should update the stepExecution in repository
}
catch (Throwable e) {
stepExecution.addFailureException(e);
stepExecution.setStatus(BatchStatus.FAILED);
// The receiver should update the stepExecution in repository
}
return stepExecution;
}
}
package org.springframework.batch.integration.partition;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.JobInterruptedException;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.explore.JobExplorer;
import org.springframework.batch.core.step.NoSuchStepException;
import org.springframework.batch.core.step.StepLocator;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.annotation.ServiceActivator;
/**
* A {@link MessageEndpoint} that can handle a {@link StepExecutionRequest} and return a
* {@link StepExecution} as the result. Typically these need to be aggregated into a
* response to a partition handler.
*
* @author Dave Syer
*
*/
@MessageEndpoint
public class StepExecutionRequestHandler {
private JobExplorer jobExplorer;
private StepLocator stepLocator;
/**
* Used to locate a {@link Step} to execute for each request.
* @param stepLocator a {@link StepLocator}
*/
public void setStepLocator(StepLocator stepLocator) {
this.stepLocator = stepLocator;
}
/**
* An explorer that should be used to check for {@link StepExecution} completion.
* @param jobExplorer a {@link JobExplorer} that is linked to the shared repository
* used by all remote workers.
*/
public void setJobExplorer(JobExplorer jobExplorer) {
this.jobExplorer = jobExplorer;
}
@ServiceActivator
public StepExecution handle(StepExecutionRequest request) {
Long jobExecutionId = request.getJobExecutionId();
Long stepExecutionId = request.getStepExecutionId();
StepExecution stepExecution = jobExplorer.getStepExecution(jobExecutionId, stepExecutionId);
if (stepExecution == null) {
throw new NoSuchStepException("No StepExecution could be located for this request: " + request);
}
String stepName = request.getStepName();
Step step = stepLocator.getStep(stepName);
if (step == null) {
throw new NoSuchStepException(String.format("No Step with name [%s] could be located.", stepName));
}
try {
step.execute(stepExecution);
}
catch (JobInterruptedException e) {
stepExecution.setStatus(BatchStatus.STOPPED);
// The receiver should update the stepExecution in repository
}
catch (Throwable e) {
stepExecution.addFailureException(e);
stepExecution.setStatus(BatchStatus.FAILED);
// The receiver should update the stepExecution in repository
}
return stepExecution;
}
}

12
spring-batch-integration/src/main/java/org/springframework/batch/integration/step/DelegateStep.java
View File

@@ -22,11 +22,11 @@ import org.springframework.batch.core.step.AbstractStep;
import org.springframework.util.Assert;
/**
* Provides a wrapper for an existing {@link Step}, delegating execution to it,
* but serving all other operations locally.
*
* Provides a wrapper for an existing {@link Step}, delegating execution to it, but
* serving all other operations locally.
*
* @author Dave Syer
*
*
*/
public class DelegateStep extends AbstractStep {
@@ -38,13 +38,13 @@ public class DelegateStep extends AbstractStep {
public void setDelegate(Step delegate) {
this.delegate = delegate;
}
/**
* Check mandatory properties (delegate).
*/
@Override
public void afterPropertiesSet() throws Exception {
Assert.state(delegate!=null, "A delegate Step must be provided");
Assert.state(delegate != null, "A delegate Step must be provided");
super.afterPropertiesSet();
}