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

Merge pull request #980 from cshannon/amq-9217-rollback 

AMQ-9217 - Fix IndividualDeadLetter strategy when retrying messages
This commit is contained in:
Christopher L. Shannon 2023-02-16 10:55:42 -05:00 committed by GitHub
parent ee14778305 459388185a
commit ae4e305f85
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 15 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
activemq-broker/src/main/java/org/apache/activemq/broker/region/policy/AbstractDeadLetterStrategy.java
View File

@ -36,7 +36,7 @@ public abstract class AbstractDeadLetterStrategy implements DeadLetterStrategy {
@Override
public void rollback(Message message) {
if (message != null && this.enableAudit) {
lookupActiveMQMessageAudit(message).rollback(message);
lookupActiveMQMessageAudit(message, true).rollback(message);
}
}
@ -45,7 +45,7 @@ public abstract class AbstractDeadLetterStrategy implements DeadLetterStrategy {
boolean result = false;
if (message != null) {
result = true;
if (enableAudit && lookupActiveMQMessageAudit(message).isDuplicate(message)) {
if (enableAudit && lookupActiveMQMessageAudit(message, false).isDuplicate(message)) {
result = false;
LOG.debug("Not adding duplicate to DLQ: {}, dest: {}", message.getMessageId(), message.getDestination());
}
@ -115,5 +115,5 @@ public abstract class AbstractDeadLetterStrategy implements DeadLetterStrategy {
public abstract int getMaxAuditDepth();
protected abstract ActiveMQMessageAudit lookupActiveMQMessageAudit(Message message);
protected abstract ActiveMQMessageAudit lookupActiveMQMessageAudit(Message message, boolean rollback);
}

14
activemq-broker/src/main/java/org/apache/activemq/broker/region/policy/IndividualDeadLetterStrategy.java
View File

@ -202,15 +202,23 @@ public class IndividualDeadLetterStrategy extends AbstractDeadLetterStrategy {
}
@Override
protected ActiveMQMessageAudit lookupActiveMQMessageAudit(Message message) {
protected ActiveMQMessageAudit lookupActiveMQMessageAudit(Message message, boolean rollback) {
ActiveMQMessageAudit messageAudit;
synchronized(dedicatedMessageAudits) {
messageAudit = dedicatedMessageAudits.get(message.getDestination().getQualifiedName());
// Normally we want to just use the destination property on the message as the key for the map for
// caching the messageAudit object for each destination. However, when rolling back, the message
// provided here has had its destination changed to the individual DLQ destination and is no longer
// the original destination. So to find the correct messageAudit to rollback we need to use
// the originalDestination property on the message to get the correct destination that was
// used to first cache the messageAudit.
final String destinationName = rollback && message.getOriginalDestination() != null ?
message.getOriginalDestination().getQualifiedName() : message.getDestination().getQualifiedName();
messageAudit = dedicatedMessageAudits.get(destinationName);
if(messageAudit == null) {
messageAudit = new ActiveMQMessageAudit(getMaxAuditDepth(), getMaxProducersToAudit());
dedicatedMessageAudits.put(message.getDestination().getQualifiedName(), messageAudit);
dedicatedMessageAudits.put(destinationName, messageAudit);
}
return messageAudit;

2
activemq-broker/src/main/java/org/apache/activemq/broker/region/policy/SharedDeadLetterStrategy.java
View File

@ -71,7 +71,7 @@ public class SharedDeadLetterStrategy extends AbstractDeadLetterStrategy {
}
@Override
protected ActiveMQMessageAudit lookupActiveMQMessageAudit(Message message) {
protected ActiveMQMessageAudit lookupActiveMQMessageAudit(Message message, boolean rollback) {
return messageAudit;
}