Add a new Step implementation for the chunk-oriented processing model

This commit also adapts the ChunkListener interface as well as corresponding
annotations to the new implementation.

Moreover, this commit deprecates SkipWrapper and its usage in the Chunk API
as they are not used in the new implementation.

Resolves #3950

Author: fmbenhassine

spring-batch-core/src/main/java/org/springframework/batch/core/annotation/AfterChunk.java

@@ -16,21 +16,21 @@
 package org.springframework.batch.core.annotation;
 
 import org.springframework.batch.core.listener.ChunkListener;
-import org.springframework.batch.core.scope.context.ChunkContext;
+import org.springframework.batch.item.Chunk;
 
 import java.lang.annotation.ElementType;
 import java.lang.annotation.Retention;
 import java.lang.annotation.RetentionPolicy;
 import java.lang.annotation.Target;
 
 /**
- * Marks a method to be called after a chunk is executed.<br>
- * <br>
- * Expected signature: void afterChunk(ChunkContext context)
+ * Marks a method to be called after a chunk is processed. <br>
+ * Expected signature: void afterChunk(Chunk)
  *
  * @author Lucas Ward
+ * @author Mahmoud Ben Hassine
  * @since 2.0
- * @see ChunkListener#afterChunk(ChunkContext context)
+ * @see ChunkListener#afterChunk(Chunk)
  */
 @Retention(RetentionPolicy.RUNTIME)
 @Target({ ElementType.METHOD })

spring-batch-core/src/main/java/org/springframework/batch/core/annotation/BeforeChunk.java

@@ -22,15 +22,17 @@
 
 import org.springframework.batch.core.listener.ChunkListener;
 import org.springframework.batch.core.scope.context.ChunkContext;
+import org.springframework.batch.item.Chunk;
 
 /**
  * Marks a method to be called before a chunk is executed. <br>
  * <br>
- * Expected signature: void beforeChunk(ChunkContext context)
+ * Expected signature: void beforeChunk(Chunk)
  *
  * @author Lucas Ward
+ * @author Mahmoud Ben Hassine
  * @since 2.0
- * @see ChunkListener#beforeChunk(ChunkContext context)
+ * @see ChunkListener#beforeChunk(Chunk)
  */
 @Retention(RetentionPolicy.RUNTIME)
 @Target({ ElementType.METHOD })

spring-batch-core/src/main/java/org/springframework/batch/core/annotation/OnChunkError.java

@@ -0,0 +1,38 @@
+/*
+ * Copyright 2025-present 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.core.annotation;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+import org.springframework.batch.core.listener.ChunkListener;
+import org.springframework.batch.item.Chunk;
+
+/**
+ * Marks a method to be called after a chunk has failed. <br>
+ * Expected signature: void onChunkError(Exception, Chunk)
+ *
+ * @author Mahmoud Ben Hassine
+ * @since 6.0
+ * @see ChunkListener#onChunkError(Exception, Chunk)
+ */
+@Retention(RetentionPolicy.RUNTIME)
+@Target({ ElementType.METHOD })
+public @interface OnChunkError {
+
+}

spring-batch-core/src/main/java/org/springframework/batch/core/listener/ChunkListener.java

@@ -16,6 +16,7 @@
 package org.springframework.batch.core.listener;
 
 import org.springframework.batch.core.scope.context.ChunkContext;
+import org.springframework.batch.item.Chunk;
 
 /**
  * Listener interface for the lifecycle of a chunk. A chunk can be thought of as a
@@ -30,24 +31,32 @@
  * @author Parikshit Dutta
  * @author Injae Kim
  */
-public interface ChunkListener extends StepListener {
+public interface ChunkListener<I, O> extends StepListener {
 
 	/**
 	 * The key for retrieving the rollback exception.
+	 * @deprecated since 6.0 with no replacement. Scheduled for removal in 6.2 or later.
 	 */
+	@Deprecated(since = "6.0", forRemoval = true)
 	String ROLLBACK_EXCEPTION_KEY = "sb_rollback_exception";
 
 	/**
 	 * Callback before the chunk is executed, but inside the transaction.
 	 * @param context The current {@link ChunkContext}
+	 * @deprecated since 6.0, use {@link #beforeChunk(Chunk)} instead. Scheduled for
+	 * removal in 6.2 or later.
 	 */
+	@Deprecated(since = "6.0", forRemoval = true)
 	default void beforeChunk(ChunkContext context) {
 	}
 
 	/**
 	 * Callback after the chunk is executed, outside the transaction.
 	 * @param context The current {@link ChunkContext}
+	 * @deprecated since 6.0, use {@link #afterChunk(Chunk)} instead. Scheduled for
+	 * removal in 6.2 or later.
 	 */
+	@Deprecated(since = "6.0", forRemoval = true)
 	default void afterChunk(ChunkContext context) {
 	}
 
@@ -61,8 +70,37 @@ default void afterChunk(ChunkContext context) {
 	 * from here.</em>
 	 * @param context the chunk context containing the exception that caused the
 	 * underlying rollback.
+	 * @deprecated since 6.0, use {@link #onChunkError(Exception,Chunk)} instead.
+	 * Scheduled for removal in 6.2 or later.
 	 */
+	@Deprecated(since = "6.0", forRemoval = true)
 	default void afterChunkError(ChunkContext context) {
 	}
 
+	/**
+	 * Callback before the chunk is processed, inside the transaction.
+	 * @since 6.0
+	 */
+	default void beforeChunk(Chunk<I> chunk) {
+	}
+
+	/**
+	 * Callback after the chunk is written, inside the transaction.
+	 * @since 6.0
+	 */
+	default void afterChunk(Chunk<O> chunk) {
+	}
+
+	/**
+	 * Callback if an exception occurs while processing or writing a chunk, inside the
+	 * transaction, which is about to be rolled back. <em>As a result, you should use
+	 * {@code PROPAGATION_REQUIRES_NEW} for any transactional operation that is called
+	 * here.</em>
+	 * @param exception the exception that caused the underlying rollback.
+	 * @param chunk the processed chunk
+	 * @since 6.0
+	 */
+	default void onChunkError(Exception exception, Chunk<O> chunk) {
+	}
+
 }

spring-batch-core/src/main/java/org/springframework/batch/core/listener/CompositeChunkListener.java

@@ -20,14 +20,15 @@
 import java.util.List;
 
 import org.springframework.batch.core.scope.context.ChunkContext;
+import org.springframework.batch.item.Chunk;
 import org.springframework.core.Ordered;
 
 /**
  * @author Lucas Ward
  * @author Mahmoud Ben Hassine
  *
  */
-public class CompositeChunkListener implements ChunkListener {
+public class CompositeChunkListener<I, O> implements ChunkListener<I, O> {
 
 	private final OrderedComposite<ChunkListener> listeners = new OrderedComposite<>();
 
@@ -74,7 +75,10 @@ public void register(ChunkListener chunkListener) {
 	 * Call the registered listeners in reverse order.
 	 *
 	 * @see ChunkListener#afterChunk(ChunkContext context)
+	 * @deprecated since 6.0, use {@link #afterChunk(Chunk)} instead. Scheduled for
+	 * removal in 6.2 or later.
 	 */
+	@Deprecated(since = "6.0", forRemoval = true)
 	@Override
 	public void afterChunk(ChunkContext context) {
 		for (Iterator<ChunkListener> iterator = listeners.reverse(); iterator.hasNext();) {
@@ -83,12 +87,28 @@ public void afterChunk(ChunkContext context) {
 		}
 	}
 
+	/**
+	 * Call the registered listeners in reverse order.
+	 *
+	 * @see ChunkListener#afterChunk(Chunk)
+	 */
+	@Override
+	public void afterChunk(Chunk chunk) {
+		for (Iterator<ChunkListener> iterator = listeners.reverse(); iterator.hasNext();) {
+			ChunkListener listener = iterator.next();
+			listener.afterChunk(chunk);
+		}
+	}
+
 	/**
 	 * Call the registered listeners in order, respecting and prioritizing those that
 	 * implement {@link Ordered}.
 	 *
 	 * @see ChunkListener#beforeChunk(ChunkContext context)
+	 * @deprecated since 6.0, use {@link #beforeChunk(Chunk)} instead. Scheduled for
+	 * removal in 6.2 or later.
 	 */
+	@Deprecated(since = "6.0", forRemoval = true)
 	@Override
 	public void beforeChunk(ChunkContext context) {
 		for (Iterator<ChunkListener> iterator = listeners.iterator(); iterator.hasNext();) {
@@ -97,11 +117,28 @@ public void beforeChunk(ChunkContext context) {
 		}
 	}
 
+	/**
+	 * Call the registered listeners in order, respecting and prioritizing those that
+	 * implement {@link Ordered}.
+	 *
+	 * @see ChunkListener#beforeChunk(Chunk chunk)
+	 */
+	@Override
+	public void beforeChunk(Chunk chunk) {
+		for (Iterator<ChunkListener> iterator = listeners.iterator(); iterator.hasNext();) {
+			ChunkListener listener = iterator.next();
+			listener.beforeChunk(chunk);
+		}
+	}
+
 	/**
 	 * Call the registered listeners in reverse order.
 	 *
 	 * @see ChunkListener#afterChunkError(ChunkContext context)
+	 * @deprecated since 6.0, use {@link #onChunkError(Exception,Chunk)} instead.
+	 * Scheduled for removal in 6.2 or later.
 	 */
+	@Deprecated(since = "6.0", forRemoval = true)
 	@Override
 	public void afterChunkError(ChunkContext context) {
 		for (Iterator<ChunkListener> iterator = listeners.reverse(); iterator.hasNext();) {
@@ -110,4 +147,17 @@ public void afterChunkError(ChunkContext context) {
 		}
 	}
 
+	/**
+	 * Call the registered listeners in reverse order.
+	 *
+	 * @see ChunkListener#onChunkError(Exception, Chunk)
+	 */
+	@Override
+	public void onChunkError(Exception exception, Chunk chunk) {
+		for (Iterator<ChunkListener> iterator = listeners.reverse(); iterator.hasNext();) {
+			ChunkListener listener = iterator.next();
+			listener.onChunkError(exception, chunk);
+		}
+	}
+
 }