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

Fix incorrect behaviour of RepositoryItemReader#jumpToItem() on restart

Browse Source 
* Corrects unit tests with expected behavior.
* Creates integration tests highlighting Hibernate lazy-loading failure if the reader has requested the page on open().

Issue #1074
This commit is contained in:
Antoine Kapps
2016-12-20 00:04:29 +01:00
committed by Mahmoud Ben Hassine
parent 613f425336
commit 95f11facf4
10 changed files with 529 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
spring-batch-infrastructure/src/main/java/org/springframework/batch/item/data/RepositoryItemReader.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2012-2019 the original author or authors.
* Copyright 2012-2020 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.
@@ -71,6 +71,7 @@ import org.springframework.util.MethodInvoker;
* </p>
*
* @author Michael Minella
* @author Antoine Kapps
* @since 2.2
*/
public class RepositoryItemReader<T> extends AbstractItemCountingItemStreamItemReader<T> implements InitializingBean {
@@ -91,7 +92,7 @@ public class RepositoryItemReader<T> extends AbstractItemCountingItemStreamItemR
private volatile List<T> results;
private Object lock = new Object();
private final Object lock = new Object();
private String methodName;
@@ -156,20 +157,24 @@ public class RepositoryItemReader<T> extends AbstractItemCountingItemStreamItemR
protected T doRead() throws Exception {
synchronized (lock) {
if(results == null || current >= results.size()) {
boolean nextPageNeeded = (results != null && current >= results.size());
if (results == null || nextPageNeeded) {
if (logger.isDebugEnabled()) {
logger.debug("Reading page " + page);
}
results = doPageRead();
current = 0;
page ++;
if(results.size() <= 0) {
return null;
}
if (nextPageNeeded) {
current = 0;
}
}
if(current < results.size()) {
@@ -186,11 +191,8 @@ public class RepositoryItemReader<T> extends AbstractItemCountingItemStreamItemR
@Override
protected void jumpToItem(int itemLastIndex) throws Exception {
synchronized (lock) {
page = (itemLastIndex - 1) / pageSize;
current = (itemLastIndex - 1) % pageSize;
results = doPageRead();
page++;
page = itemLastIndex / pageSize;
current = itemLastIndex % pageSize;
}
}

175
spring-batch-infrastructure/src/test/java/org/springframework/batch/item/data/RepositoryItemReaderTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2013-2014 the original author or authors.
* Copyright 2013-2020 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.
@@ -16,7 +16,9 @@
package org.springframework.batch.item.data;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import org.junit.Before;
@@ -35,12 +37,17 @@ import org.springframework.data.domain.Sort;
import org.springframework.data.domain.Sort.Direction;
import org.springframework.data.repository.PagingAndSortingRepository;
import static java.util.Collections.singletonList;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@SuppressWarnings("serial")
@@ -54,8 +61,7 @@ public class RepositoryItemReaderTests {
@Before
public void setUp() throws Exception {
MockitoAnnotations.initMocks(this);
sorts = new HashMap<>();
sorts.put("id", Direction.ASC);
sorts = Collections.singletonMap("id", Direction.ASC);
reader = new RepositoryItemReader<>();
reader.setRepository(repository);
reader.setPageSize(1);
@@ -66,9 +72,10 @@ public class RepositoryItemReaderTests {
@Test
public void testAfterPropertiesSet() throws Exception {
try {
new RepositoryItemReader<>().afterPropertiesSet();
new RepositoryItemReader<>().afterPropertiesSet();
fail();
} catch (IllegalStateException e) {
// expected
}
try {
@@ -77,6 +84,7 @@ public class RepositoryItemReaderTests {
reader.afterPropertiesSet();
fail();
} catch (IllegalStateException iae) {
// expected
}
try {
@@ -86,6 +94,7 @@ public class RepositoryItemReaderTests {
reader.afterPropertiesSet();
fail();
} catch (IllegalStateException iae) {
// expected
}
try {
@@ -95,6 +104,7 @@ public class RepositoryItemReaderTests {
reader.afterPropertiesSet();
fail();
} catch (IllegalStateException iae) {
// expected
}
reader = new RepositoryItemReader<>();
@@ -120,14 +130,13 @@ public class RepositoryItemReaderTests {
}
@Test
@SuppressWarnings("serial")
public void testDoReadFirstReadResults() throws Exception {
ArgumentCaptor<PageRequest> pageRequestContainer = ArgumentCaptor.forClass(PageRequest.class);
final Object result = new Object();
when(repository.findAll(pageRequestContainer.capture())).thenReturn(new PageImpl<>(new ArrayList<Object>(){{
add(result);
}}));
when(repository.findAll(pageRequestContainer.capture())).thenReturn(new PageImpl<>(singletonList(
result
)));
assertEquals(result, reader.doRead());
@@ -139,15 +148,14 @@ public class RepositoryItemReaderTests {
}
@Test
@SuppressWarnings("serial")
public void testDoReadFirstReadSecondPage() throws Exception {
ArgumentCaptor<PageRequest> pageRequestContainer = ArgumentCaptor.forClass(PageRequest.class);
final Object result = new Object();
when(repository.findAll(pageRequestContainer.capture())).thenReturn(new PageImpl<>(new ArrayList<Object>() {{
add(new Object());
}})).thenReturn(new PageImpl<>(new ArrayList<Object>(){{
add(result);
}}));
when(repository.findAll(pageRequestContainer.capture())).thenReturn(new PageImpl<>(singletonList(
new Object()
))).thenReturn(new PageImpl<>(singletonList(
result
)));
assertFalse(reader.doRead() == result);
assertEquals(result, reader.doRead());
@@ -160,15 +168,14 @@ public class RepositoryItemReaderTests {
}
@Test
@SuppressWarnings("serial")
public void testDoReadFirstReadExhausted() throws Exception {
ArgumentCaptor<PageRequest> pageRequestContainer = ArgumentCaptor.forClass(PageRequest.class);
final Object result = new Object();
when(repository.findAll(pageRequestContainer.capture())).thenReturn(new PageImpl<>(new ArrayList<Object>() {{
add(new Object());
}})).thenReturn(new PageImpl<>(new ArrayList<Object>(){{
add(result);
}})).thenReturn(new PageImpl<>(new ArrayList<>()));
when(repository.findAll(pageRequestContainer.capture())).thenReturn(new PageImpl<>(singletonList(
new Object()
))).thenReturn(new PageImpl<>(singletonList(
result
))).thenReturn(new PageImpl<>(new ArrayList<>()));
assertFalse(reader.doRead() == result);
assertEquals(result, reader.doRead());
@@ -182,15 +189,21 @@ public class RepositoryItemReaderTests {
}
@Test
@SuppressWarnings("serial")
public void testJumpToItem() throws Exception {
reader.setPageSize(100);
final List<Object> objectList = fillWithNewObjects(100);
ArgumentCaptor<PageRequest> pageRequestContainer = ArgumentCaptor.forClass(PageRequest.class);
when(repository.findAll(pageRequestContainer.capture())).thenReturn(new PageImpl<>(new ArrayList<Object>() {{
add(new Object());
}}));
when(repository.findAll(pageRequestContainer.capture())).thenReturn(new PageImpl<>(
objectList
));
reader.jumpToItem(485);
// no page requested at this stage
verify(repository, never()).findAll(any(Pageable.class));
// the page must only actually be fetched on the next "doRead()" call
final Object o = reader.doRead();
assertSame("Fetched object should be at index 85 in the current page", o, objectList.get(85));
Pageable pageRequest = pageRequestContainer.getValue();
assertEquals(400, pageRequest.getOffset());
@@ -199,6 +212,34 @@ public class RepositoryItemReaderTests {
assertEquals("id: ASC", pageRequest.getSort().toString());
}
@Test
public void testJumpToItemFirstItemOnPage() throws Exception {
reader.setPageSize(50);
final List<Object> objectList = fillWithNewObjects(50);
ArgumentCaptor<PageRequest> pageRequestContainer = ArgumentCaptor.forClass(PageRequest.class);
when(repository.findAll(pageRequestContainer.capture())).thenReturn(new PageImpl<>(
objectList
));
reader.jumpToItem(150);
verify(repository, never()).findAll(any(Pageable.class));
assertSame("Fetched object should be the first one in the current page", objectList.get(0), reader.doRead());
Pageable pageRequest = pageRequestContainer.getValue();
assertEquals(150, pageRequest.getOffset());
assertEquals(3, pageRequest.getPageNumber());
assertEquals(50, pageRequest.getPageSize());
}
private static List<Object> fillWithNewObjects(int nb) {
List<Object> result = new ArrayList<>();
for (int i = 0; i < nb; i++) {
result.add(new TestItem(i));
}
return result;
}
@Test
public void testInvalidMethodName() throws Exception {
reader.setMethodName("thisMethodDoesNotExist");
@@ -215,17 +256,16 @@ public class RepositoryItemReaderTests {
public void testDifferentTypes() throws Exception {
TestRepository differentRepository = mock(TestRepository.class);
RepositoryItemReader<String> reader = new RepositoryItemReader<>();
sorts = new HashMap<>();
sorts.put("id", Direction.ASC);
sorts = Collections.singletonMap("id", Direction.ASC);
reader.setRepository(differentRepository);
reader.setPageSize(1);
reader.setSort(sorts);
reader.setMethodName("findFirstNames");
ArgumentCaptor<PageRequest> pageRequestContainer = ArgumentCaptor.forClass(PageRequest.class);
when(differentRepository.findFirstNames(pageRequestContainer.capture())).thenReturn(new PageImpl<>(new ArrayList<String>(){{
add("result");
}}));
when(differentRepository.findFirstNames(pageRequestContainer.capture())).thenReturn(new PageImpl<>(singletonList(
"result"
)));
assertEquals("result", reader.doRead());
@@ -238,47 +278,47 @@ public class RepositoryItemReaderTests {
@Test
public void testSettingCurrentItemCountExplicitly() throws Exception {
reader.setCurrentItemCount(3);
// Dataset : ("1" "2") | "3" "4" | "5" "6"
reader.setCurrentItemCount(3); // item as index 3 is : "4"
reader.setPageSize(2);
PageRequest request = PageRequest.of(1, 2, Sort.by(Direction.ASC, "id"));
when(repository.findAll(request)).thenReturn(new PageImpl<>(new ArrayList<Object>() {{
add("3");
add("4");
}}));
when(repository.findAll(request)).thenReturn(new PageImpl<>(Arrays.<Object> asList(
"3",
"4"
)));
request = PageRequest.of(2, 2, Sort.by(Direction.ASC, "id"));
when(repository.findAll(request)).thenReturn(new PageImpl<>(new ArrayList<Object>(){{
add("5");
add("6");
}}));
when(repository.findAll(request)).thenReturn(new PageImpl<>(Arrays.<Object> asList(
"5",
"6"
)));
reader.open(new ExecutionContext());
Object result = reader.read();
assertEquals("3", result);
assertEquals("4", reader.read());
assertEquals("4", result);
assertEquals("5", reader.read());
assertEquals("6", reader.read());
}
@Test
public void testSettingCurrentItemCountRestart() throws Exception {
reader.setCurrentItemCount(3);
reader.setCurrentItemCount(3); // item as index 3 is : "4"
reader.setPageSize(2);
PageRequest request = PageRequest.of(1, 2, Sort.by(Direction.ASC, "id"));
when(repository.findAll(request)).thenReturn(new PageImpl<>(new ArrayList<Object>(){{
add("3");
add("4");
}}));
when(repository.findAll(request)).thenReturn(new PageImpl<>(Arrays.<Object> asList(
"3",
"4"
)));
request = PageRequest.of(2, 2, Sort.by(Direction.ASC, "id"));
when(repository.findAll(request)).thenReturn(new PageImpl<>(new ArrayList<Object>() {{
add("5");
add("6");
}}));
when(repository.findAll(request)).thenReturn(new PageImpl<>(Arrays.<Object> asList(
"5",
"6"
)));
ExecutionContext executionContext = new ExecutionContext();
reader.open(executionContext);
@@ -287,10 +327,9 @@ public class RepositoryItemReaderTests {
reader.update(executionContext);
reader.close();
assertEquals("3", result);
assertEquals("4", result);
reader.open(executionContext);
assertEquals("4", reader.read());
assertEquals("5", reader.read());
assertEquals("6", reader.read());
}
@@ -300,16 +339,16 @@ public class RepositoryItemReaderTests {
reader.setPageSize(2);
PageRequest request = PageRequest.of(0, 2, Sort.by(Direction.ASC, "id"));
when(repository.findAll(request)).thenReturn(new PageImpl<>(new ArrayList<Object>(){{
add("1");
add("2");
}}));
when(repository.findAll(request)).thenReturn(new PageImpl<>(Arrays.<Object> asList(
"1",
"2"
)));
request = PageRequest.of(1, 2, Sort.by(Direction.ASC, "id"));
when(repository.findAll(request)).thenReturn(new PageImpl<>(new ArrayList<Object>() {{
add("3");
add("4");
}}));
when(repository.findAll(request)).thenReturn(new PageImpl<>(Arrays.<Object> asList(
"3",
"4"
)));
ExecutionContext executionContext = new ExecutionContext();
reader.open(executionContext);
@@ -325,7 +364,21 @@ public class RepositoryItemReaderTests {
assertEquals("3", reader.read());
}
public interface TestRepository extends PagingAndSortingRepository<Map<String, String>, Long> {
public interface TestRepository extends PagingAndSortingRepository<Map, Long> {
Page<String> findFirstNames(Pageable pageable);
}
// Simple object for readability
private static class TestItem {
private final int myIndex;
TestItem(int myIndex) {
this.myIndex = myIndex;
}
@Override
public String toString() {
return "TestItem at index " + myIndex;
}
}
}