diff --git a/spring-batch-core/pom.xml b/spring-batch-core/pom.xml index 2172783df..e87696d4d 100644 --- a/spring-batch-core/pom.xml +++ b/spring-batch-core/pom.xml @@ -49,6 +49,12 @@ commons-lang true + + commons-dbcp + commons-dbcp + true + test + org.easymock easymock diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java index 11ef2c26c..1fa0522e8 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/tasklet/TaskletStep.java @@ -261,12 +261,15 @@ public class TaskletStep extends AbstractStep { boolean locked = false; + Integer oldVersion = null; + boolean committed = true; + try { try { try { result = tasklet.execute(contribution, chunkContext); - if(result == null) { + if (result == null) { result = RepeatStatus.FINISHED; } } @@ -275,9 +278,29 @@ public class TaskletStep extends AbstractStep { throw e; } } + chunkListener.afterChunk(); + } finally { + + // If the step operations are asynchronous then we need + // to synchronize changes to the step execution (at a + // minimum). Take the lock *before* changing the step + // execution. + try { + semaphore.acquire(); + locked = true; + } + catch (InterruptedException e) { + stepExecution.setStatus(BatchStatus.STOPPED); + Thread.currentThread().interrupt(); + } + + // In case we need to push it back to its old value + // after a commit fails... + oldVersion = stepExecution.getVersion(); + // Apply the contribution to the step // even if unsuccessful logger.debug("Applying contribution: " + contribution); @@ -285,26 +308,26 @@ public class TaskletStep extends AbstractStep { } - // If the step operations are asynchronous then we need - // to synchronize changes to the step execution (at a - // minimum). - try { - semaphore.acquire(); - locked = true; - } - catch (InterruptedException e) { - stepExecution.setStatus(BatchStatus.STOPPED); - Thread.currentThread().interrupt(); - } - stream.update(stepExecution.getExecutionContext()); try { + // Going to attempt a commit. If it fails this flag will + // stay false and we can use that later. + committed = false; getJobRepository().updateExecutionContext(stepExecution); - transactionManager.commit(transaction); stepExecution.incrementCommitCount(); - logger.debug("Saving step execution after commit: " + stepExecution); + /* + * The step execution has to be saved before commit + * because otherwise there is a deadlock between the + * data source pool and the semaphore. As long as only + * one connection is used inside the section of this + * callback that is locked with the semaphore, the + * deadlock is avoided. + */ + logger.debug("Saving step execution before commit: " + stepExecution); getJobRepository().update(stepExecution); + transactionManager.commit(transaction); + committed = true; } catch (Exception e) { throw new FatalException("Fatal failure detected", e); @@ -352,11 +375,19 @@ public class TaskletStep extends AbstractStep { throw e; } finally { + if (!committed && oldVersion != null) { + // Wah! the commit failed. We need to rescue the step + // execution data. + stepExecution.setVersion(oldVersion); + } // only release the lock if we acquired it if (locked) { semaphore.release(); } locked = false; + if (!committed) { + getJobRepository().update(stepExecution); + } } // Check for interruption after transaction as well, so that diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/TaskletStepExceptionTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/TaskletStepExceptionTests.java index 496b259b8..43e33ac31 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/TaskletStepExceptionTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/TaskletStepExceptionTests.java @@ -235,8 +235,8 @@ public class TaskletStepExceptionTests { taskletStep.execute(stepExecution); assertEquals(UNKNOWN, stepExecution.getStatus()); - assertTrue(stepExecution.getFailureExceptions().contains(taskletException)); assertTrue(stepExecution.getFailureExceptions().contains(exception)); + assertTrue(stepExecution.getFailureExceptions().contains(taskletException)); } private static class ExceptionTasklet implements Tasklet { diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/AsyncChunkOrientedStepIntegrationTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/AsyncChunkOrientedStepIntegrationTests.java new file mode 100644 index 000000000..125531867 --- /dev/null +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/AsyncChunkOrientedStepIntegrationTests.java @@ -0,0 +1,128 @@ +/* + * 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.step.tasklet; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import org.apache.commons.dbcp.BasicDataSource; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.batch.core.BatchStatus; +import org.springframework.batch.core.Job; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobParameter; +import org.springframework.batch.core.JobParameters; +import org.springframework.batch.core.StepExecution; +import org.springframework.batch.core.job.JobSupport; +import org.springframework.batch.core.repository.JobRepository; +import org.springframework.batch.item.ItemReader; +import org.springframework.batch.item.ItemWriter; +import org.springframework.batch.item.support.ListItemReader; +import org.springframework.batch.repeat.policy.SimpleCompletionPolicy; +import org.springframework.batch.repeat.support.RepeatTemplate; +import org.springframework.batch.repeat.support.TaskExecutorRepeatTemplate; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.core.task.SimpleAsyncTaskExecutor; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.transaction.PlatformTransactionManager; + +/** + * @author Dave Syer + * + */ +@RunWith(SpringJUnit4ClassRunner.class) +@ContextConfiguration(locations = "/org/springframework/batch/core/repository/dao/sql-dao-test.xml") +public class AsyncChunkOrientedStepIntegrationTests { + + private TaskletStep step; + + private Job job; + + private List written = new ArrayList(); + + @Autowired + private PlatformTransactionManager transactionManager; + + @Autowired + private BasicDataSource dataSource; + + @Autowired + private JobRepository jobRepository; + + private RepeatTemplate chunkOperations; + + private ItemReader getReader(String[] args) { + return new ListItemReader(Arrays.asList(args)); + } + + @Before + public void onSetUp() throws Exception { + + // Force deadlock with batch waiting for DB pool and vice versa + dataSource.setMaxActive(1); + dataSource.setMaxIdle(1); + + step = new TaskletStep("stepName"); + step.setJobRepository(jobRepository); + step.setTransactionManager(transactionManager); + + // Only process one item: + chunkOperations = new RepeatTemplate(); + chunkOperations.setCompletionPolicy(new SimpleCompletionPolicy(1)); + + job = new JobSupport("FOO"); + + TaskExecutorRepeatTemplate repeatTemplate = new TaskExecutorRepeatTemplate(); + repeatTemplate.setThrottleLimit(2); + repeatTemplate.setTaskExecutor(new SimpleAsyncTaskExecutor()); + step.setStepOperations(repeatTemplate); + step.setTransactionManager(transactionManager); + + } + + @Test + public void testStatus() throws Exception { + + step.setTasklet(new TestingChunkOrientedTasklet(getReader(new String[] { "a", "b", "c", "a", "b", "c", + "a", "b", "c", "a", "b", "c" }), new ItemWriter() { + public void write(List data) throws Exception { + written.addAll(data); + } + }, chunkOperations)); + + JobExecution jobExecution = jobRepository.createJobExecution(job.getName(), new JobParameters(Collections + .singletonMap("run.id", new JobParameter(getClass().getName() + ".1")))); + StepExecution stepExecution = new StepExecution(step.getName(), jobExecution); + + jobRepository.add(stepExecution); + step.execute(stepExecution); + assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); + StepExecution lastStepExecution = jobRepository.getLastStepExecution(jobExecution.getJobInstance(), step + .getName()); + assertEquals(lastStepExecution, stepExecution); + assertFalse(lastStepExecution == stepExecution); + + } + +} diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/ChunkOrientedStepIntegrationTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/ChunkOrientedStepIntegrationTests.java index 30fbddeaa..95d64a9c1 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/ChunkOrientedStepIntegrationTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/ChunkOrientedStepIntegrationTests.java @@ -20,24 +20,20 @@ import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; import java.util.Arrays; +import java.util.Collections; import java.util.List; -import javax.sql.DataSource; - import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.batch.core.BatchStatus; import org.springframework.batch.core.Job; import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobParameter; import org.springframework.batch.core.JobParameters; import org.springframework.batch.core.StepExecution; import org.springframework.batch.core.job.JobSupport; import org.springframework.batch.core.repository.JobRepository; -import org.springframework.batch.core.repository.dao.MapJobExecutionDao; -import org.springframework.batch.core.repository.dao.MapJobInstanceDao; -import org.springframework.batch.core.repository.dao.MapStepExecutionDao; -import org.springframework.batch.core.repository.support.JobRepositoryFactoryBean; import org.springframework.batch.item.ExecutionContext; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemWriter; @@ -67,8 +63,6 @@ public class ChunkOrientedStepIntegrationTests { private PlatformTransactionManager transactionManager; @Autowired - private DataSource dataSource; - private JobRepository jobRepository; private RepeatTemplate chunkOperations; @@ -79,23 +73,10 @@ public class ChunkOrientedStepIntegrationTests { @Before public void onSetUp() throws Exception { - MapJobInstanceDao.clear(); - MapStepExecutionDao.clear(); - MapJobExecutionDao.clear(); - - JobRepositoryFactoryBean jobRepositoryFactoryBean = new JobRepositoryFactoryBean(); - jobRepositoryFactoryBean.setDatabaseType("hsql"); - jobRepositoryFactoryBean.setDataSource(dataSource); - jobRepositoryFactoryBean.setTransactionManager(transactionManager); - jobRepositoryFactoryBean.afterPropertiesSet(); - jobRepository = (JobRepository) jobRepositoryFactoryBean.getObject(); step = new TaskletStep("stepName"); step.setJobRepository(jobRepository); step.setTransactionManager(transactionManager); - RepeatTemplate template = new RepeatTemplate(); - template.setCompletionPolicy(new SimpleCompletionPolicy(1)); - step.setStepOperations(template); // Only process one item: chunkOperations = new RepeatTemplate(); @@ -122,7 +103,8 @@ public class ChunkOrientedStepIntegrationTests { } }, chunkOperations)); - JobExecution jobExecution = jobRepository.createJobExecution(job.getName(), new JobParameters()); + JobExecution jobExecution = jobRepository.createJobExecution(job.getName(), new JobParameters(Collections + .singletonMap("run.id", new JobParameter(getClass().getName() + ".1")))); StepExecution stepExecution = new StepExecution(step.getName(), jobExecution); stepExecution.setExecutionContext(new ExecutionContext() { @@ -130,11 +112,12 @@ public class ChunkOrientedStepIntegrationTests { put("foo", "bar"); } }); - + jobRepository.add(stepExecution); step.execute(stepExecution); assertEquals(BatchStatus.UNKNOWN, stepExecution.getStatus()); - StepExecution lastStepExecution = jobRepository.getLastStepExecution(jobExecution.getJobInstance(), step.getName()); + StepExecution lastStepExecution = jobRepository.getLastStepExecution(jobExecution.getJobInstance(), step + .getName()); assertEquals(lastStepExecution, stepExecution); assertFalse(lastStepExecution == stepExecution); diff --git a/spring-batch-core/src/test/resources/org/springframework/batch/core/repository/dao/data-source-context.xml b/spring-batch-core/src/test/resources/org/springframework/batch/core/repository/dao/data-source-context.xml index 0499aa2c2..ca1a9b254 100644 --- a/spring-batch-core/src/test/resources/org/springframework/batch/core/repository/dao/data-source-context.xml +++ b/spring-batch-core/src/test/resources/org/springframework/batch/core/repository/dao/data-source-context.xml @@ -10,9 +10,11 @@ - + + +