Spring Batch MultiLineItemReader with MultiResourcePartitioner

I have a File which has Multiline data like this. DataID is Start of a new record. e.g. One record is a combination of ID and concatenating below line until the start of a new record.

    >DataID1
    Line1asdfsafsdgdsfghfghfghjfgjghjgxcvmcxnvm
    Line2asdfsafsdgdsfghfghfghjfgjghjgxcvmcxnvm
    Line3asdfsafsdgdsfghfghfghjfgjghjgxcvmcxnvm
    >DataID2
    DataID2asdfsafsdgdsfghfghfghjfgjghjgxcvmcxnvm
    >DataID3
    DataID2asdfsafsdgdsfghfghfghjfgjghjgxcvmcxnvm

I was able to implement this using SingleItemPeekableItemReader and it's working fine.

I am not trying to implement partition, As we need to process multiple files. I am not sure how the partitioner is passing file info to my customer reader and how to make my SingleItemPeekableItemReader thread safe as it not working correctly

Need some inputs as I am stuck at this point

java-config

@Bean
      public Partitioner partitioner() {
          MultiResourcePartitioner partitioner = new MultiResourcePartitioner();
          partitioner.setResources(resources);          
          partitioner.partition(10);      
          return partitioner;
      }
      @Bean
      public TaskExecutor taskExecutor() {
          ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
          taskExecutor.setMaxPoolSize(4);
          taskExecutor.setCorePoolSize(4);
          taskExecutor.setQueueCapacity(8);
          taskExecutor.afterPropertiesSet();
          return taskExecutor;
      }   

      @Bean
      @Qualifier("masterStep")
      public Step masterStep() {
          return stepBuilderFactory.get("masterStep")                  
                  .partitioner("step1",partitioner())
                  .step(step1())
                  .taskExecutor(taskExecutor())                  
                  .build();
      }

     @Bean
      public MultiResourceItemReader<FieldSet> multiResourceItemReader() {
        log.info("Total Number of Files to be process {}",resources.length);
        report.setFileCount(resources.length);
        MultiResourceItemReader<FieldSet> resourceItemReader = new MultiResourceItemReader<FieldSet>();     
        resourceItemReader.setResources(resources);     
        resourceItemReader.setDelegate(reader());       
        return resourceItemReader;
      }

    @Bean
    public FlatFileItemReader<FieldSet> reader() {
         FlatFileItemReader<FieldSet> build = new FlatFileItemReaderBuilder<FieldSet>().name("fileReader")              
                .lineTokenizer(orderFileTokenizer())
                .fieldSetMapper(new FastFieldSetMapper())                   
                .recordSeparatorPolicy(new BlankLineRecordSeparatorPolicy())
                .build();        
         build.setBufferedReaderFactory(gzipBufferedReaderFactory);
         return build;
    }

    @Bean
    public SingleItemPeekableItemReader<FieldSet> readerPeek() {
        SingleItemPeekableItemReader<FieldSet> reader = new SingleItemPeekableItemReader<>();
        reader.setDelegate(multiResourceItemReader());
        return reader;
    }

    @Bean
    public MultiLineFastaItemReader itemReader() {
        MultiLineFastaItemReader itemReader = new MultiLineFastaItemReader(multiResourceItemReader());
        itemReader.setSingalPeekable(readerPeek());     
        return itemReader;
    }

    @Bean
    public PatternMatchingCompositeLineTokenizer orderFileTokenizer() {
        PatternMatchingCompositeLineTokenizer tokenizer = new PatternMatchingCompositeLineTokenizer();
        Map<String, LineTokenizer> tokenizers = new HashMap<>(2);
        tokenizers.put(">*", head());
        tokenizers.put("*", tail());
        tokenizer.setTokenizers(tokenizers);
        return tokenizer;
    }

    public DelimitedLineTokenizer head() {
        DelimitedLineTokenizer token = new DelimitedLineTokenizer();
        token.setNames("sequenceIdentifier");
        token.setDelimiter(" ");
        token.setStrict(false);
        return token;
    }

    public DelimitedLineTokenizer tail() {
        DelimitedLineTokenizer token = new DelimitedLineTokenizer();
        token.setNames("sequences");
        token.setDelimiter(" ");
        return token;
    }

    @Bean
    public FastReportWriter writer() {
        return new FastReportWriter();
    }

    @Bean
    public Job importUserJob(JobCompletionNotificationListener listener, Step step1) {
        return jobBuilderFactory.get("importUserJob")
                .incrementer(new RunIdIncrementer())
                .listener(listener)
                .flow(masterStep())
                //.flow(step1)
                .next(step2())
                .end()
                .build();
    }

    @Bean
    public Step step1() {
        return stepBuilderFactory.get("step1")
                .<Fasta, Fasta>chunk(5000)
                .reader(itemReader())
                .processor(new FastaIteamProcessor())
                //.processor(new PassThroughItemProcessor<>())
                .writer(writer())
                .build();
    }

public class MultiLineFastaItemReader implements ItemReader<Fasta>, ItemStream {

    private static final Logger log = LoggerFactory.getLogger(MultiLineFastaItemReader.class);
    private SingleItemPeekableItemReader<FieldSet> singalPeekable;

    AtomicInteger iteamCounter = new AtomicInteger(0);

    ConcurrentHashMap<String, AtomicInteger> fileNameAndCounterMap = new ConcurrentHashMap<>();

    @Autowired
    private SequenceFastaReport sequenceFastaReport;

    private MultiResourceItemReader<FieldSet> resourceItemReader;

    public MultiLineFastaItemReader(MultiResourceItemReader<FieldSet> multiResourceItemReader) {

        this.resourceItemReader = multiResourceItemReader;
    }

    public SingleItemPeekableItemReader<FieldSet> getSingalPeekable() {
        return singalPeekable;
    }

    public void setSingalPeekable(SingleItemPeekableItemReader<FieldSet> singalPeekable) {
        this.singalPeekable = singalPeekable;

    }

    @Override
    public Fasta read() throws Exception {
        FieldSet item = singalPeekable.read();
        if (item == null) {
            return null;
        }
        Fasta fastaObject = new Fasta();

        log.info("ID {} fileName {}", item.readString(0), resourceItemReader.getCurrentResource());
        fastaObject.setSequenceIdentifier(item.readString(0)
                .toUpperCase());
        fastaObject.setFileName(resourceItemReader.getCurrentResource()
                .getFilename());

        if (!fileNameAndCounterMap.containsKey(fastaObject.getFileName())) {
            fileNameAndCounterMap.put(fastaObject.getFileName(), new AtomicInteger(0));

        }

        while (true) {

            FieldSet possibleRelatedObject = singalPeekable.peek();
            if (possibleRelatedObject == null) {
                if (fastaObject.getSequenceIdentifier()
                        .length() < 1)
                    throw new InvalidParameterException("Somwthing Wrong in file");
                sequenceFastaReport.addToReport(fileNameAndCounterMap.get(fastaObject.getFileName())
                        .incrementAndGet(), fastaObject.getSequences());
                return fastaObject;
            }

            if (possibleRelatedObject.readString(0)
                    .startsWith(">")) {
                if (fastaObject.getSequenceIdentifier()
                        .length() < 1)
                    throw new InvalidParameterException("Somwthing Wrong in file");

                sequenceFastaReport.addToReport(fileNameAndCounterMap.get(fastaObject.getFileName())
                        .incrementAndGet(), fastaObject.getSequences());

                return fastaObject;
            }
            String data = fastaObject.getSequences()
                    .toUpperCase();
            fastaObject.setSequences(data + singalPeekable.read()
                    .readString(0)
                    .toUpperCase());

        }

    }

    @Override
    public void close() {
        this.singalPeekable.close();
    }

    @Override
    public void open(ExecutionContext executionContext) {
        this.singalPeekable.open(executionContext);

    }

    @Override
    public void update(ExecutionContext executionContext) {

        this.singalPeekable.update(executionContext);
    }

}