0

Issue

I've to create a Spring batch project with two jobs that can be executed independently and together. Each job has the necessary code to read from database and to write using FlatFileItemWriter and ClassifierCompositeItemWriter. I've found that if I execute the Jobs independently (-Dspring.batch.job.names=schoolJob,-Dspring.batch.job.names=studentJob), the files are generated fine, but when I execute the Jobs together (-Dspring.batch.job.names=schoolJob,studentJob), the files of a Job only have the footer and the header. There seems to be something wrong but I can't find the cause.

Some code

  • Batch config, job and steps
@Configuration
@EnableBatchProcessing
@SuppressWarnings("rawtypes, unchecked")
public class MyJobConfiguration
{
    @Autowired
    private JobBuilderFactory jobBuilderFactory;

    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    @Autowired
    private JdbcTemplate jdbcTemplate;

    @Autowired
    private ConfigurableApplicationContext applicationContext;

    @Bean
    public Step studentStep1() {
        return stepBuilderFactory.get("calculateDistinctValuesAndRegisterStudentWriters")
                                 .tasklet(new DynamicStudentWritersConfigurationTasklet(jdbcTemplate,
                                                                                        applicationContext))
                                 .build();
    }

    @Bean
    public Step schoolStep1() {
        return stepBuilderFactory.get("calculateDistinctValuesAndRegisterSchoolWriters")
                                 .tasklet(new DynamicSchoolWritersConfigurationTasklet(jdbcTemplate,
                                                                                       applicationContext))
                                 .build();
    }

    @Bean
    @JobScope
    public Step studentStep2(StudentReader reader,
                             @Qualifier("studentClassfierItemWriter")
                             ClassifierCompositeItemWriter<Student> writer) {
        SimpleStepBuilder<Student, Student> studentStep2 = stepBuilderFactory.get(
                "readWriteStudents").<Student, Student>chunk(2).reader(reader).writer(writer);

        Map<String, FlatFileItemWriter> beansOfType = applicationContext.getBeansOfType(
                FlatFileItemWriter.class);
        for (FlatFileItemWriter flatFileItemWriter : beansOfType.values())
        {
            studentStep2.stream(flatFileItemWriter);
        }

        return studentStep2.build();
    }

    @Bean
    @JobScope
    public Step schoolStep2(SchoolReader reader,
                            @Qualifier("schoolClassfierItemWriter")
                            ClassifierCompositeItemWriter<School> writer) {
        SimpleStepBuilder<School, School> schoolStep2 = stepBuilderFactory.get("readWriteSchools")
                                                                          .<School, School>chunk(2)
                                                                          .reader(reader)
                                                                          .writer(writer);

        Map<String, FlatFileItemWriter> beansOfType = applicationContext.getBeansOfType(
                FlatFileItemWriter.class);
        for (FlatFileItemWriter flatFileItemWriter : beansOfType.values())
        {
            schoolStep2.stream(flatFileItemWriter);
        }

        return schoolStep2.build();
    }

    @Bean
    public Job studentJob(Step studentStep1, Step studentStep2) {
        return jobBuilderFactory.get("studentJob").start(studentStep1).next(studentStep2).build();
    }

    @Bean
    public Job schoolJob(Step schoolStep1, Step schoolStep2) {
        return jobBuilderFactory.get("schoolJob").start(schoolStep1).next(schoolStep2).build();
    }
  • Data source configuration
@Configuration
class DatasourceConfig
{
    @Bean
    public DataSource dataSource()
    {
        String dbSchema = "/org/springframework/batch/core/schema-h2.sql";
        String initData = "data.sql";

        return new EmbeddedDatabaseBuilder().setType(EmbeddedDatabaseType.H2)
                                            .addScript(dbSchema)
                                            .addScript(initData)
                                            .build();
    }
}
  • Readers
@Component
class SchoolReader extends JdbcCursorItemReader<School>
{
    @Autowired
    private DataSource dataSource;

    @Override
    public void afterPropertiesSet() throws Exception
    {
        super.setName("schoolItemReader");
        super.setDataSource(dataSource);
        super.setSql("select * from school");
        super.setRowMapper(new BeanPropertyRowMapper<>(School.class));
        super.afterPropertiesSet();
    }
}

@Component
class StudentReader extends JdbcCursorItemReader<Student>
{
    @Autowired
    private DataSource dataSource;

    @Override
    public void afterPropertiesSet() throws Exception
    {
        super.setName("studentItemReader");
        super.setDataSource(dataSource);
        super.setSql("select * from student");
        super.setRowMapper(new BeanPropertyRowMapper<>(Student.class));
        super.afterPropertiesSet();
    }
}
  • Writers
@Configuration
public class SchoolWriter
{
    @Autowired
    private ConfigurableApplicationContext applicationContext;

    @Bean(name = "schoolClassfierItemWriter")
    @StepScope
    public ClassifierCompositeItemWriter<School> itemWriter()
    {
        Map<String, FlatFileItemWriter> beansOfType = applicationContext.getBeansOfType(
                FlatFileItemWriter.class);

        Classifier<School, FlatFileItemWriter<School>> classifier = school -> beansOfType.get(
                "school-group" + school.getGroupId() + "Writer");
        return new ClassifierCompositeItemWriterBuilder().classifier(classifier).build();
    }
}

@Configuration
public class StudentWriter
{
    @Autowired
    private ConfigurableApplicationContext applicationContext;

    @Bean(name = "studentClassfierItemWriter")
    @StepScope
    public ClassifierCompositeItemWriter<Student> itemWriter()
    {
        Map<String, FlatFileItemWriter> beansOfType = applicationContext.getBeansOfType(
                FlatFileItemWriter.class);

        Classifier<Student, FlatFileItemWriter<Student>> classifier = student -> beansOfType.get(
                "student-group" + student.getGroupId() + "Writer");
        return new ClassifierCompositeItemWriterBuilder().classifier(classifier).build();

    }
}
  • Tasklets
class DynamicSchoolWritersConfigurationTasklet implements Tasklet
{
    private JdbcTemplate                   jdbcTemplate;
    private ConfigurableApplicationContext applicationContext;

    public DynamicSchoolWritersConfigurationTasklet(JdbcTemplate jdbcTemplate,
                                                    ConfigurableApplicationContext applicationContext)
    {
        this.jdbcTemplate       = jdbcTemplate;
        this.applicationContext = applicationContext;
    }

    @Override
    public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext)
    {
        ConfigurableListableBeanFactory beanFactory = applicationContext.getBeanFactory();

        String        sql    = "select distinct(groupId) from school";
        List<Integer> groups = jdbcTemplate.queryForList(sql, Integer.class);

        for (Integer group : groups)
        {
            String name = "school-group" + group + "Writer";

            //@f:off
            MutablePropertyValues propertyValues = new MutablePropertyValues();
            propertyValues.addPropertyValue("name", name);
            propertyValues.addPropertyValue("lineAggregator", new PassThroughLineAggregator<>());
            propertyValues.addPropertyValue("resource", new FileSystemResource("school-" + group + ".txt"));
            propertyValues.addPropertyValue("headerCallback", (FlatFileHeaderCallback) writer -> writer.write("header-school"));
            propertyValues.addPropertyValue("footerCallback", (FlatFileFooterCallback) writer -> writer.write("footer-school"));
            //@f:on

            GenericBeanDefinition beanDefinition = new GenericBeanDefinition();
            beanDefinition.setBeanClassName(FlatFileItemWriter.class.getName());
            beanDefinition.setPropertyValues(propertyValues);

            BeanDefinitionRegistry registry = (BeanDefinitionRegistry) beanFactory;
            registry.registerBeanDefinition(name, beanDefinition);
        }

        return RepeatStatus.FINISHED;
    }
}

class DynamicStudentWritersConfigurationTasklet implements Tasklet
{
    private JdbcTemplate                   jdbcTemplate;
    private ConfigurableApplicationContext applicationContext;

    public DynamicStudentWritersConfigurationTasklet(JdbcTemplate jdbcTemplate,
                                                     ConfigurableApplicationContext applicationContext)
    {
        this.jdbcTemplate       = jdbcTemplate;
        this.applicationContext = applicationContext;
    }

    @Override
    public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext)
    {
        ConfigurableListableBeanFactory beanFactory = applicationContext.getBeanFactory();

        String        sql    = "select distinct(groupId) from student";
        List<Integer> groups = jdbcTemplate.queryForList(sql, Integer.class);

        for (Integer group : groups)
        {
            String name = "student-group" + group + "Writer";

            //@f:off
            MutablePropertyValues propertyValues = new MutablePropertyValues();
            propertyValues.addPropertyValue("name", name);
            propertyValues.addPropertyValue("lineAggregator", new PassThroughLineAggregator<>());
            propertyValues.addPropertyValue("resource", new FileSystemResource("student-" + group + ".txt"));
            propertyValues.addPropertyValue("headerCallback", (FlatFileHeaderCallback) writer -> writer.write("header-student"));
            propertyValues.addPropertyValue("footerCallback", (FlatFileFooterCallback) writer -> writer.write("footer-student"));
            //@f:on

            GenericBeanDefinition beanDefinition = new GenericBeanDefinition();
            beanDefinition.setBeanClassName(FlatFileItemWriter.class.getName());
            beanDefinition.setPropertyValues(propertyValues);

            BeanDefinitionRegistry registry = (BeanDefinitionRegistry) beanFactory;
            registry.registerBeanDefinition(name, beanDefinition);
        }

        return RepeatStatus.FINISHED;
    }
}
  • DAO
@Getter
@Setter
@ToString
@NoArgsConstructor
@AllArgsConstructor
public class School
{
    private int    id;
    private String name;
    private int    groupId;
}

@Getter
@Setter
@ToString
@NoArgsConstructor
@AllArgsConstructor
public class Student
{
    private int    id;
    private String name;
    private int    groupId;
}
igarciadev
  • 131
  • 3
  • 11

1 Answers1

0

This is similar to https://stackoverflow.com/a/67635289/5019386. I think you need to make your dynamic item writers step-scoped as well, something like:

propertyValues.addPropertyValue("scope", "step");

Please note that I did not try that. That said, I would really recommend making your app do one thing and do it well, ie isolate job definitions and package/run each job separately.

Mahmoud Ben Hassine
  • 28,519
  • 3
  • 32
  • 50