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

RESOLVED - BATCH-651: buffered readers don't handle volatile commit interval

Browse Source 
added test and patched the affected readers
This commit is contained in:
robokaso
2008-06-05 08:54:39 +00:00
parent e194b8da09
commit 23b5f47d82
4 changed files with 56 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/MultiResourceItemReader.java
View File

@@ -1,8 +1,8 @@
package org.springframework.batch.item;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.ListIterator;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.io.Resource;
@@ -33,11 +33,13 @@ public class MultiResourceItemReader extends ExecutionContextUserSupport impleme
private List itemBuffer = new ArrayList();
private Iterator itemBufferIterator = null;
private ListIterator itemBufferIterator = null;
private boolean shouldReadBuffer = false;
private boolean saveState = false;
private int lastMarkedBufferIndex = 0;
public MultiResourceItemReader() {
setName(MultiResourceItemReader.class.getSimpleName());
@@ -84,9 +86,15 @@ public class MultiResourceItemReader extends ExecutionContextUserSupport impleme
* @see ItemReader#mark()
*/
public void mark() throws MarkFailedException {
if (!shouldReadBuffer) {
itemBuffer.clear();
itemBufferIterator = null;
lastMarkedBufferIndex = 0;
}
else {
lastMarkedBufferIndex = itemBufferIterator.nextIndex();
}
delegate.mark();
itemBuffer.clear();
shouldReadBuffer = false;
}
/**
@@ -96,7 +104,7 @@ public class MultiResourceItemReader extends ExecutionContextUserSupport impleme
*/
public void reset() throws ResetFailedException {
shouldReadBuffer = true;
itemBufferIterator = itemBuffer.listIterator();
itemBufferIterator = itemBuffer.listIterator(lastMarkedBufferIndex);
}
/**

19
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/JdbcCursorItemReader.java
View File

@@ -512,6 +512,8 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements
private int INITIAL_POSITION = -1;
private int lastMarkedIndex;
public BufferredResultSetReader(ResultSet rs, RowMapper rowMapper, int processedRowCount) {
Assert.notNull(rs, "The ResultSet must not be null");
Assert.notNull(rowMapper, "The RowMapper must not be null");
@@ -519,6 +521,7 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements
this.rowMapper = rowMapper;
buffer = new ArrayList();
currentIndex = INITIAL_POSITION;
lastMarkedIndex = INITIAL_POSITION;
this.processedRowCount = processedRowCount;
}
@@ -536,7 +539,7 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements
if (!rs.next()) {
return null;
}
int currentRow = processedRowCount + 1;// rs.getRow();
int currentRow = processedRowCount + 1;
buffer.add(rowMapper.mapRow(rs, currentRow));
verifyCursorPosition(currentRow);
}
@@ -550,13 +553,19 @@ public class JdbcCursorItemReader extends ExecutionContextUserSupport implements
}
public void mark() throws MarkFailedException {
buffer.clear();
currentIndex = INITIAL_POSITION;
if (currentIndex == buffer.size()) {
buffer.clear();
currentIndex = INITIAL_POSITION;
lastMarkedIndex = INITIAL_POSITION;
}
else {
lastMarkedIndex = currentIndex;
}
}
public void reset() throws ResetFailedException {
processedRowCount -= buffer.size();
currentIndex = INITIAL_POSITION;
processedRowCount -= currentIndex - lastMarkedIndex;
currentIndex = lastMarkedIndex;
}
/**

41
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/xml/StaxEventItemReader.java
View File

@@ -3,8 +3,8 @@ package org.springframework.batch.item.xml;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.ListIterator;
import javax.xml.namespace.QName;
import javax.xml.stream.XMLEventReader;
@@ -61,9 +61,11 @@ public class StaxEventItemReader extends ExecutionContextUserSupport implements
private boolean saveState = false;
private List buffer = new ArrayList();
private List itemBuffer = new ArrayList();
private Iterator bufferIterator = null;
private ListIterator itemBufferIterator = null;
private int lastMarkedBufferIndex = 0;
/**
* indicates the reader has been shouldReadBuffer and should read items from
@@ -90,13 +92,13 @@ public class StaxEventItemReader extends ExecutionContextUserSupport implements
// read from buffer after rollback
if (shouldReadBuffer) {
if (bufferIterator.hasNext()) {
return bufferIterator.next();
if (itemBufferIterator.hasNext()) {
return itemBufferIterator.next();
}
else {
// buffer is exhausted, continue reading from file
shouldReadBuffer = false;
bufferIterator = null;
itemBufferIterator = null;
}
}
@@ -108,7 +110,7 @@ public class StaxEventItemReader extends ExecutionContextUserSupport implements
fragmentReader.markFragmentProcessed();
}
buffer.add(item);
itemBuffer.add(item);
if (item == null) {
currentRecordCount--;
}
@@ -118,7 +120,8 @@ public class StaxEventItemReader extends ExecutionContextUserSupport implements
public void close(ExecutionContext executionContext) {
initialized = false;
currentRecordCount = 0;
clearBuffer();
itemBuffer.clear();
itemBufferIterator = null;
try {
if (fragmentReader != null) {
fragmentReader.close();
@@ -262,9 +265,17 @@ public class StaxEventItemReader extends ExecutionContextUserSupport implements
* @see org.springframework.batch.item.AbstractItemReader#mark()
*/
public void mark() {
if (!shouldReadBuffer) {
itemBuffer.clear();
itemBufferIterator = null;
lastMarkedBufferIndex = 0;
}
else {
lastMarkedBufferIndex = itemBufferIterator.nextIndex();
}
lastCommitPointRecordCount = currentRecordCount;
clearBuffer();
shouldReadBuffer = false;
}
/*
@@ -275,7 +286,7 @@ public class StaxEventItemReader extends ExecutionContextUserSupport implements
public void reset() {
currentRecordCount = lastCommitPointRecordCount;
shouldReadBuffer = true;
bufferIterator = buffer.listIterator();
itemBufferIterator = itemBuffer.listIterator(lastMarkedBufferIndex);
fragmentReader.reset();
}
@@ -290,11 +301,5 @@ public class StaxEventItemReader extends ExecutionContextUserSupport implements
this.saveState = saveState;
}
/**
* Clear the buffer and release the iterator.
*/
private void clearBuffer() {
buffer.clear();
bufferIterator = null;
}
}

12
spring-batch-infrastructure/src/test/java/org/springframework/batch/item/CommonItemReaderTests.java
View File

@@ -64,12 +64,12 @@ public abstract class CommonItemReaderTests extends TestCase {
assertEquals(foo2, tested.read());
// TODO handle shortening the commit interval on the fly
//
// tested.mark();
//
// assertEquals(foo3, tested.read());
//
// tested.reset();
tested.mark();
assertEquals(foo3, tested.read());
tested.reset();
assertEquals(foo3, tested.read());