jms:essage-driven-channel-adapter and startup ("Dispatcher has no subscribers.")

[jakeri]

Hi,

We have a spring integration flow with about 10 to 15 different jms:message-driven-channel-adapter and some file adapters connected too.

Unfortunately, we are getting some strange error when we have messages on our jms-queues (Sonic) and we startup the server. Then we get a org.springframework.integration.message.MessageDel iveryException: Dispatcher has no subscribers.. If we do not have any messages in the queue and resend the exact same message when the server is running no problems occur.
Our server is quite big and take about 2-3 minutes to startup.

My gut feeling is that the listener start to listen way to early on the jms queue and try to dispatch messages even if Spring integration has not been able to start all beans.

Can I make the jms listener wait for complete startup?
Bug?
Mis-configuration from our side?

....
][Headers={springintegration_jms_messageId=ID:3f27f598:8fe1f0001:12B586247E9, JMS_SonicMQ_notifyUndelivered=true, springintegration_timestamp=1285679053523, springint
egration_id=ebeb55dd-fd1f-4146-b34f-7d7df568175f, springintegration_jms_redelivered=false, JMS_SonicMQ_preserveUndelivered=true}]
org.springframework.integration.message.MessageDeliveryException: Dispatcher has no subscribers.
        at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:97)
        at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:90)
        at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:43)
        at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:116)
        at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:94)
        at org.springframework.integration.channel.MessageChannelTemplate.doSend(MessageChannelTemplate.java:224)
        at org.springframework.integration.channel.MessageChannelTemplate.send(MessageChannelTemplate.java:181)
        at org.springframework.integration.channel.MessageChannelTemplate.send(MessageChannelTemplate.java:169)
        at org.springframework.integration.jms.ChannelPublishingJmsMessageListener.onMessage(ChannelPublishingJmsMessageListener.java:203)
        at org.springframework.jms.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:535)
        at org.springframework.jms.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:495)
        at org.springframework.jms.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:467)
        at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.doReceiveAndExecute(AbstractPollingMessageListenerContainer.java:323)
        at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.receiveAndExecute(AbstractPollingMessageListenerContainer.java:261)
        at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.invokeListener(DefaultMessageListenerContainer.java:1056)
        at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.executeOngoingLoop(DefaultMessageListenerContainer.java:1048)
        at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.run(DefaultMessageListenerContainer.java:947)
        at java.lang.Thread.run(Thread.java:619)

Spring integration configuration looks like this

    <bean id="containerDataIn" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="autoStartup" value="${listener.autostart}" />
        <property name="connectionFactory" ref="connectionFactory" />
        <property name="destination" ref="inputColourQueue" />
        <property name="errorHandler" ref="errorHandler" />
    </bean>

    <jms:message-driven-channel-adapter  auto-startup="${listener.autostart}" id="jmsDataIn" container="containerDataIn" channel="fileDataIn"/>


    <file:outbound-gateway id="dataMover" request-channel="fileDataIn" 
                           reply-channel="dataFileChannel" filename-generator="timeStampFileNameGenerator"
                           directory="${import.basedir}/Data/${import.started.dir}" delete-source-files="true"/> 

Spring integration 1.0.4
Spring framework 3.0.4

[Mark Fisher]

It's not directly relevant, but this issue addresses the same general issue:
https://jira.springsource.org/browse/INT-845

Basically, prior to Spring 3.0, there was not a very fine-grained option for controlling startup order. Then, we added "phase" values within "SmartLifecycle" components in Spring 3.0.

Since you are using Spring 3.0, your MessageListener container does have a phase value, but the Spring Integration 1.0.x versions are not able to build on Spring 3.0 features (they will in the upcoming Spring 2.0).

The default phase is Integer.MAX_VALUE for the MessageListener container which is the latest possible "auto-start" and that is invoked upon context refresh. That is later than bean initialization where the Spring Integration EventDrivenConsumers *in 1.0.x* are auto-starting (via afterPropertiesSet if autoStartup == true). So, it does seem that a 3.0 MessageListener container should be starting after any Spring Integration 1.0.x EventDrivenConsumers.

Could you switch on debug-level logging and see if the startup order is what you actually see?

[jakeri]

I limited logging to org.springframework.integration.

To me it looks like it starts based on how the order is in the application-context file.

    <bean id="inputReferenceQueue" class="progress.message.jclient.Queue">
        <constructor-arg value="${queue.reference}" />
    </bean>

    <bean id="containerReferenceIn" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="autoStartup" value="${queues.autostart}" />
        <property name="connectionFactory" ref="connectionFactory" />
        <property name="destination" ref="inputReferenceQueue" />
        <property name="errorHandler" ref="errorHandler" />
    </bean>

    <jms:message-driven-channel-adapter auto-startup="${queues.autostart}" id="jmsReferenceIn" container="containerReferenceIn" channel="fileReferenceIn"/>

    <!-- File -->
    <file:inbound-channel-adapter id="fileReferenceIn" auto-startup="${file.autostart}"
        directory="${import.basedir}/Reference/pickup"
        filename-pattern=".*">
        <si:poller>
            <si:interval-trigger interval="5" time-unit="SECONDS" />
        </si:poller>
    </file:inbound-channel-adapter>

    <file:outbound-gateway id="referenceMover" request-channel="fileReferenceIn" 
                           reply-channel="referenceFileChannel" filename-generator="timeStampFileNameGenerator"
                           directory="${import.basedir}/Reference/${import.started.dir}" delete-source-files="true"/>
                           
    ...
    ... 

 2010-09-29 13:22:06,425 INFO  [org.springframework.integration.gateway.SimpleMessagingGateway](WrapperSimpleAppMain) started org.springframework.integration.gateway.SimpleMessagingGateway@13d3f62
 2010-09-29 13:22:06,425 INFO  [org.springframework.integration.gateway.SimpleMessagingGateway](WrapperSimpleAppMain) started org.springframework.integration.gateway.SimpleMessagingGateway@c7816
 2010-09-29 13:22:06,425 INFO  [org.springframework.integration.gateway.SimpleMessagingGateway](WrapperSimpleAppMain) started org.springframework.integration.gateway.SimpleMessagingGateway@a0187c
 2010-09-29 13:22:06,425 INFO  [org.springframework.integration.gateway.SimpleMessagingGateway](WrapperSimpleAppMain) started org.springframework.integration.gateway.SimpleMessagingGateway@a79c31
 2010-09-29 13:22:06,425 INFO  [org.springframework.integration.gateway.GatewayProxyFactoryBean](WrapperSimpleAppMain) started availabilityBatchFileInboundPublisher
 2010-09-29 13:22:06,535 INFO  [org.springframework.integration.gateway.SimpleMessagingGateway](WrapperSimpleAppMain) started org.springframework.integration.gateway.SimpleMessagingGateway@265121
 2010-09-29 13:22:06,535 INFO  [org.springframework.integration.gateway.SimpleMessagingGateway](WrapperSimpleAppMain) started org.springframework.integration.gateway.SimpleMessagingGateway@19853c2
 2010-09-29 13:22:06,535 INFO  [org.springframework.integration.gateway.SimpleMessagingGateway](WrapperSimpleAppMain) started org.springframework.integration.gateway.SimpleMessagingGateway@1a2f9dd
 2010-09-29 13:22:06,535 INFO  [org.springframework.integration.gateway.SimpleMessagingGateway](WrapperSimpleAppMain) started org.springframework.integration.gateway.SimpleMessagingGateway@c21d01
 2010-09-29 13:22:06,535 INFO  [org.springframework.integration.gateway.GatewayProxyFactoryBean](WrapperSimpleAppMain) started availabilityBatchFileInboundPublisher
 2010-09-29 13:22:10,925 INFO  [org.springframework.integration.config.xml.DefaultConfiguringBeanFactoryPostProcessor](WrapperSimpleAppMain) No bean named 'errorChannel' has been explicitly defined. Therefore, a default PublishSubscribeChannel will be created.
 2010-09-29 13:22:10,925 INFO  [org.springframework.integration.config.xml.DefaultConfiguringBeanFactoryPostProcessor](WrapperSimpleAppMain) No bean named 'taskScheduler' has been explicitly defined. Therefore, a default SimpleTaskScheduler will be created.
 2010-09-29 13:22:36,582 INFO  [org.springframework.integration.jms.JmsMessageDrivenEndpoint](WrapperSimpleAppMain) started jmsReferenceIn
 2010-09-29 13:22:37,613 DEBUG [org.springframework.integration.jms.HeaderMappingMessageConverter](containerReferenceIn-1) converted JMS Message [progress.message.jimpl.TextMessage@174689] to integration Message [[Payload=Hello Im gonna fail startup data again][Headers={springintegration_jms_messageId=ID:3f27f598:2fd22d000d:12B5D39AB24, springintegration_timestamp=1285759357613, springintegration_id=43bbf30f-0ebc-4228-80d8-d4311c9d5d91, springintegration_jms_correlationId=Message 1, springintegration_jms_redelivered=false}]]
 2010-09-29 13:22:37,613 DEBUG [org.springframework.integration.channel.DirectChannel](containerReferenceIn-1) preSend on channel 'fileReferenceIn', message: [Payload=Hello Im gonna fail startup data again][Headers={springintegration_jms_messageId=ID:3f27f598:2fd22d000d:12B5D39AB24, springintegration_timestamp=1285759357613, springintegration_id=43bbf30f-0ebc-4228-80d8-d4311c9d5d91, springintegration_jms_correlationId=Message 1, springintegration_jms_redelivered=false}]
 2010-09-29 13:22:37,629 INFO  [org.springframework.integration.endpoint.SourcePollingChannelAdapter](WrapperSimpleAppMain) started fileReferenceIn.adapter
 2010-09-29 13:22:37,629 DEBUG [org.springframework.integration.channel.PublishSubscribeChannel](containerReferenceIn-1) preSend on channel 'errorChannel', message: [Payload=org.springframework.integration.message.MessageDeliveryException: Dispatcher has no subscribers.][Headers={springintegration_timestamp=1285759357629, springintegration_id=ee3330e7-87d9-4503-bfd5-c1dfb3ad29e0}]
 2010-09-29 13:22:37,629 DEBUG [org.springframework.integration.channel.PublishSubscribeChannel](containerReferenceIn-1) postSend (sent=false) on channel 'errorChannel', message: [Payload=org.springframework.integration.message.MessageDeliveryException: Dispatcher has no subscribers.][Headers={springintegration_timestamp=1285759357629, springintegration_id=ee3330e7-87d9-4503-bfd5-c1dfb3ad29e0}]
 2010-09-29 13:22:37,629 ERROR [org.springframework.integration.channel.MessagePublishingErrorHandler](containerReferenceIn-1) failure occurred in messaging task with message: [Payload=Hello Im gonna fail startup data again][Headers={springintegration_jms_messageId=ID:3f27f598:2fd22d000d:12B5D39AB24, springintegration_timestamp=1285759357613, springintegration_id=43bbf30f-0ebc-4228-80d8-d4311c9d5d91, springintegration_jms_correlationId=Message 1, springintegration_jms_redelivered=false}]
 org.springframework.integration.message.MessageDeliveryException: Dispatcher has no subscribers.
	at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:97)
	at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:90)
	at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:43)
	at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:116)
	at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:94)
	at org.springframework.integration.channel.MessageChannelTemplate.doSend(MessageChannelTemplate.java:224)
	at org.springframework.integration.channel.MessageChannelTemplate.send(MessageChannelTemplate.java:181)
	at org.springframework.integration.channel.MessageChannelTemplate.send(MessageChannelTemplate.java:169)
	at org.springframework.integration.jms.ChannelPublishingJmsMessageListener.onMessage(ChannelPublishingJmsMessageListener.java:203)
	at org.springframework.jms.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:535)
	at org.springframework.jms.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:495)
	at org.springframework.jms.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:467)
	at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.doReceiveAndExecute(AbstractPollingMessageListenerContainer.java:323)
	at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.receiveAndExecute(AbstractPollingMessageListenerContainer.java:261)
	at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.invokeListener(DefaultMessageListenerContainer.java:1056)
	at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.executeOngoingLoop(DefaultMessageListenerContainer.java:1048)
	at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.run(DefaultMessageListenerContainer.java:947)
	at java.lang.Thread.run(Thread.java:619)
2010-09-29 13:22:37,691 INFO  [org.springframework.integration.endpoint.EventDrivenConsumer](WrapperSimpleAppMain) started referenceMover
 2010-09-29 13:22:37,738 INFO  [org.springframework.integration.endpoint.EventDrivenConsumer](WrapperSimpleAppMain) started org.springframework.integration.config.ConsumerEndpointFactoryBean#0
 2010-09-29 13:22:37,816 INFO  [org.springframework.integration.endpoint.EventDrivenConsumer](WrapperSimpleAppMain) started org.springframework.integration.config.ConsumerEndpointFactoryBean#1
 2010-09-29 13:22:37,863 INFO  [org.springframework.integration.endpoint.EventDrivenConsumer](WrapperSimpleAppMain) started org.springframework.integration.config.ConsumerEndpointFactoryBean#2
 2010-09-29 13:22:38,004 INFO
 ...

[jakeri]

It works when moving jms:message-driven-channel-adapter to the end of the file.
It is not a solution but a way forward right now.

[Mark Fisher]

Could you try one thing... set the container's autoStartup property to FALSE, but leave the message-driven-channel-adapter's autoStartup property as TRUE?

[jakeri]

No go, same problem.

[oleg.zhurakousky]

Could you possibly isolate this into a simple but complete configuration that reproduces the problem and post this configuration? (no need to post java files if any)
I tried to reproduce it based on the config excerpts that you posted before, but can't seem to trigger the problem.

[jakeri]

I will try to isolate the problem. We have a quite big spring context that starts the integration flow.

Do you have some kind good lab-project to try to add it to?

[oleg.zhurakousky]

Well you can check out our samples:
http://blog.springsource.com/2010/09...ation-samples/
There are few JMS samples there and one of them is using outbound-channel-adapter

[jakeri]

I have been able to reproduce the problem using sample libraries from the download.
Spring integration 1.0.4 uses Spring core 2.5.6 per default. As I wrote before, our project uses Spring core 3.0.4. If you use 3.0.4 instead of 2.5.6 you will see this problem.

Debug messages from Spring Integration 1.0.4 with Spring Core 2.5.6 running JMS example. Look at started jmsin

INFO : org.springframework.integration.config.xml.DefaultConfiguringBeanFactoryPostProcessor - No bean named 'errorChannel' has been explicitly defined. Therefore, a default PublishSubscribeChannel will be created.
INFO : org.springframework.integration.config.xml.DefaultConfiguringBeanFactoryPostProcessor - No bean named 'taskScheduler' has been explicitly defined. Therefore, a default SimpleTaskScheduler will be created.
INFO : org.springframework.integration.endpoint.EventDrivenConsumer - started colourMover
INFO : org.springframework.integration.endpoint.EventDrivenConsumer - started stdout
INFO : org.springframework.integration.jms.JmsMessageDrivenEndpoint - started jmsin
INFO : org.springframework.integration.endpoint.EventDrivenConsumer - started org.springframework.integration.endpoint.EventDrivenConsumer#0
INFO : org.springframework.integration.scheduling.SimpleTaskScheduler - started org.springframework.integration.scheduling.SimpleTaskScheduler@bb0d0d

Spring Integration 1.0.4 with Spring Core 3.0.4.

INFO : org.springframework.integration.config.xml.DefaultConfiguringBeanFactoryPostProcessor - No bean named 'errorChannel' has been explicitly defined. Therefore, a default PublishSubscribeChannel will be created.
INFO : org.springframework.integration.config.xml.DefaultConfiguringBeanFactoryPostProcessor - No bean named 'taskScheduler' has been explicitly defined. Therefore, a default SimpleTaskScheduler will be created.
INFO : org.springframework.integration.jms.JmsMessageDrivenEndpoint - started jmsin
INFO : org.springframework.integration.endpoint.EventDrivenConsumer - started colourMover
INFO : org.springframework.integration.endpoint.EventDrivenConsumer - started stdout
INFO : org.springframework.integration.endpoint.EventDrivenConsumer - started org.springframework.integration.endpoint.EventDrivenConsumer#0
INFO : org.springframework.integration.scheduling.SimpleTaskScheduler - started org.springframework.integration.scheduling.SimpleTaskScheduler@64ab4d

Configuration file

<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/integration"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:beans="http://www.springframework.org/schema/beans"
	xmlns:jms="http://www.springframework.org/schema/integration/jms"
	xmlns:stream="http://www.springframework.org/schema/integration/stream"
	xmlns:file="http://www.springframework.org/schema/integration/file"
	xsi:schemaLocation="http://www.springframework.org/schema/beans
			http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
			http://www.springframework.org/schema/integration
			http://www.springframework.org/schema/integration/spring-integration-1.0.xsd
			http://www.springframework.org/schema/integration/jms
			http://www.springframework.org/schema/integration/jms/spring-integration-jms-1.0.xsd
			http://www.springframework.org/schema/integration/stream
			http://www.springframework.org/schema/integration/stream/spring-integration-stream-1.0.xsd
			http://www.springframework.org/schema/integration/file
            http://www.springframework.org/schema/integration/file/spring-integration-file-1.0.xsd">

	<beans:bean id="containerColourIn" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
		<beans:property name="autoStartup" value="true" />
		<beans:property name="connectionFactory" ref="connectionFactory" />
		<beans:property name="destination" ref="requestQueue" />
	</beans:bean>

	<jms:message-driven-channel-adapter
		id="jmsin" auto-startup="true" channel="jmsinToStdoutChannel" container="containerColourIn"/>

	<file:outbound-gateway id="colourMover"
		request-channel="jmsinToStdoutChannel" reply-channel="jmsinToStdoutChannelReply"
		directory="file:C:/TEMP/springtest" delete-source-files="true" />

	<channel id="jmsinToStdoutChannelReply"/>

	<stream:stdout-channel-adapter id="stdout"
		channel="jmsinToStdoutChannelReply" append-newline="true" />

</beans:beans>