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 <[email protected]> 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(); }
