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

This closes #3658

This commit is contained in:
Clebert Suconic 2021-07-16 09:19:03 -04:00
parent 595af7f904 c479cb558a
commit 4883285a2a
7 changed files with 826 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/DuplicateDetectionTest.java
View File

@ -20,6 +20,9 @@ import javax.transaction.xa.XAException;
import javax.transaction.xa.XAResource;
import javax.transaction.xa.Xid;
import java.util.Arrays;
import java.util.Collection;
import org.apache.activemq.artemis.api.core.ActiveMQDuplicateIdException;
import org.apache.activemq.artemis.api.core.ActiveMQException;
import org.apache.activemq.artemis.api.core.Message;
@ -42,11 +45,27 @@ import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
import org.apache.activemq.artemis.utils.UUIDGenerator;
import org.jboss.logging.Logger;
import org.junit.Assert;
import org.junit.Assume;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
@RunWith(Parameterized.class)
public class DuplicateDetectionTest extends ActiveMQTestBase {
@Parameterized.Parameters(name = "persistentCache={0}")
public static Collection<Object[]> parameters() {
return Arrays.asList(new Object[][] {
{true}, {false}
});
}
@Parameterized.Parameter(0)
public boolean persistCache;
private final Logger log = Logger.getLogger(this.getClass());
private ActiveMQServer server;
@ -217,6 +236,7 @@ public class DuplicateDetectionTest extends ActiveMQTestBase {
// we would eventually have a higher number of caches while we couldn't have time to clear previous ones
@Test
public void testShrinkCache() throws Exception {
Assume.assumeTrue("This test would restart the server", persistCache);
server.stop();
server.getConfiguration().setIDCacheSize(150);
server.start();
@ -1454,6 +1474,7 @@ public class DuplicateDetectionTest extends ActiveMQTestBase {
@Test
public void testPersistTransactional() throws Exception {
Assume.assumeTrue("This test would restart the server", persistCache);
ClientSession session = sf.createSession(false, false, false);
session.start();
@ -1709,6 +1730,8 @@ public class DuplicateDetectionTest extends ActiveMQTestBase {
@Test
public void testPersistXA1() throws Exception {
Assume.assumeTrue("This test would restart the server", persistCache);
ClientSession session = addClientSession(sf.createSession(true, false, false));
Xid xid = new XidImpl("xa1".getBytes(), 1, UUIDGenerator.getInstance().generateStringUUID().getBytes());
@ -1802,7 +1825,7 @@ public class DuplicateDetectionTest extends ActiveMQTestBase {
public void setUp() throws Exception {
super.setUp();
config = createDefaultInVMConfig().setIDCacheSize(cacheSize);
config = createDefaultInVMConfig().setIDCacheSize(cacheSize).setPersistIDCache(persistCache);
server = createServer(true, config);

62
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpSenderTest.java
View File

@ -22,6 +22,8 @@ import javax.jms.DeliveryMode;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;
import java.util.Arrays;
import java.util.Collection;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
@ -45,12 +47,30 @@ import org.apache.qpid.proton.engine.Delivery;
import org.apache.qpid.proton.engine.Sender;
import org.junit.Assert;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
/**
* Test broker behavior when creating AMQP senders
*/
@RunWith(Parameterized.class)
public class AmqpSenderTest extends AmqpClientTestSupport {
@Parameterized.Parameters(name = "persistentCache={0}")
public static Collection<Object[]> parameters() {
return Arrays.asList(new Object[][] {
{true}, {false}
});
}
@Override
protected void addConfiguration(ActiveMQServer server) {
server.getConfiguration().setPersistIDCache(persistCache);
}
@Parameterized.Parameter(0)
public boolean persistCache;
@Override
protected void addAdditionalAcceptors(ActiveMQServer server) throws Exception {
}
@ -252,6 +272,48 @@ public class AmqpSenderTest extends AmqpClientTestSupport {
connection.close();
}
@Test(timeout = 60000)
public void testDuplicateDetectionRollback() throws Exception {
ConnectionFactory factory = CFUtil.createConnectionFactory("AMQP", "tcp://localhost:5672");
try (Connection connection = factory.createConnection();
Session session = connection.createSession(true, Session.SESSION_TRANSACTED)) {
javax.jms.Queue producerQueue = session.createQueue(getQueueName());
MessageProducer producer = session.createProducer(producerQueue);
javax.jms.Message message = session.createTextMessage("test");
message.setStringProperty(Message.HDR_DUPLICATE_DETECTION_ID.toString(), "123");
producer.send(message);
session.rollback();
producer.send(message);
session.commit();
connection.start();
MessageConsumer consumer = session.createConsumer(producerQueue);
Assert.assertNotNull(consumer.receive(5000));
Assert.assertNull(consumer.receiveNoWait());
session.commit();
Queue serverQueue = server.locateQueue(getQueueName());
Wait.assertEquals(0, serverQueue::getMessageCount);
message = session.createTextMessage("test");
message.setStringProperty(Message.HDR_DUPLICATE_DETECTION_ID.toString(), "123");
producer.send(message);
boolean error = false;
try {
session.commit();
} catch (Exception e) {
error = true;
}
Assert.assertTrue(error);
}
}
@Test(timeout = 60000)
public void testSenderCreditReplenishment() throws Exception {
AtomicInteger counter = new AtomicInteger();

43
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/bridge/BridgeReconnectTest.java
View File

@ -17,6 +17,8 @@
package org.apache.activemq.artemis.tests.integration.cluster.bridge;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@ -25,6 +27,7 @@ import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.activemq.artemis.api.core.ActiveMQException;
import org.apache.activemq.artemis.api.core.ActiveMQExceptionType;
@ -43,6 +46,7 @@ import org.apache.activemq.artemis.api.core.client.ServerLocator;
import org.apache.activemq.artemis.api.core.management.QueueControl;
import org.apache.activemq.artemis.api.core.management.ResourceNames;
import org.apache.activemq.artemis.core.config.BridgeConfiguration;
import org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds;
import org.apache.activemq.artemis.core.remoting.impl.invm.InVMConnector;
import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.core.server.HandleStatus;
@ -60,11 +64,26 @@ import org.apache.activemq.artemis.core.transaction.Transaction;
import org.apache.activemq.artemis.spi.core.protocol.RemotingConnection;
import org.apache.activemq.artemis.tests.util.Wait;
import org.jboss.logging.Logger;
import org.junit.Assert;
import org.junit.Assume;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
@RunWith(Parameterized.class)
public class BridgeReconnectTest extends BridgeTestBase {
@Parameterized.Parameters(name = "persistentCache={0}")
public static Collection<Object[]> parameters() {
return Arrays.asList(new Object[][] {
{true}, {false}
});
}
@Parameterized.Parameter(0)
public boolean persistCache;
private static final Logger log = Logger.getLogger(BridgeReconnectTest.class);
private static final int NUM_MESSAGES = 100;
@ -412,8 +431,9 @@ public class BridgeReconnectTest extends BridgeTestBase {
}
// Fail bridge and reconnect same node, no backup specified
// It will keep a send blocking as if CPU was making it creep
@Test
public void testReconnectSameNodeAfterDelivery() throws Exception {
public void testReconnectSameNodeAfterDeliveryWithBlocking() throws Exception {
server0 = createActiveMQServer(0, isNetty(), server0Params);
TransportConfiguration server0tc = new TransportConfiguration(getConnector(), server0Params, "server0tc");
@ -530,6 +550,15 @@ public class BridgeReconnectTest extends BridgeTestBase {
closeServers();
assertNoMoreConnections();
HashMap<Integer, AtomicInteger> counts = countJournal(server1.getConfiguration());
if (persistCache) {
// There should be one record per message
Assert.assertEquals(numMessages, counts.get(new Integer(JournalRecordIds.DUPLICATE_ID)).intValue());
} else {
// no cache means there shouldn't be an id anywhere
Assert.assertNull(counts.get(new Integer(JournalRecordIds.DUPLICATE_ID)));
}
}
// We test that we can pause more than client failure check period (to prompt the pinger to failing)
@ -545,6 +574,7 @@ public class BridgeReconnectTest extends BridgeTestBase {
}
private void testShutdownServerCleanlyAndReconnectSameNode(final boolean sleep) throws Exception {
Assume.assumeTrue(persistCache);
server0 = createActiveMQServer(0, isNetty(), server0Params);
TransportConfiguration server0tc = new TransportConfiguration(getConnector(), server0Params, "server0tc");
@ -842,4 +872,15 @@ public class BridgeReconnectTest extends BridgeTestBase {
throw new IllegalStateException("Failed to get forwarding connection");
}
@Override
protected ActiveMQServer createActiveMQServer(final int id,
final Map<String, Object> params,
final boolean netty,
final NodeManager nodeManager) throws Exception {
ActiveMQServer server = super.createActiveMQServer(id, params, netty, nodeManager);
server.getConfiguration().setPersistIDCache(persistCache);
return server;
}
}

32
tests/smoke-tests/pom.xml
View File

@ -848,6 +848,38 @@
<configuration>${basedir}/target/classes/servers/brokerConnect/qdr</configuration>
</configuration>
</execution>
<execution>
<phase>test-compile</phase>
<id>create-bridge-transfer-A</id>
<goals>
<goal>create</goal>
</goals>
<configuration>
<role>amq</role>
<user>artemis</user>
<password>artemis</password>
<allowAnonymous>true</allowAnonymous>
<noWeb>true</noWeb>
<instance>${basedir}/target/bridgeTransfer/serverA</instance>
<configuration>${basedir}/target/classes/servers/bridgeTransfer/serverA</configuration>
</configuration>
</execution>
<execution>
<phase>test-compile</phase>
<id>create-bridge-transfer-B</id>
<goals>
<goal>create</goal>
</goals>
<configuration>
<role>amq</role>
<user>artemis</user>
<password>artemis</password>
<allowAnonymous>true</allowAnonymous>
<noWeb>true</noWeb>
<instance>${basedir}/target/bridgeTransfer/serverB</instance>
<configuration>${basedir}/target/classes/servers/bridgeTransfer/serverB</configuration>
</configuration>
</execution>
</executions>
<dependencies>
<dependency>

247
tests/smoke-tests/src/main/resources/servers/bridgeTransfer/serverA/broker.xml Normal file
View File

@ -0,0 +1,247 @@
<?xml version='1.0'?>
<!--
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.
-->
<configuration xmlns="urn:activemq"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:xi="http://www.w3.org/2001/XInclude"
xsi:schemaLocation="urn:activemq /schema/artemis-configuration.xsd">
<core xmlns="urn:activemq:core" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="urn:activemq:core ">
<name>0.0.0.0</name>
<persistence-enabled>true</persistence-enabled>
<!-- this could be ASYNCIO, MAPPED, NIO
ASYNCIO: Linux Libaio
MAPPED: mmap files
NIO: Plain Java Files
-->
<journal-type>NIO</journal-type>
<paging-directory>./data/paging</paging-directory>
<bindings-directory>./data/bindings</bindings-directory>
<journal-directory>./data/journal</journal-directory>
<large-messages-directory>./data/large-messages</large-messages-directory>
<!--
if you want to retain your journal uncomment this following configuration.
This will allow your system to keep 7 days of your data, up to 10G. Tweak it accordingly to your use case and capacity.
it is recommended to use a separate storage unit from the journal for performance considerations.
<journal-retention-directory period="7" unit="DAYS" storage-limit="10G">data/retention</journal-retention-directory>>
-->
<journal-datasync>true</journal-datasync>
<journal-min-files>2</journal-min-files>
<journal-pool-files>10</journal-pool-files>
<journal-device-block-size>4096</journal-device-block-size>
<journal-file-size>10M</journal-file-size>
<!--
You can verify the network health of a particular NIC by specifying the <network-check-NIC> element.
<network-check-NIC>theNicName</network-check-NIC>
-->
<!--
Use this to use an HTTP server to validate the network
<network-check-URL-list>http://www.apache.org</network-check-URL-list> -->
<!-- <network-check-period>10000</network-check-period> -->
<!-- <network-check-timeout>1000</network-check-timeout> -->
<!-- this is a comma separated list, no spaces, just DNS or IPs
it should accept IPV6
Warning: Make sure you understand your network topology as this is meant to validate if your network is valid.
Using IPs that could eventually disappear or be partially visible may defeat the purpose.
You can use a list of multiple IPs, and if any successful ping will make the server OK to continue running -->
<!-- <network-check-list>10.0.0.1</network-check-list> -->
<!-- use this to customize the ping used for ipv4 addresses -->
<!-- <network-check-ping-command>ping -c 1 -t %d %s</network-check-ping-command> -->
<!-- use this to customize the ping used for ipv6 addresses -->
<!-- <network-check-ping6-command>ping6 -c 1 %2$s</network-check-ping6-command> -->
<!-- how often we are looking for how many bytes are being used on the disk in ms -->
<disk-scan-period>5000</disk-scan-period>
<!-- once the disk hits this limit the system will block, or close the connection in certain protocols
that won't support flow control. -->
<max-disk-usage>90</max-disk-usage>
<!-- should the broker detect dead locks and other issues -->
<critical-analyzer>true</critical-analyzer>
<critical-analyzer-timeout>120000</critical-analyzer-timeout>
<critical-analyzer-check-period>60000</critical-analyzer-check-period>
<critical-analyzer-policy>HALT</critical-analyzer-policy>
<!-- the system will enter into page mode once you hit this limit.
This is an estimate in bytes of how much the messages are using in memory
The system will use half of the available memory (-Xmx) by default for the global-max-size.
You may specify a different value here if you need to customize it to your needs.
<global-max-size>100Mb</global-max-size>
-->
<acceptors>
<!-- useEpoll means: it will use Netty epoll if you are on a system (Linux) that supports it -->
<!-- amqpCredits: The number of credits sent to AMQP producers -->
<!-- amqpLowCredits: The server will send the # credits specified at amqpCredits at this low mark -->
<!-- amqpDuplicateDetection: If you are not using duplicate detection, set this to false
as duplicate detection requires applicationProperties to be parsed on the server. -->
<!-- amqpMinLargeMessageSize: Determines how many bytes are considered large, so we start using files to hold their data.
default: 102400, -1 would mean to disable large mesasge control -->
<!-- Note: If an acceptor needs to be compatible with HornetQ and/or Artemis 1.x clients add
"anycastPrefix=jms.queue.;multicastPrefix=jms.topic." to the acceptor url.
See https://issues.apache.org/jira/browse/ARTEMIS-1644 for more information. -->
<!-- Acceptor for every supported protocol -->
<acceptor name="artemis">tcp://0.0.0.0:61616?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;amqpMinLargeMessageSize=102400;protocols=CORE,AMQP,STOMP,HORNETQ,MQTT,OPENWIRE;useEpoll=true;amqpCredits=1000;amqpLowCredits=300;amqpDuplicateDetection=true</acceptor>
</acceptors>
<connectors>
<connector name="other-side">tcp://localhost:61617</connector>
</connectors>
<bridges>
<bridge name="otherSide">
<queue-name>bridgeQueue</queue-name>
<retry-interval>100</retry-interval>
<reconnect-attempts>-1</reconnect-attempts>
<static-connectors>
<connector-ref>other-side</connector-ref>
</static-connectors>
</bridge>
</bridges>
<security-settings>
<security-setting match="#">
<permission type="createNonDurableQueue" roles="amq"/>
<permission type="deleteNonDurableQueue" roles="amq"/>
<permission type="createDurableQueue" roles="amq"/>
<permission type="deleteDurableQueue" roles="amq"/>
<permission type="createAddress" roles="amq"/>
<permission type="deleteAddress" roles="amq"/>
<permission type="consume" roles="amq"/>
<permission type="browse" roles="amq"/>
<permission type="send" roles="amq"/>
<!-- we need this otherwise ./artemis data imp wouldn't work -->
<permission type="manage" roles="amq"/>
</security-setting>
</security-settings>
<address-settings>
<!-- if you define auto-create on certain queues, management has to be auto-create -->
<address-setting match="activemq.management#">
<dead-letter-address>DLQ</dead-letter-address>
<expiry-address>ExpiryQueue</expiry-address>
<redelivery-delay>0</redelivery-delay>
<!-- with -1 only the global-max-size is in use for limiting -->
<max-size-bytes>-1</max-size-bytes>
<message-counter-history-day-limit>10</message-counter-history-day-limit>
<address-full-policy>PAGE</address-full-policy>
<auto-create-queues>true</auto-create-queues>
<auto-create-addresses>true</auto-create-addresses>
<auto-create-jms-queues>true</auto-create-jms-queues>
<auto-create-jms-topics>true</auto-create-jms-topics>
</address-setting>
<!--default for catch all-->
<address-setting match="#">
<dead-letter-address>DLQ</dead-letter-address>
<expiry-address>ExpiryQueue</expiry-address>
<redelivery-delay>0</redelivery-delay>
<!-- with -1 only the global-max-size is in use for limiting -->
<max-size-bytes>-1</max-size-bytes>
<message-counter-history-day-limit>10</message-counter-history-day-limit>
<address-full-policy>PAGE</address-full-policy>
<auto-create-queues>true</auto-create-queues>
<auto-create-addresses>true</auto-create-addresses>
<auto-create-jms-queues>true</auto-create-jms-queues>
<auto-create-jms-topics>true</auto-create-jms-topics>
</address-setting>
</address-settings>
<addresses>
<address name="DLQ">
<anycast>
<queue name="DLQ" />
</anycast>
</address>
<address name="ExpiryQueue">
<anycast>
<queue name="ExpiryQueue" />
</anycast>
</address>
<address name="bridgeQueue">
<anycast>
<queue name="bridgeQueue"/>
</anycast>
</address>
</addresses>
<!-- Uncomment the following if you want to use the Standard LoggingActiveMQServerPlugin pluging to log in events
<broker-plugins>
<broker-plugin class-name="org.apache.activemq.artemis.core.server.plugin.impl.LoggingActiveMQServerPlugin">
<property key="LOG_ALL_EVENTS" value="true"/>
<property key="LOG_CONNECTION_EVENTS" value="true"/>
<property key="LOG_SESSION_EVENTS" value="true"/>
<property key="LOG_CONSUMER_EVENTS" value="true"/>
<property key="LOG_DELIVERING_EVENTS" value="true"/>
<property key="LOG_SENDING_EVENTS" value="true"/>
<property key="LOG_INTERNAL_EVENTS" value="true"/>
</broker-plugin>
</broker-plugins>
-->
</core>
</configuration>

229
tests/smoke-tests/src/main/resources/servers/bridgeTransfer/serverB/broker.xml Normal file
View File

@ -0,0 +1,229 @@
<?xml version='1.0'?>
<!--
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.
-->
<configuration xmlns="urn:activemq"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:xi="http://www.w3.org/2001/XInclude"
xsi:schemaLocation="urn:activemq /schema/artemis-configuration.xsd">
<core xmlns="urn:activemq:core" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="urn:activemq:core ">
<name>0.0.0.0</name>
<persistence-enabled>true</persistence-enabled>
<!-- this could be ASYNCIO, MAPPED, NIO
ASYNCIO: Linux Libaio
MAPPED: mmap files
NIO: Plain Java Files
-->
<journal-type>NIO</journal-type>
<paging-directory>./data/paging</paging-directory>
<bindings-directory>./data/bindings</bindings-directory>
<journal-directory>./data/journal</journal-directory>
<large-messages-directory>./data/large-messages</large-messages-directory>
<!--
if you want to retain your journal uncomment this following configuration.
This will allow your system to keep 7 days of your data, up to 10G. Tweak it accordingly to your use case and capacity.
it is recommended to use a separate storage unit from the journal for performance considerations.
<journal-retention-directory period="7" unit="DAYS" storage-limit="10G">data/retention</journal-retention-directory>>
-->
<journal-datasync>true</journal-datasync>
<journal-min-files>2</journal-min-files>
<journal-pool-files>10</journal-pool-files>
<journal-device-block-size>4096</journal-device-block-size>
<journal-file-size>10M</journal-file-size>
<!--
You can verify the network health of a particular NIC by specifying the <network-check-NIC> element.
<network-check-NIC>theNicName</network-check-NIC>
-->
<!--
Use this to use an HTTP server to validate the network
<network-check-URL-list>http://www.apache.org</network-check-URL-list> -->
<!-- <network-check-period>10000</network-check-period> -->
<!-- <network-check-timeout>1000</network-check-timeout> -->
<!-- this is a comma separated list, no spaces, just DNS or IPs
it should accept IPV6
Warning: Make sure you understand your network topology as this is meant to validate if your network is valid.
Using IPs that could eventually disappear or be partially visible may defeat the purpose.
You can use a list of multiple IPs, and if any successful ping will make the server OK to continue running -->
<!-- <network-check-list>10.0.0.1</network-check-list> -->
<!-- use this to customize the ping used for ipv4 addresses -->
<!-- <network-check-ping-command>ping -c 1 -t %d %s</network-check-ping-command> -->
<!-- use this to customize the ping used for ipv6 addresses -->
<!-- <network-check-ping6-command>ping6 -c 1 %2$s</network-check-ping6-command> -->
<!-- how often we are looking for how many bytes are being used on the disk in ms -->
<disk-scan-period>5000</disk-scan-period>
<!-- once the disk hits this limit the system will block, or close the connection in certain protocols
that won't support flow control. -->
<max-disk-usage>90</max-disk-usage>
<!-- should the broker detect dead locks and other issues -->
<critical-analyzer>true</critical-analyzer>
<critical-analyzer-timeout>120000</critical-analyzer-timeout>
<critical-analyzer-check-period>60000</critical-analyzer-check-period>
<critical-analyzer-policy>HALT</critical-analyzer-policy>
<!-- the system will enter into page mode once you hit this limit.
This is an estimate in bytes of how much the messages are using in memory
The system will use half of the available memory (-Xmx) by default for the global-max-size.
You may specify a different value here if you need to customize it to your needs.
<global-max-size>100Mb</global-max-size>
-->
<acceptors>
<!-- useEpoll means: it will use Netty epoll if you are on a system (Linux) that supports it -->
<!-- amqpCredits: The number of credits sent to AMQP producers -->
<!-- amqpLowCredits: The server will send the # credits specified at amqpCredits at this low mark -->
<!-- amqpDuplicateDetection: If you are not using duplicate detection, set this to false
as duplicate detection requires applicationProperties to be parsed on the server. -->
<!-- amqpMinLargeMessageSize: Determines how many bytes are considered large, so we start using files to hold their data.
default: 102400, -1 would mean to disable large mesasge control -->
<!-- Note: If an acceptor needs to be compatible with HornetQ and/or Artemis 1.x clients add
"anycastPrefix=jms.queue.;multicastPrefix=jms.topic." to the acceptor url.
See https://issues.apache.org/jira/browse/ARTEMIS-1644 for more information. -->
<!-- Acceptor for every supported protocol -->
<acceptor name="artemis">tcp://0.0.0.0:61617?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;amqpMinLargeMessageSize=102400;protocols=CORE,AMQP,STOMP,HORNETQ,MQTT,OPENWIRE;useEpoll=true;amqpCredits=1000;amqpLowCredits=300;amqpDuplicateDetection=true</acceptor>
</acceptors>
<security-settings>
<security-setting match="#">
<permission type="createNonDurableQueue" roles="amq"/>
<permission type="deleteNonDurableQueue" roles="amq"/>
<permission type="createDurableQueue" roles="amq"/>
<permission type="deleteDurableQueue" roles="amq"/>
<permission type="createAddress" roles="amq"/>
<permission type="deleteAddress" roles="amq"/>
<permission type="consume" roles="amq"/>
<permission type="browse" roles="amq"/>
<permission type="send" roles="amq"/>
<!-- we need this otherwise ./artemis data imp wouldn't work -->
<permission type="manage" roles="amq"/>
</security-setting>
</security-settings>
<address-settings>
<!-- if you define auto-create on certain queues, management has to be auto-create -->
<address-setting match="activemq.management#">
<dead-letter-address>DLQ</dead-letter-address>
<expiry-address>ExpiryQueue</expiry-address>
<redelivery-delay>0</redelivery-delay>
<!-- with -1 only the global-max-size is in use for limiting -->
<max-size-bytes>-1</max-size-bytes>
<message-counter-history-day-limit>10</message-counter-history-day-limit>
<address-full-policy>PAGE</address-full-policy>
<auto-create-queues>true</auto-create-queues>
<auto-create-addresses>true</auto-create-addresses>
<auto-create-jms-queues>true</auto-create-jms-queues>
<auto-create-jms-topics>true</auto-create-jms-topics>
</address-setting>
<!--default for catch all-->
<address-setting match="#">
<dead-letter-address>DLQ</dead-letter-address>
<expiry-address>ExpiryQueue</expiry-address>
<redelivery-delay>0</redelivery-delay>
<!-- with -1 only the global-max-size is in use for limiting -->
<max-size-bytes>-1</max-size-bytes>
<message-counter-history-day-limit>10</message-counter-history-day-limit>
<address-full-policy>PAGE</address-full-policy>
<auto-create-queues>true</auto-create-queues>
<auto-create-addresses>true</auto-create-addresses>
<auto-create-jms-queues>true</auto-create-jms-queues>
<auto-create-jms-topics>true</auto-create-jms-topics>
</address-setting>
</address-settings>
<addresses>
<address name="DLQ">
<anycast>
<queue name="DLQ" />
</anycast>
</address>
<address name="ExpiryQueue">
<anycast>
<queue name="ExpiryQueue" />
</anycast>
</address>
<address name="bridgeQueue">
<anycast>
<queue name="bridgeQueue"/>
</anycast>
</address>
</addresses>
<!-- Uncomment the following if you want to use the Standard LoggingActiveMQServerPlugin pluging to log in events
<broker-plugins>
<broker-plugin class-name="org.apache.activemq.artemis.core.server.plugin.impl.LoggingActiveMQServerPlugin">
<property key="LOG_ALL_EVENTS" value="true"/>
<property key="LOG_CONNECTION_EVENTS" value="true"/>
<property key="LOG_SESSION_EVENTS" value="true"/>
<property key="LOG_CONSUMER_EVENTS" value="true"/>
<property key="LOG_DELIVERING_EVENTS" value="true"/>
<property key="LOG_SENDING_EVENTS" value="true"/>
<property key="LOG_INTERNAL_EVENTS" value="true"/>
</broker-plugin>
</broker-plugins>
-->
</core>
</configuration>

190
tests/smoke-tests/src/test/java/org/apache/activemq/artemis/tests/smoke/bridgeTransfer/BridgeTransferingTest.java Normal file
View File

@ -0,0 +1,190 @@
/*
* 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.smoke.bridgeTransfer;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;
import java.util.Arrays;
import java.util.Collection;
import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;
import org.apache.activemq.artemis.tests.smoke.common.SmokeTestBase;
import org.apache.activemq.artemis.utils.Wait;
import org.apache.qpid.jms.JmsConnectionFactory;
import org.jboss.logging.Logger;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
@RunWith(Parameterized.class)
public class BridgeTransferingTest extends SmokeTestBase {
public static final String SERVER_NAME_0 = "bridgeTransfer/serverA";
public static final String SERVER_NAME_1 = "bridgeTransfer/serverB";
private static final Logger logger = Logger.getLogger(BridgeTransferingTest.class);
private static final String JMX_SERVER_HOSTNAME = "localhost";
private static final int JMX_SERVER_PORT = 11099;
final String theprotocol;
// As the produces sends messages, a client will be killed every X messages. This is it!
final int killServerInterval;
final int numberOfMessages;
final int commitInterval;
final int messageSize;
final boolean killBothServers;
final int minlargeMessageSize;
Process serverProcess;
Process serverProcess2;
public BridgeTransferingTest(String protocol, int commitInterval, int killServerInterval, int numberOfMessages, int messageSize, int minlargeMessageSize, boolean killBothServers) {
this.theprotocol = protocol;
this.killServerInterval = killServerInterval;
this.messageSize = messageSize;
this.commitInterval = commitInterval;
this.numberOfMessages = numberOfMessages;
this.killBothServers = killBothServers;
this.minlargeMessageSize = minlargeMessageSize;
}
@Parameterized.Parameters(name = "protocol={0}, commitInterval={1}, killInterval={2}, numberOfMessages={3}, messageSize={4}, minLargeMessageSize={5}, KillBothServers={6}")
public static Collection<Object[]> parameters() {
return Arrays.asList(new Object[][]{{"CORE", 200, 1000, 10000, 15_000, 5000, true}, {"CORE", 200, 1000, 10000, 15_000, 5000, false}});
}
public static ConnectionFactory createConnectionFactory(String protocol, String uri) {
if (protocol.toUpperCase().equals("OPENWIRE")) {
return new org.apache.activemq.ActiveMQConnectionFactory(uri);
} else if (protocol.toUpperCase().equals("AMQP")) {
if (uri.startsWith("tcp://")) {
// replacing tcp:// by amqp://
uri = "amqp" + uri.substring(3);
}
return new JmsConnectionFactory(uri);
} else if (protocol.toUpperCase().equals("CORE") || protocol.toUpperCase().equals("ARTEMIS")) {
return new org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory(uri);
} else {
throw new IllegalStateException("Unkown:" + protocol);
}
}
@Before
public void before() throws Exception {
cleanupData(SERVER_NAME_0);
cleanupData(SERVER_NAME_1);
disableCheckThread();
serverProcess = startServer(SERVER_NAME_0, 0, 30000);
serverProcess2 = startServer(SERVER_NAME_1, 1, 30000);
}
@After
public void stopServers() throws Exception {
serverProcess2.destroyForcibly();
serverProcess.destroyForcibly();
}
@Test
public void testTransfer() throws Exception {
ConnectionFactory cf = createConnectionFactory(theprotocol, "tcp://localhost:61616");
((ActiveMQConnectionFactory) cf).setMinLargeMessageSize(minlargeMessageSize);
String body;
{
StringBuffer buffer = new StringBuffer();
while (buffer.length() < messageSize) {
buffer.append(" ");
}
body = buffer.toString();
}
{
Connection connection = cf.createConnection();
Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
Queue queue = session.createQueue("bridgeQueue");
MessageProducer producer = session.createProducer(queue);
int txElement = 0;
int killElement = 0;
for (int i = 0; i < numberOfMessages; i++) {
producer.send(session.createTextMessage(body + " " + i));
if (++txElement == commitInterval) {
logger.debug("Sent " + (i + 1) + " messages");
txElement = 0;
session.commit();
}
if (++killElement == killServerInterval) {
logger.debug("Killing server at " + (i + 1));
killElement = 0;
if (killBothServers) {
serverProcess.destroyForcibly();
Wait.assertFalse(serverProcess::isAlive);
}
serverProcess2.destroyForcibly();
Wait.assertFalse(serverProcess2::isAlive);
serverProcess2 = startServer(SERVER_NAME_1, 1, 30000);
if (killBothServers) {
serverProcess = startServer(SERVER_NAME_0, 0, 30000);
}
if (killBothServers) {
connection.close();
connection = cf.createConnection();
session = connection.createSession(true, Session.SESSION_TRANSACTED);
queue = session.createQueue("bridgeQueue");
producer = session.createProducer(queue);
}
}
}
if (txElement > 0) {
session.commit();
}
}
ConnectionFactory cf2 = createConnectionFactory(theprotocol, "tcp://localhost:61617");
try (Connection connection = cf2.createConnection()) {
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Queue queue = session.createQueue("bridgeQueue");
MessageConsumer consumer = session.createConsumer(queue);
connection.start();
for (int i = 0; i < numberOfMessages; i++) {
if (i % 100 == 0) {
logger.debug("consuming " + i);
}
TextMessage message = (TextMessage) consumer.receive(5000);
Assert.assertNotNull(message);
Assert.assertEquals(body + " " + i, message.getText());
}
Assert.assertNull(consumer.receiveNoWait());
}
}
}