Apply optimistic locking for MongoDB repository

Signed-off-by: Yanming Zhou <[email protected]>

Author: quaff

spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/AbstractJdbcBatchMetadataDao.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2006-2023 the original author or authors.
+ * Copyright 2006-2025 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.
@@ -20,6 +20,7 @@
 
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.jdbc.core.JdbcOperations;
+import org.springframework.jdbc.core.simple.JdbcClient;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
@@ -29,6 +30,7 @@
  *
  * @author Robert Kasanicky
  * @author Mahmoud Ben Hassine
+ * @author Yanming Zhou
  */
 public abstract class AbstractJdbcBatchMetadataDao implements InitializingBean {
 
@@ -47,6 +49,8 @@ public abstract class AbstractJdbcBatchMetadataDao implements InitializingBean {
 
 	private JdbcOperations jdbcTemplate;
 
+	private JdbcClient jdbcClient;
+
 	protected String getQuery(String base) {
 		return StringUtils.replace(base, "%PREFIX%", tablePrefix);
 	}
@@ -66,12 +70,17 @@ public void setTablePrefix(String tablePrefix) {
 
 	public void setJdbcTemplate(JdbcOperations jdbcTemplate) {
 		this.jdbcTemplate = jdbcTemplate;
+		this.jdbcClient = JdbcClient.create(jdbcTemplate);
 	}
 
 	protected JdbcOperations getJdbcTemplate() {
 		return jdbcTemplate;
 	}
 
+	protected JdbcClient getJdbcClient() {
+		return jdbcClient;
+	}
+
 	public int getClobTypeToUse() {
 		return clobTypeToUse;
 	}
@@ -83,6 +92,7 @@ public void setClobTypeToUse(int clobTypeToUse) {
 	@Override
 	public void afterPropertiesSet() throws Exception {
 		Assert.state(jdbcTemplate != null, "JdbcOperations is required");
+		Assert.state(jdbcClient != null, "JdbcClient is required");
 	}
 
 }

spring-batch-core/src/main/java/org/springframework/batch/core/repository/dao/jdbc/JdbcExecutionContextDao.java

@@ -24,6 +24,7 @@
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
+import java.sql.Types;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.Iterator;
@@ -43,6 +44,7 @@
 import org.springframework.core.serializer.Serializer;
 import org.springframework.jdbc.core.BatchPreparedStatementSetter;
 import org.springframework.jdbc.core.RowMapper;
+import org.springframework.jdbc.core.simple.JdbcClient;
 import org.springframework.lang.NonNull;
 import org.springframework.util.Assert;
 
@@ -65,7 +67,7 @@ public class JdbcExecutionContextDao extends AbstractJdbcBatchMetadataDao implem
 	private static final String FIND_JOB_EXECUTION_CONTEXT = """
 			SELECT SHORT_CONTEXT, SERIALIZED_CONTEXT
 			FROM %PREFIX%JOB_EXECUTION_CONTEXT
-			WHERE JOB_EXECUTION_ID = ?
+			WHERE JOB_EXECUTION_ID = :executionId
 			""";
 
 	private static final String INSERT_JOB_EXECUTION_CONTEXT = """
@@ -82,7 +84,7 @@ public class JdbcExecutionContextDao extends AbstractJdbcBatchMetadataDao implem
 	private static final String FIND_STEP_EXECUTION_CONTEXT = """
 			SELECT SHORT_CONTEXT, SERIALIZED_CONTEXT
 			FROM %PREFIX%STEP_EXECUTION_CONTEXT
-			WHERE STEP_EXECUTION_ID = ?
+			WHERE STEP_EXECUTION_ID = :executionId
 			""";
 
 	private static final String INSERT_STEP_EXECUTION_CONTEXT = """
@@ -98,12 +100,12 @@ public class JdbcExecutionContextDao extends AbstractJdbcBatchMetadataDao implem
 
 	private static final String DELETE_STEP_EXECUTION_CONTEXT = """
 			DELETE FROM %PREFIX%STEP_EXECUTION_CONTEXT
-			WHERE STEP_EXECUTION_ID = ?
+			WHERE STEP_EXECUTION_ID = :executionId
 			""";
 
 	private static final String DELETE_JOB_EXECUTION_CONTEXT = """
 			DELETE FROM %PREFIX%JOB_EXECUTION_CONTEXT
-			WHERE JOB_EXECUTION_ID = ?
+			WHERE JOB_EXECUTION_ID = :executionId
 			""";
 
 	private Charset charset = StandardCharsets.UTF_8;
@@ -154,8 +156,10 @@ public ExecutionContext getExecutionContext(JobExecution jobExecution) {
 		Long executionId = jobExecution.getId();
 		Assert.notNull(executionId, "ExecutionId must not be null.");
 
-		try (Stream<ExecutionContext> stream = getJdbcTemplate().queryForStream(getQuery(FIND_JOB_EXECUTION_CONTEXT),
-				new ExecutionContextRowMapper(), executionId)) {
+		try (Stream<ExecutionContext> stream = getJdbcClient().sql(getQuery(FIND_JOB_EXECUTION_CONTEXT))
+			.param("executionId", executionId)
+			.query(new ExecutionContextRowMapper())
+			.stream()) {
 			return stream.findFirst().orElseGet(ExecutionContext::new);
 		}
 	}
@@ -165,8 +169,10 @@ public ExecutionContext getExecutionContext(StepExecution stepExecution) {
 		Long executionId = stepExecution.getId();
 		Assert.notNull(executionId, "ExecutionId must not be null.");
 
-		try (Stream<ExecutionContext> stream = getJdbcTemplate().queryForStream(getQuery(FIND_STEP_EXECUTION_CONTEXT),
-				new ExecutionContextRowMapper(), executionId)) {
+		try (Stream<ExecutionContext> stream = getJdbcClient().sql(getQuery(FIND_STEP_EXECUTION_CONTEXT))
+			.param("executionId", executionId)
+			.query(new ExecutionContextRowMapper())
+			.stream()) {
 			return stream.findFirst().orElseGet(ExecutionContext::new);
 		}
 	}
@@ -248,7 +254,7 @@ public void saveExecutionContexts(Collection<StepExecution> stepExecutions) {
 	 */
 	@Override
 	public void deleteExecutionContext(JobExecution jobExecution) {
-		getJdbcTemplate().update(getQuery(DELETE_JOB_EXECUTION_CONTEXT), jobExecution.getId());
+		getJdbcClient().sql(getQuery(DELETE_JOB_EXECUTION_CONTEXT)).param("executionId", jobExecution.getId()).update();
 	}
 
 	/**
@@ -257,7 +263,9 @@ public void deleteExecutionContext(JobExecution jobExecution) {
 	 */
 	@Override
 	public void deleteExecutionContext(StepExecution stepExecution) {
-		getJdbcTemplate().update(getQuery(DELETE_STEP_EXECUTION_CONTEXT), stepExecution.getId());
+		getJdbcClient().sql(getQuery(DELETE_STEP_EXECUTION_CONTEXT))
+			.param("executionId", stepExecution.getId())
+			.update();
 	}
 
 	@Override
@@ -286,16 +294,13 @@ private void persistSerializedContext(final Long executionId, String serializedC
 			longContext = null;
 		}
 
-		getJdbcTemplate().update(getQuery(sql), ps -> {
-			ps.setString(1, shortContext);
-			if (longContext != null) {
-				ps.setString(2, longContext);
-			}
-			else {
-				ps.setNull(2, getClobTypeToUse());
-			}
-			ps.setLong(3, executionId);
-		});
+		getJdbcClient().sql(getQuery(sql))
+		// @formatter:off
+				.param(1, shortContext, Types.VARCHAR)
+				.param(2, longContext, getClobTypeToUse())
+				.param(3, executionId, Types.BIGINT)
+		// @formatter:on
+			.update();
 	}
 
 	/**