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):
And here is the (shortened) stacktrace: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>
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 ...
Reply With Quote
Originally Posted by Mark Fisher
