[
https://issues.apache.org/jira/browse/ARTEMIS-2642?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Mike Youngstrom updated ARTEMIS-2642:
-------------------------------------
Description:
Using the Qpid JMS AMQP client attempting to receive messages with no wait with
a selector can produce very long drain times causing a Drain Timeout on the
client. If I change to using "receive()" (causing the qpid client to no longer
send drain requests) the problem goes away.
Test Case using Qpid AMQP JMS client
{code:java}
public static void main(String[] args) throws Exception {
final String queueName = "queue";
var connectionFactory =
new JmsConnectionFactory(
"tqadmin",
"admin",
"amqp://localhost:5672?jms.prefetchPolicy.all=1&jms.connectTimeout=60000&amqp.drainTimeout=10000");
connectionFactory.setExceptionListener(
e -> {
System.out.println("Got a JMSException. Terminating the VM.");
e.printStackTrace();
Runtime.getRuntime().halt(100);
});
var sendCount = new LongAdder();
var consumeCount = new LongAdder();
var consumerThread =
new Thread(
() -> {
try (var listenerContext =
connectionFactory.createContext(Session.AUTO_ACKNOWLEDGE)) {
try (var consumer =
listenerContext.createConsumer(
listenerContext.createQueue(queueName),
"selector='dude'")) {
while (!Thread.interrupted()) {
while (consumer.receiveNoWait() != null) {
consumeCount.increment();
long consumed = consumeCount.sum();
if (consumed % 100 == 0) {
System.out.println("Messages Consumed: " + consumed);
}
}
}
}
}
});
consumerThread.start();
try (var context =
connectionFactory.createContext(Session.AUTO_ACKNOWLEDGE)) {
final Message message = context.createMessage();
message.setStringProperty("selector", "dude");
var producer = context.createProducer();
var queue = context.createQueue(queueName);
while (sendCount.sum() < 100000 && !Thread.interrupted()) {
producer.send(queue, message);
sendCount.increment();
long sent = sendCount.sum();
if (sent % 100 == 0) {
System.out.println("Messages Sent: " + sent);
}
}
}
}
{code}
Error Thrown after about 2000 messages are consumed (in a default local
environment)
{code:java}
Exception in thread "Thread-0" javax.jms.JMSRuntimeException: Remote did not
respond to a drain request in time
at
org.apache.qpid.jms.exceptions.JmsExceptionSupport.createRuntimeException(JmsExceptionSupport.java:211)
at org.apache.qpid.jms.JmsConsumer.receiveNoWait(JmsConsumer.java:100)
at connections.TQTest2.lambda$1(TQTest2.java:33)
at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: org.apache.qpid.jms.JmsOperationTimedOutException: Remote did not
respond to a drain request in time
at
org.apache.qpid.jms.provider.exceptions.ProviderOperationTimedOutException.toJMSException(ProviderOperationTimedOutException.java:39)
at
org.apache.qpid.jms.provider.exceptions.ProviderOperationTimedOutException.toJMSException(ProviderOperationTimedOutException.java:1)
at
org.apache.qpid.jms.exceptions.JmsExceptionSupport.create(JmsExceptionSupport.java:80)
at
org.apache.qpid.jms.exceptions.JmsExceptionSupport.create(JmsExceptionSupport.java:112)
at org.apache.qpid.jms.JmsConnection.pull(JmsConnection.java:915)
at org.apache.qpid.jms.JmsConnection.pull(JmsConnection.java:899)
at
org.apache.qpid.jms.JmsMessageConsumer.performPullIfRequired(JmsMessageConsumer.java:726)
at
org.apache.qpid.jms.JmsMessageConsumer.dequeue(JmsMessageConsumer.java:332)
at
org.apache.qpid.jms.JmsMessageConsumer.receiveNoWait(JmsMessageConsumer.java:221)
at org.apache.qpid.jms.JmsConsumer.receiveNoWait(JmsConsumer.java:98)
... 2 more
Caused by:
org.apache.qpid.jms.provider.exceptions.ProviderOperationTimedOutException:
Remote did not respond to a drain request in time
at
org.apache.qpid.jms.provider.amqp.AmqpConsumer.lambda$1(AmqpConsumer.java:179)
at
io.netty.util.concurrent.PromiseTask$RunnableAdapter.call(PromiseTask.java:38)
at
io.netty.util.concurrent.ScheduledFutureTask.run(ScheduledFutureTask.java:127)
at
io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:163)
at
io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:416)
at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:515)
at
io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:918)
at
io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
... 1 more
{code}
was:
Using the Qpid JMS AMQP client attempting to receive messages with no wait with
a selector can produce very long drain times causing a Drain Timeout on the
client. If I change to using "receive()" (causing the qpid client to no longer
send drain requests) the problem goes away
Test Case using Qpid AMQP JMS client
{code:java}
public static void main(String[] args) throws Exception {
final String queueName = "queue";
var connectionFactory =
new JmsConnectionFactory(
"tqadmin",
"admin",
"amqp://localhost:5672?jms.prefetchPolicy.all=1&jms.connectTimeout=60000&amqp.drainTimeout=10000");
connectionFactory.setExceptionListener(
e -> {
System.out.println("Got a JMSException. Terminating the VM.");
e.printStackTrace();
Runtime.getRuntime().halt(100);
});
var sendCount = new LongAdder();
var consumeCount = new LongAdder();
var consumerThread =
new Thread(
() -> {
try (var listenerContext =
connectionFactory.createContext(Session.AUTO_ACKNOWLEDGE)) {
try (var consumer =
listenerContext.createConsumer(
listenerContext.createQueue(queueName),
"selector='dude'")) {
while (!Thread.interrupted()) {
while (consumer.receiveNoWait() != null) {
consumeCount.increment();
long consumed = consumeCount.sum();
if (consumed % 100 == 0) {
System.out.println("Messages Consumed: " + consumed);
}
}
}
}
}
});
consumerThread.start();
try (var context =
connectionFactory.createContext(Session.AUTO_ACKNOWLEDGE)) {
final Message message = context.createMessage();
message.setStringProperty("selector", "dude");
var producer = context.createProducer();
var queue = context.createQueue(queueName);
while (sendCount.sum() < 100000 && !Thread.interrupted()) {
producer.send(queue, message);
sendCount.increment();
long sent = sendCount.sum();
if (sent % 100 == 0) {
System.out.println("Messages Sent: " + sent);
}
}
}
}
{code}
Error Thrown after about 2000 messages are consumed (in a default local
environment)
{code:java}
Exception in thread "Thread-0" javax.jms.JMSRuntimeException: Remote did not
respond to a drain request in time
at
org.apache.qpid.jms.exceptions.JmsExceptionSupport.createRuntimeException(JmsExceptionSupport.java:211)
at org.apache.qpid.jms.JmsConsumer.receiveNoWait(JmsConsumer.java:100)
at connections.TQTest2.lambda$1(TQTest2.java:33)
at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: org.apache.qpid.jms.JmsOperationTimedOutException: Remote did not
respond to a drain request in time
at
org.apache.qpid.jms.provider.exceptions.ProviderOperationTimedOutException.toJMSException(ProviderOperationTimedOutException.java:39)
at
org.apache.qpid.jms.provider.exceptions.ProviderOperationTimedOutException.toJMSException(ProviderOperationTimedOutException.java:1)
at
org.apache.qpid.jms.exceptions.JmsExceptionSupport.create(JmsExceptionSupport.java:80)
at
org.apache.qpid.jms.exceptions.JmsExceptionSupport.create(JmsExceptionSupport.java:112)
at org.apache.qpid.jms.JmsConnection.pull(JmsConnection.java:915)
at org.apache.qpid.jms.JmsConnection.pull(JmsConnection.java:899)
at
org.apache.qpid.jms.JmsMessageConsumer.performPullIfRequired(JmsMessageConsumer.java:726)
at
org.apache.qpid.jms.JmsMessageConsumer.dequeue(JmsMessageConsumer.java:332)
at
org.apache.qpid.jms.JmsMessageConsumer.receiveNoWait(JmsMessageConsumer.java:221)
at org.apache.qpid.jms.JmsConsumer.receiveNoWait(JmsConsumer.java:98)
... 2 more
Caused by:
org.apache.qpid.jms.provider.exceptions.ProviderOperationTimedOutException:
Remote did not respond to a drain request in time
at
org.apache.qpid.jms.provider.amqp.AmqpConsumer.lambda$1(AmqpConsumer.java:179)
at
io.netty.util.concurrent.PromiseTask$RunnableAdapter.call(PromiseTask.java:38)
at
io.netty.util.concurrent.ScheduledFutureTask.run(ScheduledFutureTask.java:127)
at
io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:163)
at
io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:416)
at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:515)
at
io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:918)
at
io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
... 1 more
{code}
> Drain requests with a selector can cause Drain Timeouts
> -------------------------------------------------------
>
> Key: ARTEMIS-2642
> URL: https://issues.apache.org/jira/browse/ARTEMIS-2642
> Project: ActiveMQ Artemis
> Issue Type: Bug
> Components: Broker
> Affects Versions: 2.11.0
> Reporter: Mike Youngstrom
> Priority: Major
>
> Using the Qpid JMS AMQP client attempting to receive messages with no wait
> with a selector can produce very long drain times causing a Drain Timeout on
> the client. If I change to using "receive()" (causing the qpid client to no
> longer send drain requests) the problem goes away.
> Test Case using Qpid AMQP JMS client
> {code:java}
> public static void main(String[] args) throws Exception {
> final String queueName = "queue";
> var connectionFactory =
> new JmsConnectionFactory(
> "tqadmin",
> "admin",
>
> "amqp://localhost:5672?jms.prefetchPolicy.all=1&jms.connectTimeout=60000&amqp.drainTimeout=10000");
> connectionFactory.setExceptionListener(
> e -> {
> System.out.println("Got a JMSException. Terminating the VM.");
> e.printStackTrace();
> Runtime.getRuntime().halt(100);
> });
> var sendCount = new LongAdder();
> var consumeCount = new LongAdder();
> var consumerThread =
> new Thread(
> () -> {
> try (var listenerContext =
> connectionFactory.createContext(Session.AUTO_ACKNOWLEDGE)) {
> try (var consumer =
> listenerContext.createConsumer(
> listenerContext.createQueue(queueName),
> "selector='dude'")) {
> while (!Thread.interrupted()) {
> while (consumer.receiveNoWait() != null) {
> consumeCount.increment();
> long consumed = consumeCount.sum();
> if (consumed % 100 == 0) {
> System.out.println("Messages Consumed: " + consumed);
> }
> }
> }
> }
> }
> });
> consumerThread.start();
> try (var context =
> connectionFactory.createContext(Session.AUTO_ACKNOWLEDGE)) {
> final Message message = context.createMessage();
> message.setStringProperty("selector", "dude");
> var producer = context.createProducer();
> var queue = context.createQueue(queueName);
> while (sendCount.sum() < 100000 && !Thread.interrupted()) {
> producer.send(queue, message);
> sendCount.increment();
> long sent = sendCount.sum();
> if (sent % 100 == 0) {
> System.out.println("Messages Sent: " + sent);
> }
> }
> }
> }
> {code}
> Error Thrown after about 2000 messages are consumed (in a default local
> environment)
> {code:java}
> Exception in thread "Thread-0" javax.jms.JMSRuntimeException: Remote did not
> respond to a drain request in time
> at
> org.apache.qpid.jms.exceptions.JmsExceptionSupport.createRuntimeException(JmsExceptionSupport.java:211)
> at org.apache.qpid.jms.JmsConsumer.receiveNoWait(JmsConsumer.java:100)
> at connections.TQTest2.lambda$1(TQTest2.java:33)
> at java.base/java.lang.Thread.run(Thread.java:834)
> Caused by: org.apache.qpid.jms.JmsOperationTimedOutException: Remote did not
> respond to a drain request in time
> at
> org.apache.qpid.jms.provider.exceptions.ProviderOperationTimedOutException.toJMSException(ProviderOperationTimedOutException.java:39)
> at
> org.apache.qpid.jms.provider.exceptions.ProviderOperationTimedOutException.toJMSException(ProviderOperationTimedOutException.java:1)
> at
> org.apache.qpid.jms.exceptions.JmsExceptionSupport.create(JmsExceptionSupport.java:80)
> at
> org.apache.qpid.jms.exceptions.JmsExceptionSupport.create(JmsExceptionSupport.java:112)
> at org.apache.qpid.jms.JmsConnection.pull(JmsConnection.java:915)
> at org.apache.qpid.jms.JmsConnection.pull(JmsConnection.java:899)
> at
> org.apache.qpid.jms.JmsMessageConsumer.performPullIfRequired(JmsMessageConsumer.java:726)
> at
> org.apache.qpid.jms.JmsMessageConsumer.dequeue(JmsMessageConsumer.java:332)
> at
> org.apache.qpid.jms.JmsMessageConsumer.receiveNoWait(JmsMessageConsumer.java:221)
> at org.apache.qpid.jms.JmsConsumer.receiveNoWait(JmsConsumer.java:98)
> ... 2 more
> Caused by:
> org.apache.qpid.jms.provider.exceptions.ProviderOperationTimedOutException:
> Remote did not respond to a drain request in time
> at
> org.apache.qpid.jms.provider.amqp.AmqpConsumer.lambda$1(AmqpConsumer.java:179)
> at
> io.netty.util.concurrent.PromiseTask$RunnableAdapter.call(PromiseTask.java:38)
> at
> io.netty.util.concurrent.ScheduledFutureTask.run(ScheduledFutureTask.java:127)
> at
> io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:163)
> at
> io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:416)
> at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:515)
> at
> io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:918)
> at
> io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
> ... 1 more
> {code}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
