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

https://issues.apache.org/jira/browse/AMQ-6464 

Correct handling of rejected outcome to archive the message in the DLQ
This commit is contained in:
Timothy Bish 2016-10-14 09:35:17 -04:00
parent 3e237ca73a
commit a9f9d4a4d2
2 changed files with 8 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/protocol/AmqpSender.java
View File

@ -270,10 +270,11 @@ public class AmqpSender extends AmqpAbstractLink<Sender> {
}
settle(delivery, MessageAck.INDIVIDUAL_ACK_TYPE);
} else if (state instanceof Rejected) {
// re-deliver /w incremented delivery counter.
md.setRedeliveryCounter(md.getRedeliveryCounter() + 1);
LOG.trace("onDelivery: Rejected state = {}, delivery count now {}", state, md.getRedeliveryCounter());
settle(delivery, -1);
// Rejection is a terminal outcome, we poison the message for dispatch to
// the DLQ. If a custom redelivery policy is used on the broker the message
// can still be redelivered based on the configation of that policy.
LOG.trace("onDelivery: Rejected state = {}, message poisoned.", state, md.getRedeliveryCounter());
settle(delivery, MessageAck.POSION_ACK_TYPE);
} else if (state instanceof Released) {
LOG.trace("onDelivery: Released state = {}", state);
// re-deliver && don't increment the counter.

13
activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/interop/AmqpReceiverTest.java
View File

@ -514,16 +514,9 @@ public class AmqpReceiverTest extends AmqpClientTestSupport {
message.reject();
// Read the message again and validate its state
message = receiver.receive(10, TimeUnit.SECONDS);
assertNotNull("did not receive message again", message);
message.accept();
protonMessage = message.getWrappedMessage();
assertNotNull(protonMessage);
assertEquals("Unexpected updated value for AMQP delivery-count", 1, protonMessage.getDeliveryCount());
// Attempt to read the message again but should not get it.
message = receiver.receive(2, TimeUnit.SECONDS);
assertNull("shoudl not receive message again", message);
connection.close();
}