Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

AMQ-9193 - Improve broker shutdown in unit tests 

This should improve test reliability for the unit tests so brokers don't
hang around after the end of a test on error. Also increase the surefire
re-run count to 3 times before failing.

(cherry picked from commit a083ff4d23)
This commit is contained in:
Christopher L. Shannon (cshannon) 2023-01-11 06:21:21 -05:00
parent 164568aab4
commit 5e12551a4a
12 changed files with 157 additions and 107 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
Jenkinsfile vendored
View File

@ -75,7 +75,7 @@ pipeline {
echo 'Running tests'
// all tests is very very long (10 hours on Apache Jenkins)
// sh 'mvn -B -e test -pl activemq-unit-tests -Dactivemq.tests=all'
sh 'mvn -B -e -fae test -Dsurefire.rerunFailingTestsCount=2'
sh 'mvn -B -e -fae test -Dsurefire.rerunFailingTestsCount=3'
}
post {
always {

184
activemq-ra/src/test/java/org/apache/activemq/ra/MDBTest.java
View File

@ -467,102 +467,114 @@ public class MDBTest {
public void testErrorOnNoMessageDeliveryBrokerZeroPrefetchConfig() throws Exception {
final BrokerService brokerService = new BrokerService();
final String brokerUrl = "vm://zeroPrefetch?create=false";
brokerService.setBrokerName("zeroPrefetch");
brokerService.setPersistent(false);
PolicyMap policyMap = new PolicyMap();
PolicyEntry zeroPrefetchPolicy = new PolicyEntry();
zeroPrefetchPolicy.setQueuePrefetch(0);
policyMap.setDefaultEntry(zeroPrefetchPolicy);
brokerService.setDestinationPolicy(policyMap);
brokerService.start();
final AtomicReference<String> errorMessage = new AtomicReference<String>();
final var appender = new AbstractAppender("test", new AbstractFilter() {}, new MessageLayout(), false, new Property[0]) {
@Override
public void append(LogEvent event) {
if (event.getLevel().isMoreSpecificThan(Level.ERROR)) {
System.err.println("Event :" + event.getMessage().getFormattedMessage());
errorMessage.set(event.getMessage().getFormattedMessage());
try {
final String brokerUrl = "vm://zeroPrefetch?create=false";
brokerService.setBrokerName("zeroPrefetch");
brokerService.setPersistent(false);
PolicyMap policyMap = new PolicyMap();
PolicyEntry zeroPrefetchPolicy = new PolicyEntry();
zeroPrefetchPolicy.setQueuePrefetch(0);
policyMap.setDefaultEntry(zeroPrefetchPolicy);
brokerService.setDestinationPolicy(policyMap);
brokerService.start();
final AtomicReference<String> errorMessage = new AtomicReference<String>();
final var appender = new AbstractAppender("test", new AbstractFilter() {
}, new MessageLayout(), false, new Property[0]) {
@Override
public void append(LogEvent event) {
if (event.getLevel().isMoreSpecificThan(Level.ERROR)) {
System.err.println("Event :" + event.getMessage().getFormattedMessage());
errorMessage.set(event.getMessage().getFormattedMessage());
}
}
}
};
appender.start();
final var logger = org.apache.logging.log4j.core.Logger.class.cast(LogManager.getRootLogger());
logger.addAppender(appender);
logger.get().addAppender(appender, Level.INFO, new AbstractFilter() {});
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(brokerUrl);
Connection connection = factory.createConnection();
connection.start();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
MessageConsumer advisory = session.createConsumer(AdvisorySupport.getConsumerAdvisoryTopic(new ActiveMQQueue("TEST")));
ActiveMQResourceAdapter adapter = new ActiveMQResourceAdapter();
adapter.setServerUrl(brokerUrl);
adapter.start(new StubBootstrapContext());
final CountDownLatch messageDelivered = new CountDownLatch(1);
final StubMessageEndpoint endpoint = new StubMessageEndpoint() {
@Override
public void onMessage(Message message) {
super.onMessage(message);
messageDelivered.countDown();
};
};
appender.start();
ActiveMQActivationSpec activationSpec = new ActiveMQActivationSpec();
activationSpec.setDestinationType(Queue.class.getName());
activationSpec.setDestination("TEST");
activationSpec.setResourceAdapter(adapter);
activationSpec.validate();
final var logger = org.apache.logging.log4j.core.Logger.class.cast(
LogManager.getRootLogger());
logger.addAppender(appender);
logger.get().addAppender(appender, Level.INFO, new AbstractFilter() {
});
MessageEndpointFactory messageEndpointFactory = new MessageEndpointFactory() {
@Override
public MessageEndpoint createEndpoint(XAResource resource) throws UnavailableException {
endpoint.xaresource = resource;
return endpoint;
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(brokerUrl);
Connection connection = factory.createConnection();
connection.start();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
MessageConsumer advisory = session.createConsumer(
AdvisorySupport.getConsumerAdvisoryTopic(new ActiveMQQueue("TEST")));
ActiveMQResourceAdapter adapter = new ActiveMQResourceAdapter();
adapter.setServerUrl(brokerUrl);
adapter.start(new StubBootstrapContext());
final CountDownLatch messageDelivered = new CountDownLatch(1);
final StubMessageEndpoint endpoint = new StubMessageEndpoint() {
@Override
public void onMessage(Message message) {
super.onMessage(message);
messageDelivered.countDown();
}
;
};
ActiveMQActivationSpec activationSpec = new ActiveMQActivationSpec();
activationSpec.setDestinationType(Queue.class.getName());
activationSpec.setDestination("TEST");
activationSpec.setResourceAdapter(adapter);
activationSpec.validate();
MessageEndpointFactory messageEndpointFactory = new MessageEndpointFactory() {
@Override
public MessageEndpoint createEndpoint(XAResource resource)
throws UnavailableException {
endpoint.xaresource = resource;
return endpoint;
}
@Override
public boolean isDeliveryTransacted(Method method) throws NoSuchMethodException {
return true;
}
};
// Activate an Endpoint
adapter.endpointActivation(messageEndpointFactory, activationSpec);
ActiveMQMessage msg = (ActiveMQMessage) advisory.receive(4000);
if (msg != null) {
assertEquals("Prefetch size hasn't been set", 0,
((ConsumerInfo) msg.getDataStructure()).getPrefetchSize());
} else {
fail("Consumer hasn't been created");
}
@Override
public boolean isDeliveryTransacted(Method method) throws NoSuchMethodException {
return true;
}
};
// Send the broker a message to that endpoint
MessageProducer producer = session.createProducer(new ActiveMQQueue("TEST"));
producer.send(session.createTextMessage("Hello!"));
// Activate an Endpoint
adapter.endpointActivation(messageEndpointFactory, activationSpec);
connection.close();
ActiveMQMessage msg = (ActiveMQMessage)advisory.receive(4000);
if (msg != null) {
assertEquals("Prefetch size hasn't been set", 0, ((ConsumerInfo)msg.getDataStructure()).getPrefetchSize());
} else {
fail("Consumer hasn't been created");
// Wait for the message to be delivered.
assertFalse(messageDelivered.await(5000, TimeUnit.MILLISECONDS));
// Shut the Endpoint down.
adapter.endpointDeactivation(messageEndpointFactory, activationSpec);
adapter.stop();
assertNotNull("We got an error message", errorMessage.get());
assertTrue("correct message: " + errorMessage.get(),
errorMessage.get().contains("zero"));
logger.removeAppender(appender);
logger.get().removeAppender("test");
} finally {
brokerService.stop();
}
// Send the broker a message to that endpoint
MessageProducer producer = session.createProducer(new ActiveMQQueue("TEST"));
producer.send(session.createTextMessage("Hello!"));
connection.close();
// Wait for the message to be delivered.
assertFalse(messageDelivered.await(5000, TimeUnit.MILLISECONDS));
// Shut the Endpoint down.
adapter.endpointDeactivation(messageEndpointFactory, activationSpec);
adapter.stop();
assertNotNull("We got an error message", errorMessage.get());
assertTrue("correct message: " + errorMessage.get(), errorMessage.get().contains("zero"));
logger.removeAppender(appender);
logger.get().removeAppender("test");
brokerService.stop();
}
@Test

8
activemq-unit-tests/src/test/java/org/apache/activemq/ExpiredAckAsyncConsumerTest.java
View File

@ -99,8 +99,12 @@ public class ExpiredAckAsyncConsumerTest {
@After
public void tearDown() throws Exception {
connectionConsumer.close();
connection.close();
try {
connectionConsumer.close();
connection.close();
} catch (Exception e) {
//swallow any error so broker can still be stopped
}
broker.stop();
broker.waitUntilStopped();
}

6
activemq-unit-tests/src/test/java/org/apache/activemq/broker/virtual/VirtualTopicWildcardTest.java
View File

@ -63,7 +63,11 @@ public class VirtualTopicWildcardTest {
@After
public void afer() throws Exception {
connection.close();
try {
connection.close();
} catch (Exception e) {
//swallow any error so broker can still be stopped
}
brokerService.stop();
}

8
activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ2801Test.java
View File

@ -105,8 +105,12 @@ public class AMQ2801Test
@After
public void tearDown() throws Exception {
conn1.close();
conn2.close();
try {
conn1.close();
conn2.close();
} catch (Exception e) {
//swallow any error so broker can still be stopped
}
if (broker != null) {
broker.stop();
}

14
activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ3145Test.java
View File

@ -72,12 +72,16 @@ public class AMQ3145Test {
@After
public void tearDown() throws Exception {
if (consumer != null) {
consumer.close();
try {
if (consumer != null) {
consumer.close();
}
session.close();
connection.stop();
connection.close();
} catch (Exception e) {
//swallow any error so broker can still be stopped
}
session.close();
connection.stop();
connection.close();
broker.stop();
}

6
activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ3732Test.java
View File

@ -71,7 +71,11 @@ public class AMQ3732Test {
@After
public void stopBroker() throws Exception {
connection.close();
try {
connection.close();
} catch (Exception e) {
//swallow any error so broker can still be stopped
}
broker.stop();
broker.waitUntilStopped();

4
activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ6815Test.java
View File

@ -65,8 +65,10 @@ public class AMQ6815Test {
@After
public void tearDown() throws Exception {
if (connection != null) {
try {
connection.close();
} catch (Exception e) {
//swallow any error so broker can still be stopped
}
brokerService.stop();
}

8
activemq-unit-tests/src/test/java/org/apache/activemq/bugs/OutOfOrderTestCase.java
View File

@ -68,8 +68,12 @@ public class OutOfOrderTestCase extends TestCase {
}
protected void tearDown() throws Exception {
session.close();
connection.close();
try {
session.close();
connection.close();
} catch (Exception e) {
//swallow any error so broker can still be stopped
}
brokerService.stop();
}

6
activemq-unit-tests/src/test/java/org/apache/activemq/proxy/ProxyTestSupport.java
View File

@ -83,7 +83,11 @@ public class ProxyTestSupport extends BrokerTestSupport {
protected void tearDown() throws Exception {
for (Iterator<StubConnection> iter = connections.iterator(); iter.hasNext();) {
StubConnection connection = iter.next();
connection.stop();
try {
connection.stop();
} catch (Exception e) {
//swallow any error so broker can still be stopped
}
iter.remove();
}
remoteBroker.stop();

6
activemq-unit-tests/src/test/java/org/apache/activemq/usecases/JMXRemoveQueueThenSendIgnoredTest.java
View File

@ -149,7 +149,11 @@ public class JMXRemoveQueueThenSendIgnoredTest {
@After
public void tearDown() throws Exception {
connection.close();
try {
connection.close();
} catch (Exception e) {
//swallow any error so broker can still be stopped
}
brokerService.stop();
}
}

12
activemq-unit-tests/src/test/java/org/apache/activemq/usecases/TopicSubscriptionZeroPrefetchTest.java
View File

@ -285,10 +285,14 @@ public class TopicSubscriptionZeroPrefetchTest {
@After
public void tearDown() throws Exception {
consumer.close();
producer.close();
session.close();
connection.close();
try {
consumer.close();
producer.close();
session.close();
connection.close();
} catch (Exception e) {
//swallow any error so broker can still be stopped
}
brokerService.stop();
}