Ensure that connections get closed to allow the vm transport resources

to get cleaned up.

activemq-unit-tests/src/test/java/org/apache/activemq/broker/scheduler/JmsCronSchedulerTest.java

@@ -82,7 +82,7 @@ public class JmsCronSchedulerTest extends JobSchedulerTestSupport {
             producer.send(message);
             LOG.info("Message {} sent at {}", i, new Date().toString());
             producer.close();
-            //wait a couple sec so cron start time is different for next message
+            // wait a couple sec so cron start time is different for next message
             Thread.sleep(2000);
         }
         SchedulerBroker sb = (SchedulerBroker) this.broker.getBroker().getAdaptor(SchedulerBroker.class);
@@ -90,8 +90,10 @@ public class JmsCronSchedulerTest extends JobSchedulerTestSupport {
         List<Job> list = js.getAllJobs();
         assertEquals(COUNT, list.size());
         latch.await(2, TimeUnit.MINUTES);
-        //All should messages should have been received by now
+        // All should messages should have been received by now
         assertEquals(COUNT, count.get());
+
+        connection.close();
     }
 
     @Test
@@ -114,5 +116,7 @@ public class JmsCronSchedulerTest extends JobSchedulerTestSupport {
 
         assertNotNull(consumer.receiveNoWait());
         assertNull(consumer.receiveNoWait());
+
+        connection.close();
     }
 }

activemq-unit-tests/src/test/java/org/apache/activemq/broker/scheduler/JmsSchedulerTest.java

@@ -220,7 +220,6 @@ public class JmsSchedulerTest extends JobSchedulerTestSupport {
         broker.start();
         broker.waitUntilStarted();
 
-
         // consume the message
         connection = createConnection();
         connection.start();
@@ -244,9 +243,9 @@ public class JmsSchedulerTest extends JobSchedulerTestSupport {
         broker.getSystemUsage().getJobSchedulerUsage().setLimit(10 * 1024);
 
         ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("vm://localhost");
-        Connection conn = factory.createConnection();
+        Connection connection = factory.createConnection();
-        conn.start();
+        connection.start();
-        Session sess = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        Session sess = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
         final long time = 5000;
         final ProducerThread producer = new ProducerThread(sess, destination) {
             @Override
@@ -295,6 +294,6 @@ public class JmsSchedulerTest extends JobSchedulerTestSupport {
         latch.await(20000l, TimeUnit.MILLISECONDS);
 
         assertEquals("Consumer did not receive all messages.", 0, latch.getCount());
-        conn.close();
+        connection.close();
     }
 }

activemq-unit-tests/src/test/java/org/apache/activemq/broker/scheduler/JobSchedulerManagementTest.java

@@ -78,6 +78,8 @@ public class JobSchedulerManagementTest extends JobSchedulerTestSupport {
         // Now wait and see if any get delivered, none should.
         latch.await(10, TimeUnit.SECONDS);
         assertEquals(latch.getCount(), COUNT);
+
+        connection.close();
     }
 
     @Test
@@ -144,6 +146,8 @@ public class JobSchedulerManagementTest extends JobSchedulerTestSupport {
         // Now wait and see if any get delivered, none should.
         latch.await(10, TimeUnit.SECONDS);
         assertEquals(2, latch.getCount());
+
+        connection.close();
     }
 
     @Test
@@ -203,6 +207,8 @@ public class JobSchedulerManagementTest extends JobSchedulerTestSupport {
         // now check that they all got delivered
         latch.await(10, TimeUnit.SECONDS);
         assertEquals(latch.getCount(), 0);
+
+        connection.close();
     }
 
     @Test
@@ -270,6 +276,8 @@ public class JobSchedulerManagementTest extends JobSchedulerTestSupport {
         // destination.
         latch.await(20, TimeUnit.SECONDS);
         assertEquals(0, latch.getCount());
+
+        connection.close();
     }
 
     @Test
@@ -327,6 +335,8 @@ public class JobSchedulerManagementTest extends JobSchedulerTestSupport {
         // now check that they all got removed and are not delivered.
         latch.await(11, TimeUnit.SECONDS);
         assertEquals(COUNT, latch.getCount());
+
+        connection.close();
     }
 
     @Test
@@ -341,7 +351,6 @@ public class JobSchedulerManagementTest extends JobSchedulerTestSupport {
         MessageProducer producer = session.createProducer(management);
 
         try {
-
             // Send the remove request
             Message remove = session.createMessage();
             remove.setStringProperty(ScheduledMessage.AMQ_SCHEDULER_ACTION, ScheduledMessage.AMQ_SCHEDULER_ACTION_REMOVEALL);
@@ -349,6 +358,8 @@ public class JobSchedulerManagementTest extends JobSchedulerTestSupport {
             producer.send(remove);
         } catch (Exception e) {
             fail("Caught unexpected exception during remove of unscheduled message.");
+        } finally {
+            connection.close();
         }
     }
 
@@ -388,6 +399,8 @@ public class JobSchedulerManagementTest extends JobSchedulerTestSupport {
         // Now check if there are anymore, there shouldn't be
         message = browser.receive(5000);
         assertNull(message);
+
+        connection.close();
     }
 
     protected void scheduleMessage(Connection connection, long delay) throws Exception {