[AMQ-2446] Client hangs on receive call with timeout value > 0 when activemq is shutdown (CTRL-C)

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 5.2.0
• Fix Version/s: 5.3.0

Environment

Windows XP SP3, ActiveMQ 5.2, prefetchSize=0 (java.naming.provider.url = tcp://localhost:61616?jms.prefetchPolicy.queuePrefetch=0)

• Assignee: Gary Tully
• Reporter: Carl Allain
• Created: 08/Oct/09 23:49
• Updated: 20/Jul/12 20:30
• Resolved: 09/Oct/09 16:30

Description

Start ActiveMQ

create a message consumer and make a receive(30000) call for example.

stop ActiveMQ (CTRL-C)

There is an error reported on the console:

2009-10-08 19:23:54,978 [ActiveMQ Transport: tcp://localhost/127.0.0.1:61616] DEBUG org.apache.activemq.ActiveMQConnection - Async exception with no exception listener: java.io.EOFException
java.io.EOFException
at java.io.DataInputStream.readInt(DataInputStream.java:358)
at org.apache.activemq.openwire.OpenWireFormat.unmarshal(OpenWireFormat.java:269)
at org.apache.activemq.transport.tcp.TcpTransport.readCommand(TcpTransport.java:210)
at org.apache.activemq.transport.tcp.TcpTransport.doRun(TcpTransport.java:202)
at org.apache.activemq.transport.tcp.TcpTransport.run(TcpTransport.java:185)
at java.lang.Thread.run(Thread.java:595)
2009-10-08 19:23:54,978 [ActiveMQ Connection Worker: tcp://localhost/127.0.0.1:61616] DEBUG org.apache.activemq.transport.tcp.TcpTransport - Stopping transport tcp://localhost/127.0.0.1:61616

The thread making the receive call is waiting on a mutex that the ActiveMQ Connection Worker thread that reported that log has (I guess):

class MessageDispatchChannel
public MessageDispatch dequeue(long timeout) throws InterruptedException {
synchronized (mutex) {
// Wait until the consumer is ready to deliver messages.
while (timeout != 0 && !closed && (list.isEmpty() || !running)) {
if (timeout == -1)

{ mutex.wait(); <----------------------------------------------------------- }

else

{ mutex.wait(timeout); break; }

}
if (closed || !running || list.isEmpty())

{ return null; }

return list.removeFirst();
}
}

What is also strange to me is that my timeout of 30000 ms is silently converted to an infinite timeout if the preFetchSize == 0 by this code? Why?

class ActiveMQMessageConsumer
...
MessageDispatch md;
if (info.getPrefetchSize() == 0)

{ md = dequeue(-1); // We let the broker let us know when we // timeout. }

else

{ md = dequeue(timeout); }