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

Use the Chunk API consistently

Browse Source 
This commit replaces the usage of List with Chunk
where appropriate. Summary of changes:

- The Chunk class was moved from the `org.springframework.batch.core.step.item` package to the `org.springframework.batch.item` package
- The signature of the method `ItemWriter#write(List)` was changed to `ItemWriter#write(Chunk)`
- All implementations of `ItemWriter` were updated to use the Chunk API instead of List
- All methods in the `ItemWriteListener` interface were updated to use the Chunk API instead of List
- All implementations of `ItemWriteListener` were updated to use the Chunk API instead of List
- The constructor of `ChunkRequest` was changed to accept a Chunk instead of a Collection of items
- The return type of `ChunkRequest#getItems()` was changed from List to Chunk

Resolves #3954
This commit is contained in:
Mahmoud Ben Hassine
2022-08-17 21:06:09 +02:00
parent bf2e6ab0e5
commit e67c0069f1
175 changed files with 1077 additions and 763 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
spring-batch-integration/src/main/java/org/springframework/batch/integration/async/AsyncItemWriter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2006-2018 the original author or authors.
* Copyright 2006-2022 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.
@@ -23,6 +23,7 @@ import java.util.concurrent.Future;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.item.Chunk;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.ItemStreamException;
@@ -59,7 +60,7 @@ public class AsyncItemWriter<T> implements ItemStreamWriter<Future<T>>, Initiali
* delegate
* @throws Exception The exception returned by the Future if one was thrown
*/
public void write(List<? extends Future<T>> items) throws Exception {
public void write(Chunk<? extends Future<T>> items) throws Exception {
List<T> list = new ArrayList<>();
for (Future<T> future : items) {
try {
@@ -83,7 +84,7 @@ public class AsyncItemWriter<T> implements ItemStreamWriter<Future<T>>, Initiali
}
}
delegate.write(list);
delegate.write(new Chunk<>(list));
}
@Override

7
spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkMessageChannelItemWriter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2006-2021 the original author or authors.
* Copyright 2006-2022 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.
@@ -30,6 +30,7 @@ import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.StepExecutionListener;
import org.springframework.batch.item.Chunk;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.ItemStreamException;
@@ -92,7 +93,7 @@ public class ChunkMessageChannelItemWriter<T>
this.replyChannel = replyChannel;
}
public void write(List<? extends T> items) throws Exception {
public void write(Chunk<? extends T> items) throws Exception {
// Block until expecting <= throttle limit
while (localState.getExpecting() > throttleLimit) {
@@ -283,7 +284,7 @@ public class ChunkMessageChannelItemWriter<T>
return expected.get() - actual.get();
}
public <T> ChunkRequest<T> getRequest(List<? extends T> items) {
public <T> ChunkRequest<T> getRequest(Chunk<? extends T> items) {
return new ChunkRequest<>(current.incrementAndGet(), items, getJobId(), createStepContribution());
}

7
spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkProcessorChunkHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2006-2013 the original author or authors.
* Copyright 2006-2022 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.
@@ -20,7 +20,7 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.core.JobInterruptedException;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.step.item.Chunk;
import org.springframework.batch.item.Chunk;
import org.springframework.batch.core.step.item.ChunkProcessor;
import org.springframework.batch.core.step.item.FaultTolerantChunkProcessor;
import org.springframework.batch.core.step.skip.NonSkippableReadException;
@@ -40,6 +40,7 @@ import org.springframework.util.Assert;
*
* @author Dave Syer
* @author Michael Minella
* @author Mahmoud Ben Hassine
* @param <S> the type of the items in the chunk to be handled
*/
@MessageEndpoint
@@ -100,7 +101,7 @@ public class ChunkProcessorChunkHandler<S> implements ChunkHandler<S>, Initializ
*/
private Throwable process(ChunkRequest<S> chunkRequest, StepContribution stepContribution) throws Exception {
Chunk<S> chunk = new Chunk<>(chunkRequest.getItems());
Chunk chunk = chunkRequest.getItems();
Throwable failure = null;
try {
chunkProcessor.process(stepContribution, chunk);

10
spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkRequest.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2006-2007 the original author or authors.
* Copyright 2006-2022 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.
@@ -20,11 +20,13 @@ import java.io.Serializable;
import java.util.Collection;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.item.Chunk;
/**
* Encapsulation of a chunk of items to be processed remotely as part of a step execution.
*
* @author Dave Syer
* @author Mahmoud Ben Hassine
* @param <T> the type of the items to process
*/
public class ChunkRequest<T> implements Serializable {
@@ -33,13 +35,13 @@ public class ChunkRequest<T> implements Serializable {
private final long jobId;
private final Collection<? extends T> items;
private final Chunk<? extends T> items;
private final StepContribution stepContribution;
private final int sequence;
public ChunkRequest(int sequence, Collection<? extends T> items, long jobId, StepContribution stepContribution) {
public ChunkRequest(int sequence, Chunk<? extends T> items, long jobId, StepContribution stepContribution) {
this.sequence = sequence;
this.items = items;
this.jobId = jobId;
@@ -50,7 +52,7 @@ public class ChunkRequest<T> implements Serializable {
return jobId;
}
public Collection<? extends T> getItems() {
public Chunk<? extends T> getItems() {
return items;
}

6
spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/RemoteChunkHandlerFactoryBean.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2006-2021 the original author or authors.
* Copyright 2006-2022 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.
@@ -22,7 +22,7 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.StepExecutionListener;
import org.springframework.batch.core.step.item.Chunk;
import org.springframework.batch.item.Chunk;
import org.springframework.batch.core.step.item.ChunkOrientedTasklet;
import org.springframework.batch.core.step.item.ChunkProcessor;
import org.springframework.batch.core.step.item.FaultTolerantChunkProcessor;
@@ -176,7 +176,7 @@ public class RemoteChunkHandlerFactoryBean<T> implements FactoryBean<ChunkHandle
@Override
protected void write(StepContribution contribution, Chunk<T> inputs, Chunk<T> outputs)
throws Exception {
doWrite(outputs.getItems());
doWrite(outputs);
// Do not update the step contribution until the chunks are
// actually processed
updateStepContribution(contribution, stepContributionSource);

22
spring-batch-integration/src/test/java/org/springframework/batch/integration/async/AsyncItemWriterTests.java
View File

@@ -27,6 +27,7 @@ import java.util.concurrent.TimeoutException;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.batch.item.Chunk;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemStreamException;
import org.springframework.batch.item.ItemStreamWriter;
@@ -41,6 +42,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
/**
* @author mminella
* @author Mahmoud Ben Hassine
*/
class AsyncItemWriterTests {
@@ -60,7 +62,7 @@ class AsyncItemWriterTests {
@Test
void testRoseyScenario() throws Exception {
writer.setDelegate(new ListItemWriter(writtenItems));
List<FutureTask<String>> processedItems = new ArrayList<>();
Chunk<FutureTask<String>> processedItems = new Chunk<>();
processedItems.add(new FutureTask<>(new Callable<String>() {
@Override
@@ -90,7 +92,7 @@ class AsyncItemWriterTests {
@Test
void testFilteredItem() throws Exception {
writer.setDelegate(new ListItemWriter(writtenItems));
List<FutureTask<String>> processedItems = new ArrayList<>();
Chunk<FutureTask<String>> processedItems = new Chunk<>();
processedItems.add(new FutureTask<>(new Callable<String>() {
@Override
@@ -119,7 +121,7 @@ class AsyncItemWriterTests {
@Test
void testException() {
writer.setDelegate(new ListItemWriter(writtenItems));
List<FutureTask<String>> processedItems = new ArrayList<>();
Chunk<FutureTask<String>> processedItems = new Chunk<>();
processedItems.add(new FutureTask<>(new Callable<String>() {
@Override
@@ -147,7 +149,7 @@ class AsyncItemWriterTests {
void testExecutionException() {
ListItemWriter delegate = new ListItemWriter(writtenItems);
writer.setDelegate(delegate);
List<Future<String>> processedItems = new ArrayList<>();
Chunk<Future<String>> processedItems = new Chunk<>();
processedItems.add(new Future<String>() {
@@ -189,7 +191,7 @@ class AsyncItemWriterTests {
ListItemStreamWriter itemWriter = new ListItemStreamWriter(writtenItems);
writer.setDelegate(itemWriter);
List<FutureTask<String>> processedItems = new ArrayList<>();
Chunk<FutureTask<String>> processedItems = new Chunk<>();
ExecutionContext executionContext = new ExecutionContext();
writer.open(executionContext);
@@ -207,7 +209,7 @@ class AsyncItemWriterTests {
ListItemWriter itemWriter = new ListItemWriter(writtenItems);
writer.setDelegate(itemWriter);
List<FutureTask<String>> processedItems = new ArrayList<>();
Chunk<FutureTask<String>> processedItems = new Chunk<>();
ExecutionContext executionContext = new ExecutionContext();
writer.open(executionContext);
@@ -235,8 +237,8 @@ class AsyncItemWriterTests {
}
@Override
public void write(List<? extends String> items) throws Exception {
this.items.addAll(items);
public void write(Chunk<? extends String> chunk) throws Exception {
this.items.addAll(chunk.getItems());
}
}
@@ -256,8 +258,8 @@ class AsyncItemWriterTests {
}
@Override
public void write(List<? extends String> items) throws Exception {
this.items.addAll(items);
public void write(Chunk<? extends String> chunk) throws Exception {
this.items.addAll(chunk.getItems());
}
@Override

11
spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkMessageItemWriterIntegrationTests.java
View File

@@ -37,6 +37,7 @@ import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.repository.JobRestartException;
import org.springframework.batch.core.repository.support.JobRepositoryFactoryBean;
import org.springframework.batch.core.step.factory.SimpleStepFactoryBean;
import org.springframework.batch.item.Chunk;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.support.ListItemReader;
import org.springframework.beans.factory.annotation.Autowired;
@@ -161,8 +162,8 @@ class ChunkMessageItemWriterIntegrationTests {
stepExecution.getExecutionContext().putInt(ChunkMessageChannelItemWriter.EXPECTED, 6);
stepExecution.getExecutionContext().putInt(ChunkMessageChannelItemWriter.ACTUAL, 4);
// And make the back log real
requests.send(getSimpleMessage("foo", stepExecution.getJobExecution().getJobId()));
requests.send(getSimpleMessage("bar", stepExecution.getJobExecution().getJobId()));
requests.send(getSimpleMessage(stepExecution.getJobExecution().getJobId(), "foo"));
requests.send(getSimpleMessage(stepExecution.getJobExecution().getJobId(), "bar"));
step.execute(stepExecution);
waitForResults(8, 10);
@@ -190,7 +191,7 @@ class ChunkMessageItemWriterIntegrationTests {
writer.setMaxWaitTimeouts(2);
// And make the back log real
requests.send(getSimpleMessage("foo", 4321L));
requests.send(getSimpleMessage(4321L, "foo"));
step.execute(stepExecution);
assertEquals(BatchStatus.FAILED, stepExecution.getStatus());
assertEquals(ExitStatus.FAILED.getExitCode(), stepExecution.getExitStatus().getExitCode());
@@ -205,10 +206,10 @@ class ChunkMessageItemWriterIntegrationTests {
}
@SuppressWarnings({ "unchecked", "rawtypes" })
private GenericMessage<ChunkRequest> getSimpleMessage(String string, Long jobId) {
private GenericMessage<ChunkRequest> getSimpleMessage(Long jobId, String... items) {
StepContribution stepContribution = new JobExecution(new JobInstance(0L, "job"), new JobParameters())
.createStepExecution("step").createStepContribution();
ChunkRequest chunk = new ChunkRequest(0, StringUtils.commaDelimitedListToSet(string), jobId, stepContribution);
ChunkRequest chunk = new ChunkRequest(0, Chunk.of(items), jobId, stepContribution);
GenericMessage<ChunkRequest> message = new GenericMessage<>(chunk);
return message;
}

12
spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkProcessorChunkHandlerTests.java
View File

@@ -20,7 +20,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
import org.junit.jupiter.api.Test;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.step.item.Chunk;
import org.springframework.batch.item.Chunk;
import org.springframework.batch.core.step.item.ChunkProcessor;
import org.springframework.batch.test.MetaDataInstanceFactory;
import org.springframework.util.StringUtils;
@@ -33,14 +33,20 @@ class ChunkProcessorChunkHandlerTests {
@Test
void testVanillaHandleChunk() throws Exception {
// given
handler.setChunkProcessor(new ChunkProcessor<Object>() {
public void process(StepContribution contribution, Chunk<Object> chunk) throws Exception {
count += chunk.size();
}
});
StepContribution stepContribution = MetaDataInstanceFactory.createStepExecution().createStepContribution();
ChunkResponse response = handler.handleChunk(
new ChunkRequest<>(0, StringUtils.commaDelimitedListToSet("foo,bar"), 12L, stepContribution));
Chunk items = Chunk.of("foo", "bar");
ChunkRequest chunkRequest = new ChunkRequest<>(0, items, 12L, stepContribution);
// when
ChunkResponse response = handler.handleChunk(chunkRequest);
// then
assertEquals(stepContribution, response.getStepContribution());
assertEquals(12, response.getJobId().longValue());
assertTrue(response.isSuccessful());

5
spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/ChunkRequestTests.java
View File

@@ -21,16 +21,19 @@ import static org.junit.jupiter.api.Assertions.assertNotNull;
import java.util.Arrays;
import org.junit.jupiter.api.Test;
import org.springframework.batch.item.Chunk;
import org.springframework.batch.test.MetaDataInstanceFactory;
import org.springframework.util.SerializationUtils;
/**
* @author Dave Syer
* @author Mahmoud Ben Hassine
*
*/
class ChunkRequestTests {
private final ChunkRequest<String> request = new ChunkRequest<>(0, Arrays.asList("foo", "bar"), 111L,
private final ChunkRequest<String> request = new ChunkRequest<>(0, Chunk.of("foo", "bar"), 111L,
MetaDataInstanceFactory.createStepExecution().createStepContribution());
@Test

19
spring-batch-integration/src/test/java/org/springframework/batch/integration/chunk/TestItemWriter.java
View File

@@ -1,9 +1,26 @@
/*
* Copyright 2022 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
*
* https://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.integration.chunk;
import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.item.Chunk;
import org.springframework.batch.item.ItemWriter;
import org.springframework.stereotype.Component;
@@ -37,7 +54,7 @@ public class TestItemWriter<T> implements ItemWriter<T> {
*/
public static final String WAIT_ON = "wait";
public void write(List<? extends T> items) throws Exception {
public void write(Chunk<? extends T> items) throws Exception {
for (T item : items) {

3
spring-batch-integration/src/test/java/org/springframework/batch/integration/config/xml/RemoteChunkingParserTests.java
View File

@@ -25,6 +25,7 @@ import org.springframework.batch.integration.chunk.ChunkHandler;
import org.springframework.batch.integration.chunk.ChunkMessageChannelItemWriter;
import org.springframework.batch.integration.chunk.ChunkProcessorChunkHandler;
import org.springframework.batch.integration.chunk.RemoteChunkHandlerFactoryBean;
import org.springframework.batch.item.Chunk;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.support.PassThroughItemProcessor;
@@ -251,7 +252,7 @@ class RemoteChunkingParserTests {
private static class Writer implements ItemWriter<String> {
@Override
public void write(List<? extends String> items) throws Exception {
public void write(Chunk<? extends String> items) throws Exception {
//
}

7
spring-batch-integration/src/test/java/org/springframework/batch/integration/item/MessagingGatewayIntegrationTests.java
View File

@@ -22,6 +22,8 @@ import java.util.Arrays;
import java.util.List;
import org.junit.jupiter.api.Test;
import org.springframework.batch.item.Chunk;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemWriter;
import org.springframework.beans.factory.annotation.Autowired;
@@ -69,7 +71,7 @@ class MessagingGatewayIntegrationTests {
@Test
void testWriter() throws Exception {
writer.write(Arrays.asList("foo", "bar", "spam"));
writer.write(Chunk.of("foo", "bar", "spam"));
assertEquals(3, splitter.count);
assertEquals(3, service.count);
}
@@ -104,6 +106,7 @@ class MessagingGatewayIntegrationTests {
* More complex splitters might filter or enhance the items before passing them on.
*
* @author Dave Syer
* @author Mahmoud Ben Hassine
*
*/
@MessageEndpoint
@@ -113,7 +116,7 @@ class MessagingGatewayIntegrationTests {
private int count;
@Splitter
public List<String> split(List<String> input) {
public Chunk<String> split(Chunk<String> input) {
count += input.size();
return input;
}

21
spring-batch-integration/src/test/java/org/springframework/batch/integration/partition/ExampleItemWriter.java
View File

@@ -1,9 +1,26 @@
/*
* Copyright 2022 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
*
* https://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.integration.partition;
import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.item.Chunk;
import org.springframework.batch.item.ItemWriter;
/**
@@ -14,9 +31,9 @@ public class ExampleItemWriter implements ItemWriter<Object> {
private static final Log log = LogFactory.getLog(ExampleItemWriter.class);
/**
* @see ItemWriter#write(List)
* @see ItemWriter#write(Chunk)
*/
public void write(List<? extends Object> data) throws Exception {
public void write(Chunk<? extends Object> data) throws Exception {
log.info(data);
}