format code

Author: pstetsuk-exadel

spring-batch-core/src/main/java/org/springframework/batch/core/BatchStatus.java

@@ -73,7 +73,7 @@ public enum BatchStatus {
 	 */
 	UNKNOWN;
 
-    public static final Set<BatchStatus> RUNNING_STATUSES = Set.of(STARTING, STARTED, STOPPING);
+	public static final Set<BatchStatus> RUNNING_STATUSES = Set.of(STARTING, STARTED, STOPPING);
 
 	/**
 	 * Convenience method to return the higher value status of the statuses passed to the

spring-batch-core/src/main/java/org/springframework/batch/core/explore/JobExplorer.java

@@ -84,13 +84,13 @@ default JobInstance getLastJobInstance(String jobName) {
 	@Nullable
 	StepExecution getStepExecution(@Nullable Long jobExecutionId, @Nullable Long stepExecutionId);
 
-    /**
-     * Find {@link StepExecution}s by IDs and parent {@link JobExecution} ID
-     * @param jobExecutionId given job execution id
-     * @param stepExecutionIds given step execution ids
-     * @return collection of {@link StepExecution}
-     */
-    Set<StepExecution> getStepExecutions(Long jobExecutionId, Set<Long> stepExecutionIds);
+	/**
+	 * Find {@link StepExecution}s by IDs and parent {@link JobExecution} ID
+	 * @param jobExecutionId given job execution id
+	 * @param stepExecutionIds given step execution ids
+	 * @return collection of {@link StepExecution}
+	 */
+	Set<StepExecution> getStepExecutions(Long jobExecutionId, Set<Long> stepExecutionIds);
 
 	/**
 	 * @param instanceId {@link Long} The ID for the {@link JobInstance} to obtain.
@@ -175,11 +175,13 @@ default JobExecution getLastJobExecution(JobInstance jobInstance) {
 	 */
 	long getJobInstanceCount(@Nullable String jobName) throws NoSuchJobException;
 
-    /**
-     * Retrieve number of step executions that match the step execution ids and the batch statuses
-     * @param stepExecutionIds given step execution ids
-     * @param matchingBatchStatuses given batch statuses to match against
-     * @return number of {@link StepExecution} matching the criteria
-     */
-    long getStepExecutionCount(Set<Long> stepExecutionIds, Set<BatchStatus> matchingBatchStatuses);
+	/**
+	 * Retrieve number of step executions that match the step execution ids and the batch
+	 * statuses
+	 * @param stepExecutionIds given step execution ids
+	 * @param matchingBatchStatuses given batch statuses to match against
+	 * @return number of {@link StepExecution} matching the criteria
+	 */
+	long getStepExecutionCount(Set<Long> stepExecutionIds, Set<BatchStatus> matchingBatchStatuses);
+
 }

spring-batch-core/src/main/java/org/springframework/batch/core/explore/support/SimpleJobExplorer.java

@@ -144,20 +144,20 @@ public StepExecution getStepExecution(@Nullable Long jobExecutionId, @Nullable L
 		return stepExecution;
 	}
 
-    @Nullable
-    @Override
-    public Set<StepExecution> getStepExecutions(Long jobExecutionId, Set<Long> stepExecutionIds) {
-        JobExecution jobExecution = jobExecutionDao.getJobExecution(jobExecutionId);
-        if (jobExecution == null) {
-            return null;
-        }
-        getJobExecutionDependencies(jobExecution);
-        Set<StepExecution> stepExecutions = stepExecutionDao.getStepExecutions(jobExecution, stepExecutionIds);
-        stepExecutions.forEach(this::getStepExecutionDependencies);
-        return stepExecutions;
-    }
-
-    @Nullable
+	@Nullable
+	@Override
+	public Set<StepExecution> getStepExecutions(Long jobExecutionId, Set<Long> stepExecutionIds) {
+		JobExecution jobExecution = jobExecutionDao.getJobExecution(jobExecutionId);
+		if (jobExecution == null) {
+			return null;
+		}
+		getJobExecutionDependencies(jobExecution);
+		Set<StepExecution> stepExecutions = stepExecutionDao.getStepExecutions(jobExecution, stepExecutionIds);
+		stepExecutions.forEach(this::getStepExecutionDependencies);
+		return stepExecutions;
+	}
+
+	@Nullable
 	@Override
 	public JobInstance getJobInstance(@Nullable Long instanceId) {
 		return jobInstanceDao.getJobInstance(instanceId);
@@ -190,13 +190,13 @@ public long getJobInstanceCount(@Nullable String jobName) throws NoSuchJobExcept
 		return jobInstanceDao.getJobInstanceCount(jobName);
 	}
 
-    @Override
-    public long getStepExecutionCount(Set<Long> stepExecutionIds, Set<BatchStatus> matchingBatchStatuses) {
-        if (stepExecutionIds.isEmpty() || matchingBatchStatuses.isEmpty()) {
-            return 0;
-        }
-        return stepExecutionDao.countStepExecutions(stepExecutionIds, matchingBatchStatuses);
-    }
+	@Override
+	public long getStepExecutionCount(Set<Long> stepExecutionIds, Set<BatchStatus> matchingBatchStatuses) {
+		if (stepExecutionIds.isEmpty() || matchingBatchStatuses.isEmpty()) {
+			return 0;
+		}
+		return stepExecutionDao.countStepExecutions(stepExecutionIds, matchingBatchStatuses);
+	}
 
 	/**
 	 * @return instance of {@link JobInstanceDao}.

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

@@ -98,15 +98,14 @@ public void afterPropertiesSet() throws Exception {
 
 	/**
 	 * Replaces a given placeholder with a number of parameters (i.e. "?").
-	 *
 	 * @param sqlTemplate given sql template
 	 * @param placeholder placeholder that is being used for parameters
 	 * @param parameters collection of parameters with variable size
-	 *
 	 * @return sql query replaced with a number of parameters
 	 */
 	private static String createParameterizedQuery(String sqlTemplate, String placeholder, Collection<?> parameters) {
 		String params = parameters.stream().map(p -> "?").collect(Collectors.joining(", "));
 		return sqlTemplate.replace(placeholder, params);
 	}
+
 }

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

@@ -95,16 +95,17 @@ public class JdbcStepExecutionDao extends AbstractJdbcBatchMetadataDao implement
 
 	private static final String GET_STEP_EXECUTION = GET_RAW_STEP_EXECUTIONS + " AND STEP_EXECUTION_ID = ?";
 
-    private static final String GET_STEP_EXECUTIONS_BY_IDS = GET_RAW_STEP_EXECUTIONS + " and STEP_EXECUTION_ID IN (%STEP_EXECUTION_IDS%)";
+	private static final String GET_STEP_EXECUTIONS_BY_IDS = GET_RAW_STEP_EXECUTIONS
+			+ " and STEP_EXECUTION_ID IN (%STEP_EXECUTION_IDS%)";
 
-    private static final String COUNT_STEP_EXECUTIONS_BY_IDS_AND_STATUSES = """
+	private static final String COUNT_STEP_EXECUTIONS_BY_IDS_AND_STATUSES = """
 			SELECT COUNT(*)
-		    FROM %PREFIX%STEP_EXECUTION SE
-		    WHERE SE.STEP_EXECUTION_ID IN (%STEP_EXECUTION_IDS%)
-		    AND SE.STATUS IN (%STEP_STATUSES%)
+			   FROM %PREFIX%STEP_EXECUTION SE
+			   WHERE SE.STEP_EXECUTION_ID IN (%STEP_EXECUTION_IDS%)
+			   AND SE.STATUS IN (%STEP_STATUSES%)
 			""";
 
-    private static final String GET_LAST_STEP_EXECUTION = """
+	private static final String GET_LAST_STEP_EXECUTION = """
 			SELECT SE.STEP_EXECUTION_ID, SE.STEP_NAME, SE.START_TIME, SE.END_TIME, SE.STATUS, SE.COMMIT_COUNT, SE.READ_COUNT, SE.FILTER_COUNT, SE.WRITE_COUNT, SE.EXIT_CODE, SE.EXIT_MESSAGE, SE.READ_SKIP_COUNT, SE.WRITE_SKIP_COUNT, SE.PROCESS_SKIP_COUNT, SE.ROLLBACK_COUNT, SE.LAST_UPDATED, SE.VERSION, SE.CREATE_TIME, JE.JOB_EXECUTION_ID, JE.START_TIME, JE.END_TIME, JE.STATUS, JE.EXIT_CODE, JE.EXIT_MESSAGE, JE.CREATE_TIME, JE.LAST_UPDATED, JE.VERSION
 			FROM %PREFIX%JOB_EXECUTION JE
 				JOIN %PREFIX%STEP_EXECUTION SE ON SE.JOB_EXECUTION_ID = JE.JOB_EXECUTION_ID
@@ -393,12 +394,11 @@ public long countStepExecutions(JobInstance jobInstance, String stepName) {
 	@Override
 	public long countStepExecutions(Collection<Long> stepExecutionIds, Collection<BatchStatus> matchingBatchStatuses) {
 		return getJdbcTemplate().queryForObject(
-				getQuery(
-						COUNT_STEP_EXECUTIONS_BY_IDS_AND_STATUSES,
-						Map.of("%STEP_EXECUTION_IDS%", stepExecutionIds, "%STEP_STATUSES%", matchingBatchStatuses)
-				),
+				getQuery(COUNT_STEP_EXECUTIONS_BY_IDS_AND_STATUSES,
+						Map.of("%STEP_EXECUTION_IDS%", stepExecutionIds, "%STEP_STATUSES%", matchingBatchStatuses)),
 				Long.class,
-				Stream.concat(stepExecutionIds.stream(), matchingBatchStatuses.stream().map(BatchStatus::name)).toArray(Object[]::new));
+				Stream.concat(stepExecutionIds.stream(), matchingBatchStatuses.stream().map(BatchStatus::name))
+					.toArray(Object[]::new));
 	}
 
 	/**

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

@@ -64,13 +64,14 @@ public interface StepExecutionDao {
 	@Nullable
 	StepExecution getStepExecution(JobExecution jobExecution, Long stepExecutionId);
 
-    /**
-     * Get a collection of {@link StepExecution} matching job execution and step execution ids.
-     * @param jobExecution the parent job execution
-     * @param stepExecutionIds the step execution ids
-     * @return collection of {@link StepExecution}
-     */
-    Set<StepExecution> getStepExecutions(JobExecution jobExecution, Set<Long> stepExecutionIds);
+	/**
+	 * Get a collection of {@link StepExecution} matching job execution and step execution
+	 * ids.
+	 * @param jobExecution the parent job execution
+	 * @param stepExecutionIds the step execution ids
+	 * @return collection of {@link StepExecution}
+	 */
+	Set<StepExecution> getStepExecutions(JobExecution jobExecution, Set<Long> stepExecutionIds);
 
 	/**
 	 * Retrieve the last {@link StepExecution} for a given {@link JobInstance} ordered by
@@ -101,15 +102,16 @@ default long countStepExecutions(JobInstance jobInstance, String stepName) {
 		throw new UnsupportedOperationException();
 	}
 
-    /**
-     * Count {@link StepExecution} that match the ids and statuses of them - avoid loading them into memory
-     * @param stepExecutionIds given step execution ids
-     * @param matchingBatchStatuses
-     * @return the count of matching steps
-     */
-    long countStepExecutions(Collection<Long> stepExecutionIds, Collection<BatchStatus> matchingBatchStatuses);
+	/**
+	 * Count {@link StepExecution} that match the ids and statuses of them - avoid loading
+	 * them into memory
+	 * @param stepExecutionIds given step execution ids
+	 * @param matchingBatchStatuses
+	 * @return the count of matching steps
+	 */
+	long countStepExecutions(Collection<Long> stepExecutionIds, Collection<BatchStatus> matchingBatchStatuses);
 
-    /**
+	/**
 	 * Delete the given step execution.
 	 * @param stepExecution the step execution to delete
 	 * @since 5.0

spring-batch-core/src/test/java/org/springframework/batch/core/launch/support/CommandLineJobRunnerTests.java

@@ -582,6 +582,7 @@ public long getJobInstanceCount(@Nullable String jobName) throws NoSuchJobExcept
 		public long getStepExecutionCount(Set<Long> stepExecutionIds, Set<BatchStatus> matchingBatchStatuses) {
 			throw new UnsupportedOperationException();
 		}
+
 	}
 
 	public static class StubJobParametersConverter implements JobParametersConverter {

spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/MessageChannelPartitionHandler.java

@@ -251,18 +251,21 @@ protected Set<StepExecution> doHandle(StepExecution managerStepExecution,
 
 	private Set<StepExecution> pollReplies(final StepExecution managerStepExecution, final Set<StepExecution> split)
 			throws Exception {
-        Callable<Set<StepExecution>> callback = () -> {
-            Set<Long> currentStepExecutionIds = split.stream().map(StepExecution::getId).collect(Collectors.toSet());
-            long runningStepExecutions = jobExplorer.getStepExecutionCount(currentStepExecutionIds, BatchStatus.RUNNING_STATUSES);
-            if(runningStepExecutions > 0 && !split.isEmpty()) {
-                if(logger.isDebugEnabled()) {
-                    logger.debug(String.format("Currently waiting on %s out of %s partitions to finish", runningStepExecutions, split.size()));
-                }
-                return null;
-            } else {
-                return jobExplorer.getStepExecutions(managerStepExecution.getJobExecutionId(), currentStepExecutionIds);
-            }
-        };
+		Callable<Set<StepExecution>> callback = () -> {
+			Set<Long> currentStepExecutionIds = split.stream().map(StepExecution::getId).collect(Collectors.toSet());
+			long runningStepExecutions = jobExplorer.getStepExecutionCount(currentStepExecutionIds,
+					BatchStatus.RUNNING_STATUSES);
+			if (runningStepExecutions > 0 && !split.isEmpty()) {
+				if (logger.isDebugEnabled()) {
+					logger.debug(String.format("Currently waiting on %s out of %s partitions to finish",
+							runningStepExecutions, split.size()));
+				}
+				return null;
+			}
+			else {
+				return jobExplorer.getStepExecutions(managerStepExecution.getJobExecutionId(), currentStepExecutionIds);
+			}
+		};
 
 		Poller<Set<StepExecution>> poller = new DirectPoller<>(pollInterval);
 		Future<Set<StepExecution>> resultsFuture = poller.poll(callback);