Help with MultiResourceItemReader - file archiving and logging

[sameer794]

Hi,

In our application, text files within a folder are read using MultiResourceItemReader, and data is updated in a database.

everything is working fine for me but I am unable to achieve two things

1) Log the actual file name before and after it is processed
2) after the file is read, move it to an archive folder.

Below is my job configuration

<job id="samplejob" xmlns="http://www.springframework.org/schema/batch">
	<step id="processStep">
		<tasklet>
			<chunk reader="multiResourceReader" writer="customItemWriter"
				processor="validatingItemProcessor" commit-interval="1" skip-limit="4">
				<skippable-exception-classes>
					<include
						class="org.springframework.batch.item.file.FlatFileParseException" />
				</skippable-exception-classes>
				<streams>
					<stream ref="errorItemWriter" />
				</streams>
			</chunk>
		</tasklet>
	</step>
</job>

<bean id="multiResourceReader"
	class="org.springframework.batch.item.file.FlatFileParseException"
	scope="step">
	<property name="resources" value="#{jobParameters[fileInputPath]}" />
	<property name="delegate" ref="flatFileItemReader" />
</bean>

	<!--
		flatFileItemReader is a simple reader reading delimited text and we
		have a custom writer to update multiple tables.
	-->

With ItemReadListener, I get items within the files but not the path of the file read by the MRIR. I have tried different listener configurations but unable to get access to the file being processed to achieve my objectives. I know I have to write a tasklet to achieve the file archiving but unable to do so.

Any hints in this regard would be of great help to me.
And i should admit, spring batch is cool...

Thanks in advance

[Dave Syer]

That's an interesting requirement. Listening to the item-level operations is too low level anyway though, right? And you would only want to do the archiving once the step had completed successfully (otherwise a restart would have the wrong resources to read).

So what I would do is inject the same Resources[] into another bean in the same step, like a StepExecutionListener, and then do the archiving in one go at the end.

To log each file before and after is a little bit more challenging, but if your listener calls update() on the MultiResourceItemReader it will get a counter pointing to the current resource, so you can use that together with the Resources[] to locate the one that is being processed.

[sameer794]

thanks for the reply Dave. I would try your options and get back.

[smomula]

Hi Dave,

As you said, in StepExecutionListener, if i call update() on the MultiResourceItemReader, i can get the current resource. I'm trying to configure this, but it throws $proxy10 exception.

Here is the configuration which i used.
ApplicationContextProvider is a Bean which implements ApplicationContextAware to get the MultiResourceItemReader

<bean id="applicationContextProvider" class="org.springframework.sample.batch.example.ApplicationContextProvider"/>
	
<bean id="stepListener" class="org.springframework.sample.batch.example.ExampleStepExecutionListener">
		<property name="applicationContextProvider" ref="applicationContextProvider"/>
	</bean>

And here what i do in the after method of stepExecutionListener.

public ExitStatus afterStep(StepExecution stepExecution) {
		ApplicationContext applicationContext = this.applicationContextProvider.getApplicationContext();	
		
		log.info("object type is : " + applicationContext.getType("reader").getName());
		MultiResourceItemReader multiResourceItemReader = (MultiResourceItemReader) applicationContext.getBean("reader");
		
		ExecutionContext executionContext = stepExecution.getExecutionContext();
		multiResourceItemReader.update(executionContext);
		log.info("current file name (after): " + multiResourceItemReader.getCurrentResource().getFilename());
		
		return ExitStatus.COMPLETED;
	}

Here is the full Exception stack trace..

java.lang.IllegalArgumentException: Unable to invoke method: [public org.springframework.batch.core.ExitStatus org.springframework.sample.batch.example.ExampleStepExecutionListener.afterStep(org.springframework.batch.core.StepExecution)] on object: [org.springframework.sample.batch.example.ExampleStepExecutionListener@48bc64] with arguments: [[StepExecution: id=1, version=22, name=step1, status=COMPLETED, exitStatus=COMPLETED, readCount=20, filterCount=0, writeCount=20 readSkipCount=0, writeSkipCount=0, processSkipCount=0, commitCount=21, rollbackCount=0, exitDescription=]]
	at org.springframework.batch.support.SimpleMethodInvoker.invokeMethod(SimpleMethodInvoker.java:108)
	at org.springframework.batch.core.listener.MethodInvokerMethodInterceptor.invoke(MethodInvokerMethodInterceptor.java:68)
	at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:172)
	at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:202)
	at $Proxy11.afterStep(Unknown Source)
	at org.springframework.batch.core.listener.CompositeStepExecutionListener.afterStep(CompositeStepExecutionListener.java:62)
	at org.springframework.batch.core.step.AbstractStep.execute(AbstractStep.java:223)
	at org.springframework.batch.core.job.SimpleStepHandler.handleStep(SimpleStepHandler.java:135)
	at org.springframework.batch.core.job.flow.JobFlowExecutor.executeStep(JobFlowExecutor.java:61)
	at org.springframework.batch.core.job.flow.support.state.StepState.handle(StepState.java:60)
	at org.springframework.batch.core.job.flow.support.SimpleFlow.resume(SimpleFlow.java:144)
	at org.springframework.batch.core.job.flow.support.SimpleFlow.start(SimpleFlow.java:124)
	at org.springframework.batch.core.job.flow.FlowJob.doExecute(FlowJob.java:135)
	at org.springframework.batch.core.job.AbstractJob.execute(AbstractJob.java:281)
	at org.springframework.batch.core.launch.support.SimpleJobLauncher$1.run(SimpleJobLauncher.java:120)
	at org.springframework.core.task.SyncTaskExecutor.execute(SyncTaskExecutor.java:48)
	at org.springframework.batch.core.launch.support.SimpleJobLauncher.run(SimpleJobLauncher.java:114)
	at org.springframework.sample.batch.example.ExampleJobConfigurationTests.testLaunchJob(ExampleJobConfigurationTests.java:50)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
	at java.lang.reflect.Method.invoke(Method.java:585)
	at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:44)
	at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:15)
	at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:41)
	at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:20)
	at org.springframework.test.context.junit4.statements.RunBeforeTestMethodCallbacks.evaluate(RunBeforeTestMethodCallbacks.java:74)
	at org.springframework.test.context.junit4.statements.RunAfterTestMethodCallbacks.evaluate(RunAfterTestMethodCallbacks.java:82)
	at org.springframework.test.context.junit4.statements.SpringRepeat.evaluate(SpringRepeat.java:72)
	at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.runChild(SpringJUnit4ClassRunner.java:240)
	at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:50)
	at org.junit.runners.ParentRunner$3.run(ParentRunner.java:193)
	at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:52)
	at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:191)
	at org.junit.runners.ParentRunner.access$000(ParentRunner.java:42)
	at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:184)
	at org.springframework.test.context.junit4.statements.RunBeforeTestClassCallbacks.evaluate(RunBeforeTestClassCallbacks.java:61)
	at org.springframework.test.context.junit4.statements.RunAfterTestClassCallbacks.evaluate(RunAfterTestClassCallbacks.java:70)
	at org.junit.runners.ParentRunner.run(ParentRunner.java:236)
	at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.run(SpringJUnit4ClassRunner.java:180)
	at org.eclipse.jdt.internal.junit4.runner.JUnit4TestReference.run(JUnit4TestReference.java:46)
	at org.eclipse.jdt.internal.junit.runner.TestExecution.run(TestExecution.java:38)
	at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.runTests(RemoteTestRunner.java:467)
	at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.runTests(RemoteTestRunner.java:683)
	at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.run(RemoteTestRunner.java:390)
	at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.main(RemoteTestRunner.java:197)
Caused by: java.lang.reflect.InvocationTargetException
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
	at java.lang.reflect.Method.invoke(Method.java:585)
	at org.springframework.batch.support.SimpleMethodInvoker.invokeMethod(SimpleMethodInvoker.java:105)
	... 45 more
Caused by: java.lang.ClassCastException: $Proxy10
	at org.springframework.sample.batch.example.ExampleStepExecutionListener.afterStep(ExampleStepExecutionListener.java:50)
	... 50 more

It clearly says ClassCastException, i think MultiResourceItemReader is now a proxy bean, so how exactly i can get this bean to call update method.

Is there any alternative to know what is the current file that is being processed in case of multiple file processing.

Thanks in advance.

[Dave Syer]

ApplicationContextProvider is a Bean which implements ApplicationContextAware to get the MultiResourceItemReader

Not sure why you need to do that - sounds like a dependency lookup anti-pattern. Can't you just inject it?

It clearly says ClassCastException, i think MultiResourceItemReader is now a proxy bean, so how exactly i can get this bean to call update method.

The update() method is on the ItemStream interface. Use the interface instead of the concrete class.

[smomula]

Hi Dave,

If i inject MultiResourceItemReader into my stepExecutionListener, it still throws this class cast exception ($proxy10).

I've three files in a folder :
file-1.txt
file-2.txt
file-3.txt

I want to process each file one by one and print what file has been processed at end of each step and once all the files are processed, print all the file names to log.

Could you please guide me to some testcase or approach which can accomplish this.

Thanks in advance.

Thanks,
Sandeep K Momula.

[Dave Syer]

I was thinking of something simple:

public class ExampleStepExecutionListener extends StepExecutionListenerSupport {
    private Resource[] resources;
    public void setResources(Resources[] resources) { this.resources=resources; }
    public ExitStatus afterStep(StepExecution execution) {
        // do stuff with the resources...
        return execution.getExitStatus();
    }
}

I don't understand what you are doing with the reader. Maybe if you posted the listener implementation it would be clearer?

[smomula]

Hi Dave,

Thanks for the reply.

I don't understand what you are doing with the reader. Maybe if you posted the listener implementation it would be clearer?

Here is my afterStep() method of my stepExecutionListener class.

public ExitStatus afterStep(StepExecution stepExecution) {
		ApplicationContext applicationContext = this.applicationContextProvider.getApplicationContext();	
		
		log.info("object type is : " + applicationContext.getType("reader").getName());
		MultiResourceItemReader multiResourceItemReader = (MultiResourceItemReader) applicationContext.getBean("reader");
		
		ExecutionContext executionContext = stepExecution.getExecutionContext();
		multiResourceItemReader.update(executionContext);
		log.info("current file name (after): " + multiResourceItemReader.getCurrentResource().getFilename());
		
		return ExitStatus.COMPLETED;
	}

If i read the rest of this thread, the approach you mentioned to inject MultiResourceItemReader into stepExecutionListener, but it throws exception.

Here is the stepExecutionListener configuration with MultiResourceItemReader as property.

<bean id="stepListener" class="org.springframework.sample.batch.example.ExampleStepExecutionListener">
		<property name="reader" ref="multiResourceItemReader"/>
	</bean>

The above code throws exception.

I agree with the approach of injecting resources array into listener, but how to get current resource to say i finished processing file-1.txt out of
file-1.txt
file-2.txt
file-3.txt


I need to call update() method on reader class right, but i'm unable to inject that reader class into my listener. Am i doing something wrong? Please correct me.

Thanks,
Sandeep.

[Dave Syer]

update() is a method on ItemStream, so the signature of the setter should have a parameter that is an ItemStream not an ItemReader.

[anish555]

For a similar requirement, I have extended FlatFileItemReader and overriden setResource method. Within setResource method, you can have logic to log the current File name. This class is your delegate.


See below:

public class XYZFileItemReader extends FlatFileItemReader<String>{
	
	private static final Logger logger = Logger.getLogger("......");
	
		
	/**
	 * Overriden setResource method.
	 */
	@Override
	public void setResource(Resource resource) {
	logger.error("Start Processing "+resource.getFilename());
		super.setResource(resource);
	}
	}