[WIP] Use chunk API consistently

Issue spring-projects#3954

Author: fmbenhassine

spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkOrientedTasklet.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2006-2019 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.
@@ -21,13 +21,15 @@
 import org.springframework.batch.core.StepContribution;
 import org.springframework.batch.core.scope.context.ChunkContext;
 import org.springframework.batch.core.step.tasklet.Tasklet;
+import org.springframework.batch.item.Chunk;
 import org.springframework.batch.repeat.RepeatStatus;
 import org.springframework.lang.Nullable;
 
 /**
  * A {@link Tasklet} implementing variations on read-process-write item handling.
  *
  * @author Dave Syer
+ * @author Mahmoud Ben Hassine
  * @param <I> input item type
  */
 public class ChunkOrientedTasklet<I> implements Tasklet {

spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkProcessor.java

@@ -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.
@@ -17,6 +17,7 @@
 package org.springframework.batch.core.step.item;
 
 import org.springframework.batch.core.StepContribution;
+import org.springframework.batch.item.Chunk;
 
 /**
  * Interface defined for processing {@link Chunk}s.

spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkProvider.java

@@ -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.
@@ -17,6 +17,7 @@
 package org.springframework.batch.core.step.item;
 
 import org.springframework.batch.core.StepContribution;
+import org.springframework.batch.item.Chunk;
 
 /**
  * Interface for providing {@link Chunk}s to be processed, used by the

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

@@ -34,8 +34,10 @@
 import org.springframework.batch.core.step.skip.SkipLimitExceededException;
 import org.springframework.batch.core.step.skip.SkipListenerFailedException;
 import org.springframework.batch.core.step.skip.SkipPolicy;
+import org.springframework.batch.item.Chunk;
 import org.springframework.batch.item.ItemProcessor;
 import org.springframework.batch.item.ItemWriter;
+import org.springframework.batch.item.SkipWrapper;
 import org.springframework.classify.BinaryExceptionClassifier;
 import org.springframework.classify.Classifier;
 import org.springframework.retry.ExhaustedRetryException;

spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProvider.java

@@ -23,6 +23,7 @@
 import org.springframework.batch.core.step.skip.SkipListenerFailedException;
 import org.springframework.batch.core.step.skip.SkipPolicy;
 import org.springframework.batch.core.step.skip.SkipPolicyFailedException;
+import org.springframework.batch.item.Chunk;
 import org.springframework.batch.item.ItemReader;
 import org.springframework.batch.repeat.RepeatOperations;
 import org.springframework.classify.BinaryExceptionClassifier;

spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkProcessor.java

@@ -26,6 +26,7 @@
 import org.springframework.batch.core.StepListener;
 import org.springframework.batch.core.listener.MulticasterBatchListener;
 import org.springframework.batch.core.observability.BatchMetrics;
+import org.springframework.batch.item.Chunk;
 import org.springframework.batch.item.ItemProcessor;
 import org.springframework.batch.item.ItemWriter;
 import org.springframework.beans.factory.InitializingBean;

spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkProvider.java

@@ -28,6 +28,7 @@
 import org.springframework.batch.core.StepListener;
 import org.springframework.batch.core.listener.MulticasterBatchListener;
 import org.springframework.batch.core.observability.BatchMetrics;
+import org.springframework.batch.item.Chunk;
 import org.springframework.batch.item.ItemReader;
 import org.springframework.batch.repeat.RepeatCallback;
 import org.springframework.batch.repeat.RepeatContext;

spring-batch-core/src/test/java/org/springframework/batch/core/job/builder/FlowJobBuilderTests.java

@@ -44,6 +44,8 @@
 import org.springframework.batch.core.repository.JobRepository;
 import org.springframework.batch.core.repository.support.JobRepositoryFactoryBean;
 import org.springframework.batch.core.step.StepSupport;
+import org.springframework.batch.item.Chunk;
+import org.springframework.batch.item.ItemWriter;
 import org.springframework.batch.item.support.ListItemReader;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.ApplicationContext;
@@ -292,7 +294,11 @@ static class JobConfiguration {
 		public Step step(StepBuilderFactory stepBuilderFactory,
 				@Value("#{jobParameters['chunkSize']}") Integer chunkSize) {
 			return stepBuilderFactory.get("step").<Integer, Integer>chunk(chunkSize)
-					.reader(new ListItemReader<>(Arrays.asList(1, 2, 3, 4))).writer(items -> {
+					.reader(new ListItemReader<>(Arrays.asList(1, 2, 3, 4))).writer(new ItemWriter<Integer>() {
+						@Override
+						public void write(Chunk<Integer> chunk) throws Exception {
+
+						}
 					}).build();
 		}
 

spring-batch-core/src/test/java/org/springframework/batch/core/observability/BatchMetricsTests.java

@@ -38,6 +38,8 @@
 import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
 import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
 import org.springframework.batch.core.launch.JobLauncher;
+import org.springframework.batch.item.Chunk;
+import org.springframework.batch.item.ItemWriter;
 import org.springframework.batch.item.support.ListItemReader;
 import org.springframework.batch.repeat.RepeatStatus;
 import org.springframework.context.ApplicationContext;
@@ -237,17 +239,27 @@ public Step step1() {
 
 		@Bean
 		public Step step2() {
+			ItemWriter<? super Integer> itemWriter = new ItemWriter<Integer>() {
+				@Override
+				public void write(Chunk<Integer> chunk) throws Exception {
+					chunk.getItems().forEach(System.out::println);
+				}
+			};
 			return stepBuilderFactory.get("step2").<Integer, Integer>chunk(2)
-					.reader(new ListItemReader<>(Arrays.asList(1, 2, 3, 4, 5)))
-					.writer(items -> items.forEach(System.out::println)).build();
+					.reader(new ListItemReader<>(Arrays.asList(1, 2, 3, 4, 5))).writer(itemWriter).build();
 		}
 
 		@Bean
 		public Step step3() {
+			ItemWriter<? super Integer> itemWriter = new ItemWriter<Integer>() {
+				@Override
+				public void write(Chunk<Integer> chunk) throws Exception {
+					chunk.getItems().forEach(System.out::println);
+				}
+			};
 			return stepBuilderFactory.get("step3").<Integer, Integer>chunk(2)
-					.reader(new ListItemReader<>(Arrays.asList(6, 7, 8, 9, 10)))
-					.writer(items -> items.forEach(System.out::println)).faultTolerant().skip(Exception.class)
-					.skipLimit(3).build();
+					.reader(new ListItemReader<>(Arrays.asList(6, 7, 8, 9, 10))).writer(itemWriter).faultTolerant()
+					.skip(Exception.class).skipLimit(3).build();
 		}
 
 		@Bean

spring-batch-core/src/test/java/org/springframework/batch/core/step/item/AlmostStatefulRetryChunkTests.java

@@ -26,12 +26,15 @@
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.MethodSource;
 
+import org.springframework.batch.item.Chunk;
+
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 /**
  * @author Dave Syer
+ * @author Mahmoud Ben Hassine
  *
  */
 class AlmostStatefulRetryChunkTests {

spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ChunkOrientedTaskletTests.java

@@ -27,6 +27,7 @@
 import org.springframework.batch.core.StepContribution;
 import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.core.scope.context.ChunkContext;
+import org.springframework.batch.item.Chunk;
 
 /**
  * @author Dave Syer

spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessorTests.java

@@ -36,6 +36,7 @@
 import org.springframework.batch.core.listener.ItemListenerSupport;
 import org.springframework.batch.core.step.skip.AlwaysSkipItemSkipPolicy;
 import org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy;
+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;

spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProviderTests.java

@@ -28,6 +28,7 @@
 import org.springframework.batch.core.StepContribution;
 import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.core.step.skip.LimitCheckingItemSkipPolicy;
+import org.springframework.batch.item.Chunk;
 import org.springframework.batch.item.ItemReader;
 import org.springframework.batch.item.ParseException;
 import org.springframework.batch.item.UnexpectedInputException;

spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleChunkProcessorTests.java

@@ -28,6 +28,7 @@
 import org.springframework.batch.core.JobParameters;
 import org.springframework.batch.core.StepContribution;
 import org.springframework.batch.core.StepExecution;
+import org.springframework.batch.item.Chunk;
 import org.springframework.batch.item.ItemProcessor;
 import org.springframework.batch.item.ItemWriter;
 import org.springframework.lang.Nullable;

spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleChunkProviderTests.java

@@ -26,6 +26,7 @@
 import org.springframework.batch.core.JobParameters;
 import org.springframework.batch.core.StepContribution;
 import org.springframework.batch.core.StepExecution;
+import org.springframework.batch.item.Chunk;
 import org.springframework.batch.item.support.ListItemReader;
 import org.springframework.batch.repeat.support.RepeatTemplate;
 

spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SkipWrapperTests.java

@@ -21,6 +21,8 @@
 
 import org.junit.jupiter.api.Test;
 
+import org.springframework.batch.item.SkipWrapper;
+
 /**
  * @author Dave Syer
  *
@@ -41,7 +43,7 @@ void testItemWrapperT() {
 
 	/**
 	 * Test method for
-	 * {@link org.springframework.batch.core.step.item.SkipWrapper#SkipWrapper(java.lang.Object, java.lang.Throwable)}.
+	 * {@link SkipWrapper#SkipWrapper(java.lang.Object, java.lang.Throwable)}.
 	 */
 	@Test
 	void testItemWrapperTException() {
@@ -51,8 +53,7 @@ void testItemWrapperTException() {
 	}
 
 	/**
-	 * Test method for
-	 * {@link org.springframework.batch.core.step.item.SkipWrapper#toString()}.
+	 * Test method for {@link SkipWrapper#toString()}.
 	 */
 	@Test
 	void testToString() {

spring-batch-core/src/test/java/org/springframework/batch/core/step/tasklet/ChunkOrientedStepIntegrationTests.java

@@ -26,8 +26,10 @@
 import org.springframework.batch.core.StepExecution;
 import org.springframework.batch.core.job.JobSupport;
 import org.springframework.batch.core.repository.JobRepository;
+import org.springframework.batch.item.Chunk;
 import org.springframework.batch.item.ExecutionContext;
 import org.springframework.batch.item.ItemReader;
+import org.springframework.batch.item.ItemWriter;
 import org.springframework.batch.item.support.ListItemReader;
 import org.springframework.batch.repeat.policy.SimpleCompletionPolicy;
 import org.springframework.batch.repeat.support.RepeatTemplate;
@@ -87,13 +89,18 @@ void onSetUp() {
 	@Disabled
 	void testStatusForCommitFailedException() throws Exception {
 
-		step.setTasklet(new TestingChunkOrientedTasklet<>(getReader(new String[] { "a", "b", "c" }),
-				data -> TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
+		step.setTasklet(
+				new TestingChunkOrientedTasklet<>(getReader(new String[] { "a", "b", "c" }), new ItemWriter<String>() {
 					@Override
-					public void beforeCommit(boolean readOnly) {
-						throw new RuntimeException("Simulate commit failure");
+					public void write(Chunk<String> chunk) throws Exception {
+						TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
+							@Override
+							public void beforeCommit(boolean readOnly) {
+								throw new RuntimeException("Simulate commit failure");
+							}
+						});
 					}
-				}), chunkOperations));
+				}, chunkOperations));
 
 		JobExecution jobExecution = jobRepository.createJobExecution(job.getName(),
 				new JobParameters(Collections.singletonMap("run.id", new JobParameter(getClass().getName() + ".1"))));

spring-batch-core/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepIntegrationTests.java

@@ -35,6 +35,7 @@
 import org.springframework.batch.core.step.skip.AlwaysSkipItemSkipPolicy;
 import org.springframework.batch.core.step.skip.SkipLimitExceededException;
 import org.springframework.batch.core.step.skip.SkipPolicy;
+import org.springframework.batch.item.Chunk;
 import org.springframework.batch.item.ItemProcessor;
 import org.springframework.batch.item.ItemReader;
 import org.springframework.batch.item.ItemWriter;
@@ -70,9 +71,12 @@ class FaultTolerantStepIntegrationTests {
 	@BeforeEach
 	void setUp() {
 		ItemReader<Integer> itemReader = new ListItemReader<>(createItems());
-		ItemWriter<Integer> itemWriter = chunk -> {
-			if (chunk.contains(1)) {
-				throw new IllegalArgumentException();
+		ItemWriter<Integer> itemWriter = new ItemWriter<Integer>() {
+			@Override
+			public void write(Chunk<Integer> chunk) throws Exception {
+				if (chunk.getItems().contains(1)) {
+					throw new IllegalArgumentException();
+				}
 			}
 		};
 		skipPolicy = new SkipIllegalArgumentExceptionSkipPolicy();

spring-batch-core/src/main/java/org/springframework/batch/core/step/item/Chunk.java renamed to spring-batch-infrastructure/src/main/java/org/springframework/batch/item/Chunk.java

@@ -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.
@@ -14,8 +14,9 @@
  * limitations under the License.
  */
 
-package org.springframework.batch.core.step.item;
+package org.springframework.batch.item;
 
+import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -26,13 +27,14 @@
  * Encapsulation of a list of items to be processed and possibly a list of failed items to
  * be skipped. To mark an item as skipped clients should iterate over the chunk using the
  * {@link #iterator()} method, and if there is a failure call
- * {@link org.springframework.batch.core.step.item.Chunk.ChunkIterator#remove()} on the
- * iterator. The skipped items are then available through the chunk.
+ * {@link Chunk.ChunkIterator#remove()} on the iterator. The skipped items are then
+ * available through the chunk.
  *
  * @author Dave Syer
+ * @author Mahmoud Ben Hassine
  * @since 2.0
  */
-public class Chunk<W> implements Iterable<W> {
+public class Chunk<W> implements Iterable<W>, Serializable {
 
 	private List<W> items = new ArrayList<>();
 

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemWriter.java

@@ -36,8 +36,9 @@
  * @author Dave Syer
  * @author Lucas Ward
  * @author Taeik Lim
+ * @author Mahmoud Ben Hassine
  */
-@FunctionalInterface
+// TODO in v5.2 add @FunctionalInterface once write(List<? extends T> items) is removed
 public interface ItemWriter<T> {
 
 	/**
@@ -46,7 +47,23 @@ public interface ItemWriter<T> {
 	 * @param items items to be written
 	 * @throws Exception if there are errors. The framework will catch the exception and
 	 * convert or rethrow it as appropriate.
+	 * @deprecated since 5.0 in favor of {@link ItemWriter#write(Chunk)} instead. Will be
+	 * removed in v5.2
 	 */
-	void write(List<? extends T> items) throws Exception;
+	@Deprecated(since = "5.0")
+	default void write(List<? extends T> items) throws Exception {
+		write(new Chunk<>(items));
+	}
+
+	/**
+	 * Write items to a target. Will not be called with any null items in normal
+	 * operation.
+	 * @param items to write
+	 * @throws Exception if there are errors. The framework will catch the exception and
+	 * convert or rethrow it as appropriate.
+	 */
+	default void write(Chunk<T> items) throws Exception {
+
+	}
 
 }

spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SkipWrapper.java renamed to spring-batch-infrastructure/src/main/java/org/springframework/batch/item/SkipWrapper.java

@@ -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.
@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.batch.core.step.item;
+package org.springframework.batch.item;
 
 import org.springframework.lang.Nullable;