Fix output chunk end property in ChunkProcessor implementations

Resolves #4560

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2006-2023 the original author or authors.
+ * Copyright 2006-2024 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.
@@ -306,7 +306,9 @@ public class FaultTolerantChunkProcessor<I, O> extends SimpleChunkProcessor<I, O
 				break;
 			}
 		}
-
+		if (inputs.isEnd()) {
+			outputs.setEnd();
+		}
 		return outputs;
 
 	}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2006-2022 the original author or authors.
+ * Copyright 2006-2024 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.
@@ -340,6 +340,9 @@ public class SimpleChunkProcessor<I, O> implements ChunkProcessor<I>, Initializi
 				iterator.remove();
 			}
 		}
+		if (inputs.isEnd()) {
+			outputs.setEnd();
+		}
 		return outputs;
 	}
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2008-2023 the original author or authors.
+ * Copyright 2008-2024 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.
@@ -18,6 +18,7 @@ package org.springframework.batch.core.step.item;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
 
 import java.util.ArrayList;
@@ -97,6 +98,16 @@ class FaultTolerantChunkProcessorTests {
 		assertEquals(1, contribution.getFilterCount());
 	}
 
+	@Test
+	void testTransformChunkEnd() throws Exception {
+		Chunk<String> inputs = new Chunk<>(Arrays.asList("1", "2"));
+		inputs.setEnd();
+		processor.initializeUserData(inputs);
+		Chunk<String> outputs = processor.transform(contribution, inputs);
+		assertEquals(Arrays.asList("1", "2"), outputs.getItems());
+		assertTrue(outputs.isEnd());
+	}
+
 	@Test
 	void testFilterCountOnSkip() throws Exception {
 		processor.setProcessSkipPolicy(new AlwaysSkipItemSkipPolicy());

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2008-2023 the original author or authors.
+ * Copyright 2008-2024 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,6 +16,7 @@
 package org.springframework.batch.core.step.item;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -76,4 +77,15 @@ class SimpleChunkProcessorTests {
 		assertEquals(2, contribution.getWriteCount());
 	}
 
+	@Test
+	void testTransform() throws Exception {
+		Chunk<String> inputs = new Chunk<>();
+		inputs.add("foo");
+		inputs.add("bar");
+		inputs.setEnd();
+		Chunk<String> outputs = processor.transform(contribution, inputs);
+		assertEquals(Arrays.asList("foo", "bar"), outputs.getItems());
+		assertTrue(outputs.isEnd());
+	}
+
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2006-2023 the original author or authors.
+ * Copyright 2006-2024 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.
@@ -154,6 +154,13 @@ public class Chunk<W> implements Iterable<W>, Serializable {
 
 	/**
 	 * Flag to indicate if the source data is exhausted.
+	 *
+	 * <p>
+	 * Note: This may return false if the last chunk has the same number of items as the
+	 * configured commit interval. Consequently, in such cases,there will be a last empty
+	 * chunk that won't be processed. It is recommended to consider this behavior when
+	 * utilizing this method.
+	 * </p>
 	 * @return true if there is no more data to process
 	 */
 	public boolean isEnd() {