You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by cs...@apache.org on 2022/02/09 11:36:15 UTC
[activemq] branch activemq-5.16.x updated: [AMQ-8397] Add unit tests for AMQ-8397
This is an automated email from the ASF dual-hosted git repository.
cshannon pushed a commit to branch activemq-5.16.x
in repository https://gitbox.apache.org/repos/asf/activemq.git
The following commit(s) were added to refs/heads/activemq-5.16.x by this push:
new acc3c69 [AMQ-8397] Add unit tests for AMQ-8397
acc3c69 is described below
commit acc3c694ab5b88d4437017e94e7b4f60d5d0efc4
Author: Matt Pavlovich <ma...@hyte.io>
AuthorDate: Tue Feb 8 19:09:25 2022 -0600
[AMQ-8397] Add unit tests for AMQ-8397
(cherry picked from commit 9d8ee0d3b73d214aedcb0e806daf32c3c2ff69db)
---
.../java/org/apache/activemq/PolicyEntryTest.java | 33 +++++
...Test-policy-sendDuplicateFromStoreToDLQ-mod.xml | 36 +++++
...ntryTest-policy-sendDuplicateFromStoreToDLQ.xml | 36 +++++
.../policy/SendDuplicateFromStoreToDLQTest.java | 156 +++++++++++++++++++++
4 files changed, 261 insertions(+)
diff --git a/activemq-runtime-config/src/test/java/org/apache/activemq/PolicyEntryTest.java b/activemq-runtime-config/src/test/java/org/apache/activemq/PolicyEntryTest.java
index 120a4c6..b486c30 100644
--- a/activemq-runtime-config/src/test/java/org/apache/activemq/PolicyEntryTest.java
+++ b/activemq-runtime-config/src/test/java/org/apache/activemq/PolicyEntryTest.java
@@ -24,6 +24,7 @@ import org.junit.Test;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
public class PolicyEntryTest extends RuntimeConfigTestSupport {
@@ -45,6 +46,18 @@ public class PolicyEntryTest extends RuntimeConfigTestSupport {
}
@Test
+ public void testModSendDuplicateFromStoreToDLQ() throws Exception {
+ final String brokerConfig = configurationSeed + "-policy-ml-broker";
+ applyNewConfig(brokerConfig, configurationSeed + "-policy-sendDuplicateFromStoreToDLQ");
+ startBroker(brokerConfig);
+ assertTrue("broker alive", brokerService.isStarted());
+
+ verifyBooleanField("AMQ.8397", "sendDuplicateFromStoreToDLQ", true);
+ applyNewConfig(brokerConfig, configurationSeed + "-policy-sendDuplicateFromStoreToDLQ-mod", SLEEP);
+ verifyBooleanField("AMQ.8397", "sendDuplicateFromStoreToDLQ", false);
+ }
+
+ @Test
public void testAddNdMod() throws Exception {
final String brokerConfig = configurationSeed + "-policy-ml-broker";
applyNewConfig(brokerConfig, configurationSeed + "-policy-ml");
@@ -100,6 +113,26 @@ public class PolicyEntryTest extends RuntimeConfigTestSupport {
verifyQueueLimit("queue.child.test2", 4194304);
}
+ private void verifyBooleanField(String dest, String fieldName, boolean value) throws Exception {
+ ActiveMQConnection connection = new ActiveMQConnectionFactory("vm://localhost").createActiveMQConnection();
+ try {
+ connection.start();
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ session.createConsumer(session.createQueue(dest));
+
+ switch(fieldName) {
+ case "sendDuplicateFromStoreToDLQ":
+ assertEquals(value, brokerService.getRegionBroker().getDestinationMap().get(new ActiveMQQueue(dest)).isSendDuplicateFromStoreToDLQ());
+ break;
+ default:
+ fail("Unsupported field specified: " + fieldName);
+ }
+
+ } finally {
+ connection.close();
+ }
+ }
+
private void verifyQueueLimit(String dest, int memoryLimit) throws Exception {
ActiveMQConnection connection = new ActiveMQConnectionFactory("vm://localhost").createActiveMQConnection();
try {
diff --git a/activemq-runtime-config/src/test/resources/org/apache/activemq/policyEntryTest-policy-sendDuplicateFromStoreToDLQ-mod.xml b/activemq-runtime-config/src/test/resources/org/apache/activemq/policyEntryTest-policy-sendDuplicateFromStoreToDLQ-mod.xml
new file mode 100644
index 0000000..0984792
--- /dev/null
+++ b/activemq-runtime-config/src/test/resources/org/apache/activemq/policyEntryTest-policy-sendDuplicateFromStoreToDLQ-mod.xml
@@ -0,0 +1,36 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+<beans
+ xmlns="http://www.springframework.org/schema/beans"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd
+ http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd">
+
+ <broker xmlns="http://activemq.apache.org/schema/core" start="false" persistent="false">
+ <plugins>
+ <runtimeConfigurationPlugin checkPeriod="1000" />
+ </plugins>
+ <destinationPolicy>
+ <policyMap>
+ <policyEntries>
+ <policyEntry queue="AMQ.8397" sendDuplicateFromStoreToDLQ="false"/>
+ </policyEntries>
+ </policyMap>
+ </destinationPolicy>
+ </broker>
+</beans>
diff --git a/activemq-runtime-config/src/test/resources/org/apache/activemq/policyEntryTest-policy-sendDuplicateFromStoreToDLQ.xml b/activemq-runtime-config/src/test/resources/org/apache/activemq/policyEntryTest-policy-sendDuplicateFromStoreToDLQ.xml
new file mode 100644
index 0000000..cbe6ccb
--- /dev/null
+++ b/activemq-runtime-config/src/test/resources/org/apache/activemq/policyEntryTest-policy-sendDuplicateFromStoreToDLQ.xml
@@ -0,0 +1,36 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+<beans
+ xmlns="http://www.springframework.org/schema/beans"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd
+ http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd">
+
+ <broker xmlns="http://activemq.apache.org/schema/core" start="false" persistent="false">
+ <plugins>
+ <runtimeConfigurationPlugin checkPeriod="1000" />
+ </plugins>
+ <destinationPolicy>
+ <policyMap>
+ <policyEntries>
+ <policyEntry queue="AMQ.8397"/>
+ </policyEntries>
+ </policyMap>
+ </destinationPolicy>
+ </broker>
+</beans>
diff --git a/activemq-unit-tests/src/test/java/org/apache/activemq/broker/policy/SendDuplicateFromStoreToDLQTest.java b/activemq-unit-tests/src/test/java/org/apache/activemq/broker/policy/SendDuplicateFromStoreToDLQTest.java
new file mode 100644
index 0000000..adfc88d
--- /dev/null
+++ b/activemq-unit-tests/src/test/java/org/apache/activemq/broker/policy/SendDuplicateFromStoreToDLQTest.java
@@ -0,0 +1,156 @@
+/**
+ * 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.broker.policy;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
+import java.io.File;
+import java.util.List;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Queue;
+import javax.jms.Session;
+
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.region.Subscription;
+import org.apache.activemq.broker.region.policy.IndividualDeadLetterStrategy;
+import org.apache.activemq.broker.region.policy.PolicyEntry;
+import org.apache.activemq.broker.region.policy.PolicyMap;
+import org.apache.activemq.command.ActiveMQQueue;
+import org.apache.activemq.store.kahadb.KahaDBPersistenceAdapter;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+
+/**
+ * This unit test is to test that setting the property "sendDuplicateFromStoreToDLQ" on
+ * PolicyEntry works correctly.
+ *
+ */
+public class SendDuplicateFromStoreToDLQTest {
+ BrokerService broker;
+ ConnectionFactory factory;
+ Connection connection;
+ Session session;
+ MessageProducer producer;
+
+ @Before
+ public void setUp() throws Exception {
+ broker = new BrokerService();
+
+ File testDataDir = new File("target/activemq-data/AMQ-8397");
+ broker.setDataDirectoryFile(testDataDir);
+ broker.setUseJmx(true);
+ broker.setDeleteAllMessagesOnStartup(true);
+ broker.getSystemUsage().getMemoryUsage().setLimit(1024l * 1024 * 64);
+ KahaDBPersistenceAdapter persistenceAdapter = new KahaDBPersistenceAdapter();
+ persistenceAdapter.setDirectory(new File(testDataDir, "kahadb"));
+ broker.setPersistenceAdapter(persistenceAdapter);
+ broker.addConnector("tcp://localhost:0");
+ broker.start();
+ factory = new ActiveMQConnectionFactory(broker.getTransportConnectors()
+ .get(0).getConnectUri().toString());
+ connection = factory.createConnection();
+ connection.start();
+ session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ session.close();
+ connection.stop();
+ connection.close();
+ broker.deleteAllMessages();
+ broker.stop();
+ }
+
+ @Test
+ public void testSendDuplicateFromStoreToDLQTrue() throws Exception {
+ applySendDuplicateFromStoreToDLQPolicy(true);
+ doProcessSendDuplicateFromStoreToDLQ(true);
+ }
+
+ @Test
+ public void testSendDuplicateFromStoreToDLQFalse() throws Exception {
+ applySendDuplicateFromStoreToDLQPolicy(false);
+ doProcessSendDuplicateFromStoreToDLQ(false);
+ }
+
+ protected void doProcessSendDuplicateFromStoreToDLQ(boolean enabled) throws Exception {
+ createQueue("AMQ.8397");
+ org.apache.activemq.broker.region.Queue queue = (org.apache.activemq.broker.region.Queue)broker.getDestination(new ActiveMQQueue("AMQ.8397"));
+ assertEquals(Long.valueOf(0l), Long.valueOf(queue.getDestinationStatistics().getMessages().getCount()));
+ assertEquals(Long.valueOf(0l), Long.valueOf(queue.getDestinationStatistics().getDuplicateFromStore().getCount()));
+
+ MessageConsumer messageConsumer = session.createConsumer(session.createQueue("AMQ.8397"));
+ producer.send(session.createTextMessage("Hello world!"));
+
+ int loopCount=0;
+ int maxLoops=50;
+ boolean found=false;
+ Message recvMessage = null;
+ do {
+ recvMessage = messageConsumer.receive(200l);
+ if(recvMessage != null) {
+ found = true;
+ }
+ loopCount++;
+ } while(!found && loopCount < maxLoops);
+
+ assertNotNull(recvMessage);
+ List<Subscription> queueSubscriptions = queue.getConsumers();
+ assertNotNull(queueSubscriptions);
+ assertEquals(Integer.valueOf(1), Integer.valueOf(queueSubscriptions.size()));
+ queue.duplicateFromStore((org.apache.activemq.command.Message) recvMessage, queueSubscriptions.get(0));
+
+ org.apache.activemq.broker.region.Queue dlq = (org.apache.activemq.broker.region.Queue)broker.getDestination(new ActiveMQQueue("ActiveMQ.DLQ.Queue.AMQ.8397"));
+
+ if(enabled) {
+ assertEquals(Long.valueOf(0l), Long.valueOf(queue.getDestinationStatistics().getMessages().getCount()));
+ assertEquals(Long.valueOf(1l), Long.valueOf(queue.getDestinationStatistics().getDuplicateFromStore().getCount()));
+ assertEquals(Long.valueOf(1l), Long.valueOf(dlq.getDestinationStatistics().getMessages().getCount()));
+ assertEquals(Long.valueOf(0l), Long.valueOf(dlq.getDestinationStatistics().getDuplicateFromStore().getCount()));
+ } else {
+ assertEquals(Long.valueOf(0l), Long.valueOf(queue.getDestinationStatistics().getMessages().getCount()));
+ assertEquals(Long.valueOf(1l), Long.valueOf(queue.getDestinationStatistics().getDuplicateFromStore().getCount()));
+ assertEquals(Long.valueOf(0l), Long.valueOf(dlq.getDestinationStatistics().getMessages().getCount()));
+ assertEquals(Long.valueOf(0l), Long.valueOf(dlq.getDestinationStatistics().getDuplicateFromStore().getCount()));
+ }
+ }
+
+ private PolicyMap applySendDuplicateFromStoreToDLQPolicy(boolean enabled) {
+ PolicyMap policyMap = new PolicyMap();
+ PolicyEntry defaultEntry = new PolicyEntry();
+ defaultEntry.setSendDuplicateFromStoreToDLQ(enabled);
+ defaultEntry.setDeadLetterStrategy(new IndividualDeadLetterStrategy());
+ policyMap.setDefaultEntry(defaultEntry);
+ broker.setDestinationPolicy(policyMap);
+ return policyMap;
+ }
+
+ private void createQueue(String queueName) throws Exception {
+ Queue queue = session.createQueue(queueName);
+ producer = session.createProducer(queue);
+ }
+}