diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/Entity.java b/spring-batch-core/src/main/java/org/springframework/batch/core/Entity.java index 0d4d35767..85232185d 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/Entity.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/Entity.java @@ -85,7 +85,7 @@ public class Entity implements Serializable { @Override public String toString() { - return ClassUtils.getShortName(getClass()) + ": id=" + getId(); + return String.format("%s: id=%d, version=%d", ClassUtils.getShortName(getClass()), id, version); } /** diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/MapStepExecutionDao.java b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/MapStepExecutionDao.java index 1a12a7b30..e160114d2 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/MapStepExecutionDao.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/MapStepExecutionDao.java @@ -21,13 +21,13 @@ import java.util.Collections; import java.util.Comparator; import java.util.List; import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicLong; import org.springframework.batch.core.Entity; import org.springframework.batch.core.JobExecution; import org.springframework.batch.core.StepExecution; import org.springframework.batch.support.SerializationUtils; +import org.springframework.batch.support.transaction.TransactionAwareProxyFactory; import org.springframework.dao.OptimisticLockingFailureException; import org.springframework.util.Assert; import org.springframework.util.ReflectionUtils; @@ -37,9 +37,11 @@ import org.springframework.util.ReflectionUtils; */ public class MapStepExecutionDao implements StepExecutionDao { - private Map> executionsByJobExecutionId = new ConcurrentHashMap>(); + private Map> executionsByJobExecutionId = TransactionAwareProxyFactory + .createAppendOnlyTransactionalMap(); - private Map executionsByStepExecutionId = new ConcurrentHashMap(); + private Map executionsByStepExecutionId = TransactionAwareProxyFactory + .createAppendOnlyTransactionalMap(); private AtomicLong currentId = new AtomicLong(); @@ -51,15 +53,16 @@ public class MapStepExecutionDao implements StepExecutionDao { private static StepExecution copy(StepExecution original) { return (StepExecution) SerializationUtils.deserialize(SerializationUtils.serialize(original)); } - + private static void copy(final StepExecution sourceExecution, final StepExecution targetExecution) { - // Cheaper than full serialization is a reflective field copy, which is fine for volatile storage + // Cheaper than full serialization is a reflective field copy, which is + // fine for volatile storage ReflectionUtils.doWithFields(StepExecution.class, new ReflectionUtils.FieldCallback() { public void doWith(Field field) throws IllegalArgumentException, IllegalAccessException { field.setAccessible(true); field.set(targetExecution, field.get(sourceExecution)); } - }); + }); } public void saveStepExecution(StepExecution stepExecution) { @@ -70,7 +73,7 @@ public class MapStepExecutionDao implements StepExecutionDao { Map executions = executionsByJobExecutionId.get(stepExecution.getJobExecutionId()); if (executions == null) { - executions = new ConcurrentHashMap(); + executions = TransactionAwareProxyFactory.createAppendOnlyTransactionalMap(); executionsByJobExecutionId.put(stepExecution.getJobExecutionId(), executions); } @@ -100,8 +103,10 @@ public class MapStepExecutionDao implements StepExecutionDao { } stepExecution.incrementVersion(); - copy(stepExecution, persistedExecution); - executions.put(stepExecution.getId(), persistedExecution); + StepExecution copy = new StepExecution(stepExecution.getStepName(), stepExecution.getJobExecution()); + copy(stepExecution, copy); + executions.put(stepExecution.getId(), copy); + executionsByStepExecutionId.put(stepExecution.getId(), copy); } } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/FatalStepExecutionException.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/FatalStepExecutionException.java new file mode 100644 index 000000000..4aac34109 --- /dev/null +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/FatalStepExecutionException.java @@ -0,0 +1,34 @@ +/* + * Copyright 2006-2010 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.batch.core.step; + +import org.springframework.batch.core.UnexpectedJobExecutionException; + +/** + * @author Dave Syer + * + */ +public class FatalStepExecutionException extends UnexpectedJobExecutionException { + + /** + * @param msg the message to send to the caller + * @param nested the cause of this exception + */ + public FatalStepExecutionException(String msg, Throwable nested) { + super(msg, nested); + } + +} diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBean.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBean.java index 698084c7c..4b0c54bd8 100755 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBean.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantStepFactoryBean.java @@ -28,6 +28,7 @@ import org.springframework.batch.classify.Classifier; import org.springframework.batch.classify.SubclassClassifier; import org.springframework.batch.core.JobInterruptedException; import org.springframework.batch.core.Step; +import org.springframework.batch.core.step.FatalStepExecutionException; import org.springframework.batch.core.step.skip.ExceptionClassifierSkipPolicy; import org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy; import org.springframework.batch.core.step.skip.NeverSkipItemSkipPolicy; @@ -323,9 +324,11 @@ public class FaultTolerantStepFactoryBean extends SimpleStepFactoryBean extends SimpleStepFactoryBean0) { + if (skipLimit > 0) { logger.info("Skip limit will be ignored because a SkipPolicy was specified explicitly"); } } 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 20eb52849..cd3bf823d 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 @@ -15,6 +15,7 @@ */ package org.springframework.batch.core.step.tasklet; +import java.lang.reflect.Field; import java.util.concurrent.Semaphore; import org.apache.commons.logging.Log; @@ -30,6 +31,7 @@ import org.springframework.batch.core.repository.JobRepository; 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.FatalStepExecutionException; import org.springframework.batch.core.step.StepInterruptionPolicy; import org.springframework.batch.core.step.ThreadStepInterruptionPolicy; import org.springframework.batch.item.ExecutionContext; @@ -51,6 +53,7 @@ import org.springframework.transaction.support.TransactionSynchronizationAdapter import org.springframework.transaction.support.TransactionSynchronizationManager; import org.springframework.transaction.support.TransactionTemplate; import org.springframework.util.Assert; +import org.springframework.util.ReflectionUtils; /** * Simple implementation of executing the step as a call to a {@link Tasklet}, @@ -305,7 +308,7 @@ public class TaskletStep extends AbstractStep { private boolean rolledBack = false; - private Integer oldVersion; + private StepExecution oldVersion; private boolean locked = false; @@ -321,7 +324,8 @@ public class TaskletStep extends AbstractStep { if (oldVersion != null) { // Wah! the commit failed. We need to rescue the step // execution data. - stepExecution.setVersion(oldVersion); + copy(oldVersion, stepExecution); + stepExecution.incrementRollbackCount(); } } if (status == TransactionSynchronization.STATUS_UNKNOWN) { @@ -386,7 +390,8 @@ public class TaskletStep extends AbstractStep { // In case we need to push it back to its old value // after a commit fails... - oldVersion = stepExecution.getVersion(); + oldVersion = new StepExecution(stepExecution.getStepName(), stepExecution.getJobExecution()); + copy(stepExecution, oldVersion); // Apply the contribution to the step // even if unsuccessful @@ -408,10 +413,11 @@ public class TaskletStep extends AbstractStep { catch (Exception e) { // If we get to here there was a problem saving the step // execution and we have to fail. - logger.error("JobRepository failure forcing exit with unknown status", e); + String msg = "JobRepository failure forcing exit with unknown status"; + logger.error(msg, e); stepExecution.upgradeStatus(BatchStatus.UNKNOWN); stepExecution.setTerminateOnly(); - throw e; + throw new FatalStepExecutionException(msg, e); } } @@ -419,7 +425,6 @@ public class TaskletStep extends AbstractStep { logger.debug("Rollback for Error: " + e.getClass().getName() + ": " + e.getMessage()); rollback(stepExecution); throw e; - } catch (RuntimeException e) { logger.debug("Rollback for RuntimeException: " + e.getClass().getName() + ": " + e.getMessage()); @@ -443,6 +448,14 @@ public class TaskletStep extends AbstractStep { rolledBack = true; } } + + private void copy(final StepExecution source, final StepExecution target) { + target.setVersion(source.getVersion()); + target.setWriteCount(source.getWriteCount()); + target.setFilterCount(source.getFilterCount()); + target.setRollbackCount(source.getRollbackCount()); + } + } 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 61849125d..2c41e2b9c 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 @@ -255,7 +255,7 @@ public class TaskletStepExceptionTests { taskletStep.execute(stepExecution); assertEquals(UNKNOWN, stepExecution.getStatus()); Throwable e = stepExecution.getFailureExceptions().get(0); - assertEquals("Expected exception in step execution persistence", e.getMessage()); + assertEquals("JobRepository failure forcing exit with unknown status", e.getMessage()); }