5

I am trying to launch a job in Spring Batch 2, and I need to pass some information in the job parameters, but I do not want it to count for the uniqueness of the job instance. For example, I'd want these two sets of parameters to be considered unique:

file=/my/file/path,session=1234
file=/my/file/path,session=5678

The idea is that there will be two different servers trying to start the same job, but with different sessions attached to them. I need that session number in both cases. Any ideas?

Thanks!

3 Answers 3

4

So, if 'file' is the only attribute that's supposed to be unique and 'session' is used by downstream code, then your problem matches almost exactly what I had. I had a JMSCorrelationId that i needed to store in the execution context for later use and I didn't want it to play into the job parameters' uniqueness. Per Dave Syer, this really wasn't possible, so I took the route of creating the job with the parameters (not the 'session' in your case), and then adding the 'session' attribute to the execution context before anything actually runs.

This gave me access to 'session' downstream but it was not in the job parameters so it didn't affect uniqueness.

References

https://jira.springsource.org/browse/BATCH-1412

http://forum.springsource.org/showthread.php?104440-Non-Identity-Job-Parameters&highlight=

You'll see from this forum that there's no good way to do it (per Dave Syer), but I wrote my own launcher based on the SimpleJobLauncher (in fact I delegate to the SimpleLauncher if a non-overloaded method is called) that has an overloaded method for starting a job that takes a callback interface that allows contribution of parameters to the execution context while not being 'true' job parameters. You could do something very similar.

I think the applicable LOC for you is right here: jobExecution = jobRepository.createJobExecution(job.getName(), jobParameters);

    if (contributor != null) {
        if (contributor.contributeTo(jobExecution.getExecutionContext())) {
            jobRepository.updateExecutionContext(jobExecution);
        }
    }

which is where, after execution context creatin, the execution context is added to. Hopefully this helps you in your implementation.

public class ControlMJobLauncher implements JobLauncher, InitializingBean {

    private JobRepository jobRepository;
    private TaskExecutor taskExecutor;
    private SimpleJobLauncher simpleLauncher;
    private JobFilter jobFilter;

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

    public void setTaskExecutor(TaskExecutor taskExecutor) {
        this.taskExecutor = taskExecutor;
    }

    /**
     * Optional filter to prevent job launching based on some specific criteria.
     * Jobs that are filtered out will return success to ControlM, but will not run
     */
    public void setJobFilter(JobFilter jobFilter) {
        this.jobFilter = jobFilter;
    }

    public JobExecution run(final Job job, final JobParameters jobParameters, ExecutionContextContributor contributor)
            throws JobExecutionAlreadyRunningException, JobRestartException,
            JobInstanceAlreadyCompleteException, JobParametersInvalidException, JobFilteredException {

        Assert.notNull(job, "The Job must not be null.");
        Assert.notNull(jobParameters, "The JobParameters must not be null.");

        //See if job is filtered
        if(this.jobFilter != null && !jobFilter.launchJob(job, jobParameters)) {
            throw new JobFilteredException(String.format("Job has been filtered by the filter: %s", jobFilter.getFilterName()));
        }

        final JobExecution jobExecution;

        JobExecution lastExecution = jobRepository.getLastJobExecution(job.getName(), jobParameters);
        if (lastExecution != null) {
            if (!job.isRestartable()) {
                throw new JobRestartException("JobInstance already exists and is not restartable");
            }
            logger.info(String.format("Restarting job %s instance %d", job.getName(), lastExecution.getId()));
        }

        // Check the validity of the parameters before doing creating anything
        // in the repository...
        job.getJobParametersValidator().validate(jobParameters);

        /*
         * There is a very small probability that a non-restartable job can be
         * restarted, but only if another process or thread manages to launch
         * <i>and</i> fail a job execution for this instance between the last
         * assertion and the next method returning successfully.
         */
        jobExecution = jobRepository.createJobExecution(job.getName(),
                jobParameters);

        if (contributor != null) {
            if (contributor.contributeTo(jobExecution.getExecutionContext())) {
                jobRepository.updateExecutionContext(jobExecution);
            }
        }

        try {
            taskExecutor.execute(new Runnable() {

                public void run() {
                    try {
                        logger.info("Job: [" + job
                                + "] launched with the following parameters: ["
                                + jobParameters + "]");
                        job.execute(jobExecution);
                        logger.info("Job: ["
                                + job
                                + "] completed with the following parameters: ["
                                + jobParameters
                                + "] and the following status: ["
                                + jobExecution.getStatus() + "]");
                    } catch (Throwable t) {
                        logger.warn(
                                "Job: ["
                                        + job
                                        + "] failed unexpectedly and fatally with the following parameters: ["
                                        + jobParameters + "]", t);
                        rethrow(t);
                    }
                }

                private void rethrow(Throwable t) {
                    if (t instanceof RuntimeException) {
                        throw (RuntimeException) t;
                    } else if (t instanceof Error) {
                        throw (Error) t;
                    }
                    throw new IllegalStateException(t);
                }
            });
        } catch (TaskRejectedException e) {
            jobExecution.upgradeStatus(BatchStatus.FAILED);
            if (jobExecution.getExitStatus().equals(ExitStatus.UNKNOWN)) {
                jobExecution.setExitStatus(ExitStatus.FAILED
                        .addExitDescription(e));
            }
            jobRepository.update(jobExecution);
        }

        return jobExecution;
    }

    static interface ExecutionContextContributor {
        boolean CONTRIBUTED_SOMETHING = true;
        boolean CONTRIBUTED_NOTHING = false;
        /**
         * 
         * @param executionContext
         * @return true if the exeuctioncontext was contributed to
         */
        public boolean contributeTo(ExecutionContext executionContext);
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        Assert.state(jobRepository != null, "A JobRepository has not been set.");
        if (taskExecutor == null) {
            logger.info("No TaskExecutor has been set, defaulting to synchronous executor.");
            taskExecutor = new SyncTaskExecutor();
        }
        this.simpleLauncher = new SimpleJobLauncher();
        this.simpleLauncher.setJobRepository(jobRepository);
        this.simpleLauncher.setTaskExecutor(taskExecutor);
        this.simpleLauncher.afterPropertiesSet();
    }


    @Override
    public JobExecution run(Job job, JobParameters jobParameters)
            throws JobExecutionAlreadyRunningException, JobRestartException,
            JobInstanceAlreadyCompleteException, JobParametersInvalidException {
        return simpleLauncher.run(job, jobParameters);
    }

}
3
  • Quick question, where does your JobFilter come from?
    – daveslab
    Feb 27, 2012 at 17:19
  • That job filter interface is a custom implementation we put in as yet another point to remove a possible implementation. I believe the only implementation is to check for running jobs based on a different criteria set than the parameters and throw out the job request if it finds one. You can leave that out if you'd like. If you need it I can dig up the interface for that if you'd like but you can glean the interface from the code above as it has just one method. Feb 28, 2012 at 3:15
  • Yeah, that makes sense. I figured it was a custom class, but I wanted to make sure.
    – daveslab
    Feb 28, 2012 at 14:37
4

Starting from spring batch 2.2.x, there is support for non-identifying parameters. If you are using CommandLineJobRunner, you can specify non-identifying parameters with '-' prefix. For example: java org.springframework.batch.core.launch.support.CommandLineJobRunner file=/my/file/path -session=5678

If you are using old version of spring batch, you need to migrate your database schema. See 'Migrating to 2.x.x' section at http://docs.spring.io/spring-batch/getting-started.html.

This is the Jira page of the feature https://jira.springsource.org/browse/BATCH-1412, and here are the change that implement it https://fisheye.springsource.org/changelog/spring-batch?cs=557515df45c0f596588418d53c3f2bae3781c1c3

1
  • Thank you so much. using the minus sign worked for me even from the spring batch admin console Feb 11, 2014 at 16:10
1

In more recent versions of Spring Batch (I am using spring-batch-core:4.3.3), you can use the JobParametersBuilder to specify whether a parameter is identifying or not. For example:

new JobParametersBuilder()
  .addString("identifying-param-name", paramValue1)
  .addString("non-identifying-param-name", paramValue2, false)
  .toJobParameters();

The 'false' in the third argument makes the parameter non-identifying.

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.