Startup race-condition spring-batch with spring-integration

Question

For an application I'm using spring-boot, spring-batch and spring-integration.

My problem is that, on startup, when everything is auto configured and auto wired, my spring-integration @MessageEndpoints, connected to RabbitMQ, are starting to process available messages on its queues. Based on the received messages, these @MessageEndpoints are trying to start specific spring-batch Jobs, looked up, through it's auto wired JobRegistery.

Because of all the auto configuration not all jobs are yet registered to the used JobRegistery! (A few seconds later they will be).

After all spring-batch jobs are registered to the JobRegistery then the @MessageEndpoints should start. Is this possible? Maybe thought the ContextRefreshEvent?

Answer 1

I just looked at the code and the problem appears to be that the AutomaticJobRegistrar uses the context refreshed event to load the jobs; it should really implement SmartLifecycle and start in an "early-ish" phase.

The spring integration components implement SmartLifecycle and inbound endpoints (such as the rabbit endpoints) start in a late phase.

I suggest you open a JIRA against Batch - there's a TODO in the AutomaticJobRegistrar code:

// TODO: With Spring 3 a SmartLifecycle is started automatically

As a work-around, you could set autoStartup to false on the inbound adapter(s) and use your own event listener to start them on the context refreshed event.

The listener should implement Ordered; the AutomaticJobRegistrar is Ordered.LOWEST_PRECEDENCE so you want to run with a higher precedence (lower priority).

Answer 2

Gary Russell thank you for your push in the right direction! I have solved it as follow:

1. Disable autostart for the inbound channels in my integration XML.
2. After SpringApplication.run() if start manually these inbound channels, found through getBeansOftype().

.

public static void main(final String[] args) {
    ConfigurableApplicationContext context = SpringApplication.run(SpringBatchApplication.class, args);
    startInboundChannelAdapters(context);
}    
private static void startInboundChannelAdapters(ConfigurableApplicationContext context) {
    Map<String, AmqpInboundChannelAdapter> adapters = context.getBeansOfType(AmqpInboundChannelAdapter.class);
    adapters.forEach((name, adapter) -> {
        if (!adapter.isRunning()) {
            adapter.start();
        }
    });
}