MultiResourceItemReader оставил последнее чтение файла заблокированным, поэтому не может двигаться после завершения обработки - PullRequest
Новая
       77

MultiResourceItemReader оставил последнее чтение файла заблокированным, поэтому не может двигаться после завершения обработки

0 голосов
/ 29 марта 2020

Идеи о том, почему MultiResourceItemReader оставляет последний файл заблокированным, и я не могу переместить его с помощью Move Tasklet? Ход завершает все остальные файлы, но последний из них читается.

Исключение ввода-вывода имеет: 

java.nio.file.FileSystemException: C:\Users\UGDW\MyProjects\ngsa2\oab-outside-assets-batchlauncher\input\EQ_AcctData_4321_03292020.csv -> C:\Users\UGDW\MyProjects\ngsa2\oab-outside-assets-batchlauncher\output\EQ_AcctData_4321_03292020.csv_processed: The process cannot access the file because it is being used by another process.

Пакетная конфигурация (урезанная): 

<batch:job id="stockPlanAccountDataJob">
    <batch:step id="getFilesInInputDirectory" next="fileProcessing">
        <tasklet ref="getFilesInInputDirectoryTasklet"/>
    </batch:step>

    <batch:step id="fileProcessing" next="moveFilesToOuputDirectory">                         
        <tasklet>
            <chunk reader="stockPlanAccountDataFileReader" processor="stockPlanAccountDataProcessor" writer="stockPlanConsoleItemWriter" 
                    commit-interval="20" skip-limit="20">                   
                <batch:skippable-exception-classes>
                    <batch:include class="java.lang.Exception"/>  
                    <batch:exclude class="org.springframework.batch.item.file.FlatFileParseException"/>  
                </batch:skippable-exception-classes>                        
            </chunk>
        </tasklet>
    </batch:step>

    <batch:step id="moveFilesToOuputDirectory">
        <tasklet ref="stockPlanMoveFilesTasklet"/>
    </batch:step>       
</batch:job>

<bean id="getFilesInInputDirectoryTasklet" class="simplepeekandmulti.GetFilesInInputDirectoryTasklet" scope="step"/>

<bean id="stockPlanAccountDataFileReader" class="simplepeekandmulti.StockPlanAccountDataFileReader" scope="step">
    <property name="delegate" ref="preprocessorUsingPeekable"/>
</bean>  

<bean id="preprocessorUsingPeekable" class="org.springframework.batch.item.support.SingleItemPeekableItemReader" scope="step">
    <property name="delegate" ref="multiFileResourceReader"/>
</bean>      

<bean name="multiFileResourceReader" class="org.springframework.batch.item.file.MultiResourceItemReader" scope="step"> 
    <property name="resources" value="file:#{jobExecutionContext[filepattern]}" />
    <property name="delegate" ref="genericFlatFileReader" />
    <property name="strict" value="true" />
</bean>

<bean id="genericFlatFileReader" class="org.springframework.batch.item.file.FlatFileItemReader" scope="step">
    <property name="lineMapper" ref="genericFileLineMapper"/>    
</bean>

<bean name="genericFileLineMapper" class="org.springframework.batch.item.file.mapping.PassThroughLineMapper" scope="step" />

<bean id="stockPlanAccountDataProcessor" class="simplepeekandmulti.StockPlanAccountDataProcessor" scope="step"/>

<bean id="stockPlanMoveFilesTasklet" class="simplepeekandmulti.StockPlanMoveFilesTasklet" scope="step"/>

Reader (с тупой логикой c): 

package simplepeekandmulti;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;

import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.PeekableItemReader;

import simplepeekandmulti.StockPlanAccountData;
import simplepeekandmulti.StockPlanFileInputAccountData;

public class StockPlanAccountDataFileReader implements ItemReader<StockPlanFileInputAccountData> {

    private PeekableItemReader<String> delegate;
    private AtomicLong itemsRead = new AtomicLong(0L);
    private static final String PIPE = "|";
    private static final String PIPE_SPLIT = "\\|";
    private static final int NUM_RECORDS_PER_LINE = 6;

    public PeekableItemReader<String> getDelegate() {
        return delegate;
    }

    public void setDelegate(PeekableItemReader<String> delegate) {
        this.delegate = delegate;
    }

    @Override
    public StockPlanFileInputAccountData read() throws Exception {
        String currentLine = delegate.read();
        StockPlanFileInputAccountData inputData = new StockPlanFileInputAccountData();
        int recs = 0;
        List<String> errorList = new ArrayList<>();

        while (currentLine != null) {
            if (currentLine.contains(PIPE)) {
                recs++;
                setDetailLine(currentLine, inputData, recs, errorList);
            } else {
                errorList.add(currentLine);
            }
            if ((errorList.size() % 2) == 0) {
                return inputData;
            }
            itemsRead.incrementAndGet();
            currentLine = delegate.read();
        }
        return null;
    }

    private void setDetailLine(String inputLine, StockPlanFileInputAccountData inputData, 
            int numRecs, List<String> errorList) {

        String[] entry = inputLine.split(PIPE_SPLIT);
        if (entry.length == NUM_RECORDS_PER_LINE) {
            inputData.setDataRecordsPerFile(numRecs);
            StockPlanAccountData data = new StockPlanAccountData();
            data.setExternalClientId(entry[0]);
            data.setSSN(entry[1]);
            data.setExternalParticipantId(entry[2]);
            data.setFirstName(entry[3]);
            data.setLastName(entry[4]);
            data.setDateOfBitrth(entry[5]);
            inputData.addToDataList(data);
        } else {
            errorList.add("Detail Line Is Invalid, Does NOT have 6 columns, 5 pipes: " + inputLine);
        }
    }
}

Процессор: 

package simplepeekandmulti;

import java.util.ArrayList;
import java.util.List;


import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.annotation.BeforeStep;
import org.springframework.batch.item.ItemProcessor;

import com.vanguard.inst.batch.oab.springboot.data.StockPlanFileInputAccountData;

public class StockPlanAccountDataProcessor implements ItemProcessor<StockPlanFileInputAccountData, StockPlanFileInputAccountData> {

    private StepExecution stepExecution;

    @BeforeStep
    public void beforeStep(StepExecution stepExecution) {
        this.stepExecution = stepExecution;
    }

    public StockPlanFileInputAccountData process(StockPlanFileInputAccountData item) throws Exception {

        List<String> errorList = new ArrayList<>(0);

        if (errorList.isEmpty()) {
            return item;
        } else {
            //exchangeEmailService.sendEmail(fileName, errorList);
            return null;
        }
    }

}

Writer: 

package simplepeekandmulti;

import java.util.List;

import org.springframework.batch.item.ItemWriter;
import org.springframework.stereotype.Component;

import com.vanguard.inst.batch.oab.springboot.data.StockPlanFileInputAccountData;

@Component
public class StockConsoleOutputItemWriter implements ItemWriter<StockPlanFileInputAccountData> {

    @Override
    public void write(List<? extends StockPlanFileInputAccountData> arg0) throws Exception {
        // TODO Auto-generated method stub

    }
}

Тасклет перемещения файлов (с жестко заданным именем файла): Последний файл в l oop всегда терпит неудачу. 

package simplepeekandmulti;

import java.io.IOException;
import java.nio.file.FileSystems;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.StandardCopyOption;
import java.util.ArrayList;
import java.util.List;

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;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

@Component
public class StockPlanMoveFilesTasklet implements Tasklet {

    private static final String CLASS_NAME = StockPlanMoveFilesTasklet.class.getSimpleName();

    @Value("$simplepeekandmulti-{INPUT_DIR}")
    private String inputDir;

    @Value("$simplepeekandmulti-{OUTPUT_DIR}")
    private String outputDir;

    private static final String PROCESSED = "_processed";

    public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) {

        String[] fileList = {"EQ_AcctData_3210_03302020.csv", "EQ_AcctData_4321_03302020.csv"};

        try {
            for (String fileName : fileList) {
                Path pathFrom = FileSystems.getDefault().getPath(inputDir, fileName);
                Path pathTo = FileSystems.getDefault().getPath(outputDir, fileName + PROCESSED);
                Files.move(pathFrom, pathTo, StandardCopyOption.REPLACE_EXISTING);
            }
        } catch (IOException io) {
            System.out.println(io.toString());
        }
        return RepeatStatus.FINISHED;
    }
}

CSV файлы просто имеют; дата заголовка, разделитель строк с записями, общее количество записей нижнего колонтитула

03/30/2020

3210 | 59658625 | 12000 | AADFBCJGH | LLOQMNURS | 1962-03-08

3210 | 10124602 | 12001 | AADFBCJGH | LLOQMNURS | 1962-03-08

2

03/30/2020

4321 | 5690154 | 13000 | AADFBCJGH | LLOQMNURS | 1988-10-23

4321 | 745701 | 13001 | AADFBCJGH | LLOQMNURS | 1988-10-23

2

...