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

ARTEMIS-3529 Expire should not reject messages from duplicateID

This commit is contained in:
Clebert Suconic 2021-10-13 16:21:13 -04:00 committed by clebertsuconic
parent df2c4e80e4
commit 1d84800d5a
2 changed files with 214 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java
View File

@ -3576,7 +3576,7 @@ public class QueueImpl extends CriticalComponentImpl implements Queue {
ActiveMQServerLogger.LOGGER.errorExpiringReferencesNoBindings(expiryAddress);
acknowledge(tx, ref, AckReason.EXPIRED, null);
} else {
move(expiryAddress, tx, ref, true, true);
move(expiryAddress, tx, ref, true, false);
}
} else {
if (!printErrorExpiring) {

213
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/client/MoveMessageDuplicateIDTest.java Normal file
View File

@ -0,0 +1,213 @@
/*
* 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.jms.client;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;
import java.util.Arrays;
import java.util.Collection;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.api.core.management.AddressControl;
import org.apache.activemq.artemis.core.config.Configuration;
import org.apache.activemq.artemis.core.server.Queue;
import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
import org.apache.activemq.artemis.logs.AssertionLoggerHandler;
import org.apache.activemq.artemis.tests.integration.management.ManagementControlHelper;
import org.apache.activemq.artemis.tests.util.CFUtil;
import org.apache.activemq.artemis.tests.util.JMSTestBase;
import org.apache.activemq.artemis.tests.util.Wait;
import org.junit.Assert;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
@RunWith(Parameterized.class)
public class MoveMessageDuplicateIDTest extends JMSTestBase {
@Parameterized.Parameter(0)
public String protocol = "AMQP";
@Parameterized.Parameters(name = "protocol={0}")
public static Collection<Object[]> parameters() {
return Arrays.asList(new Object[][] {
{"AMQP"},
{"CORE"},
{"OPENWIRE"}
});
}
@Override
protected Configuration createDefaultConfig(boolean netty) throws Exception {
return super.createDefaultConfig(netty).setMessageExpiryScanPeriod(50);
}
@Test
public void testTwoQueuesSingleDLQ() throws Exception {
AssertionLoggerHandler.startCapture();
try {
server.getAddressSettingsRepository().clear();
server.getAddressSettingsRepository().addMatch("#", new AddressSettings().setDeadLetterAddress(SimpleString.toSimpleString("JUNKYARD")).setExpiryAddress(SimpleString.toSimpleString("JUNKYARD")).setMaxDeliveryAttempts(1));
createQueue("JUNKYARD");
Queue junkQueue = server.locateQueue("JUNKYARD");
Assert.assertNotNull(junkQueue);
javax.jms.Queue queue1 = createQueue("q1");
javax.jms.Queue queue2 = createQueue("q2");
ConnectionFactory factory = CFUtil.createConnectionFactory(protocol, "tcp://localhost:61616");
Connection conn = factory.createConnection();
Session sess = conn.createSession(true, Session.SESSION_TRANSACTED);
MessageProducer prod1 = sess.createProducer(queue1);
MessageProducer prod2 = sess.createProducer(queue2);
for (int i = 0; i < 100; i++) {
TextMessage txt = sess.createTextMessage("txt");
txt.setStringProperty("_AMQ_DUPL_ID", "" + i);
prod1.send(txt);
prod2.send(txt);
}
sess.commit();
conn.start();
MessageConsumer consumer = sess.createConsumer(queue1);
for (int i = 0; i < 100; i++) {
TextMessage textMessage = (TextMessage) consumer.receive(5000);
Assert.assertNotNull(textMessage);
}
sess.rollback();
Assert.assertNull(consumer.receiveNoWait());
consumer.close();
Wait.assertEquals(100L, junkQueue::getMessageCount, 2000, 10);
consumer = sess.createConsumer(queue2);
for (int i = 0; i < 100; i++) {
TextMessage textMessage = (TextMessage) consumer.receive(5000);
Assert.assertNotNull(textMessage);
}
sess.rollback();
Assert.assertNull(consumer.receiveNoWait());
consumer.close();
conn.close();
Wait.assertEquals(200L, junkQueue::getMessageCount, 2000, 10);
} finally {
AssertionLoggerHandler.stopCapture();
}
}
@Test
public void testMultiplSubscriptionSingleExpire() throws Exception {
server.getAddressSettingsRepository().clear();
server.getAddressSettingsRepository().addMatch("#", new AddressSettings().setDeadLetterAddress(SimpleString.toSimpleString("DLQ")).setExpiryAddress(SimpleString.toSimpleString("DLQ")));
createQueue("DLQ");
Queue dlqServerQueue = server.locateQueue("DLQ");
org.wildfly.common.Assert.assertNotNull(dlqServerQueue);
Topic topic = createTopic("test-topic");
AddressControl control = ManagementControlHelper.createAddressControl(new SimpleString(topic.getTopicName()), mbeanServer);
ConnectionFactory cf = CFUtil.createConnectionFactory(protocol, "tcp://localhost:61616");
Connection conn = cf.createConnection();
conn.setClientID("client1");
Session sess2 = conn.createSession(true, Session.SESSION_TRANSACTED);
sess2.createDurableSubscriber(topic, "client-sub1");
sess2.createDurableSubscriber(topic, "client-sub2");
conn.close();
conn = cf.createConnection();
Session sess = conn.createSession(true, Session.SESSION_TRANSACTED);
MessageProducer prod = sess.createProducer(topic);
prod.setTimeToLive(1);
for (int i = 0; i < 100; i++) {
TextMessage txt = sess.createTextMessage("txt");
txt.setStringProperty("_AMQ_DUPL_ID", "" + i);
prod.send(txt);
}
sess.commit();
conn.close();
Wait.assertEquals(0L, control::getMessageCount, 2000, 10);
Wait.assertEquals(200L, dlqServerQueue::getMessageCount, 2000, 10);
}
@Test
public void testTwoQueuesSingleExpire() throws Exception {
server.getAddressSettingsRepository().clear();
server.getAddressSettingsRepository().addMatch("#", new AddressSettings().setDeadLetterAddress(SimpleString.toSimpleString("JUNKYARD")).setExpiryAddress(SimpleString.toSimpleString("JUNKYARD")));
createQueue("JUNKYARD");
Queue junkQueue = server.locateQueue("JUNKYARD");
org.wildfly.common.Assert.assertNotNull(junkQueue);
javax.jms.Queue queue1 = createQueue("q1");
javax.jms.Queue queue2 = createQueue("q2");
ConnectionFactory cf = CFUtil.createConnectionFactory(protocol, "tcp://localhost:61616");
Connection conn = cf.createConnection();
Session sess = conn.createSession(true, Session.SESSION_TRANSACTED);
MessageProducer prod1 = sess.createProducer(queue1);
MessageProducer prod2 = sess.createProducer(queue2);
prod1.setTimeToLive(1);
prod2.setTimeToLive(1);
for (int i = 0; i < 100; i++) {
TextMessage txt = sess.createTextMessage("txt");
txt.setStringProperty("_AMQ_DUPL_ID", "" + i);
prod1.send(txt);
}
sess.commit();
Wait.assertEquals(100L, junkQueue::getMessageCount, 2000, 10);
for (int i = 0; i < 100; i++) {
TextMessage txt = sess.createTextMessage("txt");
txt.setStringProperty("_AMQ_DUPL_ID", "" + i);
prod2.send(txt);
}
sess.commit();
conn.close();
Wait.assertEquals(200L, junkQueue::getMessageCount, 2000, 10);
}
}