(Spring boot + Spring Batch + tasklet +JobParameter) How to pass jobParameter to Tasklet

(Spring boot + Spring Batch + tasklet +JobParameter) How to pass jobParameter to Tasklet

I tried to pass jobParameter which user(anonymous) would set using tasklet(not itemReader or itemWriter, there are a lots of examples about how to pass jobParameter using itemReader & itemWriter) but I encountered below error code.

error code “no job configuration with the name was registered”

Then I found out how to fix it.

just Add the below code in your batch configuration code. The below code will register your job name. after then they can read job parameter in tasklet

@Bean
public JobRegistryBeanPostProcessor jobRegistryBeanPostProcessor( JobRegistry jobRegistry) {
JobRegistryBeanPostProcessor jobRegistryBeanPostProcessor = new JobRegistryBeanPostProcessor();
jobRegistryBeanPostProcessor.setJobRegistry(jobRegistry);
    return jobRegistryBeanPostProcessor;
}

my whole BatchConfiguration code is like

@PropertySource("classpath:application.yml")
@AutoConfigureAfter(value = { BatchAutoConfiguration.class })
@EnableBatchProcessing
@Configuration
public class SeleniumBatchConfiguration {

    @Autowired
    private JobBuilderFactory jobBuilderFactory;

    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    @Bean
    public JobRegistryBeanPostProcessor jobRegistryBeanPostProcessor(JobRegistry jobRegistry) {
    JobRegistryBeanPostProcessor jobRegistryBeanPostProcessor = new JobRegistryBeanPostProcessor();
    jobRegistryBeanPostProcessor.setJobRegistry(jobRegistry);
return jobRegistryBeanPostProcessor;
    }

    @Bean
    public Job job() {
        return jobBuilderFactory.get("your job name")        .start(requestForSignUpSMT3rdPartyStep()).build();
    }

    @Bean
    public Step step() {
        return stepBuilderFactory.get("your step name")
                .allowStartIfComplete(true)
                .tasklet(new RequestForSignUpSMT3rdPartyTasklet())
                .build();
    }

(sorry that my code look bad in blog)

Set jobParameter using jobParameterBuilder

It is just example. I use this jobParameter to run job using JobLauncher.

    User user = new User();
    user.setComment("This is Comment");
    user.setEmail("sooyoung32@gmail.com");
    user.setFirstName("NARASIMHAN");
    user.setLastName("NARASIMHAN");
    user.setEsiid("10443720003427889");


JobParameters parameters = new JobParametersBuilder()
                                                        .addDate("date", new Date()) //date will ba a key for jobInstance
                                                        .addString("firstName", user.getFirstName())
                                                        .addString("lastName", user.getLastName())
                                                        .addString("esiid", user.getEsiid())
                                                        .addString("comment", user.getComment())
                                                        .addString("email", user.getEmail())
                                                        .toJobParameters();

how to get JobParameter(which we set above) in Tasklek

When you make tasklet class you should implement Tasklet. Then you must implement “execute” method that have SetpContribution and ChunkContext as parameters. We use ChunkContext to get jobParameter.

String date = chunkContext.getStepContext().getStepExecution().getJobParameters().getString("date");
        String email = chunkContext.getStepContext().getStepExecution().getJobParameters().getString("email");
        String firstName = chunkContext.getStepContext().getStepExecution().getJobParameters().getString("firstName");
        String lastName = chunkContext.getStepContext().getStepExecution().getJobParameters().getString("lastName");
        String esiid = chunkContext.getStepContext().getStepExecution().getJobParameters().getString("esiid");
        String comment = chunkContext.getStepContext().getStepExecution().getJobParameters().getString("comment");

Spring Boot + Spring Batch (Spring Batch Integration : Remote Partitioning) + Mysql + JPA +Gradle

Spring Boot + Spring Batch (Spring Batch Integration : Remote Partitioning) + Mysql + JPA +Gradle

• batch 설정 javaConfig, integration 설정 xml
• Job : resources 인 person.txt의 내용을 DB(mysql)에 저장시킨다.
• Batch : Job > Step > tasklet > Chunk (read-process-write) 의 포함관계
• partitioning 이란 Job이 step1 을 실행시키고, step1 이 partionHandler를 이용해 partition 한 후 step2 실행. 실질적인 일은 step2 (reader, processor, writer)가 한다.
• remote partioning은 서로 다른 서버에 job과 스텝이 있을때, master 서버와 slave 서버를 분리시킨 후 integration을 이용해 통신.
• Master 서버에서는 job 등록 -> step1 수행 -> partitioner 까지. 후에 메세지 큐를 이용해 Slave로 보냄
• Slave 서버에서는 master에서 보낸 메시지 큐에서 partition한 step2를 실질적으로 수행.(read, process, write)
• 그림
  
  참고 [http://docs.spring.io/spring-batch/trunk/reference/html/springBatchIntegration.html]

Master 코드

패키지 구조

• src/main/java
  • kr.co.kware.batch.test
    • MasterApplicaion.java
  • kr.co.kware.batch.test.model
    • Person.java
  • kr.co.kware.batch.test.partition
    • MaterPartitioner.java
  • kr.co.kware.batch.test.tasklet
    • MasterTasklet.java
  • kr.co.kware.batch.test.config
    • MasterBatchConfiguration.java
• src/main/resouces
  • application.properties
  • integration-master.xml
  • PersonData.txt (first_name, last_name, year 데이터만 있음)

build.gradle

buildscript {
    ext {
        springBootVersion = '1.2.3.RELEASE'
    }

    repositories {
        jcenter()
    }

    dependencies {
        classpath("org.springframework.boot:spring-boot-gradle-plugin:${springBootVersion}")
    }
}

apply plugin: 'java'
apply plugin: 'idea'
apply plugin: 'eclipse'
apply plugin: 'spring-boot'

sourceCompatibility = 1.7
targetCompatibility = 1.7

jar {
    baseName = 'batch-master-mysql'
    version = '0.0.1'
}

repositories {
    jcenter()

    maven { url 'https://repo.spring.io/release' }
    maven { url 'https://repo.spring.io/snapshot' }
    maven { url 'https://repo.spring.io/milestone' }

    maven { url 'https://repository.jboss.org/nexus/content/repositories/releases' }
    maven { url 'https://oss.sonatype.org/content/repositories/releases' }
    maven { url 'https://oss.sonatype.org/content/repositories/snapshots' }
    maven { url 'http://repo.maven.apache.org/maven2' }
}

dependencies {

    compile("org.springframework.boot:spring-boot-autoconfigure")
    compile("org.springframework.boot:spring-boot-starter-data-jpa")
    compile("org.springframework.boot:spring-boot-starter-actuator")
    compile("org.springframework.boot:spring-boot-starter-logging")
    compile("org.springframework.boot:spring-boot-starter-aop")

    compile("org.springframework.boot:spring-boot-starter-batch")
    compile("org.springframework.batch:spring-batch-infrastructure")
    compile("org.springframework.batch:spring-batch-integration")

    compile("mysql:mysql-connector-java")

    compile("org.springframework.boot:spring-boot-starter-integration")
    compile("org.springframework.integration:spring-integration-jms")
    compile("org.springframework.integration:spring-integration-amqp")

    compile("org.apache.activemq:activemq-broker:5.11.1")

    testCompile("org.springframework.boot:spring-boot-starter-test")
}

application.properties

spring.jpa.hibernate.ddl-auto=create
spring.jpa.show-sql=false

spring.datasource.driver-class-name= com.mysql.jdbc.Driver
spring.datasource.url=jdbc:mysql://localhost:3306/
spring.datasource.username=
spring.datasource.password=
spring.datasource.platform=mysql
spring.datasource.initialize=false

spring.jpa.database=MYSQL
spring.jpa.open-in-view=false
spring.jpa.show-sql=true
spring.jpa.generate-ddl=false
spring.jpa.hibernate.ddl-auto=create-drop
spring.jpa.hibernate.naming-strategy=org.hibernate.cfg.ImprovedNamingStrategy
spring.jpa.properties.hibernate.dialect= org.hibernate.dialect.MySQL5Dialect

spring.activemq.broker-url=
spring.activemq.user=
spring.activemq.password=

spring.batch.initializer.enabled=false
spring.batch.grid.size=5
spring.batch.admin.job.configuration.file.dir=target/config # /META-INF/spring/batch/bootstrap/integration/configuration-context.xml
spring.batch.admin.job.service.reaper.interval= 60000

batch.job.configuration.file.dir= target/config
batch.job.service.reaper.interval= 60000

logging.file=./logs/master.log
logging.level=INFO

integration_master.xml

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
       xmlns:p="http://www.springframework.org/schema/p"
       xmlns:int="http://www.springframework.org/schema/integration"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
         http://www.springframework.org/schema/beans/spring-beans.xsd
         http://www.springframework.org/schema/integration/jms
         http://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd
         http://www.springframework.org/schema/integration
         http://www.springframework.org/schema/integration/spring-integration.xsd">

       <int:channel id="requestsChannel" />

       <int-jms:outbound-channel-adapter connection-factory="jmsConnectionFactory"
                                         channel="requestsChannel"
                                         destination-name="requestsQueue" />

       <int:channel id="replyChannel"/>

       <int-jms:message-driven-channel-adapter connection-factory="jmsConnectionFactory"
                                               channel="replyChannel"
                                               destination-name="replyQueue"/>

       <int:channel id="aggregatedReplyChannel">
           <int:queue/>
       </int:channel>

       <int:aggregator ref="remotePartitionHandler"
                       input-channel="replyChannel"
                       output-channel="aggregatedReplyChannel"
                       send-timeout="3600000"/>

</beans>

Person.java

import javax.persistence.Column;
import javax.persistence.Entity;
import javax.persistence.GeneratedValue;
import javax.persistence.GenerationType;
import javax.persistence.Id;

@Entity(name="person")
public class Person {

    @Id
    @GeneratedValue(strategy=GenerationType.IDENTITY)
    private int id = 1;

    @Column(name="first_name")
    private String firstName;
    @Column(name="last_name")
    private String lastName;
    @Column(name="year")
    private int year;
    public int getId() {
        return id;
    }
    public void setId(int id) {
        this.id = id;
    }
    public String getFirstName() {
        return firstName;
    }
    public void setFirstName(String firstName) {
        this.firstName = firstName;
    }
    public String getLastName() {
        return lastName;
    }
    public void setLastName(String lastName) {
        this.lastName = lastName;
    }
    public int getYear() {
        return year;
    }
    public void setYear(int year) {
        this.year = year;
    }
}

MasterApplication.java

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.ImportResource;

@SpringBootApplication
@ImportResource("classpath:integration-master.xml")
public class MasterApplication {
    public static void main(String[] args) {
         ApplicationContext context= SpringApplication.run(MasterApplication.class, args);
          for (String name : context.getBeanDefinitionNames()) {
                System.out.println("BEAN NAME : " + name);
            }
    }
}

MasterBatchCnfiguration.java

import org.springframework.batch.core.Job;
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.support.RunIdIncrementer;
import org.springframework.batch.core.partition.PartitionHandler;
import org.springframework.batch.integration.partition.MessageChannelPartitionHandler;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.orm.jpa.EntityScan;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.PropertySource;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import spring.boot.batch.mysql.test.model.Person;
import spring.boot.batch.mysql.test.partition.MasterPartitioner;
import spring.boot.batch.mysql.test.step.MasterTasklet;

@SpringBootApplication
@EnableBatchProcessing
@EntityScan(basePackageClasses = Person.class)
@PropertySource("classpath:application.properties")
public class MasterBatchConfiguration {

    /*
     * Load the properties
     */
    @Value("${spring.batch.grid.size}")
    private int gridSize;

    @Autowired
    private JobBuilderFactory jobBuilderFactory;


    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    @Autowired
    private PollableChannel aggregatedReplyChannel;

    @Autowired
    private MessageChannel requestsChannel;


    @Bean
    public Job importPerson(Step step1) {
        return jobBuilderFactory.get("importPerson")
                    .incrementer(new RunIdIncrementer()) 
                    .start(partitionStep())
                    .build();

    }

    @Bean
    public Step partitionStep() {
        return stepBuilderFactory.get("partitionStep")
                .partitioner(step1())
                .partitioner("step1", new MasterPartitioner(MasterPartitioner.DEFAULT_TEST_VALUE_SIZE))
                .partitionHandler(remotePartitionHandler())
                .build();
    }

    @Bean
    public Step step1() {
        return stepBuilderFactory.get("step1")
                                  .tasklet(new MasterTasklet())
                                  .build();
    }

    @Bean
    public PartitionHandler remotePartitionHandler() {
        MessageChannelPartitionHandler partitionHandler = new MessageChannelPartitionHandler();
        partitionHandler.setStepName("step1");
        partitionHandler.setGridSize(this.gridSize);
        partitionHandler.setReplyChannel(this.aggregatedReplyChannel);

        MessagingTemplate messagingTemplate = new MessagingTemplate(this.requestsChannel);

        partitionHandler.setMessagingOperations(messagingTemplate);

        return partitionHandler;
    }
}

MasterTasklet.java

• partitioner에서 job을 전달받은 step2는 slave에서 처리되야하기 때문에, 여기서는 비어있는 tasklet을 넣어준다. 이게 없으면 step2를 찾을수 없다는 에러..ㅠㅠ

import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;

public class MasterTasklet implements Tasklet {
    @Override
    public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
        return RepeatStatus.FINISHED;
    }
}

MasterPartitioner : 어떻게 파티션할것인가.

import org.springframework.batch.core.partition.support.Partitioner;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.beans.factory.annotation.Value;

import java.util.HashMap;
import java.util.Map;

public class MasterPartitioner implements Partitioner {
    public static long DEFAULT_TEST_VALUE_SIZE = 10L;

    private long testValueSize = DEFAULT_TEST_VALUE_SIZE;

    public MasterPartitioner(long testValueSize) {
        this.testValueSize = testValueSize;
    }

    @Override
    public Map<String, ExecutionContext> partition(int gridSize) {
        long targetSize = testValueSize / gridSize;

        Map<String, ExecutionContext> result = new HashMap<>();
        for (int i = 0; i < gridSize; i++) {
            ExecutionContext value = new ExecutionContext();
            result.put("partition" + i, value);

            value.putLong("minValue", i * targetSize);
            value.putLong("maxValue", (i + 1) * targetSize -1);
        }

        return result;
    }
}

Slave 코드

패키지 구조

• src/main/java
  • kr.co.kware.batch.test.slave
    • SlaveApplication.java (마스터와 상동, xml 파일명을 integration-slave.xml 변경)
  • kr.co.kware.batch.test.slave.model
    • Person.java (마스터와 상동)
  • kr.co.kware.batch.test.slave.config
    • SlaveBatchConfiguration.java
  • kr.co.kware.batch.test.slave.processor
    • PersonProcessor.java
  • kr.co.kware.batch.test.slave.tasklet
    • SlaveTasklet.java
• src/main/resources
  • application.properties (마스터와 상동)
  • PersonData.txt
  • integration-slave.xml

integration-slave.xml

<?xml version="1.0" encoding="UTF-8"?>

<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
       xmlns:p="http://www.springframework.org/schema/p"
       xmlns:int="http://www.springframework.org/schema/integration"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
         http://www.springframework.org/schema/beans/spring-beans.xsd
         http://www.springframework.org/schema/integration/jms
         http://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd
         http://www.springframework.org/schema/integration
         http://www.springframework.org/schema/integration/spring-integration.xsd">

    <int:channel id="requestChannel"/>

    <int-jms:message-driven-channel-adapter connection-factory="jmsConnectionFactory"
                                            destination-name="requestsQueue"
                                            channel="requestChannel"/>

    <int:channel id="replyChannel"/>

    <int-jms:outbound-channel-adapter connection-factory="jmsConnectionFactory"
                                      destination-name="replyQueue"
                                      channel="replyChannel"/>

    <int:service-activator input-channel="requestChannel"
                           output-channel="replyChannel"
                           ref="remoteStepExecutionRequestHandler"/>
</beans>

# SlaveBachConfiguration

import javax.sql.DataSource;

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.configuration.annotation.StepScope;
import org.springframework.batch.core.explore.JobExplorer;
import org.springframework.batch.core.explore.support.JobExplorerFactoryBean;
import org.springframework.batch.integration.partition.BeanFactoryStepLocator;
import org.springframework.batch.integration.partition.StepExecutionRequestHandler;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.database.JpaItemWriter;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper;
import org.springframework.batch.item.file.mapping.DefaultLineMapper;
import org.springframework.batch.item.file.transform.DelimitedLineTokenizer;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.ClassPathResource;
import org.springframework.core.io.support.ResourcePatternResolver;
import org.springframework.orm.jpa.JpaVendorAdapter;
import org.springframework.orm.jpa.LocalContainerEntityManagerFactoryBean;

import srping.boot.batch.mysql.test.model.Person;
import srping.boot.batch.mysql.test.processor.PersonItemProcessor;

@Configuration
@EnableBatchProcessing
public class SlaveBatchConfiguration {
    private static final String OVERRIDDEN_BY_EXPRESSION = null;
    @Autowired
    private DataSource dataSource;

    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    @Autowired
    private BeanFactory beanFactory;

    @Autowired
    private ResourcePatternResolver resourcePatternResolver;

    @Autowired
    private JobBuilderFactory jobBuilders;

    @Autowired
    private StepBuilderFactory stepBuilders;

    @Autowired
    private JpaVendorAdapter JpaVendorAdapter;

    @Autowired
    LocalContainerEntityManagerFactoryBean entityManageFactory;

    @Bean
    @StepScope
    public FlatFileItemReader<Person> reader(){
        FlatFileItemReader<Person> reader = new FlatFileItemReader<Person>();

        reader.setResource(new ClassPathResource("PersonData.txt"));
        reader.setLineMapper(new DefaultLineMapper<Person>() {{
            setLineTokenizer(new DelimitedLineTokenizer() {{
                setNames(new String[] {"firstName","lastName","year"});
            }});
            setFieldSetMapper(new BeanWrapperFieldSetMapper<Person>() {{
                setTargetType(Person.class);
            }});
        }});
        System.out.println(">>> reader" );
        return reader;
    }

    @Bean
    public ItemProcessor<Person, Person> processor() {
        System.out.println(">>> processor");
        return new PersonItemProcessor();
    }

    @Bean
    public ItemWriter<Person> writer() {
        JpaItemWriter<Person> itemWriter = new JpaItemWriter<Person>();
        itemWriter.setEntityManagerFactory(this.entityManageFactory.getObject());
        return itemWriter;
    }

    @Bean
    public Step step1() {
        return stepBuilderFactory.get("step1")
                                 .<Person, Person> chunk(10)
                                 .reader(reader())
                                 .processor(processor())
                                 .writer(writer())
                                 .build();
    }

    @Bean
    public StepExecutionRequestHandler remoteStepExecutionRequestHandler() {
        StepExecutionRequestHandler stepExecutionRequestHandler = new StepExecutionRequestHandler();
        BeanFactoryStepLocator beanFactoryStepLocator = new BeanFactoryStepLocator();
        beanFactoryStepLocator.setBeanFactory(this.beanFactory);

        stepExecutionRequestHandler.setStepLocator(beanFactoryStepLocator);

        JobExplorer jobExplorer = null;
        try {
            JobExplorerFactoryBean jobExplorerFactoryBean = new JobExplorerFactoryBean();
            jobExplorerFactoryBean.setDataSource(this.dataSource);
            jobExplorerFactoryBean.afterPropertiesSet();

            jobExplorer = jobExplorerFactoryBean.getObject();
        } catch (Exception e) {
            e.printStackTrace();
        }

        stepExecutionRequestHandler.setJobExplorer(jobExplorer);

        return stepExecutionRequestHandler;
    }
}

PersonItemProcessor

import org.springframework.batch.item.ItemProcessor;

import srping.boot.batch.mysql.test.model.Person;

public class PersonItemProcessor implements ItemProcessor<Person, Person>{

    @Override
    public Person process(final Person item) throws Exception {
        return item;
    }

}

SlvaeTasklet.java

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;

public class TestSlaveTasklet implements Tasklet {
    private static final Logger LOG = LoggerFactory.getLogger(TestSlaveTasklet.class);

    @Override
    public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
        LOG.info("step execution context : " + chunkContext.getStepContext().getStepExecutionContext());

        return RepeatStatus.FINISHED;
    }
}

Spring Batch - Read multiple file

Spring Batch - Read multiple file

tag : ReaderNotOpenException or reader must be open before it can be read

목적

Partitioner 에서 여러 파일을 읽어 한 DB에 저장시키고 싶었다.

• partioner 코드 : Resource에서 .csv 파일을 모두 읽어서 DB에 저장 file:src/main/resources/*.csv 밑에는 3개의 csv 파일이 있다.

@Bean
public Partitioner partitioner() {
        System.out.println(">>> partitioner");
        MultiResourcePartitioner partitioner = new MultiResourcePartitioner();
        Resource[] resources;
        try {
            resources = resourcePatternResolver.getResources("file:src/main/resources/*.csv");
            //System.out.println(">>> resources : "+ resources.length);
        } catch(IOException e) {
            throw new RuntimeException("I/O problems when resolving the input file pattern", e);
        }
        partitioner.setResources(resources);
        return partitioner;

    }

문제

• 수정 전 reader code : ItemReader를 반환하고 sample-data.csv를 읽으라고 hard coding 되어있다. 이렇게 실행 시키니 위에 resource에서는 3개의 파일이 있으니 sample-data.csv <-resource 만 3번 읽는다..ㅠㅠ

@Bean
@StepScope
public ItemReader<Person> reader(){
        FlatFileItemReader<Person> reader = new FlatFileItemReader<Person>();


        reader.setResource(new FileSystemResource("sample-data.csv");
        reader.setStrict(true);
        reader.setLineMapper(new DefaultLineMapper<Person>() {{
            setLineTokenizer(new DelimitedLineTokenizer() {{
                setNames(new String[] {"firstName", "lastName"});
            }});
            setFieldSetMapper(new BeanWrapperFieldSetMapper<Person>() {{
                setTargetType(Person.class);
            }});
        }});
        System.out.println(">>> reader" );
        return reader;
    }

해결

partitioner 메소드는 resource를 잘 읽기 때문에 reader에서 여러파일을 읽으려먼 아래와 같이 수정한다.

@Bean
@StepScope
public FlatFileItemReader<Person> reader(@Value("#{stepExecutionContext[fileName]}") String fileName){
        FlatFileItemReader<Person> reader = new FlatFileItemReader<Person>();

        System.out.println(">>> file :" + fileName.substring(5));

        reader.setResource(new FileSystemResource(fileName.substring(5)));

        ... 

        return reader

• @StepScope는 reader가 매 StepExecution 마다 생성함을 지시한다. reader는 singleton 이기 때문에 매 StepExecution 마다 생성해 줘야 한다.
  
• reader를 생성하는 메소드는 파일경로를 담는 String fileName을 가지고 있고, 여기에 @Value annotaiton을 이용하여 fileName 이란 이름으로 stepExecutionContext를 주입한다. (fileName이란 변수명이 중요하다.) (지금 나는 stepExecutionContext에서 여러 파일을 읽기 때문에 stepExecutionContext를 사용했지만, job parameter나, jobExecutionContext도 같은 방법으로 사용한다)
  
• reader.setResource(new FileSystemResource(fileName))을 넣어준다.
  
• 마지막으로 return type을 ItemReader -> FlatFileItemReader 로 바꿔준다!!!
  
• 그렇지 않으면 “ReaderNotOpenException or reader must be open before it can be read” 에러가 발생한다. (이게 제일 어려웠음 ㅠㅠ) reader가 읽히지 전에 open 되어야 하고 때문에 reader는 null값이다.
  원인은 이거다(이것 일 것이다..). itemReader interface안에는 read()라는 추상 메소드 하나가 있다. 파일 하나를 읽을때는 우리가 하드코딩한 (reader.setResource(new FileSystemResource(“sample-data.csv”)) 파일만 읽으면 되기때문에 read만 있는것이 문제되지 않는다.하지만 여러 파일을 읽을 때는 파일 하나를 읽고 닫은후 다시 새로운 파일을 읽어야 한다. 추상메소드 read()를 구현하는 곳에는 open() , update() close()가 없기 때문에 위와같은 에러가 발생한것이다.
  반면 FlatFileItemReader에는 ResourceAwareItemReaderItemStream를 구현하고 있는데 ResourceAwareItemReaderItemStream 는 ItemStreamReader 를 상속 받고 있는데, ItemStreamReader에는 ItemStream, ItemReader를 또 상속받고있다. 그리고 ItemStream 안에는 open(), update(), close()를 구현하고 있어서 FlatFileItemReader에서 override하는 것이다!!!!

결론

문서를 제대로 읽자..ㅎㅎ 사실 내 수준에서 어려운 문제기도 했지만 문서를 제대로 읽지 않아 에러를 잡는데 더 오래걸렸다. 물론 문제를 해결하는 과정에서 많은것을 배웠지만.
원인이라고 적어둔 것은 나와 팀장님의 추측임을 밝힌다. 좀 더 명확한 해답을 알고 있다면 댓글이나 이메일로 알려주시길 바란다..ㅠㅠ
하..여튼 풀었다.

---

참조

• https://blog.codecentric.de/en/2013/06/spring-batch-2-2-javaconfig-part-2-jobparameters-executioncontext-and-stepscope/
• http://stackoverflow.com/questions/23847661/spring-batch-org-springframework-batch-item-readernotopenexception-reader-must/30096121#30096121