Skip to content

Commit 9f82e3e

Browse files
mattrpavmattrpav
committed
[AMQ-9554] Add RedeliveryPolicy setting to ignore browsers by default
1 parent e45ee4a commit 9f82e3e

File tree

3 files changed

+226
-8
lines changed

3 files changed

+226
-8
lines changed

‎activemq-client/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java

Lines changed: 12 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -548,13 +548,18 @@ private void poisonAck(MessageDispatch md, String cause) throws JMSException {
548548

549549
private boolean redeliveryExceeded(MessageDispatch md) {
550550
try {
551-
return session.getTransacted()
552-
&& redeliveryPolicy != null
553-
&& redeliveryPolicy.isPreDispatchCheck()
554-
&& redeliveryPolicy.getMaximumRedeliveries() != RedeliveryPolicy.NO_MAXIMUM_REDELIVERIES
555-
&& md.getRedeliveryCounter() > redeliveryPolicy.getMaximumRedeliveries()
556-
// redeliveryCounter > x expected after resend via brokerRedeliveryPlugin
557-
&& md.getMessage().getProperty("redeliveryDelay") == null;
551+
if(!session.getTransacted() || redeliveryPolicy == null || !redeliveryPolicy.isPreDispatchCheck()) {
552+
return false;
553+
}
554+
555+
if(info.isBrowser() && redeliveryPolicy.isQueueBrowserIgnored()) {
556+
return false;
557+
}
558+
559+
return redeliveryPolicy.getMaximumRedeliveries() != RedeliveryPolicy.NO_MAXIMUM_REDELIVERIES
560+
&& md.getRedeliveryCounter() > redeliveryPolicy.getMaximumRedeliveries()
561+
// redeliveryCounter > x expected after resend via brokerRedeliveryPlugin
562+
&& md.getMessage().getProperty("redeliveryDelay") == null;
558563
} catch (Exception ignored) {
559564
return false;
560565
}

‎activemq-client/src/main/java/org/apache/activemq/RedeliveryPolicy.java

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -46,6 +46,7 @@ public class RedeliveryPolicy extends DestinationMapEntry implements Cloneable,
4646
protected double backOffMultiplier = 5.0;
4747
protected long redeliveryDelay = initialRedeliveryDelay;
4848
protected boolean preDispatchCheck = true;
49+
protected boolean queueBrowserIgnored = true;
4950

5051
public RedeliveryPolicy() {
5152
}
@@ -165,4 +166,12 @@ public void setPreDispatchCheck(boolean preDispatchCheck) {
165166
public boolean isPreDispatchCheck() {
166167
return preDispatchCheck;
167168
}
169+
170+
public void setQueueBrowserIgnore(boolean queueBrowserIgnored) {
171+
this.queueBrowserIgnored = queueBrowserIgnored;
172+
}
173+
174+
public boolean isQueueBrowserIgnored() {
175+
return queueBrowserIgnored;
176+
}
168177
}

‎activemq-unit-tests/src/test/java/org/apache/activemq/usecases/QueueBrowsingTest.java

Lines changed: 205 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,12 +17,15 @@
1717
package org.apache.activemq.usecases;
1818

1919
import static org.junit.Assert.assertEquals;
20+
import static org.junit.Assert.assertNull;
2021
import static org.junit.Assert.assertTrue;
21-
22+
import static org.junit.Assert.fail;
2223

2324
import java.io.IOException;
2425
import java.net.URI;
2526
import java.util.Enumeration;
27+
import java.util.Set;
28+
import java.util.concurrent.atomic.AtomicInteger;
2629

2730
import jakarta.jms.Connection;
2831
import jakarta.jms.JMSException;
@@ -35,8 +38,14 @@
3538
import org.apache.activemq.ActiveMQConnectionFactory;
3639
import org.apache.activemq.broker.BrokerService;
3740
import org.apache.activemq.broker.TransportConnector;
41+
import org.apache.activemq.broker.jmx.DestinationView;
42+
import org.apache.activemq.broker.region.Destination;
43+
import org.apache.activemq.broker.region.Queue;
44+
import org.apache.activemq.broker.region.QueueMessageReference;
45+
import org.apache.activemq.broker.region.policy.IndividualDeadLetterStrategy;
3846
import org.apache.activemq.broker.region.policy.PolicyEntry;
3947
import org.apache.activemq.broker.region.policy.PolicyMap;
48+
import org.apache.activemq.command.ActiveMQMessage;
4049
import org.apache.activemq.command.ActiveMQQueue;
4150
import org.junit.After;
4251
import org.junit.Before;
@@ -68,6 +77,10 @@ public void startBroker() throws Exception {
6877

6978
connectUri = connector.getConnectUri();
7079
factory = new ActiveMQConnectionFactory(connectUri);
80+
factory.setWatchTopicAdvisories(false);
81+
factory.getRedeliveryPolicy().setInitialRedeliveryDelay(0l);
82+
factory.getRedeliveryPolicy().setRedeliveryDelay(0l);
83+
factory.getRedeliveryPolicy().setMaximumRedeliveryDelay(0l);
7184
}
7285

7386
public BrokerService createBroker() throws IOException {
@@ -217,4 +230,195 @@ public void testMemoryLimit() throws Exception {
217230
browser.close();
218231
assertTrue("got at least maxPageSize", received >= maxPageSize);
219232
}
233+
234+
@Test // https://issues.apache.org/jira/browse/AMQ-9554
235+
public void testBrowseRedeliveryMaxRedelivered() throws Exception {
236+
browseRedelivery(0, true);
237+
}
238+
239+
@Test // Ignore https://issues.apache.org/jira/browse/AMQ-9554
240+
public void testBrowseRedeliveryIgnored() throws Exception {
241+
browseRedelivery(1, false);
242+
}
243+
244+
protected void browseRedelivery(int browseExpected, boolean dlqDlqExpected) throws Exception {
245+
IndividualDeadLetterStrategy individualDeadLetterStrategy = new IndividualDeadLetterStrategy();
246+
individualDeadLetterStrategy.setQueuePrefix("");
247+
individualDeadLetterStrategy.setQueueSuffix(".dlq");
248+
individualDeadLetterStrategy.setUseQueueForQueueMessages(true);
249+
broker.getDestinationPolicy().getDefaultEntry().setDeadLetterStrategy(individualDeadLetterStrategy);
250+
broker.getDestinationPolicy().getDefaultEntry().setPersistJMSRedelivered(true);
251+
252+
if(dlqDlqExpected) {
253+
factory.getRedeliveryPolicy().setQueueBrowserIgnore(false);
254+
}
255+
256+
String messageId = null;
257+
258+
String queueName = "browse.redeliverd.tx";
259+
String dlqQueueName = "browse.redeliverd.tx.dlq";
260+
String dlqDlqQueueName = "browse.redeliverd.tx.dlq.dlq";
261+
262+
ActiveMQQueue queue = new ActiveMQQueue(queueName + "?consumer.prefetchSize=0");
263+
ActiveMQQueue queueDLQ = new ActiveMQQueue(dlqQueueName + "?consumer.prefetchSize=0");
264+
ActiveMQQueue queueDLQDLQ = new ActiveMQQueue(dlqDlqQueueName);
265+
266+
broker.getAdminView().addQueue(queueName);
267+
broker.getAdminView().addQueue(dlqQueueName);
268+
269+
DestinationView dlqQueueView = broker.getAdminView().getBroker().getQueueView(dlqQueueName);
270+
DestinationView queueView = broker.getAdminView().getBroker().getQueueView(queueName);
271+
272+
verifyQueueStats(0l, 0l, 0l, dlqQueueView);
273+
verifyQueueStats(0l, 0l, 0l, queueView);
274+
275+
Connection connection = factory.createConnection();
276+
connection.start();
277+
Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
278+
MessageProducer producer = session.createProducer(queue);
279+
280+
Message sendMessage = session.createTextMessage("Hello world!");
281+
producer.send(sendMessage);
282+
messageId = sendMessage.getJMSMessageID();
283+
session.commit();
284+
producer.close();
285+
286+
verifyQueueStats(0l, 0l, 0l, dlqQueueView);
287+
verifyQueueStats(1l, 0l, 1l, queueView);
288+
289+
// Redeliver message to DLQ
290+
Message message = null;
291+
MessageConsumer consumer = session.createConsumer(queue);
292+
int rollbackCount = 0;
293+
do {
294+
message = consumer.receive(2000l);
295+
if(message != null) {
296+
session.rollback();
297+
rollbackCount++;
298+
}
299+
} while (message != null);
300+
301+
assertEquals(Integer.valueOf(7), Integer.valueOf(rollbackCount));
302+
verifyQueueStats(1l, 0l, 1l, dlqQueueView);
303+
verifyQueueStats(1l, 1l, 0l, queueView);
304+
305+
session.commit();
306+
consumer.close();
307+
308+
// Increment redelivery counter on the message in the DLQ
309+
// Close the consumer to force broker to dispatch
310+
Message messageDLQ = null;
311+
MessageConsumer consumerDLQ = session.createConsumer(queueDLQ);
312+
int dlqRollbackCount = 0;
313+
int dlqRollbackCountLimit = 5;
314+
do {
315+
messageDLQ = consumerDLQ.receive(2000l);
316+
if(messageDLQ != null) {
317+
session.rollback();
318+
session.close();
319+
consumerDLQ.close();
320+
session = connection.createSession(true, Session.SESSION_TRANSACTED);
321+
consumerDLQ = session.createConsumer(queueDLQ);
322+
dlqRollbackCount++;
323+
}
324+
} while (messageDLQ != null && dlqRollbackCount < dlqRollbackCountLimit);
325+
session.commit();
326+
consumerDLQ.close();
327+
328+
// Browse in tx mode works when we are at the edge of maxRedeliveries
329+
// aka browse does not increment redeliverCounter as expected
330+
Queue brokerQueueDLQ = resolveQueue(broker, queueDLQ);
331+
332+
for(int i=0; i<16; i++) {
333+
QueueBrowser browser = session.createBrowser(queueDLQ);
334+
Enumeration<?> enumeration = browser.getEnumeration();
335+
ActiveMQMessage activemqMessage = null;
336+
int received = 0;
337+
while (enumeration.hasMoreElements()) {
338+
activemqMessage = (ActiveMQMessage)enumeration.nextElement();
339+
received++;
340+
}
341+
browser.close();
342+
assertEquals(Integer.valueOf(1), Integer.valueOf(received));
343+
assertEquals(Integer.valueOf(6), Integer.valueOf(activemqMessage.getRedeliveryCounter()));
344+
345+
// Confirm broker-side redeliveryCounter
346+
QueueMessageReference queueMessageReference = brokerQueueDLQ.getMessage(messageId);
347+
assertEquals(Integer.valueOf(6), Integer.valueOf(queueMessageReference.getRedeliveryCounter()));
348+
}
349+
350+
session.close();
351+
connection.close();
352+
353+
// Change redelivery max and the browser will fail
354+
factory.getRedeliveryPolicy().setMaximumRedeliveries(3);
355+
final Connection browseConnection = factory.createConnection();
356+
browseConnection.start();
357+
358+
final AtomicInteger browseCounter = new AtomicInteger(0);
359+
final AtomicInteger jmsExceptionCounter = new AtomicInteger(0);
360+
361+
final Session browseSession = browseConnection.createSession(true, Session.SESSION_TRANSACTED);
362+
363+
Thread browseThread = new Thread() {
364+
public void run() {
365+
366+
QueueBrowser browser = null;
367+
try {
368+
browser = browseSession.createBrowser(queueDLQ);
369+
Enumeration<?> enumeration = browser.getEnumeration();
370+
if(Thread.currentThread().isInterrupted()) {
371+
Thread.currentThread().interrupt();
372+
}
373+
while (enumeration.hasMoreElements()) {
374+
Message message = (Message)enumeration.nextElement();
375+
if(message != null) {
376+
browseCounter.incrementAndGet();
377+
}
378+
}
379+
} catch (JMSException e) {
380+
jmsExceptionCounter.incrementAndGet();
381+
} finally {
382+
if(browser != null) { try { browser.close(); } catch (JMSException e) { jmsExceptionCounter.incrementAndGet(); } }
383+
if(browseSession != null) { try { browseSession.close(); } catch (JMSException e) { jmsExceptionCounter.incrementAndGet(); } }
384+
if(browseConnection != null) { try { browseConnection.close(); } catch (JMSException e) { jmsExceptionCounter.incrementAndGet(); } }
385+
}
386+
}
387+
};
388+
browseThread.start();
389+
Thread.sleep(2000l);
390+
browseThread.interrupt();
391+
392+
assertEquals(Integer.valueOf(browseExpected), Integer.valueOf(browseCounter.get()));
393+
assertEquals(Integer.valueOf(0), Integer.valueOf(jmsExceptionCounter.get()));
394+
395+
// ActiveMQConsumer sends a poison ack, messages gets moved to .dlq.dlq AND remains on the .dlq
396+
DestinationView dlqDlqQueueView = broker.getAdminView().getBroker().getQueueView(dlqDlqQueueName);
397+
verifyQueueStats(1l, 1l, 0l, queueView);
398+
verifyQueueStats(1l, 0l, 1l, dlqQueueView);
399+
400+
if(dlqDlqExpected) {
401+
verifyQueueStats(1l, 0l, 1l, dlqDlqQueueView);
402+
} else {
403+
assertNull(dlqDlqQueueView);
404+
}
405+
}
406+
protected static void verifyQueueStats(long enqueueCount, long dequeueCount, long queueSize, DestinationView queueView) {
407+
assertEquals(Long.valueOf(enqueueCount), Long.valueOf(queueView.getEnqueueCount()));
408+
assertEquals(Long.valueOf(dequeueCount), Long.valueOf(queueView.getDequeueCount()));
409+
assertEquals(Long.valueOf(queueSize), Long.valueOf(queueView.getQueueSize()));
410+
}
411+
412+
protected static Queue resolveQueue(BrokerService brokerService, ActiveMQQueue activemqQueue) throws Exception {
413+
Set<Destination> destinations = brokerService.getBroker().getDestinations(activemqQueue);
414+
if(destinations == null || destinations.isEmpty()) {
415+
return null;
416+
}
417+
418+
if(destinations.size() > 1) {
419+
fail("Expected one-and-only one queue for: " + activemqQueue);
420+
}
421+
422+
return (Queue)destinations.iterator().next();
423+
}
220424
}

0 commit comments

Comments
 (0)