Add ability to customize the job parameters converter in the default batch configuration

Before this commit, it was impossible to customize the job parameters converter
without overriding the entire`jobOperator()` method.

This commit makes it possible to override `getJobParametersConverter()`
or define a bean of type `JobParametersConverter` to customize the job
parameters converter used by the job operator.

Resolves #4650

Author: fmbenhassine

spring-batch-core/src/main/java/org/springframework/batch/core/configuration/annotation/BatchRegistrar.java

@@ -257,6 +257,12 @@ private void registerJobOperator(BeanDefinitionRegistry registry, EnableBatchPro
 		beanDefinitionBuilder.addPropertyReference("jobExplorer", "jobExplorer");
 		beanDefinitionBuilder.addPropertyReference("jobRegistry", "jobRegistry");
 
+		// set optional properties
+		String jobParametersConverterRef = batchAnnotation.jobParametersConverterRef();
+		if (registry.containsBeanDefinition(jobParametersConverterRef)) {
+			beanDefinitionBuilder.addPropertyReference("jobParametersConverter", jobParametersConverterRef);
+		}
+
 		registry.registerBeanDefinition("jobOperator", beanDefinitionBuilder.getBeanDefinition());
 	}
 

spring-batch-core/src/main/java/org/springframework/batch/core/configuration/annotation/EnableBatchProcessing.java

@@ -28,6 +28,7 @@
 import org.springframework.batch.core.configuration.support.ApplicationContextFactory;
 import org.springframework.batch.core.configuration.support.AutomaticJobRegistrar;
 import org.springframework.batch.core.configuration.support.ScopeConfiguration;
+import org.springframework.batch.core.converter.JobParametersConverter;
 import org.springframework.batch.core.launch.JobLauncher;
 import org.springframework.batch.core.launch.support.TaskExecutorJobLauncher;
 import org.springframework.batch.core.repository.JobRepository;
@@ -272,4 +273,11 @@
 	 */
 	String conversionServiceRef() default "conversionService";
 
+	/**
+	 * Set the {@link JobParametersConverter} to use in the job operator.
+	 * @return the bean name of the job parameters converter to use. Defaults to
+	 * {@literal jobParametersConverter}
+	 */
+	String jobParametersConverterRef() default "jobParametersConverter";
+
 }

spring-batch-core/src/main/java/org/springframework/batch/core/configuration/support/DefaultBatchConfiguration.java

@@ -27,6 +27,8 @@
 import org.springframework.batch.core.configuration.BatchConfigurationException;
 import org.springframework.batch.core.configuration.JobRegistry;
 import org.springframework.batch.core.converter.DateToStringConverter;
+import org.springframework.batch.core.converter.DefaultJobParametersConverter;
+import org.springframework.batch.core.converter.JobParametersConverter;
 import org.springframework.batch.core.converter.LocalDateTimeToStringConverter;
 import org.springframework.batch.core.converter.LocalDateToStringConverter;
 import org.springframework.batch.core.converter.LocalTimeToStringConverter;
@@ -237,6 +239,7 @@ public JobOperator jobOperator(JobRepository jobRepository, JobExplorer jobExplo
 		jobOperatorFactoryBean.setJobExplorer(jobExplorer);
 		jobOperatorFactoryBean.setJobRegistry(jobRegistry);
 		jobOperatorFactoryBean.setJobLauncher(jobLauncher);
+		jobOperatorFactoryBean.setJobParametersConverter(getJobParametersConverter());
 		try {
 			jobOperatorFactoryBean.afterPropertiesSet();
 			return jobOperatorFactoryBean.getObject();
@@ -465,6 +468,15 @@ protected TaskExecutor getTaskExecutor() {
 		return new SyncTaskExecutor();
 	}
 
+	/**
+	 * Return the {@link JobParametersConverter} to use in the job operator. Defaults to
+	 * {@link DefaultJobParametersConverter}
+	 * @return the {@link JobParametersConverter} to use in the job operator.
+	 */
+	protected JobParametersConverter getJobParametersConverter() {
+		return new DefaultJobParametersConverter();
+	}
+
 	/**
 	 * Return the conversion service to use in the job repository and job explorer. This
 	 * service is used to convert job parameters from String literal to typed values and

spring-batch-core/src/test/java/org/springframework/batch/core/configuration/annotation/BatchRegistrarTests.java

@@ -28,6 +28,9 @@
 import org.springframework.batch.core.JobKeyGenerator;
 import org.springframework.batch.core.configuration.JobRegistry;
 import org.springframework.batch.core.configuration.support.JobRegistrySmartInitializingSingleton;
+import org.springframework.batch.core.converter.DefaultJobParametersConverter;
+import org.springframework.batch.core.converter.JobParametersConverter;
+import org.springframework.batch.core.converter.JsonJobParametersConverter;
 import org.springframework.batch.core.explore.JobExplorer;
 import org.springframework.batch.core.launch.JobLauncher;
 import org.springframework.batch.core.launch.JobOperator;
@@ -204,6 +207,31 @@ public void testCustomJobKeyGeneratorConfiguration() {
 				jobKeyGenerator.getClass());
 	}
 
+	@Test
+	@DisplayName("When no JobParametersConverter is provided the default implementation should be used")
+	public void testDefaultJobParametersConverterConfiguration() {
+		AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(JobConfiguration.class);
+
+		JobOperator jobOperator = context.getBean(JobOperator.class);
+		JobParametersConverter jobParametersConverter = (JobParametersConverter) ReflectionTestUtils
+			.getField(jobOperator, "jobParametersConverter");
+
+		Assertions.assertEquals(DefaultJobParametersConverter.class, jobParametersConverter.getClass());
+	}
+
+	@Test
+	@DisplayName("When a custom JobParametersConverter implementation is found then it should be used")
+	public void testCustomJobParametersConverterConfiguration() {
+		AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(
+				CustomJobParametersConverterConfiguration.class);
+
+		JobOperator jobOperator = context.getBean(JobOperator.class);
+		JobParametersConverter jobParametersConverter = (JobParametersConverter) ReflectionTestUtils
+			.getField(jobOperator, "jobParametersConverter");
+
+		Assertions.assertEquals(JsonJobParametersConverter.class, jobParametersConverter.getClass());
+	}
+
 	@Configuration
 	@EnableBatchProcessing
 	public static class JobConfigurationWithoutDataSource {
@@ -328,6 +356,30 @@ public String generateKey(Object source) {
 
 	}
 
+	@Configuration
+	@EnableBatchProcessing
+	public static class CustomJobParametersConverterConfiguration {
+
+		@Bean
+		public DataSource dataSource() {
+			return new EmbeddedDatabaseBuilder().setType(EmbeddedDatabaseType.HSQL)
+				.addScript("/org/springframework/batch/core/schema-hsqldb.sql")
+				.generateUniqueName(true)
+				.build();
+		}
+
+		@Bean
+		public JdbcTransactionManager transactionManager(DataSource dataSource) {
+			return new JdbcTransactionManager(dataSource);
+		}
+
+		@Bean
+		public JobParametersConverter jobParametersConverter() {
+			return new JsonJobParametersConverter();
+		}
+
+	}
+
 	private PlatformTransactionManager getTransactionManagerSetOnJobRepository(JobRepository jobRepository) {
 		Advised target = (Advised) jobRepository; // proxy created by
 													// AbstractJobRepositoryFactoryBean