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 2022/04/07 14:33:17 UTC
[activemq-artemis] 01/03: NO-JIRA Making AmqpFlowControlFailTest more accurate
This is an automated email from the ASF dual-hosted git repository.
clebertsuconic pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git
commit bad1c2658278b716d9266a90d9e4993733ab3b1f
Author: Clebert Suconic <cl...@apache.org>
AuthorDate: Tue Apr 5 18:18:27 2022 -0400
NO-JIRA Making AmqpFlowControlFailTest more accurate
this test was relying on internal details such as number of credits on the link.
The test was flaky and eventually failing or hunging.
---
.../artemis/core/postoffice/impl/PostOfficeImpl.java | 19 +++++++++++++++++++
.../integration/amqp/AmqpFlowControlFailTest.java | 20 +++++++++++---------
2 files changed, 30 insertions(+), 9 deletions(-)
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java
index af4f3a175e..d6866856f6 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java
@@ -1846,13 +1846,32 @@ public class PostOfficeImpl implements PostOffice, NotificationListener, Binding
super(scheduledExecutorService, executor, checkPeriod, timeUnit, onDemand);
}
+ volatile CountDownLatch inUseLatch;
+
+
+ @Override
+ public void stop() {
+ super.stop();
+ // this will do a best effort to stop the current latch.
+ // no big deal if it failed. this is just to optimize this component stop.
+ CountDownLatch latch = inUseLatch;
+ if (latch != null) {
+ latch.countDown();
+ }
+ }
+
+
@Override
public void run() {
// The reaper thread should be finished case the PostOffice is gone
// This is to avoid leaks on PostOffice between stops and starts
for (Queue queue : iterableOf(getLocalQueues())) {
+ if (!isStarted()) {
+ break;
+ }
try {
CountDownLatch latch = new CountDownLatch(1);
+ this.inUseLatch = latch;
queue.expireReferences(latch::countDown);
// the idea is in fact to block the Reaper while the Queue is executing reaping.
// This would avoid another eventual expiry to be called if the period for reaping is too small
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpFlowControlFailTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpFlowControlFailTest.java
index a3f9c500e5..e640a0e22f 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpFlowControlFailTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpFlowControlFailTest.java
@@ -19,13 +19,13 @@ package org.apache.activemq.artemis.tests.integration.amqp;
import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.core.settings.impl.AddressFullMessagePolicy;
import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
-import org.apache.activemq.artemis.tests.util.Wait;
import org.apache.activemq.transport.amqp.client.AmqpClient;
import org.apache.activemq.transport.amqp.client.AmqpConnection;
import org.apache.activemq.transport.amqp.client.AmqpMessage;
import org.apache.activemq.transport.amqp.client.AmqpReceiver;
import org.apache.activemq.transport.amqp.client.AmqpSender;
import org.apache.activemq.transport.amqp.client.AmqpSession;
+import org.junit.Assert;
import org.junit.Test;
import org.junit.experimental.runners.Enclosed;
import org.junit.runner.RunWith;
@@ -35,6 +35,7 @@ import java.io.IOException;
import java.util.Arrays;
import java.util.Collection;
import java.util.Map;
+import java.util.concurrent.TimeUnit;
import org.apache.qpid.proton.amqp.Symbol;
import org.apache.qpid.proton.amqp.messaging.Accepted;
@@ -77,10 +78,10 @@ public class AmqpFlowControlFailTest {
}
- @Test(timeout = 60000)
+ @Test(timeout = 10_000)
public void testAddressFullDisposition() throws Exception {
AmqpClient client = createAmqpClient(getBrokerAmqpConnectionURI());
- AmqpConnection connection = addConnection(client.connect());
+ AmqpConnection connection = client.connect();
try {
AmqpSession session = connection.createSession();
AmqpSender sender = session.createSender(getQueueName(), null, null, outcomes);
@@ -95,6 +96,7 @@ public class AmqpFlowControlFailTest {
rejected = true;
assertTrue(String.format("Unexpected message expected %s to contain %s", e.getMessage(), expectedMessage),
e.getMessage().contains(expectedMessage));
+ break;
}
}
@@ -112,10 +114,10 @@ public class AmqpFlowControlFailTest {
AmqpFlowControlFailTest.configureAddressPolicy(server);
}
- @Test(timeout = 60000)
+ @Test
public void testMesagesNotSent() throws Exception {
AmqpClient client = createAmqpClient(getBrokerAmqpConnectionURI());
- AmqpConnection connection = addConnection(client.connect());
+ AmqpConnection connection = client.connect();
int messagesSent = 0;
try {
AmqpSession session = connection.createSession();
@@ -130,22 +132,22 @@ public class AmqpFlowControlFailTest {
messagesSent++;
} catch (IOException e) {
rejected = true;
+ break;
}
}
assertTrue(rejected);
rejected = false;
- assertEquals(0, sender.getSender().getCredit());
AmqpSession session2 = connection.createSession();
AmqpReceiver receiver = session2.createReceiver(getQueueName());
receiver.flow(messagesSent);
for (int i = 0; i < messagesSent; i++) {
- AmqpMessage receive = receiver.receive();
+ AmqpMessage receive = receiver.receive(5, TimeUnit.SECONDS);
+ Assert.assertNotNull(receive);
receive.accept();
}
receiver.close();
session2.close();
- Wait.assertEquals(1000, sender.getSender()::getCredit);
for (int i = 0; i < 1000; i++) {
final AmqpMessage message = new AmqpMessage();
byte[] payload = new byte[100];
@@ -154,10 +156,10 @@ public class AmqpFlowControlFailTest {
sender.send(message);
} catch (IOException e) {
rejected = true;
+ break;
}
}
assertTrue(rejected);
- assertEquals(0, sender.getSender().getCredit());
} finally {
connection.close();
}