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

RESOLVED - issue BATCH-920, BATCH-974: ChunkContext in StepContextRepeatCallback is unused

Browse Source
This commit is contained in:
dsyer
2008-12-30 13:40:18 +00:00
parent e4ca63a927
commit 7d7480578d
17 changed files with 169 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
spring-batch-core/pom.xml
View File

@@ -123,6 +123,16 @@
<!-- optional for Java 6 -->
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.apache.log4j</groupId>
<artifactId>com.springsource.org.apache.log4j</artifactId>
<optional>false</optional>
</dependency>
<dependency>
<groupId>org.apache.log4j</groupId>
<artifactId>com.springsource.org.apache.log4j</artifactId>
<optional>true</optional>
</dependency>
</dependencies>
<build>

44
spring-batch-core/src/main/java/org/springframework/batch/core/scope/context/ChunkContext.java
View File

@@ -1,10 +1,52 @@
package org.springframework.batch.core.scope.context;
import java.util.Arrays;
import org.springframework.core.AttributeAccessorSupport;
/**
* @author Dave Syer
*
*
*/
public class ChunkContext extends AttributeAccessorSupport {
private final StepContext stepContext;
private boolean complete = false;
/**
* @param stepContext the current step context
*/
public ChunkContext(StepContext stepContext) {
this.stepContext = stepContext;
}
/**
* @return the current step context
*/
public StepContext getStepContext() {
return stepContext;
}
/**
* @return true if there is no more processing to be done on this chunk
*/
public boolean isComplete() {
return complete;
}
/**
* Setter for the flag to signal complete processing of a chunk.
*/
public void setComplete() {
this.complete = true;
}
/* (non-Javadoc)
* @see java.lang.Object#toString()
*/
@Override
public String toString() {
return String.format("ChunkContext: attributes=%s, complete=%b, stepContext=%s", Arrays.asList(attributeNames()), complete, stepContext);
}
}

19
spring-batch-core/src/main/java/org/springframework/batch/core/scope/context/StepContextRepeatCallback.java
View File

@@ -60,17 +60,16 @@ public abstract class StepContextRepeatCallback implements RepeatCallback {
ChunkContext chunkContext = attributeQueue.poll();
if (chunkContext == null) {
chunkContext = new ChunkContext();
chunkContext = new ChunkContext(stepContext);
}
int start = chunkContext.attributeNames().length;
try {
return doInStepContext(context, stepContext);
return doInChunkContext(context, chunkContext);
}
finally {
// Still some stuff to do with the data in this chunk,
// pass it back
if (chunkContext.attributeNames().length > start) {
// pass it back.
if (!chunkContext.isComplete()) {
attributeQueue.add(chunkContext);
}
StepSynchronizationManager.close();
@@ -78,21 +77,21 @@ public abstract class StepContextRepeatCallback implements RepeatCallback {
}
/**
* Do the work required for this portion of the step. The
* {@link StepContext} provided is managed by the base class, so that if
* Do the work required for this chunk of the step. The
* {@link ChunkContext} provided is managed by the base class, so that if
* there is still work to do for the task in hand state can be stored here.
* In a multi-threaded client, the base class ensures that only one thread
* at a time can be working on each instance of {@link StepContext}. Workers
* at a time can be working on each instance of {@link ChunkContext}. Workers
* should signal that they are finished with a context by removing all the
* attributes they have added. If a worker does not remove them another
* thread might see stale state.
*
* @param context the current {@link RepeatContext}
* @param stepContext the step context in which to carry out the work
* @param chunkContext the chunk context in which to carry out the work
* @return the repeat status from the execution
* @throws Exception implementations can throw an exception if anything goes
* wrong
*/
public abstract RepeatStatus doInStepContext(RepeatContext context, StepContext stepContext) throws Exception;
public abstract RepeatStatus doInChunkContext(RepeatContext context, ChunkContext chunkContext) throws Exception;
}

11
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkOrientedTasklet.java
View File

@@ -2,9 +2,9 @@ package org.springframework.batch.core.step.item;
import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.core.AttributeAccessor;
/**
* A {@link Tasklet} implementing variations on read-process-write item
@@ -41,20 +41,21 @@ public class ChunkOrientedTasklet<I> implements Tasklet {
this.buffering = buffering;
}
public RepeatStatus execute(StepContribution contribution, AttributeAccessor attributes) throws Exception {
public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
@SuppressWarnings("unchecked")
Chunk<I> inputs = (Chunk<I>) attributes.getAttribute(INPUTS_KEY);
Chunk<I> inputs = (Chunk<I>) chunkContext.getAttribute(INPUTS_KEY);
if (inputs == null) {
inputs = chunkProvider.provide(contribution);
if (buffering) {
attributes.setAttribute(INPUTS_KEY, inputs);
chunkContext.setAttribute(INPUTS_KEY, inputs);
}
}
chunkProcessor.process(contribution, inputs);
attributes.removeAttribute(INPUTS_KEY);
chunkContext.removeAttribute(INPUTS_KEY);
chunkContext.setComplete();
chunkProvider.postProcess(contribution, inputs);
if (!inputs.isEnd()) {
contribution.setExitStatus(ExitStatus.FINISHED);

6
spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/CallableTaskletAdapter.java
View File

@@ -19,9 +19,9 @@ import java.util.concurrent.Callable;
import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.AttributeAccessor;
import org.springframework.util.Assert;
/**
@@ -55,9 +55,9 @@ public class CallableTaskletAdapter implements Tasklet, InitializingBean {
/**
* Execute the provided Callable and return its {@link ExitStatus}. Ignores
* the {@link StepContribution} and the attributes.
* @see Tasklet#execute(StepContribution, AttributeAccessor)
* @see Tasklet#execute(StepContribution, ChunkContext)
*/
public RepeatStatus execute(StepContribution contribution, AttributeAccessor attributes) throws Exception {
public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
return callable.call();
}

6
spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/MethodInvokingTaskletAdapter.java
View File

@@ -17,9 +17,9 @@ package org.springframework.batch.core.step.tasklet;
import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.item.adapter.AbstractMethodInvokingDelegator;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.core.AttributeAccessor;
/**
* A {@link Tasklet} that wraps a method in a POJO. By default the return
@@ -40,9 +40,9 @@ public class MethodInvokingTaskletAdapter extends AbstractMethodInvokingDelegato
* an {@link ExitStatus} by invoking a method in the delegate POJO. Ignores
* the {@link StepContribution} and the attributes.
*
* @see Tasklet#execute(StepContribution, AttributeAccessor)
* @see Tasklet#execute(StepContribution, ChunkContext)
*/
public RepeatStatus execute(StepContribution contribution, AttributeAccessor attributes) throws Exception {
public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
contribution.setExitStatus(mapResult(invokeDelegateMethod()));
return RepeatStatus.FINISHED;
}

4
spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/SystemCommandTasklet.java
View File

@@ -12,9 +12,9 @@ import org.springframework.batch.core.JobInterruptedException;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.listener.StepExecutionListenerSupport;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.AttributeAccessor;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
import org.springframework.util.Assert;
@@ -66,7 +66,7 @@ public class SystemCommandTasklet extends StepExecutionListenerSupport implement
* Execute system command and map its exit code to {@link ExitStatus} using
* {@link SystemProcessExitCodeMapper}.
*/
public RepeatStatus execute(StepContribution contribution, AttributeAccessor attributes) throws Exception {
public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
FutureTask<Integer> systemCommandTask = new FutureTask<Integer>(new Callable<Integer>() {

6
spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/Tasklet.java
View File

@@ -16,8 +16,8 @@
package org.springframework.batch.core.step.tasklet;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.core.AttributeAccessor;
/**
* Strategy for processing in a step.
@@ -35,11 +35,11 @@ public interface Tasklet {
*
* @param contribution mutable state to be passed back to update the current
* step execution
* @param attributes attributes shared between invocations but not between
* @param chunkContext attributes shared between invocations but not between
* restarts
* @return an {@link RepeatStatus} indicating whether processing is
* continuable.
*/
RepeatStatus execute(StepContribution contribution, AttributeAccessor attributes) throws Exception;
RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception;
}

8
spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java
View File

@@ -25,7 +25,7 @@ import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.StepExecutionListener;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.scope.context.StepContext;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.scope.context.StepContextRepeatCallback;
import org.springframework.batch.core.step.AbstractStep;
import org.springframework.batch.core.step.StepInterruptionPolicy;
@@ -219,9 +219,9 @@ public class TaskletStep extends AbstractStep {
stepOperations.iterate(new StepContextRepeatCallback(stepExecution) {
@Override
public RepeatStatus doInStepContext(RepeatContext repeatContext, StepContext stepContext) throws Exception {
public RepeatStatus doInChunkContext(RepeatContext repeatContext, ChunkContext chunkContext) throws Exception {
StepExecution stepExecution = stepContext.getStepExecution();
StepExecution stepExecution = chunkContext.getStepContext().getStepExecution();
StepContribution contribution = stepExecution.createStepContribution();
@@ -238,7 +238,7 @@ public class TaskletStep extends AbstractStep {
try {
try {
result = tasklet.execute(contribution, stepContext);
result = tasklet.execute(contribution, chunkContext);
}
finally {
// Apply the contribution to the step

4
spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/TestTasklet.java
View File

@@ -2,14 +2,14 @@ package org.springframework.batch.core.configuration.xml;
import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.core.AttributeAccessor;
public class TestTasklet extends AbstractTestComponent implements Tasklet {
public RepeatStatus execute(StepContribution contribution,
AttributeAccessor attributes) throws Exception {
ChunkContext chunkContext) throws Exception {
executed = true;
contribution.setExitStatus(ExitStatus.FINISHED);
return RepeatStatus.FINISHED;

63
spring-batch-core/src/test/java/org/springframework/batch/core/scope/context/ChunkContextTests.java Normal file
View File

@@ -0,0 +1,63 @@
/*
* Copyright 2006-2007 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.batch.core.scope.context;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import java.util.Collections;
import org.junit.Test;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobInstance;
import org.springframework.batch.core.JobParameter;
import org.springframework.batch.core.JobParameters;
/**
* @author Dave Syer
*
*/
public class ChunkContextTests {
private ChunkContext context = new ChunkContext(new StepContext(new JobExecution(new JobInstance(0L,
new JobParameters(Collections.singletonMap("foo", new JobParameter("bar"))), "job"), 1L)
.createStepExecution("foo")));
@Test
public void testGetStepContext() {
StepContext stepContext = context.getStepContext();
assertNotNull(stepContext);
assertEquals("bar", context.getStepContext().getJobParameters().get("foo"));
}
@Test
public void testIsComplete() {
assertFalse(context.isComplete());
context.setComplete();
assertTrue(context.isComplete());
}
@Test
public void testToString() {
String value = context.toString();
assertTrue("Wrong toString: "+value, value.contains("stepContext="));
assertTrue("Wrong toString: "+value, value.contains("complete=false"));
assertTrue("Wrong toString: "+value, value.contains("attributes=[]"));
}
}

13
spring-batch-core/src/test/java/org/springframework/batch/core/scope/context/StepContextRepeatCallbackTests.java
View File

@@ -24,7 +24,6 @@ import org.junit.Test;
import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.scope.context.StepContext;
import org.springframework.batch.core.scope.context.StepContextRepeatCallback;
import org.springframework.batch.core.scope.context.StepSynchronizationManager;
import org.springframework.batch.repeat.RepeatContext;
@@ -49,8 +48,8 @@ public class StepContextRepeatCallbackTests {
public void testDoInIteration() throws Exception {
StepContextRepeatCallback callback = new StepContextRepeatCallback(stepExecution) {
@Override
public RepeatStatus doInStepContext(RepeatContext context, StepContext stepContext) throws Exception {
assertEquals(Long.valueOf(123), stepContext.getStepExecution().getId());
public RepeatStatus doInChunkContext(RepeatContext context, ChunkContext chunkContext) throws Exception {
assertEquals(Long.valueOf(123), chunkContext.getStepContext().getStepExecution().getId());
return RepeatStatus.FINISHED;
}
};
@@ -63,13 +62,13 @@ public class StepContextRepeatCallbackTests {
StepSynchronizationManager.register(stepExecution);
StepContextRepeatCallback callback = new StepContextRepeatCallback(stepExecution) {
@Override
public RepeatStatus doInStepContext(RepeatContext context, StepContext stepContext) throws Exception {
public RepeatStatus doInChunkContext(RepeatContext context, ChunkContext chunkContext) throws Exception {
if (addedAttribute) {
removedAttribute = stepContext.hasAttribute("foo");
stepContext.removeAttribute("foo");
removedAttribute = chunkContext.hasAttribute("foo");
chunkContext.removeAttribute("foo");
} else {
addedAttribute = true;
stepContext.setAttribute("foo", "bar");
chunkContext.setAttribute("foo", "bar");
}
return RepeatStatus.FINISHED;
}

5
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ChunkOrientedTaskletTests.java
View File

@@ -24,8 +24,7 @@ import org.springframework.batch.core.JobInstance;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.repeat.context.RepeatContextSupport;
import org.springframework.core.AttributeAccessor;
import org.springframework.batch.core.scope.context.ChunkContext;
/**
* @author Dave Syer
@@ -33,7 +32,7 @@ import org.springframework.core.AttributeAccessor;
*/
public class ChunkOrientedTaskletTests {
private AttributeAccessor context = new RepeatContextSupport(null);
private ChunkContext context = new ChunkContext(null);
@Test
public void testHandle() throws Exception {

13
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/TaskletStepExceptionTests.java
View File

@@ -26,6 +26,7 @@ import org.springframework.batch.core.repository.JobExecutionAlreadyRunningExcep
import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.JobRestartException;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.core.step.tasklet.TaskletStep;
import org.springframework.batch.item.ExecutionContext;
@@ -34,7 +35,6 @@ import org.springframework.batch.item.ItemStreamException;
import org.springframework.batch.item.ItemStreamSupport;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
import org.springframework.core.AttributeAccessor;
import org.springframework.transaction.TransactionException;
import org.springframework.transaction.support.DefaultTransactionStatus;
@@ -127,7 +127,7 @@ public class TaskletStepExceptionTests {
} });
taskletStep.setTasklet(new Tasklet() {
public RepeatStatus execute(StepContribution contribution, AttributeAccessor attributes) throws Exception {
public RepeatStatus execute(StepContribution contribution, ChunkContext attributes) throws Exception {
return RepeatStatus.FINISHED;
}
@@ -139,7 +139,7 @@ public class TaskletStepExceptionTests {
}
@Test
/**
/*
* Exception in afterStep is ignored (only logged).
*/
public void testAfterStepFAilure() throws Exception {
@@ -189,7 +189,7 @@ public class TaskletStepExceptionTests {
taskletStep.setTasklet(new Tasklet() {
public RepeatStatus execute(StepContribution contribution, AttributeAccessor attributes) throws Exception {
public RepeatStatus execute(StepContribution contribution, ChunkContext attributes) throws Exception {
return RepeatStatus.FINISHED;
}
@@ -207,6 +207,7 @@ public class TaskletStepExceptionTests {
final RuntimeException exception = new RuntimeException();
taskletStep.setJobRepository(new UpdateCountingJobRepository() {
boolean firstCall = true;
@Override
public void update(StepExecution arg0) {
if (firstCall) {
@@ -216,7 +217,7 @@ public class TaskletStepExceptionTests {
throw exception;
}
});
taskletStep.execute(stepExecution);
assertEquals(UNKNOWN, stepExecution.getStatus());
assertTrue(stepExecution.getFailureExceptions().contains(taskletException));
@@ -225,7 +226,7 @@ public class TaskletStepExceptionTests {
private static class ExceptionTasklet implements Tasklet {
public RepeatStatus execute(StepContribution contribution, AttributeAccessor attributes) throws Exception {
public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
throw taskletException;
}
}

4
spring-batch-integration/src/test/java/org/springframework/batch/integration/job/TestTasklet.java
View File

@@ -17,9 +17,9 @@ package org.springframework.batch.integration.job;
import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.core.AttributeAccessor;
/**
* @author Dave Syer
@@ -31,7 +31,7 @@ public class TestTasklet implements Tasklet {
* (non-Javadoc)
*
*/
public RepeatStatus execute(StepContribution contribution, AttributeAccessor attributes) throws Exception {
public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
contribution.setExitStatus(ExitStatus.FINISHED);
return RepeatStatus.FINISHED;
}

4
spring-batch-samples/src/test/java/org/springframework/batch/sample/common/ErrorLogTasklet.java
View File

@@ -8,9 +8,9 @@ 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.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.core.AttributeAccessor;
import org.springframework.jdbc.core.simple.SimpleJdbcTemplate;
public class ErrorLogTasklet implements Tasklet, StepExecutionListener {
@@ -21,7 +21,7 @@ public class ErrorLogTasklet implements Tasklet, StepExecutionListener {
private String jobName;
private String stepName;
public RepeatStatus execute(StepContribution contribution, AttributeAccessor attributes) throws Exception {
public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
this.simpleJdbcTemplate.update("insert into ERROR_LOG values ('"+jobName+"', '"+stepName+"', 'Some records were skipped!')");
return RepeatStatus.FINISHED;
}

4
spring-batch-test/src/test/java/org/springframework/batch/test/sample/SampleTasklet.java
View File

@@ -1,10 +1,10 @@
package org.springframework.batch.test.sample;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.AttributeAccessor;
import org.springframework.jdbc.core.simple.SimpleJdbcTemplate;
public class SampleTasklet implements Tasklet {
@@ -18,7 +18,7 @@ public class SampleTasklet implements Tasklet {
this.id = id;
}
public RepeatStatus execute(StepContribution contribution, AttributeAccessor attributes) throws Exception {
public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
this.jdbcTemplate.update("insert into TESTS(ID, NAME) values (?, 'SampleTasklet" + id + "')", id);
return RepeatStatus.FINISHED;
}