title: How to optimize my performances using FlatFileItemReader and Asynchronous Processors
body: <p>I have a simple csv file with ~400,000 line(one column only)
  It takes me alot of time to read the records and process them</p>
  
  <p>the processor validating records against couchbase</p>
  
  <p>the writer - writing into remote topic
  Takes me around 30 mins. thats insane. </p>
  
  <p>I read that flatfileItemreader is not thread safe. so my chunk value is 1. </p>
  
  <p>I read the Asynchronous processing could assist. but I cant see any improvements.</p>
  
  <p>Thats my code:</p>
  
  <pre><code>@Configuration
  @EnableBatchProcessing
  public class NotificationFileProcessUploadedFileJob {
  
  
      @Value("${expected.snid.header}")
      public String snidHeader;
  
      @Value("${num.of.processing.chunks.per.file}")
      public int numOfProcessingChunksPerFile;
  
      @Autowired
      private InfrastructureConfigurationConfig infrastructureConfigurationConfig;
  
      private static final String OVERRIDDEN_BY_EXPRESSION = null;
  
  
      @Inject
      private JobBuilderFactory jobs;
  
      @Inject
      private StepBuilderFactory stepBuilderFactory;
  
      @Inject
      ExecutionContextPromotionListener executionContextPromotionListener;
  
  
      @Bean
      public Job processUploadedFileJob() throws Exception {
          return this.jobs.get("processUploadedFileJob").start((processSnidUploadedFileStep())).build();
  
      }
  
      @Bean
      public Step processSnidUploadedFileStep() {
          return stepBuilderFactory.get("processSnidFileStep")
                  .&lt;PushItemDTO, PushItemDTO&gt;chunk(numOfProcessingChunksPerFile)
                  .reader(snidFileReader(OVERRIDDEN_BY_EXPRESSION))
                  .processor(asyncItemProcessor())
                  .writer(asyncItemWriter())
              //    .throttleLimit(20)
               //   .taskJobExecutor(infrastructureConfigurationConfig.taskJobExecutor())
  
  
                          //     .faultTolerant()
                          //   .skipLimit(10) //default is set to 0
                          //     .skip(MySQLIntegrityConstraintViolationException.class)
                  .build();
      }
  
      @Inject
      ItemWriter writer;
  
      @Bean
      public AsyncItemWriter asyncItemWriter() {
          AsyncItemWriter asyncItemWriter=new AsyncItemWriter();
          asyncItemWriter.setDelegate(writer);
          return asyncItemWriter;
      }
  
  
      @Bean
      @Scope(value = "step", proxyMode = ScopedProxyMode.INTERFACES)
      public ItemStreamReader&lt;PushItemDTO&gt; snidFileReader(@Value("#{jobParameters[filePath]}") String filePath) {
          FlatFileItemReader&lt;PushItemDTO&gt; itemReader = new FlatFileItemReader&lt;PushItemDTO&gt;();
          itemReader.setLineMapper(snidLineMapper());
          itemReader.setLinesToSkip(1);
          itemReader.setResource(new FileSystemResource(filePath));
          return itemReader;
      }
  
  
      @Bean
      public AsyncItemProcessor asyncItemProcessor() {
  
          AsyncItemProcessor&lt;PushItemDTO, PushItemDTO&gt; asyncItemProcessor = new AsyncItemProcessor();
  
          asyncItemProcessor.setDelegate(processor(OVERRIDDEN_BY_EXPRESSION, OVERRIDDEN_BY_EXPRESSION, OVERRIDDEN_BY_EXPRESSION,
                  OVERRIDDEN_BY_EXPRESSION, OVERRIDDEN_BY_EXPRESSION, OVERRIDDEN_BY_EXPRESSION, OVERRIDDEN_BY_EXPRESSION));
          asyncItemProcessor.setTaskExecutor(infrastructureConfigurationConfig.taskProcessingExecutor());
  
          return asyncItemProcessor;
  
      }
  
      @Scope(value = "step", proxyMode = ScopedProxyMode.INTERFACES)
      @Bean
      public ItemProcessor&lt;PushItemDTO, PushItemDTO&gt; processor(@Value("#{jobParameters[pushMessage]}") String pushMessage,
                                                               @Value("#{jobParameters[jobId]}") String jobId,
                                                               @Value("#{jobParameters[taskId]}") String taskId,
                                                               @Value("#{jobParameters[refId]}") String refId,
                                                               @Value("#{jobParameters[url]}") String url,
                                                               @Value("#{jobParameters[targetType]}") String targetType,
                                                               @Value("#{jobParameters[gameType]}") String gameType) {
          return new PushItemProcessor(pushMessage, jobId, taskId, refId, url, targetType, gameType);
      }
  
      @Bean
      public LineMapper&lt;PushItemDTO&gt; snidLineMapper() {
          DefaultLineMapper&lt;PushItemDTO&gt; lineMapper = new DefaultLineMapper&lt;PushItemDTO&gt;();
          DelimitedLineTokenizer lineTokenizer = new DelimitedLineTokenizer();
          lineTokenizer.setDelimiter(",");
          lineTokenizer.setStrict(true);
          lineTokenizer.setStrict(true);
          String[] splittedHeader = snidHeader.split(",");
          lineTokenizer.setNames(splittedHeader);
          BeanWrapperFieldSetMapper&lt;PushItemDTO&gt; fieldSetMapper = new BeanWrapperFieldSetMapper&lt;PushItemDTO&gt;();
          fieldSetMapper.setTargetType(PushItemDTO.class);
  
          lineMapper.setLineTokenizer(lineTokenizer);
          lineMapper.setFieldSetMapper(new PushItemFieldSetMapper());
          return lineMapper;
      }
  }
  
  
   @Bean
      @Override
      public SimpleAsyncTaskExecutor taskProcessingExecutor() {
          SimpleAsyncTaskExecutor simpleAsyncTaskExecutor = new SimpleAsyncTaskExecutor();
          simpleAsyncTaskExecutor.setConcurrencyLimit(300);
          return simpleAsyncTaskExecutor;
      }
  </code></pre>
  
  <p>How do you think I could improve the processing performances and make them faster?
  thank you</p>
  
  <p>ItemWriter code:</p>
  
  <pre><code> @Bean
      public ItemWriter writer() {
          return new KafkaWriter();
      }
  
  
  public class KafkaWriter implements ItemWriter&lt;PushItemDTO&gt; {
  
  
      private static final Logger logger = LoggerFactory.getLogger(KafkaWriter.class);
  
      @Autowired
      KafkaProducer kafkaProducer;
  
      @Override
      public void write(List&lt;? extends PushItemDTO&gt; items) throws Exception {
  
          for (PushItemDTO item : items) {
              try {
                  logger.debug("Writing to kafka=" + item);
                  sendMessageToKafka(item);
              } catch (Exception e) {
                  logger.error("Error writing item=" + item.toString(), e);
              }
          }
      }
  </code></pre>