diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkOrientedStep.java b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkOrientedStep.java index 57c470c2a1..1269da01e4 100644 --- a/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkOrientedStep.java +++ b/spring-batch-core/src/main/java/org/springframework/batch/core/step/item/ChunkOrientedStep.java @@ -90,6 +90,7 @@ * @author Mahmoud Ben Hassine * @author Andrey Litvitski * @author xeounxzxu + * @author Minchul Son * @since 6.0 */ public class ChunkOrientedStep extends AbstractStep { @@ -375,6 +376,11 @@ protected void doExecute(StepExecution stepExecution) throws Exception { // Skip update during rollback to avoid OptimisticLockingFailureException if (transactionStatus.isRollbackOnly()) { + // Explicitly mark as locally rollback-only to prevent + // UnexpectedRollbackException when the transaction manager + // (eg JpaTransactionManager) has marked it as globally rollback-only + // (eg after a JPA flush failure) but not locally rollback-only. + transactionStatus.setRollbackOnly(); chunkTransactionEvent.transactionStatus = BatchMetrics.STATUS_ROLLED_BACK; chunkTransactionEvent.commit(); return; @@ -408,17 +414,26 @@ private void processChunkConcurrently(TransactionStatus status, StepContribution try { if (tracker.isScanMode()) { logger.info("Executing scan in new transaction after rollback"); - Chunk pendingChunk = tracker.getPendingChunk(); - if (pendingChunk != null) { + O item = tracker.pollNextScanItem(); + if (item != null) { + Chunk singleItemChunk = new Chunk<>(item); ChunkScanEvent chunkScanEvent = new ChunkScanEvent(stepExecution.getStepName(), stepExecution.getId()); chunkScanEvent.begin(); - scan(pendingChunk, contribution); + compositeChunkListener.beforeChunk(singleItemChunk); + scan(singleItemChunk, contribution); + if (!status.isRollbackOnly()) { + compositeChunkListener.afterChunk(singleItemChunk); + } chunkScanEvent.skipCount = contribution.getSkipCount(); chunkScanEvent.commit(); + } + if (!tracker.hasPendingScanItems()) { logger.info("Chunk scan completed"); tracker.exitScanMode(); - stepExecution.incrementCommitCount(); + if (!status.isRollbackOnly()) { + stepExecution.incrementCommitCount(); + } } return; } @@ -486,19 +501,26 @@ private void processChunkSequentially(TransactionStatus status, StepContribution try { if (tracker.isScanMode()) { logger.info("Executing scan in new transaction after rollback"); - Chunk pendingChunk = tracker.getPendingChunk(); - if (pendingChunk != null) { + O item = tracker.pollNextScanItem(); + if (item != null) { + Chunk singleItemChunk = new Chunk<>(item); ChunkScanEvent chunkScanEvent = new ChunkScanEvent(stepExecution.getStepName(), stepExecution.getId()); chunkScanEvent.begin(); - compositeChunkListener.beforeChunk(new Chunk<>()); - scan(pendingChunk, contribution); - compositeChunkListener.afterChunk(pendingChunk); + compositeChunkListener.beforeChunk(singleItemChunk); + scan(singleItemChunk, contribution); + if (!status.isRollbackOnly()) { + compositeChunkListener.afterChunk(singleItemChunk); + } chunkScanEvent.skipCount = contribution.getSkipCount(); chunkScanEvent.commit(); + } + if (!tracker.hasPendingScanItems()) { logger.info("Chunk scan completed"); tracker.exitScanMode(); - stepExecution.incrementCommitCount(); + if (!status.isRollbackOnly()) { + stepExecution.incrementCommitCount(); + } } return; } @@ -838,12 +860,12 @@ static ChunkTracker create() { private boolean scanMode; - @Nullable private Chunk pendingChunk; + @Nullable private LinkedList pendingScanItems; void init() { this.moreItems = true; this.scanMode = false; - this.pendingChunk = null; + this.pendingScanItems = null; } void reset() { @@ -856,20 +878,24 @@ boolean moreItems() { void enterScanMode(Chunk chunk) { this.scanMode = true; - this.pendingChunk = new Chunk<>(chunk.getItems()); + this.pendingScanItems = new LinkedList<>(chunk.getItems()); } boolean isScanMode() { return this.scanMode; } - @Nullable Chunk getPendingChunk() { - return this.pendingChunk; + @Nullable O pollNextScanItem() { + return (this.pendingScanItems != null) ? this.pendingScanItems.poll() : null; + } + + boolean hasPendingScanItems() { + return this.pendingScanItems != null && !this.pendingScanItems.isEmpty(); } void exitScanMode() { this.scanMode = false; - this.pendingChunk = null; + this.pendingScanItems = null; } } diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ChunkOrientedStepScanModeIntegrationTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ChunkOrientedStepScanModeIntegrationTests.java index 0bb07daae2..978461e65b 100644 --- a/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ChunkOrientedStepScanModeIntegrationTests.java +++ b/spring-batch-core/src/test/java/org/springframework/batch/core/step/item/ChunkOrientedStepScanModeIntegrationTests.java @@ -17,8 +17,11 @@ import java.util.ArrayList; import java.util.List; +import java.util.Objects; import java.util.concurrent.CopyOnWriteArrayList; +import javax.sql.DataSource; + import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; @@ -33,6 +36,7 @@ import org.springframework.batch.core.step.StepExecution; import org.springframework.batch.core.step.builder.ChunkOrientedStepBuilder; import org.springframework.batch.core.step.skip.AlwaysSkipItemSkipPolicy; +import org.springframework.batch.infrastructure.item.ItemWriter; import org.springframework.batch.infrastructure.item.support.ListItemReader; import org.springframework.context.ApplicationContext; import org.springframework.context.annotation.AnnotationConfigApplicationContext; @@ -40,13 +44,17 @@ import org.springframework.context.annotation.Configuration; import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.datasource.ConnectionHolder; import org.springframework.jdbc.support.JdbcTransactionManager; import org.springframework.test.jdbc.JdbcTestUtils; +import org.springframework.transaction.support.TransactionSynchronization; +import org.springframework.transaction.support.TransactionSynchronizationManager; /** * Tests for scan mode functionality in {@link ChunkOrientedStep}. * * @author KMGeon + * @author MinChul Son */ class ChunkOrientedStepScanModeIntegrationTests { @@ -282,6 +290,265 @@ public Step step(JobRepository jobRepository, JdbcTransactionManager transaction } + // Issue https://github.com/spring-projects/spring-batch/issues/5377 + @Test + void testSkipPolicyWithJpaLikeRollbackOnlyBehaviorInSequentialMode() throws Exception { + // given + ApplicationContext context = new AnnotationConfigApplicationContext(TestConfiguration.class, + JpaLikeRollbackOnlySequentialStepConfiguration.class); + JobOperator jobOperator = context.getBean(JobOperator.class); + Job job = context.getBean(Job.class); + JdbcTemplate jdbcTemplate = context.getBean(JdbcTemplate.class); + + // when + JobParameters jobParameters = new JobParametersBuilder().toJobParameters(); + JobExecution jobExecution = jobOperator.start(job, jobParameters); + + // then + // Without the fix, item "3" would fail with "Transaction silently rolled back + // because it has been marked as rollback-only" (JPA-like behavior) causing the + // step to fail with a NonSkippableWriteException. With the fix, each scan item + // runs in its own transaction so item "3" succeeds. + Assertions.assertEquals(ExitStatus.COMPLETED.getExitCode(), jobExecution.getExitStatus().getExitCode()); + StepExecution stepExecution = jobExecution.getStepExecutions().iterator().next(); + Assertions.assertEquals(3, stepExecution.getReadCount()); + Assertions.assertEquals(2, stepExecution.getWriteCount()); + Assertions.assertEquals(1, stepExecution.getWriteSkipCount()); + Assertions.assertEquals(1, + jdbcTemplate.queryForObject("SELECT COUNT(*) FROM delivery WHERE item_number = '1'", Integer.class)); + Assertions.assertEquals(1, + jdbcTemplate.queryForObject("SELECT COUNT(*) FROM delivery WHERE item_number = '3'", Integer.class)); + Assertions.assertEquals(2, JdbcTestUtils.countRowsInTable(jdbcTemplate, "delivery")); + } + + // Issue https://github.com/spring-projects/spring-batch/issues/5377 + @Test + void testSkipPolicyWithJpaLikeRollbackOnlyBehaviorInConcurrentMode() throws Exception { + // given + ApplicationContext context = new AnnotationConfigApplicationContext(TestConfiguration.class, + JpaLikeRollbackOnlyConcurrentStepConfiguration.class); + JobOperator jobOperator = context.getBean(JobOperator.class); + Job job = context.getBean(Job.class); + JdbcTemplate jdbcTemplate = context.getBean(JdbcTemplate.class); + + // when + JobParameters jobParameters = new JobParametersBuilder().toJobParameters(); + JobExecution jobExecution = jobOperator.start(job, jobParameters); + + // then + Assertions.assertEquals(ExitStatus.COMPLETED.getExitCode(), jobExecution.getExitStatus().getExitCode()); + StepExecution stepExecution = jobExecution.getStepExecutions().iterator().next(); + Assertions.assertEquals(3, stepExecution.getReadCount()); + Assertions.assertEquals(2, stepExecution.getWriteCount()); + Assertions.assertEquals(1, stepExecution.getWriteSkipCount()); + Assertions.assertEquals(1, + jdbcTemplate.queryForObject("SELECT COUNT(*) FROM delivery WHERE item_number = '1'", Integer.class)); + Assertions.assertEquals(1, + jdbcTemplate.queryForObject("SELECT COUNT(*) FROM delivery WHERE item_number = '3'", Integer.class)); + Assertions.assertEquals(2, JdbcTestUtils.countRowsInTable(jdbcTemplate, "delivery")); + } + + /** + * Simulates JPA-like rollback-only behavior: when a write fails, the transaction is + * marked as rollback-only (as JPA/Hibernate does after a flush failure). Subsequent + * writes in the same transaction would then throw "Transaction silently rolled back + * because it has been marked as rollback-only". + */ + private static ItemWriter jpaLikeWriter(JdbcTemplate jdbcTemplate) { + ThreadLocal rollbackOnly = ThreadLocal.withInitial(() -> false); + return chunk -> { + // Simulate JPA behavior: if the current "session" is rollback-only, + // throw as JPA/Hibernate would before even attempting the write + if (rollbackOnly.get()) { + throw new RuntimeException( + "Transaction silently rolled back because it has been marked as rollback-only"); + } + for (String item : chunk) { + if ("2".equals(item)) { + // Simulate JPA flush failure: mark session rollback-only and register + // cleanup synchronization (reset after transaction completion) + rollbackOnly.set(true); + TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() { + @Override + public void afterCompletion(int status) { + rollbackOnly.set(false); + } + }); + throw new RuntimeException("Simulated JPA constraint violation for item: " + item); + } + jdbcTemplate.update("INSERT INTO delivery (item_number) VALUES (?)", item); + } + }; + } + + /** + * Simulates JPA-like global rollback-only behavior by marking the underlying + * {@link ConnectionHolder} as rollback-only when an item fails. This causes + * {@link org.springframework.transaction.support.AbstractPlatformTransactionManager} + * to detect a globally-rolled-back transaction during commit and, without the fix in + * {@code ChunkOrientedStep.doExecute()}, throw an + * {@link org.springframework.transaction.UnexpectedRollbackException}. The fix + * explicitly calls {@code transactionStatus.setRollbackOnly()} before returning from + * the lambda so that the local rollback-only flag is set and the transaction manager + * performs a clean rollback instead. + */ + private static ItemWriter globalRollbackOnlyWriter(JdbcTemplate jdbcTemplate) { + DataSource dataSource = Objects.requireNonNull(jdbcTemplate.getDataSource()); + return chunk -> { + for (String item : chunk) { + if ("2".equals(item)) { + // Simulate JPA/Hibernate flush failure: mark the underlying + // connection + // holder as rollback-only (global), exactly as Hibernate does via + // JdbcResourceLocalTransactionCoordinatorImpl after a flush error. + ConnectionHolder holder = (ConnectionHolder) TransactionSynchronizationManager + .getResource(dataSource); + if (holder != null) { + holder.setRollbackOnly(); + } + throw new RuntimeException("Simulated JPA flush failure for item: " + item); + } + jdbcTemplate.update("INSERT INTO delivery (item_number) VALUES (?)", item); + } + }; + } + + // Issue https://github.com/spring-projects/spring-batch/issues/5377 + @Test + void testUnexpectedRollbackExceptionPreventedInSequentialScanMode() throws Exception { + // given + ApplicationContext context = new AnnotationConfigApplicationContext(TestConfiguration.class, + GlobalRollbackOnlySequentialStepConfiguration.class); + JobOperator jobOperator = context.getBean(JobOperator.class); + Job job = context.getBean(Job.class); + JdbcTemplate jdbcTemplate = context.getBean(JdbcTemplate.class); + + // when + JobParameters jobParameters = new JobParametersBuilder().toJobParameters(); + JobExecution jobExecution = jobOperator.start(job, jobParameters); + + // then + // Without the fix in doExecute(), the scan transaction for item "2" is globally + // rollback-only (ConnectionHolder.setRollbackOnly()) but not locally + // rollback-only. + // AbstractPlatformTransactionManager.commit() would then call + // processRollback(defStatus, unexpected=true) and throw + // UnexpectedRollbackException, + // causing the step to fail. With the fix, doExecute() calls + // transactionStatus.setRollbackOnly() before returning from the lambda so that + // the transaction manager performs a clean rollback and the step completes + // normally. + Assertions.assertEquals(ExitStatus.COMPLETED.getExitCode(), jobExecution.getExitStatus().getExitCode()); + StepExecution stepExecution = jobExecution.getStepExecutions().iterator().next(); + Assertions.assertEquals(3, stepExecution.getReadCount()); + Assertions.assertEquals(2, stepExecution.getWriteCount()); + Assertions.assertEquals(1, stepExecution.getWriteSkipCount()); + Assertions.assertEquals(2, stepExecution.getRollbackCount()); + Assertions.assertEquals(1, + jdbcTemplate.queryForObject("SELECT COUNT(*) FROM delivery WHERE item_number = '1'", Integer.class)); + Assertions.assertEquals(1, + jdbcTemplate.queryForObject("SELECT COUNT(*) FROM delivery WHERE item_number = '3'", Integer.class)); + Assertions.assertEquals(2, JdbcTestUtils.countRowsInTable(jdbcTemplate, "delivery")); + } + + // Issue https://github.com/spring-projects/spring-batch/issues/5377 + @Test + void testUnexpectedRollbackExceptionPreventedInConcurrentScanMode() throws Exception { + // given + ApplicationContext context = new AnnotationConfigApplicationContext(TestConfiguration.class, + GlobalRollbackOnlyConcurrentStepConfiguration.class); + JobOperator jobOperator = context.getBean(JobOperator.class); + Job job = context.getBean(Job.class); + JdbcTemplate jdbcTemplate = context.getBean(JdbcTemplate.class); + + // when + JobParameters jobParameters = new JobParametersBuilder().toJobParameters(); + JobExecution jobExecution = jobOperator.start(job, jobParameters); + + // then + Assertions.assertEquals(ExitStatus.COMPLETED.getExitCode(), jobExecution.getExitStatus().getExitCode()); + StepExecution stepExecution = jobExecution.getStepExecutions().iterator().next(); + Assertions.assertEquals(3, stepExecution.getReadCount()); + Assertions.assertEquals(2, stepExecution.getWriteCount()); + Assertions.assertEquals(1, stepExecution.getWriteSkipCount()); + Assertions.assertEquals(2, stepExecution.getRollbackCount()); + Assertions.assertEquals(1, + jdbcTemplate.queryForObject("SELECT COUNT(*) FROM delivery WHERE item_number = '1'", Integer.class)); + Assertions.assertEquals(1, + jdbcTemplate.queryForObject("SELECT COUNT(*) FROM delivery WHERE item_number = '3'", Integer.class)); + Assertions.assertEquals(2, JdbcTestUtils.countRowsInTable(jdbcTemplate, "delivery")); + } + + @Configuration + static class GlobalRollbackOnlySequentialStepConfiguration { + + @Bean + public Step step(JobRepository jobRepository, JdbcTransactionManager transactionManager, + JdbcTemplate jdbcTemplate) { + List items = List.of("1", "2", "3"); + return new ChunkOrientedStepBuilder(jobRepository, 3).reader(new ListItemReader<>(items)) + .writer(globalRollbackOnlyWriter(jdbcTemplate)) + .transactionManager(transactionManager) + .faultTolerant() + .skipPolicy(new AlwaysSkipItemSkipPolicy()) + .build(); + } + + } + + @Configuration + static class GlobalRollbackOnlyConcurrentStepConfiguration { + + @Bean + public Step step(JobRepository jobRepository, JdbcTransactionManager transactionManager, + JdbcTemplate jdbcTemplate) { + List items = List.of("1", "2", "3"); + return new ChunkOrientedStepBuilder(jobRepository, 3).reader(new ListItemReader<>(items)) + .writer(globalRollbackOnlyWriter(jdbcTemplate)) + .transactionManager(transactionManager) + .taskExecutor(new SimpleAsyncTaskExecutor()) + .faultTolerant() + .skipPolicy(new AlwaysSkipItemSkipPolicy()) + .build(); + } + + } + + @Configuration + static class JpaLikeRollbackOnlySequentialStepConfiguration { + + @Bean + public Step step(JobRepository jobRepository, JdbcTransactionManager transactionManager, + JdbcTemplate jdbcTemplate) { + List items = List.of("1", "2", "3"); + return new ChunkOrientedStepBuilder(jobRepository, 3).reader(new ListItemReader<>(items)) + .writer(jpaLikeWriter(jdbcTemplate)) + .transactionManager(transactionManager) + .faultTolerant() + .skipPolicy(new AlwaysSkipItemSkipPolicy()) + .build(); + } + + } + + @Configuration + static class JpaLikeRollbackOnlyConcurrentStepConfiguration { + + @Bean + public Step step(JobRepository jobRepository, JdbcTransactionManager transactionManager, + JdbcTemplate jdbcTemplate) { + List items = List.of("1", "2", "3"); + return new ChunkOrientedStepBuilder(jobRepository, 3).reader(new ListItemReader<>(items)) + .writer(jpaLikeWriter(jdbcTemplate)) + .transactionManager(transactionManager) + .taskExecutor(new SimpleAsyncTaskExecutor()) + .faultTolerant() + .skipPolicy(new AlwaysSkipItemSkipPolicy()) + .build(); + } + + } + @Configuration static class TrackingWriterStepConfiguration {