You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by cl...@apache.org on 2023/05/18 13:36:29 UTC
[activemq-artemis] branch main updated: ARTEMIS-4284 - sync operwire remove consumer with the operation context to ensure prefetched messages are available to the next consumer in order
This is an automated email from the ASF dual-hosted git repository.
clebertsuconic pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git
The following commit(s) were added to refs/heads/main by this push:
new b664022a1e ARTEMIS-4284 - sync operwire remove consumer with the operation context to ensure prefetched messages are available to the next consumer in order
b664022a1e is described below
commit b664022a1ed772e74b518dd5d12f44ca53fb2f16
Author: Gary Tully <ga...@gmail.com>
AuthorDate: Thu May 18 13:17:10 2023 +0100
ARTEMIS-4284 - sync operwire remove consumer with the operation context to ensure prefetched messages are available to the next consumer in order
This closes #4483
---
.../core/protocol/openwire/amq/AMQConsumer.java | 5 ++
.../PrefetchRedeliveryCountOpenwireTest.java | 96 ++++++++++++++++++++++
2 files changed, 101 insertions(+)
diff --git a/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/amq/AMQConsumer.java b/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/amq/AMQConsumer.java
index fee1f6c849..8397e5f84e 100644
--- a/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/amq/AMQConsumer.java
+++ b/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/amq/AMQConsumer.java
@@ -416,6 +416,11 @@ public class AMQConsumer {
if (delayedDispatchPrompter != null) {
delayedDispatchPrompter.cancel(false);
}
+ if (info.getPrefetchSize() > 1) {
+ // because response required is false on a RemoveConsumerCommand, a new consumer could miss canceled prefetched messages
+ // we await the operation context completion before handling a subsequent command
+ session.getCoreSession().getSessionContext().waitCompletion();
+ }
}
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/PrefetchRedeliveryCountOpenwireTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/PrefetchRedeliveryCountOpenwireTest.java
new file mode 100644
index 0000000000..5fef3c4a9c
--- /dev/null
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/openwire/PrefetchRedeliveryCountOpenwireTest.java
@@ -0,0 +1,96 @@
+/*
+ * 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.openwire;
+
+import javax.jms.Connection;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Queue;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+
+import java.util.Map;
+
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.artemis.api.core.QueueConfiguration;
+import org.apache.activemq.artemis.api.core.RoutingType;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
+import org.apache.activemq.command.ActiveMQQueue;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class PrefetchRedeliveryCountOpenwireTest extends OpenWireTestBase {
+
+ @Override
+ public void setUp() throws Exception {
+ realStore = true;
+ super.setUp();
+ }
+
+ @Override
+ protected void configureAddressSettings(Map<String, AddressSettings> addressSettingsMap) {
+ super.configureAddressSettings(addressSettingsMap);
+ // force send to dlq early
+ addressSettingsMap.get("#").setMaxDeliveryAttempts(2);
+ }
+
+ @Test(timeout = 60_000)
+ public void testConsumerSingleMessageLoop() throws Exception {
+ Connection exConn = null;
+
+ SimpleString durableQueue = new SimpleString("exampleQueue");
+ this.server.createQueue(new QueueConfiguration(durableQueue).setRoutingType(RoutingType.ANYCAST).setExclusive(true));
+
+ try {
+ ActiveMQConnectionFactory exFact = new ActiveMQConnectionFactory();
+ exFact.setWatchTopicAdvisories(false);
+
+ Queue queue = new ActiveMQQueue("exampleQueue");
+
+ exConn = exFact.createConnection();
+
+ exConn.start();
+
+ Session session = exConn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+
+ MessageProducer producer = session.createProducer(queue);
+
+ TextMessage message = session.createTextMessage("This is a text message");
+
+ int numMessages = 20;
+ for (int i = 0; i < numMessages; i++) {
+ producer.send(message);
+ }
+
+ for (int i = 0; i < numMessages; i++) {
+ // consumer per message
+ MessageConsumer messageConsumer = session.createConsumer(queue);
+
+ TextMessage messageReceived = (TextMessage) messageConsumer.receive(5000);
+ Assert.assertNotNull(messageReceived);
+
+ assertEquals("This is a text message", messageReceived.getText());
+ messageConsumer.close();
+ }
+ } finally {
+ if (exConn != null) {
+ exConn.close();
+ }
+ }
+ }
+}