diff --git a/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepIntegrationTests.java b/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepIntegrationTests.java index bf74493b2..0cfa99307 100644 --- a/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepIntegrationTests.java +++ b/spring-batch-core-tests/src/test/java/org/springframework/batch/core/test/step/FaultTolerantStepIntegrationTests.java @@ -1,3 +1,18 @@ +/* + * Copyright 2010-2019 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.test.step; import java.util.ArrayList; @@ -17,6 +32,7 @@ import org.springframework.batch.core.StepExecution; import org.springframework.batch.core.configuration.annotation.StepBuilderFactory; import org.springframework.batch.core.repository.JobRepository; import org.springframework.batch.core.step.builder.FaultTolerantStepBuilder; +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.ItemProcessor; @@ -195,6 +211,53 @@ public class FaultTolerantStepIntegrationTests { assertEquals(1, stepExecution.getProcessSkipCount()); } + @Test(timeout = 3000) + public void testExceptionInProcessAndWriteDuringChunkScan() throws Exception { + // Given + ListItemReader itemReader = new ListItemReader<>(Arrays.asList(1, 2, 3)); + + ItemProcessor itemProcessor = new ItemProcessor() { + @Override + public Integer process(Integer item) throws Exception { + if (item.equals(2)) { + throw new Exception("Error during process item " + item); + } + return item; + } + }; + + ItemWriter itemWriter = new ItemWriter() { + @Override + public void write(List items) throws Exception { + if (items.contains(3)) { + throw new Exception("Error during write"); + } + } + }; + + Step step = new StepBuilderFactory(jobRepository, transactionManager).get("step") + .chunk(5) + .reader(itemReader) + .processor(itemProcessor) + .writer(itemWriter) + .faultTolerant() + .skipPolicy(new AlwaysSkipItemSkipPolicy()) + .build(); + + // When + StepExecution stepExecution = execute(step); + + // Then + assertEquals(BatchStatus.COMPLETED, stepExecution.getStatus()); + assertEquals(ExitStatus.COMPLETED, stepExecution.getExitStatus()); + assertEquals(3, stepExecution.getReadCount()); + assertEquals(1, stepExecution.getWriteCount()); + assertEquals(1, stepExecution.getWriteSkipCount()); + assertEquals(1, stepExecution.getProcessSkipCount()); + assertEquals(3, stepExecution.getRollbackCount()); + assertEquals(2, stepExecution.getCommitCount()); + } + private List createItems() { List items = new ArrayList<>(TOTAL_ITEMS); for (int i = 1; i <= TOTAL_ITEMS; i++) { 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 7d31bd018..644d70c93 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-2018 the original author or authors. + * Copyright 2006-2019 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. @@ -580,8 +580,7 @@ public class FaultTolerantChunkProcessor extends SimpleChunkProcessor.ChunkIterator inputIterator = inputs.iterator(); Chunk.ChunkIterator outputIterator = outputs.iterator(); - //BATCH-2442 : do not scan skipped items - if (!inputs.getSkips().isEmpty()) { + if (!inputs.getSkips().isEmpty() && inputs.getItems().size() != outputs.getItems().size()) { if (outputIterator.hasNext()) { outputIterator.remove(); return;