Extending CompletionPolicy to complete/commit per resource from MultiResourceItemReader

Question

My problem is I don't understand the various spring batch contexts. The reference documentation explains how to pass data to future steps. But how do I pass data between the reader and writer components within a step. The step context. Is there a chunk context? I have used before the execution context while writing a Partitioner before. But those were executing in parallel.

I now need to do an ordered operation. It's basically a jdbc import job but each file needs to be committed in order else their are foreign key constraint volitions.

The simplest place I'd be able get the line count for an individual file resource is in MultiResourceItemReader before it delegates to ItemReader. But having had a look at various CompletionPolicy implementations they only seem to have access to the RepeatContext. How can I store a value in the RepeatContext from the MultiResourceItemReader so that my CompletionPolicy can access it and commit after the specific file resource line count.

An example of how to extend abstract CountingCompletionPolicy and where how to store data from MultiResourceItemReader would be helpful.

Or perhaps there is a better way to appraoch this type of job.

<!-- <job id="job" restartable="${restartable}" xmlns="http://www.springframework.org/schema/batch"> -->
    <batch:job id="job" restartable="true"
        xmlns="http://www.springframework.org/schema/batch">
        <batch:step id="step1-unzipFile">
            <batch:tasklet ref="unzipFileTasklet" />
            <batch:next on="COMPLETED" to="step2-import" />
        </batch:step>
        <batch:step id="step2-import"> <!-- we can't use a commit-interval="${commitInterval} cause it messes with 2nd pass import processing if the commit ends up being the middle of the file -->
            <batch:tasklet>
                <batch:chunk reader="multiResourceReader" writer="itemWriter" chunk-completion-policy="completionPolicy"/>  
            </batch:tasklet>
            <!-- <batch:next on="COMPLETED" to="step3-fileCleanUp" /> -->
        </batch:step>
        <!-- <batch:step id="step3-fileCleanUp">
            <batch:tasklet ref="fileCleanUpTasklet" />
        </batch:step> -->
    </batch:job>




    <bean id="multiResourceReader" class="springbatch.iimport.extended.SequentialLoaderMultiFileResourceItemReader" scope="step">
        <property name="resourceDirectoryPath" value="${importTempDirectoryBasePath}/#{jobParameters['jobKey']}/"/>
        <property name="delegate" ref="itemReader"/>
    </bean>

    <bean id="itemReader" class="org.springframework.batch.item.file.FlatFileItemReader">
        <property name="lineMapper" ref="lineMapper" />
    </bean>

    <bean id="lineMapper" class="springbatch.iimport.extended.JsonToTupleLineMapper"/>

    <bean id="itemWriter" class="springbatch.iimport.extended.TupleJdbcBatchItemWriter" scope="step">
        <property name="moduleDataSource" ref="moduleDataSource"/>
        <property name="dataSource" ref="dataSource"/>
        <property name="jobKey" value="#{jobParameters['jobKey']}"/>
        <property name="jobDef" value="#{jobParameters['jobDef']}"/>
    </bean>

    <bean id="completionPolicy" class="?"/> 

    <!-- tasklets -->

    <bean id="unzipFileTasklet" class="springbatch.iimport.tasklets.UnZipFile" scope="step">
        <!-- the temp directory the files are unzipped to end up being #{jobParameters['importZipFileName']} -->
        <property name="importZipFileName" value="${uploadDir}/#{jobParameters['importZipFileName']}" />
        <property name="jobKey" value="#{jobParameters['jobKey']}"/>
        <property name="importTempDirectoryBasePath" value="${importTempDirectoryBasePath}" />
     </bean>

Luca Basso Ricci Luca Basso Ricci · Accepted Answer · 2015-04-16T08:15:23

You can write your own CompletionPolicy which look-ahead itemReader (using a PeekableItemReader) and return current chunk as 'completed' if next call to itemReader.next() returns null (means EOF for current file).
Remember: this solution can leads you to memory problems due to full in-memory file content reading.

public class EOFCompletionPolicy extends CompletionPolicySupport
{
    private EOFCompletionContext cc;
    private PeekableItemReader<Object> reader;

    public void setReader(PeekableItemReader<Object> forseeingReader)
    {
        this.reader = forseeingReader;
    }

    @Override
    public boolean isComplete(RepeatContext context)
    {
        return this.cc.isComplete();
    }

    @Override
    public RepeatContext start(RepeatContext context)
    {
        this.cc = new EOFCompletionContext(context);
        return cc;
    }

    @Override
    public void update(RepeatContext context)
    {
        this.cc.update();
    }

    protected class EOFCompletionContext extends RepeatContextSupport
    {
        boolean eof = false;
        public EOFCompletionContext (RepeatContext context)
        {
            super(context);
        }

        public void update()
        {
            final Object next;
            try
            {
                next = reader.peek();
            }
            catch (Exception e)
            {
                throw new NonTransientResourceException("Unable to peek", e);
            }
            // EOF?
            this.eof = (next == null);
        }

        public boolean isComplete() {
            return this.eof;
        }
    }
}

Extending CompletionPolicy to complete/commit per resource from MultiResourceItemReader

1 Answers