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

BATCH-1572: try to copy more of step execution for rollback

Browse Source
This commit is contained in:
dsyer
2010-09-01 08:05:09 +00:00
parent 9d62761d24
commit 2ac38f6781
6 changed files with 76 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-batch-core/src/main/java/org/springframework/batch/core/Entity.java
View File

@@ -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);
}
/**

23
spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/MapStepExecutionDao.java
View File

@@ -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<Long, Map<Long, StepExecution>> executionsByJobExecutionId = new ConcurrentHashMap<Long, Map<Long, StepExecution>>();
private Map<Long, Map<Long, StepExecution>> executionsByJobExecutionId = TransactionAwareProxyFactory
.createAppendOnlyTransactionalMap();
private Map<Long, StepExecution> executionsByStepExecutionId = new ConcurrentHashMap<Long, StepExecution>();
private Map<Long, StepExecution> 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<Long, StepExecution> executions = executionsByJobExecutionId.get(stepExecution.getJobExecutionId());
if (executions == null) {
executions = new ConcurrentHashMap<Long, StepExecution>();
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);
}
}

34
spring-batch-core/src/main/java/org/springframework/batch/core/step/FatalStepExecutionException.java Normal file
View File

@@ -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);
}
}

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

@@ -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<T, S> extends SimpleStepFactoryBean<T,
@Override
protected void applyConfiguration(TaskletStep step) {
addNonSkippableExceptionIfMissing(SkipLimitExceededException.class, NonSkippableReadException.class,
SkipListenerFailedException.class, SkipPolicyFailedException.class, RetryException.class, JobInterruptedException.class, Error.class);
addNonRetryableExceptionIfMissing(SkipLimitExceededException.class, NonSkippableReadException.class, TransactionException.class,
SkipListenerFailedException.class, SkipPolicyFailedException.class, RetryException.class, JobInterruptedException.class, Error.class);
SkipListenerFailedException.class, SkipPolicyFailedException.class, RetryException.class,
JobInterruptedException.class, Error.class);
addNonRetryableExceptionIfMissing(SkipLimitExceededException.class, NonSkippableReadException.class,
TransactionException.class, FatalStepExecutionException.class, SkipListenerFailedException.class,
SkipPolicyFailedException.class, RetryException.class, JobInterruptedException.class, Error.class);
super.applyConfiguration(step);
}
@@ -373,7 +376,7 @@ public class FaultTolerantStepFactoryBean<T, S> extends SimpleStepFactoryBean<T,
if (!skippableExceptionClasses.isEmpty()) {
logger.info("Skippable exceptions will be ignored because a SkipPolicy was specified explicitly");
}
if (skipLimit>0) {
if (skipLimit > 0) {
logger.info("Skip limit will be ignored because a SkipPolicy was specified explicitly");
}
}

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

@@ -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());
}
}

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

@@ -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());
}