From b3cc665ad3080b1fa5134d5937f7b2054fabb02f Mon Sep 17 00:00:00 2001 From: Mustafa Yanar Date: Mon, 4 Mar 2024 22:07:58 +0300 Subject: [PATCH] Fix output chunk end property in ChunkProcessor implementations Resolves #4560 --- .../step/item/FaultTolerantChunkProcessor.java | 6 ++++-- .../batch/core/step/item/SimpleChunkProcessor.java | 5 ++++- .../item/FaultTolerantChunkProcessorTests.java | 13 ++++++++++++- .../core/step/item/SimpleChunkProcessorTests.java | 14 +++++++++++++- .../java/org/springframework/batch/item/Chunk.java | 9 ++++++++- 5 files changed, 41 insertions(+), 6 deletions(-) diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java index f1eb7321f5..ecb797111c 100755 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessor.java +++ b/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 @@ else if (shouldSkip(itemProcessSkipPolicy, e, contribution.getStepSkipCount())) break; } } - + if (inputs.isEnd()) { + outputs.setEnd(); + } return outputs; } diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkProcessor.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkProcessor.java index 813b6eb403..101945bb22 100755 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/SimpleChunkProcessor.java +++ b/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 @@ protected Chunk transform(StepContribution contribution, Chunk inputs) thr iterator.remove(); } } + if (inputs.isEnd()) { + outputs.setEnd(); + } return outputs; } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessorTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessorTests.java index 5070b277a4..d30e06917e 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/FaultTolerantChunkProcessorTests.java +++ b/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 @@ 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 @@ public String process(String item) throws Exception { assertEquals(1, contribution.getFilterCount()); } + @Test + void testTransformChunkEnd() throws Exception { + Chunk inputs = new Chunk<>(Arrays.asList("1", "2")); + inputs.setEnd(); + processor.initializeUserData(inputs); + Chunk outputs = processor.transform(contribution, inputs); + assertEquals(Arrays.asList("1", "2"), outputs.getItems()); + assertTrue(outputs.isEnd()); + } + @Test void testFilterCountOnSkip() throws Exception { processor.setProcessSkipPolicy(new AlwaysSkipItemSkipPolicy()); diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleChunkProcessorTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleChunkProcessorTests.java index e9a7e0e678..5ebcb49ced 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/SimpleChunkProcessorTests.java +++ b/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 @@ void testProcess() throws Exception { assertEquals(2, contribution.getWriteCount()); } + @Test + void testTransform() throws Exception { + Chunk inputs = new Chunk<>(); + inputs.add("foo"); + inputs.add("bar"); + inputs.setEnd(); + Chunk outputs = processor.transform(contribution, inputs); + assertEquals(Arrays.asList("foo", "bar"), outputs.getItems()); + assertTrue(outputs.isEnd()); + } + } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/Chunk.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/Chunk.java index 4cdfa1c7e6..aababa36c2 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/Chunk.java +++ b/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 int size() { /** * Flag to indicate if the source data is exhausted. + * + *

+ * 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. + *

* @return true if there is no more data to process */ public boolean isEnd() {