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

ARTEMIS-3234 Additional test with expiry and paging 

The issue found by this test has been previously fixed as part of ARTEMIS-3234

There's no additional fixing needed. This is just to commit the test to avoid future regressions.
This commit is contained in:
Clebert Suconic 2021-08-23 11:14:12 -04:00 committed by clebertsuconic
parent 7e7150ffdc
commit d15680d8d9
1 changed files with 249 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

249
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/MessagesExpiredPagingTest.java Normal file
View File

@ -0,0 +1,249 @@
/*
* 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.paging;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.activemq.artemis.api.core.QueueConfiguration;
import org.apache.activemq.artemis.api.core.RoutingType;
import org.apache.activemq.artemis.core.config.Configuration;
import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.core.server.Queue;
import org.apache.activemq.artemis.core.server.impl.AddressInfo;
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.ActiveMQTestBase;
import org.apache.activemq.artemis.tests.util.CFUtil;
import org.apache.activemq.artemis.tests.util.Wait;
import org.jboss.logging.Logger;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
public class MessagesExpiredPagingTest extends ActiveMQTestBase {
private static final Logger logger = Logger.getLogger(MessagesExpiredPagingTest.class);
private static final String ADDRESS = "MessagesExpiredPagingTest";
private static final int NUMBER_OF_QUEUES = 10;
private AtomicBoolean running = new AtomicBoolean(true);
ActiveMQServer server;
protected static final int PAGE_MAX = 100 * 1024;
protected static final int PAGE_SIZE = 10 * 1024;
Queue[] queues = new Queue[NUMBER_OF_QUEUES];
Thread[] peskyExpires = new Thread[NUMBER_OF_QUEUES];
@After
@Override
public void tearDown() throws Exception {
running.set(false);
for (Thread t : peskyExpires) {
t.join(5000);
Assert.assertFalse(t.isAlive());
}
super.tearDown();
}
@Before
@Override
public void setUp() throws Exception {
super.setUp();
Configuration config = createDefaultConfig(0, true).setJournalSyncNonTransactional(false);
config.setMessageExpiryScanPeriod(-1);
server = createServer(true, config, PAGE_SIZE, PAGE_MAX);
server.getAddressSettingsRepository().clear();
AddressSettings defaultSetting = new AddressSettings().setPageSizeBytes(PAGE_SIZE).setMaxSizeBytes(PAGE_MAX).setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE).setAutoCreateAddresses(false).setAutoCreateQueues(false);
server.getAddressSettingsRepository().addMatch("#", defaultSetting);
server.start();
server.addAddressInfo(new AddressInfo(ADDRESS).addRoutingType(RoutingType.MULTICAST));
for (int i = 0; i < 10; i++) {
Queue queue = server.createQueue(new QueueConfiguration("q" + i).setRoutingType(RoutingType.MULTICAST).setAddress(ADDRESS));
queues[i] = queue;
peskyExpires[i] = new Thread(() -> {
while (running.get()) {
try {
// I am exagerating calls into expireReferences
// as I am trying to break things.
// Notice I'm not using the scan expiry from PostOffice, but I'm calling this from multiple places trying to really break stuff.
queue.expireReferences();
Thread.sleep(10);
} catch (Throwable e) {
logger.warn(e.getMessage(), e);
}
}
});
}
}
@Test
public void testSendReceiveCORE() throws Exception {
testSendReceive("CORE");
}
@Test
public void testSendReceiveAMQP() throws Exception {
testSendReceive("AMQP");
}
@Test
public void testSendReceiveOpenWire() throws Exception {
testSendReceive("OPENWIRE");
}
public void testSendReceive(String protocol) throws Exception {
ConnectionFactory factory = CFUtil.createConnectionFactory(protocol, "tcp://localhost:61616");
Consumer[] consumers = new Consumer[NUMBER_OF_QUEUES];
for (int i = 0; i < NUMBER_OF_QUEUES; i++) {
consumers[i] = new Consumer(factory, "q" + i);
consumers[i].start();
}
try (Connection connection = factory.createConnection()) {
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
javax.jms.Topic jmsTopic = session.createTopic(ADDRESS);
MessageProducer producer = session.createProducer(jmsTopic);
for (int i = 0; i < 10; i++) {
producer.send(session.createTextMessage("hello"));
}
// just validating basic queue consumption working
for (Consumer c : consumers) {
Wait.assertEquals(10, c.consumed::get);
}
for (Consumer c : consumers) {
c.consumedDelta.set(0);
}
producer.setTimeToLive(10);
for (int i = 0; i < 1000; i++) {
if (i > 0 && i % 100 == 0) {
for (Consumer c : consumers) {
Wait.waitFor(() -> !queues[0].getPagingStore().isPaging(), 1000, 100, false);
producer.setTimeToLive(TimeUnit.HOURS.toMillis(1));
producer.send(session.createTextMessage("hello"));
Wait.assertTrue(() -> c.consumedDelta.get() > 0);
producer.setTimeToLive(10);
c.consumedDelta.set(0);
}
queues[0].getPagingStore().forceAnotherPage();
}
producer.send(session.createTextMessage("hello"));
}
producer.setTimeToLive(300);
for (int i = 0; i < 1000; i++) {
if (i > 0 && i % 100 == 0) {
queues[0].getPagingStore().forceAnotherPage();
}
producer.send(session.createTextMessage("hello"));
}
producer.setTimeToLive(TimeUnit.HOURS.toMillis(1));
producer.send(session.createTextMessage("hello again")); // something not expiring
for (Consumer c : consumers) {
Wait.assertTrue(() -> c.consumedDelta.get() > 0);
}
running.set(false);
// first check just to make sure topics are being consumed regularly
for (Consumer c : consumers) {
c.join(5000);
Assert.assertFalse(c.isAlive());
Assert.assertEquals(0, c.errors.get());
}
}
}
private class Consumer extends Thread {
final ConnectionFactory factory;
AtomicInteger consumedDelta = new AtomicInteger(0);
AtomicInteger consumed = new AtomicInteger(0);
AtomicInteger errors = new AtomicInteger(0);
final String queueName;
Consumer(ConnectionFactory factory, String queueName) {
this.factory = factory;
this.queueName = queueName;
}
@Override
public void run() {
try {
int rec = 0;
try (Connection connection = factory.createConnection()) {
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
javax.jms.Queue jmsQueue = session.createQueue(ADDRESS + "::" + queueName);
MessageConsumer consumer = session.createConsumer(jmsQueue);
connection.start();
while (running.get()) {
Message message = consumer.receive(500);
if (message != null) {
consumed.incrementAndGet();
consumedDelta.incrementAndGet();
Thread.sleep(2);
}
if (rec++ > 10) {
rec = 0;
consumer.close();
consumer = session.createConsumer(jmsQueue);
}
}
}
} catch (Throwable e) {
errors.incrementAndGet();
}
}
}
}