activemq-users mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From javaG <zenma...@gmail.com>
Subject Strange problem in activeMQ or Camel resulting in wrong message type
Date Tue, 22 Oct 2013 18:26:37 GMT

There is a strange problem either in activemq or camel which breaks
everything when it happens. This problem cannot be reproduced but 
happens randomly once in a while. What happens is that in a request/reply
pattern, the reply is of type ActiveMQMessage instead of the expected 
ActiveMQObjectMessage. I had to spend hours to write up this post and
produce the logs and gather the informaiton. I hope some one can provide an
answer.
The log at the bottom of this post captures the essence of this problem
which I was able to get after enable DEBUG camel logging. 
In my application a request is sent from the producer client once every two
seconds and response comes back from the consumer client in about 50 ms. 
All clients and activemq live in their own JVM communicating via tcp.

This is part of the code from consumer client ( service prodiver) that sends
response message:
List result = myService.getList(param)
ObjectMessage response = session.createObjectMessage()
response.setJMSCorrelationID(message.getJMSCorrelationID())
response.setObject(retValue)
replyProducer.send(message.getJMSReplyTo(), response)

As you can see this will send ActiveMQObjectMessage message type.

In the log the message that is sent 15:26:12 is exactly what it should be
but the one sent 15:26:14 has last two lines at 15:26:14:808 different. Both
the messages should be exactly 
same (except with differnt correlationid and reply temp queues) as they
involves exact same request and exact same reply payload and go exact 
same routing. Also note that persistent is set to true in the last line of
the log in second message. This is also wrong. All messages are
non persistent.


This is the routing and beans in camel.xml:


      <route>
                        
                        <from uri="jms:q1" />
                        <doTry>
                                <validate><simple>${body[subsc]} regex
'[a-z0-9]{8}' </simple> </validate>
                                <choice>
                                        <when>
                                                <simple>${body[key1]} regex
'^(v1)$'</simple>
                                                <choice>
                                                    <when>
                                                          
<simple>${body[key2]} == 'v2'</simple>
                                                               <to
uri="jms:q1.service1" />
                                                     </when>
                                                </choice>        
                                        </when>
                                        <otherwise>
                                        <to uri="jms:invalid" />
                                        <stop />
                                        </otherwise>
                                </choice>
                        <doCatch>
                               
<exception>org.apache.camel.ValidationException</exception>
                                <to uri="jms:invalid"/>
                        </doCatch>
                        </doTry>
                                
    </route>
	
		<bean id="confact" class="org.apache.activemq.ActiveMQConnectionFactory">
            <property name="brokerURL"
value="vm://amq-broker?create=false"/>
            <property name="userName" value="${activemq.username}"/>
            <property name="password" value="${activemq.password}"/>
    </bean>
	
    <bean id="activemq"
class="org.apache.activemq.camel.component.ActiveMQComponent" >
        <property name="connectionFactory" ref="confact" />
     
    </bean>
	
	<bean id="pooledConnectionFactory" 
          class="org.apache.activemq.pool.PooledConnectionFactory" 
          init-method="start" 
          destroy-method="stop">
        <property name="maxConnections" value="8" />
        <property name="connectionFactory" ref="confact" />
         <property name="expiryTimeout" value="-1" />  
        <property name="idleTimeout" value="-1" />
    </bean>

    <bean id="jmsConfig" 
class="org.apache.camel.component.jms.JmsConfiguration">
        <property name="connectionFactory" ref="pooledConnectionFactory"/>
        <property name="concurrentConsumers" value="8"/>
    </bean>

    <bean id="jms"
class="org.apache.activemq.camel.component.ActiveMQComponent">
        <property name="configuration" ref="jmsConfig"/>
    </bean>
	
	
	<bean id="myDeadLetterErrorHandler"
class="org.apache.camel.builder.DeadLetterChannelBuilder">
			<property name="deadLetterUri" value="jms:queue:deadlc"/>
			<property name="redeliveryPolicy" ref="myRedeliveryPolicyConfig"/>
		</bean>
 
		<bean id="myRedeliveryPolicyConfig"
class="org.apache.camel.processor.RedeliveryPolicy">
			<property name="maximumRedeliveries" value="3"/>
			<property name="redeliveryDelay" value="5000"/>
		</bean>
	

This is the log:
15:26:12,565 | DEBUG | Received message of type [class
org.apache.activemq.command.ActiveMQMapMessage] from consumer
[PooledMessageConsumer { ActiveMQMessageConsumer {
value=ID:myhost-59213-1382385397272-3:2:5:1, started=true } }] of session
[PooledSession { ActiveMQSession
{id=ID:myhost-59213-1382385397272-3:2:5,started=true} }] |
org.apache.camel.component.jms.DefaultJmsMessageListenerContainer | Camel
(camel) thread #5 - JmsConsumer[q1]
15:26:12,566 | DEBUG | Endpoint[jms://q1] consumer received JMS message:
ActiveMQMapMessage {commandId = 673, responseRequired = false, messageId =
ID:myhost-59240-1382385651460-1:1:1:1:668, originalDestination = null,
originalTransactionId = null, producerId =
ID:myhost-59240-1382385651460-1:1:1:1, destination = queue://q1,
transactionId = null, expiration = 0, timestamp = 1382387172562, arrival =
0, brokerInTime = 1382387172564, brokerOutTime = 1382387172564,
correlationId = 6fc99f4b-037f-480e-bde9-9724fa33d013, replyTo =
temp-queue://ID:myhost-59240-1382385651460-3:1:668, persistent = false, type
= null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId =
null, compressed = false, userID = null, content =
org.apache.activemq.util.ByteSequence@1254fcd, marshalledProperties = null,
dataStructure = null, redeliveryCounter = 0, size = 1138, properties = null,
readOnlyProperties = true, readOnlyBody = true, droppable = false}
ActiveMQMapMessage{ theTable = {} } |
org.apache.camel.component.jms.EndpointMessageListener | Camel (camel)
thread #5 - JmsConsumer[q1]
15:26:12,566 | DEBUG | Received Message has JMSCorrelationID
[6fc99f4b-037f-480e-bde9-9724fa33d013] |
org.apache.camel.component.jms.EndpointMessageListener | Camel (camel)
thread #5 - JmsConsumer[q1]
15:26:12,577 | DEBUG | Validation succeed for
Exchange[JmsMessage[JmsMessageID:
ID:myhost-59240-1382385651460-1:1:1:1:668]] with Predicate[Simple:
${body[subsc]} regex '[a-z0-9]{8}' or  ${body[spSubscriber]} regex
'[a-z0-9]{8}'] |
org.apache.camel.processor.validation.PredicateValidatingProcessor | Camel
(camel) thread #5 - JmsConsumer[q1]
15:26:12,584 | DEBUG | #0 - Simple: ${body[key1]} regex '^v1$' matches: true
for: Exchange[JmsMessage[JmsMessageID:
ID:myhost-59240-1382385651460-1:1:1:1:668]] |
org.apache.camel.processor.ChoiceProcessor | Camel (camel) thread #5 -
JmsConsumer[q1]
15:26:12,590 | DEBUG | #0 - Simple: ${body[key2]} == 'v2' matches: true for:
Exchange[JmsMessage[JmsMessageID:
ID:myhost-59240-1382385651460-1:1:1:1:668]] |
org.apache.camel.processor.ChoiceProcessor | Camel (camel) thread #5 -
JmsConsumer[q1]
15:26:12,590 | DEBUG | >>>> Endpoint[jms://q1.service1]
Exchange[JmsMessage[JmsMessageID:
ID:myhost-59240-1382385651460-1:1:1:1:668]] |
org.apache.camel.processor.SendProcessor | Camel (camel) thread #5 -
JmsConsumer[q1]
15:26:12,590 | DEBUG | Executing callback on JMS Session: PooledSession {
ActiveMQSession {id=ID:myhost-59213-1382385397272-3:2:5,started=true} } |
org.apache.camel.component.jms.JmsConfiguration$CamelJmsTemplate | Camel
(camel) thread #5 - JmsConsumer[q1]
15:26:12,591 | DEBUG | Using JMSReplyTo destination:
temp-queue://ID:myhost-59213-1382385397272-3:3:8 |
org.apache.camel.component.jms.JmsProducer | Camel (camel) thread #5 -
JmsConsumer[q1]
15:26:12,591 | DEBUG | Using JMSCorrelationID:
6fc99f4b-037f-480e-bde9-9724fa33d013 |
org.apache.camel.component.jms.JmsProducer | Camel (camel) thread #5 -
JmsConsumer[q1]
15:26:12,591 | DEBUG | Sending JMS message to: queue://q1.service1 with
message: ActiveMQMapMessage {commandId = 0, responseRequired = false,
messageId = null, originalDestination = null, originalTransactionId = null,
producerId = null, destination = null, transactionId = null, expiration = 0,
timestamp = 0, arrival = 0, brokerInTime = 0, brokerOutTime = 0,
correlationId = 6fc99f4b-037f-480e-bde9-9724fa33d013, replyTo =
temp-queue://ID:myhost-59213-1382385397272-3:3:8, persistent = false, type =
null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId =
null, compressed = false, userID = null, content = null,
marshalledProperties = null, dataStructure = null, redeliveryCounter = 0,
size = 0, properties =
{breadcrumbId=ID:myhost-59240-1382385651460-1:1:1:1:668,
CamelJmsDeliveryMode=1}, readOnlyProperties = false, readOnlyBody = false,
droppable = false} ActiveMQMapMessage{ theTable =
{key1=v1,key2=v2,subc=34xy} } |
org.apache.camel.component.jms.JmsConfiguration | Camel (camel) thread #5 -
JmsConsumer[q1]
15:26:12,595 | DEBUG | Received message of type [class
org.apache.activemq.command.ActiveMQObjectMessage] from consumer
[PooledMessageConsumer { ActiveMQMessageConsumer {
value=ID:myhost-59213-1382385397272-3:3:8:1, started=true } }] of session
[PooledSession { ActiveMQSession
{id=ID:myhost-59213-1382385397272-3:3:8,started=true} }] |
org.apache.camel.component.jms.DefaultJmsMessageListenerContainer | Camel
(camel) thread #17 - TemporaryQueueReplyManager[temporary]
15:26:12,596 | DEBUG | Received reply message with correlationID
[6fc99f4b-037f-480e-bde9-9724fa33d013] -> ActiveMQObjectMessage {commandId =
851, responseRequired = false, messageId =
ID:myhost-59320-1382386269086-3:1:1:1:423, originalDestination = null,
originalTransactionId = null, producerId =
ID:myhost-59320-1382386269086-3:1:1:1, destination =
temp-queue://ID:myhost-59213-1382385397272-3:3:8, transactionId = null,
expiration = 0, timestamp = 1382387172594, arrival = 0, brokerInTime =
1382387172595, brokerOutTime = 1382387172595, correlationId =
6fc99f4b-037f-480e-bde9-9724fa33d013, replyTo = null, persistent = false,
type = null, priority = 4, groupID = null, groupSequence = 0,
targetConsumerId = null, compressed = false, userID = null, content =
org.apache.activemq.util.ByteSequence@1a5022f, marshalledProperties = null,
dataStructure = null, redeliveryCounter = 0, size = 1701, properties = null,
readOnlyProperties = true, readOnlyBody = true, droppable = false} |
org.apache.camel.component.jms.reply.TemporaryQueueReplyManager | Camel
(camel) thread #17 - TemporaryQueueReplyManager[temporary]
15:26:12,597 | DEBUG | Reply received. Setting reply as OUT message:
[{dn1=23578, dn2=670}] |
org.apache.camel.component.jms.reply.TemporaryQueueReplyManager | Camel
(camel) thread #17 - TemporaryQueueReplyManager[temporary]
15:26:12,692 | DEBUG | Executing callback on JMS Session: PooledSession {
ActiveMQSession {id=ID:myhost-59213-1382385397272-3:2:5,started=true} } |
org.apache.camel.component.jms.JmsConfiguration$CamelJmsTemplate | Camel
(camel) thread #5 - JmsConsumer[q1]
15:26:12,693 | DEBUG | Endpoint[jms://q1] sending reply JMS message
[correlationId:6fc99f4b-037f-480e-bde9-9724fa33d013]: ActiveMQObjectMessage
{commandId = 0, responseRequired = false, messageId = null,
originalDestination = null, originalTransactionId = null, producerId = null,
destination = null, transactionId = null, expiration = 0, timestamp = 0,
arrival = 0, brokerInTime = 0, brokerOutTime = 0, correlationId =
6fc99f4b-037f-480e-bde9-9724fa33d013, replyTo = null, persistent = false,
type = null, priority = 4, groupID = null, groupSequence = 0,
targetConsumerId = null, compressed = false, userID = null, content =
org.apache.activemq.util.ByteSequence@10f7f33, marshalledProperties = null,
dataStructure = null, redeliveryCounter = 0, size = 0, properties =
{CamelJmsDeliveryMode=1}, readOnlyProperties = false, readOnlyBody = false,
droppable = false} | org.apache.camel.component.jms.EndpointMessageListener
| Camel (camel) thread #5 - JmsConsumer[q1]
15:26:12,694 | DEBUG | Sending JMS message to:
temp-queue://ID:myhost-59240-1382385651460-3:1:668 with message:
ActiveMQObjectMessage {commandId = 0, responseRequired = false, messageId =
null, originalDestination = null, originalTransactionId = null, producerId =
null, destination = null, transactionId = null, expiration = 0, timestamp =
0, arrival = 0, brokerInTime = 0, brokerOutTime = 0, correlationId =
6fc99f4b-037f-480e-bde9-9724fa33d013, replyTo = null, persistent = false,
type = null, priority = 4, groupID = null, groupSequence = 0,
targetConsumerId = null, compressed = false, userID = null, content =
org.apache.activemq.util.ByteSequence@10f7f33, marshalledProperties = null,
dataStructure = null, redeliveryCounter = 0, size = 0, properties =
{CamelJmsDeliveryMode=1}, readOnlyProperties = false, readOnlyBody = false,
droppable = false} | org.apache.camel.component.jms.JmsConfiguration | Camel
(camel) thread #5 - JmsConsumer[q1]



15:26:14,775 | DEBUG | Received message of type [class
org.apache.activemq.command.ActiveMQMapMessage] from consumer
[PooledMessageConsumer { ActiveMQMessageConsumer {
value=ID:myhost-59213-1382385397272-3:2:4:1, started=true } }] of session
[PooledSession { ActiveMQSession
{id=ID:myhost-59213-1382385397272-3:2:4,started=true} }] |
org.apache.camel.component.jms.DefaultJmsMessageListenerContainer | Camel
(camel) thread #4 - JmsConsumer[q1]
15:26:14,776 | DEBUG | Endpoint[jms://q1] consumer received JMS message:
ActiveMQMapMessage {commandId = 674, responseRequired = false, messageId =
ID:myhost-59240-1382385651460-1:1:1:1:669, originalDestination = null,
originalTransactionId = null, producerId =
ID:myhost-59240-1382385651460-1:1:1:1, destination = queue://q1,
transactionId = null, expiration = 0, timestamp = 1382387174774, arrival =
0, brokerInTime = 1382387174775, brokerOutTime = 1382387174775,
correlationId = a1115df5-affb-4f88-a909-dd877451a178, replyTo =
temp-queue://ID:myhost-59240-1382385651460-3:1:669, persistent = false, type
= null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId =
null, compressed = false, userID = null, content =
org.apache.activemq.util.ByteSequence@909572, marshalledProperties = null,
dataStructure = null, redeliveryCounter = 0, size = 1138, properties = null,
readOnlyProperties = true, readOnlyBody = true, droppable = false}
ActiveMQMapMessage{ theTable = {} } |
org.apache.camel.component.jms.EndpointMessageListener | Camel (camel)
thread #4 - JmsConsumer[q1]
15:26:14,776 | DEBUG | Received Message has JMSCorrelationID
[a1115df5-affb-4f88-a909-dd877451a178] |
org.apache.camel.component.jms.EndpointMessageListener | Camel (camel)
thread #4 - JmsConsumer[q1]
15:26:14,784 | DEBUG | Validation succeed for
Exchange[JmsMessage[JmsMessageID:
ID:myhost-59240-1382385651460-1:1:1:1:669]] with Predicate[Simple:
${body[subsc]} regex '[a-z0-9]{8}'] |
org.apache.camel.processor.validation.PredicateValidatingProcessor | Camel
(camel) thread #4 - JmsConsumer[q1]
15:26:14,791 | DEBUG | #0 - Simple: ${body[key1]} regex '^v1$' matches: true
for: Exchange[JmsMessage[JmsMessageID:
ID:myhost-59240-1382385651460-1:1:1:1:669]] |
org.apache.camel.processor.ChoiceProcessor | Camel (camel) thread #4 -
JmsConsumer[q1]
15:26:14,798 | DEBUG | #0 - Simple: ${body[key2]} == 'v2' matches: true for:
Exchange[JmsMessage[JmsMessageID:
ID:myhost-59240-1382385651460-1:1:1:1:669]] |
org.apache.camel.processor.ChoiceProcessor | Camel (camel) thread #4 -
JmsConsumer[q1]
15:26:14,798 | DEBUG | >>>> Endpoint[jms://q1.service1]
Exchange[JmsMessage[JmsMessageID:
ID:myhost-59240-1382385651460-1:1:1:1:669]] |
org.apache.camel.processor.SendProcessor | Camel (camel) thread #4 -
JmsConsumer[q1]
15:26:14,798 | DEBUG | Executing callback on JMS Session: PooledSession {
ActiveMQSession {id=ID:myhost-59213-1382385397272-3:2:4,started=true} } |
org.apache.camel.component.jms.JmsConfiguration$CamelJmsTemplate | Camel
(camel) thread #4 - JmsConsumer[q1]
15:26:14,799 | DEBUG | Using JMSReplyTo destination:
temp-queue://ID:myhost-59213-1382385397272-3:3:8 |
org.apache.camel.component.jms.JmsProducer | Camel (camel) thread #4 -
JmsConsumer[q1]
15:26:14,799 | DEBUG | Using JMSCorrelationID:
a1115df5-affb-4f88-a909-dd877451a178 |
org.apache.camel.component.jms.JmsProducer | Camel (camel) thread #4 -
JmsConsumer[q1]
15:26:14,799 | DEBUG | Sending JMS message to: queue://q1.service1 with
message: ActiveMQMapMessage {commandId = 0, responseRequired = false,
messageId = null, originalDestination = null, originalTransactionId = null,
producerId = null, destination = null, transactionId = null, expiration = 0,
timestamp = 0, arrival = 0, brokerInTime = 0, brokerOutTime = 0,
correlationId = a1115df5-affb-4f88-a909-dd877451a178, replyTo =
temp-queue://ID:myhost-59213-1382385397272-3:3:8, persistent = false, type =
null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId =
null, compressed = false, userID = null, content = null,
marshalledProperties = null, dataStructure = null, redeliveryCounter = 0,
size = 0, properties =
{breadcrumbId=ID:myhost-59240-1382385651460-1:1:1:1:669,
CamelJmsDeliveryMode=1}, readOnlyProperties = false, readOnlyBody = false,
droppable = false} ActiveMQMapMessage{ theTable =
{key1=v1,key2=v2,subc=34xy} } |
org.apache.camel.component.jms.JmsConfiguration | Camel (camel) thread #4 -
JmsConsumer[q1]
15:26:14,804 | DEBUG | Received message of type [class
org.apache.activemq.command.ActiveMQObjectMessage] from consumer
[PooledMessageConsumer { ActiveMQMessageConsumer {
value=ID:myhost-59213-1382385397272-3:3:8:1, started=true } }] of session
[PooledSession { ActiveMQSession
{id=ID:myhost-59213-1382385397272-3:3:8,started=true} }] |
org.apache.camel.component.jms.DefaultJmsMessageListenerContainer | Camel
(camel) thread #17 - TemporaryQueueReplyManager[temporary]
15:26:14,805 | DEBUG | Received reply message with correlationID
[a1115df5-affb-4f88-a909-dd877451a178] -> ActiveMQObjectMessage {commandId =
853, responseRequired = false, messageId =
ID:myhost-59320-1382386269086-3:1:1:1:424, originalDestination = null,
originalTransactionId = null, producerId =
ID:myhost-59320-1382386269086-3:1:1:1, destination =
temp-queue://ID:myhost-59213-1382385397272-3:3:8, transactionId = null,
expiration = 0, timestamp = 1382387174802, arrival = 0, brokerInTime =
1382387174803, brokerOutTime = 1382387174803, correlationId =
a1115df5-affb-4f88-a909-dd877451a178, replyTo = null, persistent = false,
type = null, priority = 4, groupID = null, groupSequence = 0,
targetConsumerId = null, compressed = false, userID = null, content =
org.apache.activemq.util.ByteSequence@1b36775, marshalledProperties = null,
dataStructure = null, redeliveryCounter = 0, size = 1701, properties = null,
readOnlyProperties = true, readOnlyBody = true, droppable = false} |
org.apache.camel.component.jms.reply.TemporaryQueueReplyManager | Camel
(camel) thread #17 - TemporaryQueueReplyManager[temporary]
15:26:14,806 | DEBUG | Reply received. Setting reply as OUT message:
[{dn1=23578, dn2=670}] |
org.apache.camel.component.jms.reply.TemporaryQueueReplyManager | Camel
(camel) thread #17 - TemporaryQueueReplyManager[temporary]
15:26:14,808 | DEBUG | Executing callback on JMS Session: PooledSession {
ActiveMQSession {id=ID:myhost-59213-1382385397272-3:2:4,started=true} } |
org.apache.camel.component.jms.JmsConfiguration$CamelJmsTemplate | Camel
(camel) thread #4 - JmsConsumer[q1]
15:26:14,808 | DEBUG | Endpoint[jms://q1] sending reply JMS message
[correlationId:a1115df5-affb-4f88-a909-dd877451a178]: ActiveMQMessage
{commandId = 0, responseRequired = false, messageId = null,
originalDestination = null, originalTransactionId = null, producerId = null,
destination = null, transactionId = null, expiration = 0, timestamp = 0,
arrival = 0, brokerInTime = 0, brokerOutTime = 0, correlationId =
a1115df5-affb-4f88-a909-dd877451a178, replyTo = null, persistent = true,
type = null, priority = 0, groupID = null, groupSequence = 0,
targetConsumerId = null, compressed = false, userID = null, content = null,
marshalledProperties = null, dataStructure = null, redeliveryCounter = 0,
size = 0, properties = null, readOnlyProperties = false, readOnlyBody =
false, droppable = false} |
org.apache.camel.component.jms.EndpointMessageListener | Camel (camel)
thread #4 - JmsConsumer[q1]
15:26:14,808 | DEBUG | Sending JMS message to:
temp-queue://ID:myhost-59240-1382385651460-3:1:669 with message:
ActiveMQMessage {commandId = 0, responseRequired = false, messageId = null,
originalDestination = null, originalTransactionId = null, producerId = null,
destination = null, transactionId = null, expiration = 0, timestamp = 0,
arrival = 0, brokerInTime = 0, brokerOutTime = 0, correlationId =
a1115df5-affb-4f88-a909-dd877451a178, replyTo = null, persistent = true,
type = null, priority = 0, groupID = null, groupSequence = 0,
targetConsumerId = null, compressed = false, userID = null, content = null,
marshalledProperties = null, dataStructure = null, redeliveryCounter = 0,
size = 0, properties = null, readOnlyProperties = false, readOnlyBody =
false, droppable = false} | org.apache.camel.component.jms.JmsConfiguration
| Camel (camel) thread #4 - JmsConsumer[q1]




--
View this message in context: http://activemq.2283324.n4.nabble.com/Strange-problem-in-activeMQ-or-Camel-resulting-in-wrong-message-type-tp4673120.html
Sent from the ActiveMQ - User mailing list archive at Nabble.com.

Mime
View raw message