title: How to launch spring batch job from integration flow?
body: <p>I have a problem with Spring Integration and Spring Batch. I want to pass csv file from sftp to a batch job then convert information into POJO and pass to output. How can I do this? I have next configuration:</p>
  
  <pre><code>@Configuration
  @RequiredArgsConstructor
  @Setter
  class BatchJobConfig {
      private final JobBuilderFactory jobBuilderFactory;
      private final StepBuilderFactory stepBuilderFactory;
      private final DataSource dataSource;
      private final PlatformTransactionManager transactionManager;
  
      @Bean
      public Job readCSVFilesJob() {
          return jobBuilderFactory
                  .get("readCSVFilesJob")
                  .incrementer(new RunIdIncrementer())
                  .start(step())
                  .build();
      }
  
      @Bean
      public Step step() {
          return stepBuilderFactory.get("step").&lt;Bill, Bill&gt;chunk(7)
                  .reader(reader())
                  .writer(writer())
                  .build();
      }
  
      @Bean
      @StepScope
      public FlatFileItemReader&lt;Bill&gt; reader() {
          FlatFileItemReader&lt;Bill&gt; reader = new FlatFileItemReader&lt;&gt;();
          reader.setResource(new FileSystemResource("/info"));
          reader.setLinesToSkip(1);
          reader.setStrict(false);
          reader.setLineMapper(new DefaultLineMapper&lt;Bill&gt;() {{
                  setLineTokenizer(new DelimitedLineTokenizer() {{
                          setNames("first-name", "last-name", "amount");
                          setDelimiter(";");
                      }
                  });
  
                  setFieldSetMapper(new RecordFieldSetMapper());
              }
          });
          return reader;
      }
  
      @Bean
      public ConsoleItemWriter&lt;Bill&gt; writer() {
          return new ConsoleItemWriter&lt;&gt;();
      }
  
      @Bean
      protected JobRepository createJobRepository() throws Exception {
          JobRepositoryFactoryBean factory = new JobRepositoryFactoryBean();
          factory.setDataSource(dataSource);
          factory.setTransactionManager(transactionManager);
          factory.setIsolationLevelForCreate("ISOLATION_SERIALIZABLE");
          factory.setTablePrefix("BATCH_");
          factory.setMaxVarCharLength(1000);
          return factory.getObject();
      }
  }
  </code></pre>
  
  <p>Writer</p>
  
  <pre><code>public class ConsoleItemWriter&lt;T&gt; implements ItemWriter&lt;T&gt; {
      @Override
      public void write(List&lt;? extends T&gt; items) throws Exception {
          for (T item : items) {
              System.out.println(item);
          }
      }
  }
  </code></pre>
  
  <p>Integration config</p>
  
  <pre><code>@Configuration
  @RequiredArgsConstructor
  public class SftpConfig {
      private final Job job;
      private final JobRepository jobRepository;
  
      @Value("${sftp.host}")
      private String sftpHost;
  
      @Value("${sftp.user}")
      private String sftpUser;
  
      @Value("${sftp.password}")
      private String sftpPassword;
  
      @Value("${sftp.port}")
      private int sftpPort;
  
      @Value("${poller.trigger}")
      private int pollerTrigger;
  
      @Bean(name = PollerMetadata.DEFAULT_POLLER)
      public PollerMetadata defaultPoller() {
          PollerMetadata pollerMetadata = new PollerMetadata();
          pollerMetadata.setTrigger(new PeriodicTrigger(pollerTrigger));
          return pollerMetadata;
      }
  
      @Bean
      SftpInboundFileSynchronizer sftpInboundFileSynchronizer() {
          SftpInboundFileSynchronizer fileSynchronizer = new SftpInboundFileSynchronizer(sftpSessionFactory());
          fileSynchronizer.setDeleteRemoteFiles(false);
          fileSynchronizer.setRemoteDirectory("/info");
          return fileSynchronizer;
      }
  
      @Bean
      @InboundChannelAdapter("sftpChannel")
      public MessageSource&lt;File&gt; sftpMessageSource() {
          SftpInboundFileSynchronizingMessageSource source = new SftpInboundFileSynchronizingMessageSource(sftpInboundFileSynchronizer());
          source.setLocalDirectory(new File("/tmp/info"));
          source.setAutoCreateLocalDirectory(true);
          return source;
      }
  
      @Bean
      public SessionFactory&lt;LsEntry&gt; sftpSessionFactory() {
          DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(true);
          factory.setHost(sftpHost);
          factory.setPort(sftpPort);
          factory.setUser(sftpUser);
          factory.setPassword(sftpPassword);
          factory.setAllowUnknownKeys(true);
          return new CachingSessionFactory&lt;&gt;(factory);
      }
  
      @Bean
      @ServiceActivator(inputChannel = "sftpChannel")
      public MessageHandler handler() {
          return message -&gt; System.out.println("transferred");
      }
  
      @Bean
      public FileMessageToJobRequest fileMessageToJobRequest() {
          FileMessageToJobRequest fileMessageToJobRequest = new FileMessageToJobRequest();
          fileMessageToJobRequest.setFileParameterName("input.file.name");
          fileMessageToJobRequest.setJob(job);
          return fileMessageToJobRequest;
      }
  
      @Bean
      public JobLaunchingGateway jobLaunchingGateway() {
          SimpleJobLauncher simpleJobLauncher = new SimpleJobLauncher();
          simpleJobLauncher.setJobRepository(jobRepository);
          simpleJobLauncher.setTaskExecutor(new SyncTaskExecutor());
          JobLaunchingGateway jobLaunchingGateway = new JobLaunchingGateway(simpleJobLauncher);
  
          return jobLaunchingGateway;
      }
  
      @Bean
      public IntegrationFlow integrationFlow() {
          return IntegrationFlows.from(Files.inboundAdapter(new File("/info")).
                          filter(new SimplePatternFileListFilter("*.csv")),
                  c -&gt; c.poller(Pollers.fixedRate(1000).maxMessagesPerPoll(1))).
                  handle(fileMessageToJobRequest()).
                  handle(jobLaunchingGateway()).
                  log(LoggingHandler.Level.WARN, "headers.id + ': ' + payload").
                  get();
      }
  }
  </code></pre>
  
  <p>Model</p>
  
  <pre><code>@Data
  @AllArgsConstructor
  @NoArgsConstructor
  @Accessors(chain = true)
  public class Bill {
      private String firstName;
      private String lastName;
      private String amount;
  }
  </code></pre>
  
  <p>Mapper</p>
  
  <pre><code>import org.springframework.batch.item.file.mapping.FieldSetMapper;
  import org.springframework.batch.item.file.transform.FieldSet;
  
  public class RecordFieldSetMapper implements FieldSetMapper&lt;Bill&gt; {
  
      @Override
      public Bill mapFieldSet(FieldSet fieldSet) {
  
          return new Bill()
                  .setFirstName(fieldSet.readString("first-name"))
                  .setLastName(fieldSet.readString("last-name"));
      }
  }
  </code></pre>
  
  <p>Request</p>
  
  <pre><code>@Setter
  public class FileMessageToJobRequest {
      private Job job;
      private String fileParameterName;
  
      @Transformer
      public JobLaunchRequest toJobLaunchRequest(Message&lt;File&gt; message) {
          JobParametersBuilder jobParametersBuilder = new JobParametersBuilder();
          jobParametersBuilder.addString(fileParameterName, message.getPayload().getAbsolutePath());
  
          return new JobLaunchRequest(job, jobParametersBuilder.toJobParameters());
      }
  }
  </code></pre>
  
  <p>Individually everything works. But in output I dont see information about my records. IntegrationFlow cant call batch job.</p>