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

ARTEMIS-2750 Fixing out of order cancellations in AMQP

This commit is contained in:
Clebert Suconic 2020-05-04 12:22:23 -04:00
parent fde126d867
commit 7438d7ff07
3 changed files with 35 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AMQPConnectionContext.java
View File

@ -544,8 +544,6 @@ public class AMQPConnectionContext extends ProtonInitializable implements EventH
// We scheduled it for later, as that will work through anything that's pending on the current deliveries.
runNow(() -> {
link.close();
link.free();
ProtonDeliveryHandler linkContext = (ProtonDeliveryHandler) link.getContext();
if (linkContext != null) {
@ -555,7 +553,15 @@ public class AMQPConnectionContext extends ProtonInitializable implements EventH
log.error(e.getMessage(), e);
}
}
flush();
/// we have to perform the link.close after the linkContext.close is finished.
// linkeContext.close will perform a few executions on the netty loop,
// this has to come next
runLater(() -> {
link.close();
link.free();
flush();
});
});
}

28
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/ProtonServerSenderContext.java
View File

@ -569,8 +569,23 @@ public class ProtonServerSenderContext extends ProtonInitializable implements Pr
*/
@Override
public void close(boolean remoteLinkClose) throws ActiveMQAMQPException {
// we need to mark closed first to make sure no more adds are accepted
closed = true;
// MessageReferences are sent to the Connection executor (Netty Loop)
// as a result the returning references have to be done later after they
// had their chance to finish and clear the runnable
connection.runLater(() -> {
try {
internalClose(remoteLinkClose);
} catch (Exception e) {
log.warn(e.getMessage(), e);
}
});
}
private void internalClose(boolean remoteLinkClose) throws ActiveMQAMQPException {
try {
closed = true;
protonSession.removeSender(sender);
sessionSPI.closeSender(brokerConsumer);
// if this is a link close rather than a connection close or detach, we need to delete
@ -836,6 +851,7 @@ public class ProtonServerSenderContext extends ProtonInitializable implements Pr
void resume() {
connection.runNow(this::deliver);
}
void deliver() {
// This is discounting some bytes due to Transfer payload
@ -994,11 +1010,11 @@ public class ProtonServerSenderContext extends ProtonInitializable implements Pr
}
private static SimpleString createQueueName(boolean useCoreSubscriptionNaming,
String clientId,
String pubId,
boolean shared,
boolean global,
boolean isVolatile) {
String clientId,
String pubId,
boolean shared,
boolean global,
boolean isVolatile) {
if (useCoreSubscriptionNaming) {
final boolean durable = !isVolatile;
final String subscriptionName = pubId.contains("|") ? pubId.split("\\|")[0] : pubId;

5
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/MessageReferenceImpl.java
View File

@ -114,7 +114,10 @@ public class MessageReferenceImpl extends LinkedListImpl.Node<MessageReferenceIm
@Override
public void onDelivery(Consumer<? super MessageReference> onDelivery) {
assert this.onDelivery == null;
// I am keeping this commented out as a documentation feature:
// a Message reference may eventually be taken back before the connection.run was finished.
// as a result it may be possible to have this.onDelivery != null here due to cancellations.
// assert this.onDelivery == null;
this.onDelivery = onDelivery;
}