Hi,
Using 5.3-SNAPSHOT, messages still remain in the queue waiting for last
redelivery of bad messages, but AsyncError Exception dissapears
If it is userful, I am using CACHE_CONSUMER with
DefaultMessageListenerContainer. I can not use any other cache level to test
if this is the error cause because redeliveries only work with this cache
level.
Thanks,
Diego
Gary Tully wrote:
>
> can you verify if this behavior exists on trunk by using a 5.3-SNAPSHOT?
> If
> it does, the please open a jira issue and attach your test case. thanks.
>
> 2009/7/30 Diego Rodríguez Martín <[email protected]>
>
>> Hi,
>>
>> I'm trying to setup a config where message redelivery is important.
>> Every
>> message read from the queue is processed sending an http request to
>> another
>> host, so it can fail due to a lot of causes, and I want the system to
>> retry
>> the post later.
>> I have made some tests and this is what is happening:
>> There are, in this test, 100 messages in the queue. 2 messages are
>> going to make the http request to a non existent host and exit with
>> RuntimeException to rollback, ("bad messages"). 98 messages are going to
>> succeed, ("good messages"). This is configurable because the host is
>> extracted from a bdd with a message attribute as the key. What I expected
>> to
>> happen at the end of the test is 98 good messages processed and the 2 bad
>> messages remaining in the queue waiting for a retry until the max
>> redeliveries in RedeliveryPolicy is reached and then sent to DLQ. This is
>> going to happen after some minutes because the redelivery policy is
>> initialRedeliveryDelay 10000, maximumRedeliveries 6,
>> useExponentialBackOff
>> true, backOffMultiplier 3.
>> What is happening instead is that after a while with no activity, I
>> can
>> see 10 messages still in the original queue, that means that 8 good
>> messages
>> are stucked in the queue. Then the 2 bad messages are retried until
>> maximumRedeliveries is reached, some minutes later, sent to DLQ, and then
>> the 8 good messages are processed succesfully.
>> This means there is something that is retaining the rollbacked
>> messages, not letting other messages in the queue to be processed
>> First I thougth of prefetchPolicy, as the messages could be retained
>> by
>> this internal queue. Setting queuePrefetch to 1, didn't made any change.
>> Then I thougth of DefaultMessageListenerContainer, because its processing
>> threads could be blocked by this rollback.
>> Also, when there are some messages waiting to be redelivered, if I
>> desinstall the DefaultMessageListenerContainer consumers, I get an
>> exception
>> in the ActiveMQ broker, probably related with the connection that is
>> retaining the rollbacked messages:
>> 30 jul 2009 14:17:45,140 ERROR [ActiveMQ Transport:
>> tcp:///127.0.0.1:3656]
>> org.apache.activemq.broker.TransportConnection.Service - sync error
>> occurred:
>> javax.jms.JMSException: Transaction
>> 'TX:ID:mymachine-4579-1248953104078-0:5207:52' has not been started.
>> javax.jms.JMSException: Transaction
>> 'TX:ID:mymachine-4579-1248953104078-0:5207:52' has not been started.
>> at
>> org.apache.activemq.broker.TransactionBroker.getTransaction(TransactionBroker.java:270)
>> at
>> org.apache.activemq.broker.TransactionBroker.acknowledge(TransactionBroker.java:190)
>> at
>> org.apache.activemq.broker.BrokerFilter.acknowledge(BrokerFilter.java:74)
>> at
>> org.apache.activemq.broker.BrokerFilter.acknowledge(BrokerFilter.java:74)
>> at
>> org.apache.activemq.broker.BrokerFilter.acknowledge(BrokerFilter.java:74)
>> at
>> org.apache.activemq.broker.MutableBrokerFilter.acknowledge(MutableBrokerFilter.java:85)
>> at
>> org.apache.activemq.broker.TransportConnection.processMessageAck(TransportConnection.java:456)
>> at
>> org.apache.activemq.command.MessageAck.visit(MessageAck.java:205)
>> at
>> org.apache.activemq.broker.TransportConnection.service(TransportConnection.java:305)
>> at
>> org.apache.activemq.broker.TransportConnection$1.onCommand(TransportConnection.java:179)
>> at
>> org.apache.activemq.transport.TransportFilter.onCommand(TransportFilter.java:68)
>> at
>> org.apache.activemq.transport.WireFormatNegotiator.onCommand(WireFormatNegotiator.java:143)
>> at
>> org.apache.activemq.transport.InactivityMonitor.onCommand(InactivityMonitor.java:206)
>> at
>> org.apache.activemq.transport.TransportSupport.doConsume(TransportSupport.java:84)
>> at
>> org.apache.activemq.transport.tcp.TcpTransport.doRun(TcpTransport.java:203)
>> at
>> org.apache.activemq.transport.tcp.TcpTransport.run(TcpTransport.java:185)
>> at java.lang.Thread.run(Thread.java:595) I don't know how
>> to
>> test who's causing this problem, DefaultMessageListenerContainer or
>> ActiveMQ. Maybe is DefaultMessageListenerContainer consumer threads, but
>> the
>> exception in the broker means ActiveMQ could be also the problem.
>> Anyone can help me to narrow the problem?
>> I am using ActiveMQ 5.2 in a window box, and Spring
>> DefaultMessageListenerContainer for the consumers with this config. In
>> you
>> need more config data, please ask
>> <bean id="jmsConnectionFactory"
>> class="org.apache.activemq.spring.ActiveMQConnectionFactory"
>> lazy-init="true">
>> <property name="brokerURL"
>> value="failover:(tcp://localhost:6275?soTimeout=60000&connectionTimeout=30000)?initialReconnectDelay=20000&randomize=false&maxReconnectAttempts=-1&useExponentialBackOff=false&reconnectDelayExponent=2"/>
>> <property name="redeliveryPolicy">
>> <bean class="org.apache.activemq.RedeliveryPolicy">
>> <property name="initialRedeliveryDelay" value="10000" />
>> <property name="maximumRedeliveries" value="6" />
>> <property name="useExponentialBackOff" value="true" />
>> <property name="backOffMultiplier" value="3" />
>> </bean>
>> </property>
>> <property name="prefetchPolicy">
>> <bean class="org.apache.activemq.ActiveMQPrefetchPolicy">
>> <property name="queuePrefetch" value="1" />
>> </bean>
>> </property>
>> <property name="copyMessageOnSend" value="false"/>
>> <property name="userName" value="amquser"/>
>> <property name="password" value="amquserpassword"/>
>> </bean>
>> <bean id="inputQueueListenerContainer"
>> class="org.springframework.jms.listener.DefaultMessageListenerContainer"
>> destroy-method="shutdown" lazy-init="true">
>> <property name="connectionFactory">
>> <ref bean="jmsConnectionFactory" />
>> </property>
>> <property name="destination">
>> <ref bean="inputQueue" />
>> </property>
>> <property name="messageListener">
>> <ref bean="inputQueueMessageListener" />
>> </property>
>> <property name="sessionTransacted" value="true" />
>> <property name="concurrentConsumers" value="2" />
>> <property name="maxConcurrentConsumers" value="4" />
>> </bean>
>>
>> --
>> -------------------------------------------------------------
>> http://www.programacionenjava.com
>> -------------------------------------------------------------
>>
>>
>
>
> --
> http://blog.garytully.com
>
> Open Source Integration
> http://fusesource.com
>
>
--
View this message in context:
http://www.nabble.com/Redeliveries%2C-DLQ%2C-AsycError-tp24737910p24754258.html
Sent from the ActiveMQ - User mailing list archive at Nabble.com.
