From c3128aa144b08cd291e3ed00630015a00fa4fc01 Mon Sep 17 00:00:00 2001 From: Clebert Suconic Date: Tue, 10 Mar 2020 19:27:04 -0400 Subject: [PATCH] ARTEMIS-2642 Fixing Drain Timeout Issue on AMQP The drain control has to immediately flush otherwise a next flow control event may remove the previous status from Proton. So, this really cannot wait the next executor, and it has to be done immediately. --- .../proton/ProtonServerSenderContext.java | 2 +- .../amqp/proton/handler/ProtonHandler.java | 3 - .../integration/amqp/DrainTimeoutTest.java | 88 +++++++++++++++++++ 3 files changed, 89 insertions(+), 4 deletions(-) create mode 100644 tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/DrainTimeoutTest.java diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/ProtonServerSenderContext.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/ProtonServerSenderContext.java index eac0f10a54..bf425d809b 100644 --- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/ProtonServerSenderContext.java +++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/ProtonServerSenderContext.java @@ -1024,7 +1024,7 @@ public class ProtonServerSenderContext extends ProtonInitializable implements Pr public void reportDrained() { connection.requireInHandler(); sender.drained(); - connection.flush(); + connection.instantFlush(); } public AMQPSessionContext getSessionContext() { diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/handler/ProtonHandler.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/handler/ProtonHandler.java index 62667ddce3..b874f4eade 100644 --- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/handler/ProtonHandler.java +++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/handler/ProtonHandler.java @@ -174,9 +174,6 @@ public class ProtonHandler extends ProtonInitializable implements SaslListener { sasl.setListener(this); } - - - /** When processing large messages, we require to flush bytes every processing */ public void instantFlush() { this.flushInstantly = true; // This will perform event handling, and at some point the flushBytes will be called diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/DrainTimeoutTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/DrainTimeoutTest.java new file mode 100644 index 0000000000..b2232fd3ea --- /dev/null +++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/DrainTimeoutTest.java @@ -0,0 +1,88 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.activemq.artemis.tests.integration.amqp; + +import javax.jms.JMSConsumer; +import javax.jms.JMSContext; +import javax.jms.JMSProducer; +import javax.jms.Message; +import javax.jms.Queue; +import javax.jms.Session; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.LongAdder; + +import org.apache.qpid.jms.JmsConnectionFactory; +import org.junit.Assert; +import org.junit.Test; + +public class DrainTimeoutTest extends AmqpClientTestSupport { + + final int NUMBER_OF_MESSAGES = 1000; + + @Test(timeout = 300_000) + public void testFlowControl() throws Exception { + final AtomicInteger errors = new AtomicInteger(0); + final String queueName = getQueueName(); + JmsConnectionFactory connectionFactory = + new JmsConnectionFactory( + "amqp://localhost:5672?jms.prefetchPolicy.all=1&jms.connectTimeout=60000&amqp.drainTimeout=1000"); + LongAdder sendCount = new LongAdder(); + LongAdder consumeCount = new LongAdder(); + Thread consumerThread = + new Thread( + () -> { + try (JMSContext listenerContext = + connectionFactory.createContext(Session.AUTO_ACKNOWLEDGE)) { + try (JMSConsumer consumer = + listenerContext.createConsumer( + listenerContext.createQueue(queueName))) { + while (!Thread.interrupted()) { + while (true) { + if (consumer.receiveNoWait() == null) { + break; + } + consumeCount.increment(); + if (consumeCount.sum() == NUMBER_OF_MESSAGES) { + return; + } + } + } + } + } catch (Exception e) { + e.printStackTrace(System.out); + errors.incrementAndGet(); + } + }); + consumerThread.start(); + try (JMSContext context = connectionFactory.createContext(Session.AUTO_ACKNOWLEDGE)) { + final Message message = context.createMessage(); + message.setStringProperty("selector", "dude"); + JMSProducer producer = context.createProducer(); + Queue queue = context.createQueue(queueName); + while (sendCount.sum() < NUMBER_OF_MESSAGES && !Thread.interrupted()) { + producer.send(queue, message); + sendCount.increment(); + long sent = sendCount.sum(); + } + } + + consumerThread.join(); + + Assert.assertEquals(0, errors.get()); + } +}