I like to test my class:
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecutionListener;
import org.springframework.batch.core.JobParameter;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.core.launch.support.SimpleJobLauncher;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.database.Order;
import org.springframework.batch.item.database.builder.JdbcPagingItemReaderBuilder;
import org.springframework.batch.item.database.support.OraclePagingQueryProvider;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase;
import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import javax.sql.DataSource;
import java.util.HashMap;
import java.util.Map;
@Configuration
@Slf4j
@RequiredArgsConstructor
@EnableScheduling
@EnableBatchProcessing
public class JobLDSExecutorConfiguration {
private final KafkaTemplate<String, String> kafkaTemplate;
private final StepBuilderFactory stepBuilderFactory;
private final JobBuilderFactory jobBuilderFactory;
private final DataSource dataSource;
private final JobRepository jobRepository;
private static final int CHUNK_SIZE = 1;
@Value("${kafka.rrss-correspondances-services-consommateur.topic}")
private String topic;
@Value("${task.thread.limit}")
private int limteThread;
@Value("${task.size}")
private int size;
@Value("${task.pageSize}")
private int pageSize;
private ChunkExecutionListener chunkListener() {
return new ChunkExecutionListener();
}
@Bean
public ItemReader<LdsOds> ldstJdbcItemReader() {
return new JdbcPagingItemReaderBuilder<LdsOds>()
.name("pagingItemReaderLDS")
.fetchSize(size)
.dataSource(dataSource)
.pageSize(pageSize)
.queryProvider(queryProvider())
.rowMapper(new LdsOdsRowMapper(TypeDonnee.LDS))
.build();
}
private OraclePagingQueryProvider queryProvider() {
OraclePagingQueryProvider provider =
new OraclePagingQueryProvider();
provider.setSelectClause(SQL_SELECT_LDS);
provider.setFromClause(SQL_FROM_LDS);
Map<String, Order> sortKeys = new HashMap<>(1);
sortKeys.put("SERVICE_DELIVERY_LOCATION_ID", Order.ASCENDING);
provider.setSortKeys(sortKeys);
return provider;
}
@Bean
public ItemWriter<LdsOds> kafkaLDSItemWriter() {
return items -> {
for (LdsOds item : items) {
kafkaTemplate.send(topic, item.getId().toString(), new ObjectMapper().writeValueAsString(item));
}
};
}
@Bean
public TaskExecutor taskLDSExecutor() {
SimpleAsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor();
taskExecutor.setConcurrencyLimit(limteThread);
return taskExecutor;
}
@Bean
public Step loadLds() {
return stepBuilderFactory.get("loadLds")
.<LdsOds, LdsOds>chunk(CHUNK_SIZE)
.reader(ldstJdbcItemReader())
.writer(kafkaLDSItemWriter())
.taskExecutor(taskLDSExecutor())
.listener(chunkListener())
.throttleLimit(limteThread)
.build();
}
@Bean
public JobExecutionListener jobLDSListener() {
return new JobCompletionNotificationListener();
}
@Bean(name = "jobLDSExtraction")
public Job jobLDSExtraction() {
return jobBuilderFactory.get("jobLDSExtraction")
.incrementer(new RunIdIncrementer())
.listener(jobLDSListener())
.start(loadLds())
.build();
}
@Bean
public TaskExecutor taskExecutor() {
ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
taskExecutor.setCorePoolSize(15);
taskExecutor.setMaxPoolSize(20);
taskExecutor.setQueueCapacity(30);
return taskExecutor;
}
@Bean
public JobLauncher jobLauncher() throws Exception {
SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
jobLauncher.setTaskExecutor(taskExecutor()); // Or below line
jobLauncher.setJobRepository(jobRepository);
jobLauncher.afterPropertiesSet();
return jobLauncher;
}
@Scheduled(fixedRate = 10000)
public void executerMajIdentifiantLDS() {
Map<String, JobParameter> confMap = new HashMap<>();
confMap.put("time", new JobParameter(System.currentTimeMillis()));
JobParameters jobParameters = new JobParameters(confMap);
try {
jobLauncher().run(jobLDSExtraction(), jobParameters);
} catch (Exception ex) {
log.error("Erreur pendant l'execution du jobLDSExtraction {}", ex.getMessage());
}
}
}
On my test
@Test
void testStep1() {
JobExecution jobExecution = jobLauncherTestUtils.launchStep("loadLds");
Assert.assertEquals(new ExitStatus("COMPLETED").getExitCode() ,jobExecution.getExitStatus().getExitCode());
}
But I have an error:
Error creating bean with name 'jobLDSExecutorConfiguration': Unsatisfied dependency expressed through constructor parameter 3; nested exception is org.springframework.beans.factory.NoSuchBeanDefinitionException: No qualifying bean of type 'javax.sql.DataSource' available: expected at least 1 bean which qualifies as autowire candidate. Dependency annotations: {}
I would like to know if it's possible to MOCK or Inject datasource and how to test it.