Spring Batch With Partitioner

@Configuration
public class ExportMasterListCsvJobConfig {

    public static final String JOB_NAME = "exportMasterListCsv";

    @Autowired
    public JobBuilderFactory jobBuilderFactory;

    @Autowired
    public StepBuilderFactory stepBuilderFactory;

    /**
     * Create object representing the master list file.
     *
     * @param outDir
     * @param jobId
     * @return
     * @throws IOException
     */
    @Bean
    @JobScope
    public FileSystemResource masterListFile(
            @Value("${out.dir}") String outDir,
            @Value("#{jobExecution.id}") Long jobId
            ) throws IOException {

        return new FileSystemResource(new File(outDir, "masterList" + jobId + ".csv"));
    }

    /**
     * Create job to export the master list from the online pricing staging db.
     *
     * @param readStgDbAndExportMasterListStep read online staging db step and export to csv
     * @return job to export the master list from the online pricing staging db.
     */
    @Bean
    public Job exportMasterListCsvJob(@Qualifier("partitionStep") Step partitionStep) {

        return jobBuilderFactory.get(JOB_NAME)
                .incrementer(new RunIdIncrementer())
                .flow(partitionStep)
                .end()
                .build();
    }

    @Bean
     public Step partitionStep(
             @Qualifier("readStgDbAndExportMasterListStep") Step readStgDbAndExportMasterListStep,
             @Qualifier("taskExecutor") TaskExecutor taskExecutor,
             @Qualifier("partitioner") Partitioner partitioner){

        return stepBuilderFactory.get("partitionStep")
           .partitioner(readStgDbAndExportMasterListStep)
           .partitioner("readStgDbAndExportMasterListStep", partitioner)
           .taskExecutor(taskExecutor)
           .build();
     }

    @Bean
    @JobScope
    public Partitioner partitioner(){
        return new DivisionPartitioner();
    }

    /**
     * Read from online pricing staging db.
     *
     * @param chunkSize
     * @param queryOnlineStagingDbReaderForMasterList
     * @param masterListOutputProcessor
     * @param masterListFileWriter
     * @return step to read from online pricing staging db.
     */
    @Bean
    @JobScope
    public Step readStgDbAndExportMasterListStep(
            @Value("${exportMasterListCsv.generateMasterListRows.chunkSize}") int chunkSize,
            @Qualifier("queryOnlineStagingDbReaderForMasterList") ItemReader<MasterList> queryOnlineStagingDbReaderForMasterList,
            @Qualifier("masterListOutputProcessor") ItemProcessor<MasterList,MasterList> masterListOutputProcessor,
            @Qualifier("masterListFileWriter") ItemWriter<MasterList> masterListFileWriter) {

        return  stepBuilderFactory.get("readStgDbAndExportMasterListStep")
                    .<MasterList,MasterList>chunk(chunkSize)
                    .reader(queryOnlineStagingDbReaderForMasterList)
                    .processor(masterListOutputProcessor)
                    .writer(masterListFileWriter)
                    .build();

    }


    @Bean
    @JobScope
    public TaskExecutor taskExecutor() {
      return new SimpleAsyncTaskExecutor();
    }

    /**
     * Query and map rows from online pricing staging db into a master list object.
     *
     * @param onlineStagingDb
     * @param masterListSql
     * @return
     */
    @Bean
    @StepScope
    public ItemReader<MasterList> queryOnlineStagingDbReaderForMasterList(
            DataSource onlineStagingDb,
            @Value("${exportMasterListCsv.generateMasterListRows.masterListSql}") String masterListSql) {

        JdbcCursorItemReader<MasterList> reader = new JdbcCursorItemReader<>();

        reader.setDataSource(onlineStagingDb);
        reader.setSql(masterListSql);

        reader.setPreparedStatementSetter(new PreparedStatementSetter() {
            public void setValues(PreparedStatement ps) throws SQLException {
                ps.setInt(1, 45);
            }
        });
        reader.setRowMapper(new RowMapper<MasterList>() {
            @Override
            public MasterList mapRow(ResultSet resultSet, int i) throws SQLException {
                MasterList masterList = new MasterList();
                masterList.setL1(resultSet.getString(DaoConstants.COLUMN_HEADER_LEVEL_ONE));
                masterList.setL2(resultSet.getString(DaoConstants.COLUMN_HEADER_LEVEL_TWO));
                masterList.setL2Name(resultSet.getString(DaoConstants.COLUMN_HEADER_LEVEL_TWO_NAME));
                masterList.setBrand(resultSet.getString(DaoConstants.COLUMN_HEADER_BRAND));
                masterList.setDivisionId(resultSet.getInt(DaoConstants.COLUMN_HEADER_DIVISION_ID));
                masterList.setSellingUnitRetail(resultSet.getDouble(DaoConstants.COLUMN_HEADER_SELLING_UNIT_RETAIL));
                masterList.setClearanceInd(resultSet.getBoolean(DaoConstants.COLUMN_HEADER_CLEARANCE_IND));
                masterList.setPromoRetail(resultSet.getDouble(DaoConstants.COLUMN_HEADER_PROMO_RETAIL));
                masterList.setPromoId(resultSet.getInt(DaoConstants.COLUMN_HEADER_PROMO_ID));
                masterList.setPromoName(resultSet.getString(DaoConstants.COLUMN_HEADER_PROMO_NAME));
                masterList.setPromoCompId(resultSet.getInt(DaoConstants.COLUMN_HEADER_PROMO_COMP_ID));
                masterList.setPromoCompName(resultSet.getString(DaoConstants.COLUMN_HEADER_PROMO_COMP_NAME));
                masterList.setRpmPromoCompDetailId(resultSet.getInt(DaoConstants.COLUMN_HEADER_RPM_PROMO_COMP_DETAIL_ID));
                masterList.setRpmPromoCompDetailState(resultSet.getString(DaoConstants.COLUMN_HEADER_RPM_PROMO_COMP_DETAIL_STATE));

                Timestamp startDateTs = resultSet.getTimestamp(DaoConstants.COLUMN_HEADER_RPM_PROMO_COMP_DETAIL_START_DATE);
                java.util.Date startDate = null;
                if (startDateTs != null){
                    startDate = new java.util.Date(startDateTs.getTime());
                }
                masterList.setRpmPromoCompDetailStartDate(startDate);

                Timestamp endDateTs = resultSet.getTimestamp(DaoConstants.COLUMN_HEADER_RPM_PROMO_COMP_DETAIL_END_DATE);
                java.util.Date endDate = null;
                if (endDateTs != null){
                    endDate = new java.util.Date(endDateTs.getTime());
                }
                masterList.setRpmPromoCompDetailEndDate(endDate);

                return masterList;
            }
        });
        return reader;
    }

    /**
     * Do some extra processing on master list rows.
     *
     * @return master list row with transformations applied.
     */
    @Bean
    @JobScope
    public ItemProcessor<MasterList,MasterList> masterListOutputProcessor() {

        return new ItemProcessor<MasterList, MasterList>() {

            @Override
            public MasterList process(MasterList masterList) throws Exception {
                masterList.setPromoIdPromoCompId(masterList.getPromoId() + " / " + masterList.getPromoCompId());
                return masterList;
            }
        };
    }

    /**
     * Write out the the masterlist.csv
     *
     * @param masterListFile
     * @param executionContext
     * @return
     */
    @Bean
    @StepScope
    public ItemWriter<MasterList> masterListFileWriter(
            FileSystemResource masterListFile,
            @Value("#{stepExecutionContext}")Map<String, Object> executionContext) {

        FlatFileItemWriter<MasterList> writer = new FlatFileItemWriter<>();
        writer.setResource(masterListFile);
        writer.setHeaderCallback(new FlatFileHeaderCallback() {

            public void writeHeader(Writer writer) throws IOException {

                StringBuilder builder = new StringBuilder();

                builder.append(FlatFileConstants.MSTRLST_COLUMN_HEADER_LEVEL_ONE + FlatFileConstants.OUTPUT_FILE_DELIM);
                builder.append(FlatFileConstants.MSTRLST_COLUMN_HEADER_LEVEL_TWO + FlatFileConstants.OUTPUT_FILE_DELIM);
                builder.append(FlatFileConstants.MSTRLST_COLUMN_HEADER_CLEARANCE + FlatFileConstants.OUTPUT_FILE_DELIM);
                builder.append(FlatFileConstants.MSTRLST_COLUMN_HEADER_WAS + FlatFileConstants.OUTPUT_FILE_DELIM);
                builder.append(FlatFileConstants.MSTRLST_COLUMN_HEADER_NOW + FlatFileConstants.OUTPUT_FILE_DELIM);
                builder.append(FlatFileConstants.MSTRLST_COLUMN_HEADER_PERCENT_OFF + FlatFileConstants.OUTPUT_FILE_DELIM);
                builder.append(FlatFileConstants.MSTRLST_COLUMN_HEADER_BRAND + FlatFileConstants.OUTPUT_FILE_DELIM);
                builder.append(FlatFileConstants.MSTRLST_COLUMN_HEADER_PRODUCT + FlatFileConstants.OUTPUT_FILE_DELIM);
                builder.append(FlatFileConstants.MSTRLST_COLUMN_HEADER_OFFER + FlatFileConstants.OUTPUT_FILE_DELIM);
                builder.append(FlatFileConstants.MSTRLST_COLUMN_HEADER_MARKETING_STATEMENT + FlatFileConstants.OUTPUT_FILE_DELIM);
                builder.append(FlatFileConstants.MSTRLST_COLUMN_HEADER_DINKUS + FlatFileConstants.OUTPUT_FILE_DELIM);
                builder.append(FlatFileConstants.MSTRLST_COLUMN_HEADER_PROMO_ID + FlatFileConstants.OUTPUT_FILE_DELIM); // really promo comp id
                builder.append(FlatFileConstants.MSTRLST_COLUMN_HEADER_PROMO_PROMO_COMPONENT_ID + FlatFileConstants.OUTPUT_FILE_DELIM);
                builder.append(FlatFileConstants.MSTRLST_COLUMN_HEADER_PROMO_PROMO_START_DATE + FlatFileConstants.OUTPUT_FILE_DELIM);
                builder.append(FlatFileConstants.MSTRLST_COLUMN_HEADER_PROMO_PROMO_END_DATE);
                writer.write(builder.toString());
            }
        });

        FormatterLineAggregator<MasterList> lineAggregator = new FormatterLineAggregator<>();
        lineAggregator.setFormat(
                "%s" + FlatFileConstants.OUTPUT_FILE_DELIM +                    // l1
                "%s" + FlatFileConstants.OUTPUT_FILE_DELIM +                    // l2
                "%s" + FlatFileConstants.OUTPUT_FILE_DELIM +                    // clearance ind
                "%.2f" + FlatFileConstants.OUTPUT_FILE_DELIM +                  // was
                "%.2f" + FlatFileConstants.OUTPUT_FILE_DELIM +                  // now
                "%.0f" + FlatFileConstants.OUTPUT_FILE_DELIM +                  // percent off
                "%s" + FlatFileConstants.OUTPUT_FILE_DELIM +                    // brand
                "%s" + FlatFileConstants.OUTPUT_FILE_DELIM +                    // product name
                "%s" + FlatFileConstants.OUTPUT_FILE_DELIM +                    // offer name
                "%s" + FlatFileConstants.OUTPUT_FILE_DELIM +                    // marketing statement
                "%s" + FlatFileConstants.OUTPUT_FILE_DELIM +                    // dinkus
                "%d" + FlatFileConstants.OUTPUT_FILE_DELIM +                    // promo comp id
                "%s"+ FlatFileConstants.OUTPUT_FILE_DELIM +                     // promo promo comp id
                "%te/%<tm/%<tY %<tT" + FlatFileConstants.OUTPUT_FILE_DELIM +    // start date
                "%te/%<tm/%<tY %<tT");                                          // end date
        writer.setLineAggregator(lineAggregator);

        BeanWrapperFieldExtractor<MasterList> extractor = new BeanWrapperFieldExtractor<MasterList>();
        extractor.setNames(new String[] {
                "l1",
                "l2",
                "clearanceInd",
                "sellingUnitRetail",
                "promoRetail",
                "percentOff",
                "brand",
                "l2Name",
                "promoCompName",
                "marketingStatement",
                "dinkus",
                "promoCompId",
                "promoIdPromoCompId",
                "rpmPromoCompDetailStartDate",
                "rpmPromoCompDetailEndDate"});
        lineAggregator.setFieldExtractor(extractor);

        writer.setForceSync(true);
        writer.open(new ExecutionContext(executionContext));
        return writer;

    }

}