Spring batch in memory and OutOfMemoryError

Some time ago I started to use Spring batch for various processing of data. While it was ok for simple usage, when I pushed large file for processing into my system it died with OutOfMemoryError.
So I took the heap dump, analyzed it, and saw some Spring Batch classes that were taking huge amount of my heap.

I googled around and it seems it's not very common to use Spring batch in-memory mode (which is done by registering MapJobRepositoryFactoryBean as job-repository), and OutOfMemory is a quite often problem in those rare cases. Seems like the MapJobRepositoryFactoryBean keeps the track of all your jobs, data, job history and who knows what else in memory. So even with not so complicated jobs, eventually you get the OOM. MapJobRepositoryFactoryBean has some nice clear methods that could be used to clear the data, but how to access it? Not so obvious, so here's the way I did it. If anybody has better idea, please share :)
First, let's create our own JobRepositoryFactoryBean, that will give us access to clear method

public class MyMapJobRepositoryFactoryBean extends MapJobRepositoryFactoryBean {

    @Override
    public Object getObject() throws Exception {
        return new MySimpleJobRepository(createJobInstanceDao(), createJobExecutionDao(), createStepExecutionDao(), createExecutionContextDao());
    }

        public class MySimpleJobRepository extends SimpleJobRepository implements CleanableJobRepository {

        MySimpleJobRepository(final JobInstanceDao jobInstanceDao,
                final JobExecutionDao jobExecutionDao,
                final StepExecutionDao stepExecutionDao,
                final ExecutionContextDao ecDao) {
            super(jobInstanceDao, jobExecutionDao, stepExecutionDao, ecDao);
        }

        @Override
        public void clean() {
            ((MapJobInstanceDao) getJobInstanceDao()).clear();
            ((MapJobExecutionDao) getJobExecutionDao()).clear();
            ((MapStepExecutionDao) getStepExecutionDao()).clear();
            ((MapExecutionContextDao) getExecutionContextDao()).clear();
        }

    }

} 

The CleanableJobRepository is a simple interface

public interface CleanableJobRepository {

    void clean();
}

Now we can use it as our job repository

<bean id="jobRepository"
        class="nadircode.MyMapJobRepositoryFactoryBean" />

The last thing is to do the actual cleanup. Since I already run several jobs, I simply added a new one, that does the cleanup

public class BatchExecutionContextCleanupJob extends QuartzJobBase implements StatefulJob {

    private CleanableJobRepository jobRepository;

    @Override
    protected void executeInternal(final JobExecutionContext arg0) throws JobExecutionException {
        try {
            @SuppressWarnings("unchecked")
            List<JobExecutionContext> currentlyExecutingJobs = scheduler.getCurrentlyExecutingJobs();
            if (onlyThisJobIsRunning(currentlyExecutingJobs)) {
                jobRepository.clean();
            }
        } catch (SchedulerException ex) {
            throw new JobExecutionException(ex);
        }
    }
  //make sure that only this job is running, to not to clear data that still may be used by other jobs!
    private boolean onlyThisJobIsRunning(final List<JobExecutionContext> currentlyExecutingJobs) {
        if (currentlyExecutingJobs.size() > 1) {
            return false;
        }
        JobExecutionContext jobExecutionContext = currentlyExecutingJobs.get(0);
        return jobExecutionContext.getJobDetail().getJobClass().equals(this.getClass());
    }

    public void setJobRepository(final CleanableJobRepository jobRepository) {
        this.jobRepository = jobRepository;
    }

}

Simply register the job in your scheduler, and your memory will be cleaned.
QuartzJobBase is my own QuartzJob extension that gives you access to scheduler.