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

BATCH-2122: Updated so that offset would be correctly handled at chunk borders

Browse Source
This commit is contained in:
Michael Minella
2013-11-18 16:26:39 -06:00
parent 0a0d62f4d4
commit cb4c08eee9
3 changed files with 13 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkMonitor.java
View File

@@ -23,7 +23,6 @@ import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.ItemStreamException;
import org.springframework.batch.item.ItemStreamSupport;
import org.springframework.batch.item.support.CompositeItemStream;
import org.springframework.util.ClassUtils;
/**
* Manage the offset data between the last successful commit and updates made to
@@ -59,10 +58,10 @@ public class ChunkMonitor extends ItemStreamSupport {
private ItemReader<?> reader;
public ChunkMonitor() {
this.setExecutionContextName(ChunkMonitor.class.getName());
}
public ChunkMonitor() {
this.setExecutionContextName(ChunkMonitor.class.getName());
}
/**
* @param stream the stream to set
*/
@@ -101,7 +100,7 @@ public class ChunkMonitor extends ItemStreamSupport {
@Override
public void close() throws ItemStreamException {
super.close();
super.close();
holder.set(null);
if (streamsRegistered) {
stream.close();
@@ -110,7 +109,7 @@ public class ChunkMonitor extends ItemStreamSupport {
@Override
public void open(ExecutionContext executionContext) throws ItemStreamException {
super.open(executionContext);
super.open(executionContext);
if (streamsRegistered) {
stream.open(executionContext);
ChunkMonitorData data = new ChunkMonitorData(executionContext.getInt(getExecutionContextKey(OFFSET), 0), 0);
@@ -127,12 +126,14 @@ public class ChunkMonitor extends ItemStreamSupport {
throw new ItemStreamException("Could not position reader with offset: " + data.offset, e);
}
}
resetOffset();
}
}
@Override
public void update(ExecutionContext executionContext) throws ItemStreamException {
super.update(executionContext);
super.update(executionContext);
if (streamsRegistered) {
ChunkMonitorData data = getData();
if (data.offset == 0) {

4
spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java
View File

@@ -256,7 +256,7 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
}
if (output == null) {
// No need to re-process filtered items
iterator.remove();
iterator.remove();
}
return output;
}
@@ -313,7 +313,6 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
@Override
protected void write(final StepContribution contribution, final Chunk<I> inputs, final Chunk<O> outputs)
throws Exception {
@SuppressWarnings("unchecked")
final UserData<O> data = (UserData<O>) inputs.getUserData();
final AtomicReference<RetryContext> contextHolder = new AtomicReference<RetryContext>();
@@ -568,6 +567,7 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
if (outputs.isEmpty()) {
data.scanning(false);
inputs.setBusy(false);
chunkMonitor.resetOffset();
return;
}

3
spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ChunkMonitorTests.java
View File

@@ -55,7 +55,7 @@ public class ChunkMonitorTests {
monitor.registerItemStream(new ItemStreamSupport() {
@Override
public void close() {
super.close();
super.close();
closed = true;
}
});
@@ -98,6 +98,7 @@ public class ChunkMonitorTests {
executionContext.putInt(ChunkMonitor.class.getName() + ".OFFSET", 2);
monitor.open(executionContext);
assertEquals(2, count);
assertEquals(0, monitor.getOffset());
}
@Test