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

Refactored to copy checkpoint object before persisting to prevent side effects when batch artifacts reuse checkpoint objects

Browse Source
This commit is contained in:
Michael Minella
2013-09-18 17:44:41 -05:00
parent eef0577979
commit c98ed8dfe7
5 changed files with 396 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files

140
spring-batch-infrastructure/src/main/java/org/springframework/batch/jsr/item/CheckpointSupport.java Normal file
View File

@@ -0,0 +1,140 @@
/*
* Copyright 2013 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.jsr.item;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import javax.batch.api.chunk.ItemReader;
import javax.batch.api.chunk.ItemWriter;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemStreamException;
import org.springframework.batch.item.ItemStreamSupport;
import org.springframework.util.Assert;
/**
* Provides support for JSR-352 checkpointing. Checkpoint objects are copied prior
* to being added to the {@link ExecutionContext} for persistence by the framework.
* If the checkpoint object cannot be copied and further changes occur to the same
* instance, side effects may occur. In cases like this, it is recommended that a
* copy of the object being acted upon in the reader/writer is returned via the
* {@link ItemReader#checkpointInfo()} or {@link ItemWriter#checkpointInfo()} calls.
*
* @author Michael Minella
* @since 3.0
*/
public abstract class CheckpointSupport extends ItemStreamSupport{
private final Log logger = LogFactory.getLog(this.getClass());
private final String checkpointKey;
/**
* @param checkpointKey key to store the checkpoint object with in the {@link ExecutionContext}
*/
public CheckpointSupport(String checkpointKey) {
Assert.hasText(checkpointKey);
this.checkpointKey = checkpointKey;
}
/* (non-Javadoc)
* @see org.springframework.batch.item.ItemStreamSupport#open(org.springframework.batch.item.ExecutionContext)
*/
@Override
public void open(ExecutionContext executionContext)
throws ItemStreamException {
try {
doOpen((Serializable) executionContext.get(getExecutionContextKey(checkpointKey)));
} catch (Exception e) {
throw new ItemStreamException(e);
}
}
/**
* Used to open a batch artifact with previously saved checkpoint information.
*
* @param checkpoint previously saved checkpoint object
* @throws Exception
*/
protected abstract void doOpen(Serializable checkpoint) throws Exception;
/* (non-Javadoc)
* @see org.springframework.batch.item.ItemStreamSupport#update(org.springframework.batch.item.ExecutionContext)
*/
@Override
public void update(ExecutionContext executionContext)
throws ItemStreamException {
try {
executionContext.put(getExecutionContextKey(checkpointKey), deepCopy(doCheckpoint()));
} catch (Exception e) {
throw new ItemStreamException(e);
}
}
/**
* Used to provide a {@link Serializable} representing the current state of the
* batch artifact.
*
* @return the current state of the batch artifact
* @throws Exception
*/
protected abstract Serializable doCheckpoint() throws Exception;
/* (non-Javadoc)
* @see org.springframework.batch.item.ItemStreamSupport#close()
*/
@Override
public void close() throws ItemStreamException {
try {
doClose();
} catch (Exception e) {
throw new ItemStreamException(e);
}
}
/**
* Used to close the underlying batch artifact
*
* @throws Exception
*/
protected abstract void doClose() throws Exception;
private Object deepCopy(Serializable orig) {
Object obj = orig;
try {
ByteArrayOutputStream bos = new ByteArrayOutputStream();
ObjectOutputStream out = new ObjectOutputStream(bos);
out.writeObject(orig);
out.flush();
out.close();
ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(bos.toByteArray()));
obj = in.readObject();
} catch (Exception e) {
logger.warn("Unable to copy checkpoint object. Updating the instance passed may cause side effects");
}
return obj;
}
}

88
spring-batch-infrastructure/src/main/java/org/springframework/batch/jsr/item/ItemReaderAdapter.java
View File

@@ -1,64 +1,84 @@
/*
* Copyright 2013 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.jsr.item;
import java.io.Serializable;
import javax.batch.api.chunk.ItemReader;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemStreamException;
import org.springframework.batch.item.ItemStreamSupport;
import org.springframework.batch.item.NonTransientResourceException;
import org.springframework.batch.item.ParseException;
import org.springframework.batch.item.UnexpectedInputException;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
/**
* Adapter that wraps an {@link ItemReader} for use by Spring Batch. All calls are delegated as appropriate
* to the corresponding method on the delegate.
*
* @author Michael Minella
* @since 3.0
*/
@SuppressWarnings("rawtypes")
public class ItemReaderAdapter extends ItemStreamSupport implements org.springframework.batch.item.ItemReader {
public class ItemReaderAdapter extends CheckpointSupport implements org.springframework.batch.item.ItemReader {
private static final String CHECKPOINT_KEY = "reader.checkpoint";
private ItemReader delegate;
/**
* @param reader the {@link ItemReader} implementation to delegate to
*/
public ItemReaderAdapter(ItemReader reader) {
super(CHECKPOINT_KEY);
Assert.notNull(reader, "An ItemReader implementation is required");
this.delegate = reader;
setExecutionContextName(ClassUtils.getShortName(delegate.getClass()));
}
@Override
public void open(ExecutionContext executionContext)
throws ItemStreamException {
try {
delegate.open((Serializable) executionContext.get(getExecutionContextKey(CHECKPOINT_KEY)));
} catch (Exception e) {
throw new ItemStreamException(e);
}
}
@Override
public void update(ExecutionContext executionContext)
throws ItemStreamException {
try {
Serializable checkpoint = delegate.checkpointInfo();
executionContext.put(getExecutionContextKey(CHECKPOINT_KEY), checkpoint);
} catch (Exception e) {
throw new ItemStreamException(e);
}
}
@Override
public void close() throws ItemStreamException {
try {
delegate.close();
} catch (Exception e) {
throw new ItemStreamException(e);
}
}
/* (non-Javadoc)
* @see org.springframework.batch.item.ItemReader#read()
*/
@Override
public Object read() throws Exception, UnexpectedInputException, ParseException,
NonTransientResourceException {
return delegate.readItem();
}
/* (non-Javadoc)
* @see org.springframework.batch.jsr.item.CheckpointSupport#doClose()
*/
@Override
protected void doClose() throws Exception{
delegate.close();
}
/* (non-Javadoc)
* @see org.springframework.batch.jsr.item.CheckpointSupport#doCheckpoint()
*/
@Override
protected Serializable doCheckpoint() throws Exception {
return delegate.checkpointInfo();
}
/* (non-Javadoc)
* @see org.springframework.batch.jsr.item.CheckpointSupport#doOpen(java.io.Serializable)
*/
@Override
protected void doOpen(Serializable checkpoint) throws Exception {
delegate.open(checkpoint);
}
}

89
spring-batch-infrastructure/src/main/java/org/springframework/batch/jsr/item/ItemWriterAdapter.java
View File

@@ -1,3 +1,18 @@
/*
* Copyright 2013 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.jsr.item;
import java.io.Serializable;
@@ -5,58 +20,64 @@ import java.util.List;
import javax.batch.api.chunk.ItemWriter;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemStreamException;
import org.springframework.batch.item.ItemStreamSupport;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
/**
* Adapter that wraps an {@link ItemWriter} for use by Spring Batch. All calls are delegated as appropriate
* to the corresponding method on the delegate.
*
* @author Michael Minella
* @since 3.0
*/
@SuppressWarnings("rawtypes")
public class ItemWriterAdapter extends ItemStreamSupport implements org.springframework.batch.item.ItemWriter {
public class ItemWriterAdapter extends CheckpointSupport implements org.springframework.batch.item.ItemWriter {
private static final String CHECKPOINT_KEY = "writer.checkpoint";
private ItemWriter delegate;
/**
* @param writer a {@link ItemWriter} to delegate calls to
*/
public ItemWriterAdapter(ItemWriter writer) {
super(CHECKPOINT_KEY);
Assert.notNull(writer, "An ItemWriter implementation is required");
this.delegate = writer;
super.setExecutionContextName(ClassUtils.getShortName(delegate.getClass()));
}
@Override
public void open(ExecutionContext executionContext)
throws ItemStreamException {
try {
delegate.open((Serializable) executionContext.get(getExecutionContextKey(CHECKPOINT_KEY)));
} catch (Exception e) {
throw new ItemStreamException(e);
}
}
@Override
public void update(ExecutionContext executionContext)
throws ItemStreamException {
try {
Serializable checkpoint = delegate.checkpointInfo();
executionContext.put(getExecutionContextKey(CHECKPOINT_KEY), checkpoint);
} catch (Exception e) {
throw new ItemStreamException(e);
}
}
@Override
public void close() throws ItemStreamException {
try {
delegate.close();
} catch (Exception e) {
throw new ItemStreamException(e);
}
}
/* (non-Javadoc)
* @see org.springframework.batch.item.ItemWriter#write(java.util.List)
*/
@Override
@SuppressWarnings("unchecked")
public void write(List items) throws Exception {
delegate.writeItems(items);
}
/* (non-Javadoc)
* @see org.springframework.batch.jsr.item.CheckpointSupport#doOpen(java.io.Serializable)
*/
@Override
protected void doOpen(Serializable checkpoint) throws Exception {
delegate.open(checkpoint);
}
/* (non-Javadoc)
* @see org.springframework.batch.jsr.item.CheckpointSupport#doCheckpoint()
*/
@Override
protected Serializable doCheckpoint() throws Exception {
Serializable checkpointInfo = delegate.checkpointInfo();
return checkpointInfo;
}
/* (non-Javadoc)
* @see org.springframework.batch.jsr.item.CheckpointSupport#doClose()
*/
@Override
protected void doClose() throws Exception{
delegate.close();
}
}