[jira] [Created] (ARTEMIS-2642) Drain requests with a selector can cause Drain Timeouts

["Mike Youngstrom (Jira)" <ji...@apache.org>]

Mike Youngstrom created ARTEMIS-2642:
----------------------------------------

             Summary: 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


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.

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&transport.trustAll=true&transport.verifyHost=false&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 timeException 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 moreCaused 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)