Announcement Announcement Module
Collapse
No announcement yet.
publish-subscribe-channel tries to use QueueSession Page Title Module
Move Remove Collapse
X
Conversation Detail Module
Collapse
  • Filter
  • Time
  • Show
Clear All
new posts

  • publish-subscribe-channel tries to use QueueSession

    Hi,

    I configured some queues an one topic. For some time the messages proceed as expected, but after a while messages published via the topic fail, since a queue session is used:
    Can some please help to check was happens here?

    Here is my configuration (partially):

    Code:
    <?xml version="1.0" encoding="UTF-8"?>
    <bean:beans xmlns="http://www.springframework.org/schema/integration"
    	xmlns:bean="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    	xmlns:context="http://www.springframework.org/schema/context"
    	xmlns:jms="http://www.springframework.org/schema/integration/jms"
    	xmlns:stream="http://www.springframework.org/schema/integration/stream"
    
    	xsi:schemaLocation="
    		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
    		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd
    		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.0.xsd
    		http://www.springframework.org/schema/integration/jms http://www.springframework.org/schema/integration/jms/spring-integration-jms-2.0.xsd
    		http://www.springframework.org/schema/integration/stream http://www.springframework.org/schema/integration/stream/spring-integration-stream.xsd">
    
    
    	<bean:bean id="connectionFactory"
    		class="org.springframework.jms.connection.CachingConnectionFactory">
    		<bean:property name="targetConnectionFactory">
    			<bean:bean class="org.apache.activemq.ActiveMQConnectionFactory">
    				<bean:property name="brokerURL" value="${jms.brokerURL}" />
    			</bean:bean>
    		</bean:property>
    		<bean:property name="sessionCacheSize" value="${jms.sessionCacheSize}" />
    		<bean:property name="cacheProducers" value="${jms.cacheProducers}" />
    		<bean:property name="cacheConsumers" value="${jms.cacheConsumers}" />
    	</bean:bean>
    
    	<bean:bean id="requestQueue" class="org.apache.activemq.command.ActiveMQQueue">
    		<bean:constructor-arg value="${jms.queueName.request}" />
    	</bean:bean>
    
    	<bean:bean id="replyQueue" class="org.apache.activemq.command.ActiveMQQueue">
    		<bean:constructor-arg value="${jms.queueName.reply}" />
    	</bean:bean>
    
    
    	<bean:bean id="asyncRequestQueue" class="org.apache.activemq.command.ActiveMQQueue">
    		<bean:constructor-arg value="${jms.queueName.async}" />
    	</bean:bean>
    
    
    	<chain input-channel="requests" output-channel="jms.replies">
    		<transformer>
    			<bean:bean
    				class="org.springframework.integration.json.ObjectToJsonTransformer" />
    		</transformer>
    		<jms:outbound-gateway request-destination="requestQueue"
    			reply-destination="replyQueue" explicit-qos-enabled="true"
    			time-to-live="${jms.request.timeToLive}" receive-timeout="${jms.receive.timeout}"  />
    	</chain>
    
    	<channel id="async.map.requests" />
    
    	<chain input-channel="async.requests" output-channel="async.map.requests">
    		<transformer>
    			<bean:bean
    				class="org.springframework.integration.json.ObjectToJsonTransformer" />
    		</transformer>
    	</chain>
    
    
    	<jms:outbound-channel-adapter id="jmsAsyncOut"
    		destination="asyncRequestQueue" channel="async.map.requests" />
    
    	<!-- activeMQ topic -->
    	
    	<bean:bean id="cacheMgmtTopic" class="org.apache.activemq.command.ActiveMQTopic">
    		<bean:constructor-arg value="${jms.cacheMgmt.cacheMgmtTopic}" />
    	</bean:bean>
    
    	<!-- Channel to publish cache management events -->
    	
    	<jms:publish-subscribe-channel id="toCacheMgmtTopic"
    		auto-startup="true" connection-factory="connectionFactory" topic="cacheMgmtTopic"
    		selector="FALSE" />
    	<!-- Channel to receive cache management events -->
    	
    	<jms:message-driven-channel-adapter
    		channel="fromCacheMgmtTopic" destination="cacheMgmtTopic"
    		connection-factory="connectionFactory" extract-payload="true" />
    
    </bean:beans>
    And here is the (shortened) stacktrace:
    Code:
    Caused by: org.springframework.integration.MessageDeliveryException: failed to send Message to channel 'toCacheMgmtTopic'
            at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:165) ~[spring-integration-core-2.0.5.RELEASE.jar:na]
            at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:128) ~[spring-integration-core-2.0.5.RELEASE.jar:na]
            at com.gamigo.gas.general.service.CacheActionPublisher.notifySubscribersToPurgeCaches(CacheActionPublisher.java:29) ~[CacheActionPublisher.class:na]
            at com.gamigo.gas.general.service.CacheActionPublisher$$FastClassByCGLIB$$2084119b.invoke(<generated>) ~[cglib-2.2.jar:na]
            at net.sf.cglib.proxy.MethodProxy.invoke(MethodProxy.java:191) ~[cglib-2.2.jar:na]
            at org.springframework.aop.framework.Cglib2AopProxy$CglibMethodInvocation.invokeJoinpoint(Cglib2AopProxy.java:688) ~[spring-aop-3.0.3.RELEASE.jar:3.0.3.RELEASE]
            at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:150) [spring-aop-3.0.3.RELEASE.jar:3.0.3.RELEASE]
            at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:166) [spring-aop-3.0.3.RELEASE.jar:3.0.3.RELEASE]
            at org.springframework.aop.interceptor.ExposeInvocationInterceptor.invoke(ExposeInvocationInterceptor.java:89) [spring-aop-3.0.3.RELEASE.jar:3.0.3.RELEASE]
            at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:172) [spring-aop-3.0.3.RELEASE.jar:3.0.3.RELEASE]
            at org.springframework.aop.framework.Cglib2AopProxy$DynamicAdvisedInterceptor.intercept(Cglib2AopProxy.java:621) ~[spring-aop-3.0.3.RELEASE.jar:3.0.3.RELEASE]
            at com.gamigo.gas.general.service.CacheActionPublisher$$EnhancerByCGLIB$$fc0f42a.notifySubscribersToPurgeCaches(<generated>) ~[cglib-2.2.jar:na]
            at com.gamigo.gas.general.service.SystemMgmtServiceImpl.clearAllCaches(SystemMgmtServiceImpl.java:53) ~[SystemMgmtServiceImpl.class:na]
            at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:1.6.0_21]
            at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39) ~[na:1.6.0_21]
            at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25) ~[na:1.6.0_21]
            at java.lang.reflect.Method.invoke(Method.java:597) ~[na:1.6.0_21]
            at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:309) [spring-aop-3.0.3.RELEASE.jar:3.0.3.RELEASE]
            at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:183) [spring-aop-3.0.3.RELEASE.jar:3.0.3.RELEASE]
            at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:150) [spring-aop-3.0.3.RELEASE.jar:3.0.3.RELEASE]
            at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:166) [spring-aop-3.0.3.RELEASE.jar:3.0.3.RELEASE]
            at org.springframework.aop.aspectj.MethodInvocationProceedingJoinPoint.proceed(MethodInvocationProceedingJoinPoint.java:80) ~[spring-aop-3.0.3.RELEASE.jar:3.0.3.RELEASE]
            at com.gamigo.gas.general.ValidationAspect.invoke(ValidationAspect.java:30) ~[ValidationAspect.class:na]
            at sun.reflect.GeneratedMethodAccessor174.invoke(Unknown Source) ~[na:na]
            at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25) ~[na:1.6.0_21]
            at java.lang.reflect.Method.invoke(Method.java:597) ~[na:1.6.0_21]
            at org.springframework.aop.aspectj.AbstractAspectJAdvice.invokeAdviceMethodWithGivenArgs(AbstractAspectJAdvice.java:621) ~[spring-aop-3.0.3.RELEASE.jar:3.0.3.RELEASE]
            at org.springframework.aop.aspectj.AbstractAspectJAdvice.invokeAdviceMethod(AbstractAspectJAdvice.java:610) ~[spring-aop-3.0.3.RELEASE.jar:3.0.3.RELEASE]
            at org.springframework.aop.aspectj.AspectJAroundAdvice.invoke(AspectJAroundAdvice.java:65) ~[spring-aop-3.0.3.RELEASE.jar:3.0.3.RELEASE]
            at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:161) [spring-aop-3.0.3.RELEASE.jar:3.0.3.RELEASE]
            at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:166) [spring-aop-3.0.3.RELEASE.jar:3.0.3.RELEASE]
            at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:166) [spring-aop-3.0.3.RELEASE.jar:3.0.3.RELEASE]
            at org.springframework.aop.interceptor.ExposeInvocationInterceptor.invoke(ExposeInvocationInterceptor.java:89) [spring-aop-3.0.3.RELEASE.jar:3.0.3.RELEASE]
            at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:172) [spring-aop-3.0.3.RELEASE.jar:3.0.3.RELEASE]
            at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:202) [spring-aop-3.0.3.RELEASE.jar:3.0.3.RELEASE]
            at $Proxy564.clearAllCaches(Unknown Source) ~[na:na]
            at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:1.6.0_21]
            at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39) ~[na:1.6.0_21]
            at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25) ~[na:1.6.0_21]
            at java.lang.reflect.Method.invoke(Method.java:597) ~[na:1.6.0_21]
            at org.apache.cxf.service.invoker.AbstractInvoker.performInvocation(AbstractInvoker.java:173) ~[cxf-rt-core-2.3.3.jar:2.3.3]
            at org.apache.cxf.service.invoker.AbstractInvoker.invoke(AbstractInvoker.java:89) ~[cxf-rt-core-2.3.3.jar:2.3.3]
            ... 76 common frames omitted ...

  • #2
    Is there any chance you can provide the next "caused by" in that stack trace?

    Comment


    • #3
      Originally posted by Mark Fisher View Post
      Is there any chance you can provide the next "caused by" in that stack trace?
      Hi Mark,
      thanks for your interest.

      here is the rest of the stacktrace:
      Code:
      Caused by: org.springframework.jms.InvalidDestinationException: Topics are not supported by a QueueSession; nested exception is javax.jms.InvalidDestinationException: Topics are not supported by a QueueSession
              at org.springframework.jms.support.JmsUtils.convertJmsAccessException(JmsUtils.java:285) ~[spring-jms-3.0.5.RELEASE.jar:3.0.5.RELEASE]
              at org.springframework.jms.support.JmsAccessor.convertJmsAccessException(JmsAccessor.java:168) ~[spring-jms-3.0.5.RELEASE.jar:3.0.5.RELEASE]
              at org.springframework.jms.core.JmsTemplate.execute(JmsTemplate.java:469) ~[spring-jms-3.0.5.RELEASE.jar:3.0.5.RELEASE]
              at org.springframework.jms.core.JmsTemplate.send(JmsTemplate.java:534) ~[spring-jms-3.0.5.RELEASE.jar:3.0.5.RELEASE]
              at org.springframework.jms.core.JmsTemplate.convertAndSend(JmsTemplate.java:612) ~[spring-jms-3.0.5.RELEASE.jar:3.0.5.RELEASE]
              at org.springframework.jms.core.JmsTemplate.convertAndSend(JmsTemplate.java:604) ~[spring-jms-3.0.5.RELEASE.jar:3.0.5.RELEASE]
              at org.springframework.integration.jms.AbstractJmsChannel.doSend(AbstractJmsChannel.java:45) ~[spring-integration-jms-2.0.5.RELEASE.jar:na]
              at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:157) ~[spring-integration-core-2.0.5.RELEASE.jar:na]
              ... 117 common frames omitted
      Caused by: javax.jms.InvalidDestinationException: Topics are not supported by a QueueSession
              at org.apache.activemq.ActiveMQQueueSession.createProducer(ActiveMQQueueSession.java:203) ~[activemq-core-5.3.2.jar:5.3.2]
              at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:1.6.0_21]
              at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39) ~[na:1.6.0_21]
              at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25) ~[na:1.6.0_21]
              at java.lang.reflect.Method.invoke(Method.java:597) ~[na:1.6.0_21]
              at org.springframework.jms.connection.CachingConnectionFactory$CachedSessionInvocationHandler.invoke(CachingConnectionFactory.java:344) ~[spring-jms-3.0.5.RELEASE.jar:3.0.5.RELEASE]
              at $Proxy664.createProducer(Unknown Source) ~[na:na]
              at org.springframework.jms.core.JmsTemplate.doCreateProducer(JmsTemplate.java:971) ~[spring-jms-3.0.5.RELEASE.jar:3.0.5.RELEASE]
              at org.springframework.jms.core.JmsTemplate.createProducer(JmsTemplate.java:952) ~[spring-jms-3.0.5.RELEASE.jar:3.0.5.RELEASE]
              at org.springframework.jms.core.JmsTemplate.doSend(JmsTemplate.java:563) ~[spring-jms-3.0.5.RELEASE.jar:3.0.5.RELEASE]
              at org.springframework.jms.core.JmsTemplate$3.doInJms(JmsTemplate.java:536) ~[spring-jms-3.0.5.RELEASE.jar:3.0.5.RELEASE]
              at org.springframework.jms.core.JmsTemplate.execute(JmsTemplate.java:466) ~[spring-jms-3.0.5.RELEASE.jar:3.0.5.RELEASE]
              ... 122 common frames omitted
      stacktrace also as attachments.

      Comment


      • #4
        Was there a resolution to this question? I am having the same problem. Thanks.

        Comment


        • #5
          Can you try using a separate instance of a CachingConnectionFactory for the publish-subscribe-channel and message-driven-channel-adapter that are using a Topic instead of Queue?

          Comment


          • #6
            I can definitely try that. I'm just curious on what triggers this problem. I have a test program that uses a single connection factory and has both topics and queues being used and don't have this problem.

            Comment

            Working...
            X