You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by cl...@apache.org on 2020/03/11 15:44:53 UTC
[activemq-artemis] branch master updated: ARTEMIS-2642 Fixing Drain
Timeout Issue on AMQP
This is an automated email from the ASF dual-hosted git repository.
clebertsuconic pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git
The following commit(s) were added to refs/heads/master by this push:
new c3128aa ARTEMIS-2642 Fixing Drain Timeout Issue on AMQP
new dc173d5 This closes #3012
c3128aa is described below
commit c3128aa144b08cd291e3ed00630015a00fa4fc01
Author: Clebert Suconic <cl...@apache.org>
AuthorDate: Tue Mar 10 19:27:04 2020 -0400
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.
---
.../amqp/proton/ProtonServerSenderContext.java | 2 +-
.../amqp/proton/handler/ProtonHandler.java | 3 -
.../tests/integration/amqp/DrainTimeoutTest.java | 88 ++++++++++++++++++++++
3 files changed, 89 insertions(+), 4 deletions(-)
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 eac0f10..bf425d8 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 62667dd..b874f4e 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 0000000..b2232fd
--- /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());
+ }
+}