This is a proposal for replacing the current message return-to-publisher mechanism with a mechanism based on the exchange-binding-queue model.
Basic proposal
1. Drop Basic.Return from the protocol.
2. Add 'return_key' property to the message.
3. Add new 'return' exchange type routing on return_key instead of routing_key.
4. Add new system exchange called 'amq.return' of 'return' type.
5. Specify that all undeliverable messages will be redirected to 'amq.return' exchange.
Request/reply use case
In this use case application sends a message and want to get either a reply from other party or a rejected message returned in case other party is offline/unfunctional.
1. Create a temporary queue for rejected messages (thus you will get broker-wide unique name)
2. Bind the queue to 'amq.return' exchange with condition 'return-key = return-queue-name'
3. Consume from the queue
4. When sending a message set its return key to return-queue-name.
Dead-letter queue
This use case shows how a queue that stores undeliverable messages can be implemented.
1. Declare a persistent queue.
2. Bind it to 'amq.return' exchange with condition 'return-key = *'
Discussion
Mandatory flag
IMO this flag is misplaced. What it's saying is 'turn return delivery of undeliverable messages on/off'. This should not be a part of Basic.Publish command, in fact it shouldn't be visible from message publishing application at all. Consider dead-letter queue scenario. Message publisher shouldn't be able to say "don't pass this message anywhere in case it cannot be delivered (including dead-letter queue)." Specifying what should be stored in dead-message queue is an administrative task and is not a responsibility of message sender.
In fact, in current proposal there's no need for the flag at all. You either want undeliverable messages back, in which case you'll subscribe to amq.return exchange, or you don't, in which case you simply don't subscribe.
Immediate flag
Immediate flag is a different story. It specifies how a queue should treat the message, so it should be grouped with properties like 'expiration' and 'priority', i.e. basic message properties. (Maybe we can consider 'immediate' being zero expiration, however, expiration is a property ported directly from JMS so it would require some investigation.)
However, it should be specified more clearly, what does 'immediate' mean.
- Does it mean that that the message will be sent back when it cannot be delivered to any consumer system-wide.
- Or does it mean that every queue that gets the message and cannot deliver it to a consumer will send it back separately ?
I, personally, would vote for b. option.
How should 'return' exchange work
Consider a scenario where you would like to subscribe for all rejected messages from a specific exchange. That would require to create a binding that matches the messages not only on 'return_key' property, but also on 'exchange' property.
This example demostrates a need for more complex matching then one on 'return_key'. Rejected message should contain several properties and 'return' exchange should allow matching on these. Here are some examples:
1. exchange name
2. rejection reason code (unroutable, no-consumers, expired)
3. queue name
etc.
Question: What happens with message properties when it is routed to 'return' exchange? Are they left untouched or modified somehow? (immediate, expiration, priority…)
Strip-content property
I would suggest to add one more property to a message. Property called 'strip-content' that would be used by return exchange. It would mean that message body has to be truncated when passing it through return exchange.
This way we can avoid returning large amount of data to sender in cases where the only thing sender cares about is whether the message was delivered or not.
Meta-rejection
What happens when a message is rejected by 'return' exchange? Dropped in all cases? Anything else that can be done with it?
Identifying AMQP broker world-wide
In our clustering architecture there's a need to identify sender application instance uniquelly so that returned messages can be delivered to it unambiguosly. Same issue will probably arrise in other use cases as well.
We would like to use '<application-instance-name>@<globally-unique-broker-name>' syntax for this. Maybe even some global registration of server names should be considered. However, this is an issue for a separate discussion.
Comments
pieterh 1157984185|%e %b %Y, %H:%M %Z|agohover
I know you've asked for this since we first considered the message return function but I don't think it's a useful feature. Returned contents are supposed to be exceptional and the message body contains useful data in most cases.
More problematic, doing this at the binding level would require the broker to modify the message for each binding, something that is going to be more expensive than simply dumping it to a queue.
Also, it seems to be a very specific function when in fact the need to read content headers without body seems to be more general.
What I'd suggest (if we decide we really need such a feature) is to add it to the Basic.Consume method so that it affects messages consumed off a queue.
Portfolio
You are right. I was hesistant about the idea because because it would allow sender of the message determine how rejected messages are processed (I've argued against this in 'Mandatory flag' section). Adding the flag to Basic.Consume is much more clear solution.
Consider the scenario where broker handles both request/reply messages as well as pub/sub messages. Request messages unable to reach the destination should be returned to sender, whereas publications should be dropped silently. (This is the scenario we encouter now at JPMC.)
Having single 'amq.return' exchange instance would mean that each undeliverable publication would be matched against all the bindings created for request messages.
Imagine a heavy load of publications and say 300 destinations for request/reply. It would mean matching each dropped message (maybe hundreds or even thousands per second) would be matched against 300 bindings before being dropped.
One of the possible solutions is to create a separate 'return' exchange instance for each object able to reject a message (currently exchange and queue).
So, for example, messages rejected by exchange 'MyExchange' (unroutable messages) would be passed to '!MyExchange' exchange (of 'return' type).
Same for queues: Undeliverable immediate/expired messages from 'MyQueue' would be routed to '~MyQueue' exchange.
In most cases this will result in rejected message being matched against none or single binding.
This will also solve the problem of subscribing for rejected messages from specific exchange/queue. There will be no need for extending the message by 'exchange' and 'queue' properties and making 'return' exchange type capable of binding on these properties.
On the other hand, it makes subscribing for all the rejected messages within virtual host (dead-letter queue scenario) impossible. You would have to subscribe for each exchange/queue separately.
I like the main thrust of the idea as presented. It seems more generic and extensible, clients will be simpler as well as more flexible.
It would be particularly nice if you could chain bindings (i.e. bind exchanges to other exchanges). This might be a way of addressing the efficiency concerns when matching routing keys by allowing some hierarchical federation. Similarly it would allow different filtering exchanges to be plugged in if stripped content was desired.
Am still chewing it over, in particular the specifc return exchange type and the return key property, but thougt it was worth expressing that I think the overall approach is definately a good one.
The model can be made even more simple and flexible this way:
1. No need for Basic.Return
2. No need for 'return' exchange type, 'amq.return' exchange and 'return-key' property.
3. Add new boolean binding argument to Queue.Bind, called say 'unroutable', meaning that the binding would be applied only if the message cannot be routed to through any of the bindings with unroutable=FALSE.
Request/reply use case
1. Create a queue for rejected messages
2. Bind the queue to exchange with unroutable=TRUE and condition 'routing-key = service-name'
3. Consume from the queue
4. When sending a message set its routing-key is set to service-name.
Interesting thing here is that rejected messages are routed by the same exchange as deliverable messages. This way routing rejected messages is governed by the same rules as routing of non-rejected messages, i.e. messages rejected by direct exchange are routed by their routing-key, messages rejected by headers exchange are routed by their headers field table, etc.
Dead-letter queue use case
1. Declare a persistent queue.
2. Bind the exchange you are interested in to the queue with unroutable=TRUE and binding args that would match any message.
A problem
This proposal doesn't solve the problem of messages rejected by queues themselves, i.e. 'immediate' messages that cannot be delivered to any consumer and expired messages.
Rethinking this idea to its logical consequences we would get to the state where we can drop the concept of 'rejected messages' from the protocol altogether.
The only thing needed is to add 'binding-priority' field (8-bit int) to Queue.Bind.
Exchanges will then work as follows:
It will try to match the message with all the bindings of the highest binding-priority and deliver the message through each matching binding.
If at least one binding was matched, routing algorithm is stopped.
If none of the bindings was matched, algorithm is repeated with bindings on second highest binding-priority.
If none of the bindings was matched, algorithm is repeated with bindings on third highest binding-priority.
etc.
If there are no more bindings, the message is simply dropped.
Advantages of this approach:
1. No 'rejected message' semantics in the protocol.
2. Allows for multi-level message rejection. Consider request/reply scenario. Service binds for requests with binding-priority=1. Sender of the request binds for the request with binding-priority=2. Dead-letter queue binds for the request with binding-priority=3. Consequently, message is delivered to the service, or - in case it is not available - it is delivered back to the sender, or - if the sender had died in the meantime - it is delivered to the dead-letter queue.
I think I like this, though it does make the concept of an exchange a little more complex. As you mention we would need to define how the immediate flag was handled.
Unlike you, in that regard I prefer option a in the original discussion above. I don't like the idea of getting a copy of my message back for every queue without subscribers - I just want to know no one can receive it right now. Option b is probably easier to implement though.
If we were to embrace more generalised sources and sinks, and had another source type- a 'subscription' (queueing not required)- that could be directly bound to an exchange, then you could get the functionality of the immediate flag through the usual unroutable handling…
It makes concept of exchange a little more complex, but on the other hand it allows us to discard all the message-rejection-specific syntax/semantics, so is worth of it, IMO.
I preferred b. option for the sake of encapsulation. If the queue rejects the immediate message how should it know whether the message was delivered by another queue or not?
Anyway, I think we can get rid of immediate flag/messages rejected by a queue altogether. Imagine that queues cannot reject messages. AFAICS there are 2 use cases that seem to contradict this:
1. JMS - it has the expiration property, that says that the message is to be dropped after some specified time. However, JMS specs give no guarantee about expired message not being delivered. So there's in fact no problem with this one.
2. Request/reply scenarion - sender of the request wants the request back in the case where there is no consumer for it. This can be accomplished through 'immediate' flag as well as through 'auto-delete' queues. Auto-delete queue is destroyed when all of its consumers are destroyed. So when last consumer quits, queue is deleted, thus making message sent to it unroutable.
The only problem with this approach is that protocol specifies that there may be a delay between last consumer quitting and queue being auto-deleted. We already had a lot of problems with this feature and at last ended with making the delay zero in out broker implementation.
So I would suggest to get rid of this auto-delete delay feature. Consequently there will be no need for immediate flag. Consequently there will be no need for queues to reject messages. Consequently it will be possible to base message-returning mechanism on binding-priority flag as described above.
I think source/sink/chained-bindigns discussion is a separate one and you should make a new wiki page for it. I would be glad to comment on the idea there.
I don't think it's necessary to over-abstract this functionality. Failure to route a message is quite a clear case and can be handled as such. We don't have multi-level failures, we don't have multiple ways of handling failures (a single exchange type and instance is enough IMO). Also we need a way for queues to (silently) process messages they can't deliver, and the convention of sending them back to a return exchange is simple and sufficient.
The simplest solution I can see is to define a standard exchange type and instance that handles all message failures for a vhost. The type of failure can be an argument used during routing. We can then bind to this exchange to collect failures and process them as wanted.
Portfolio
Regarding the 'immediate' flag, the spec should say (if it does not already) that a message will be returned at most once. Even if it's passed to many queues, it won't get returned more than once.
Portfolio
The more I am thinking about this, the more I like the idea of binding-priority solution. There are in fact no disadvantages and quite a lot of advantages:
1. Minimal changes to protocol (addition of a single flag).
2. Minimal implementation effort (modify the binding algorithm a bit).
3. No need for specific reject-reason codes or modification of rejected message in any way.
3. Efficiency. See the example of request/replies being passed through the same virtual host as pub/subs above. (This is the scenario we are using at JPMC already!) Passing all the rejected mesages through a single 'amq.return' exchange would ultimately lead to the need of separating request/replies and pubs/subs into separate virtual hosts.
4. I believe routing rejected messages through the same exchange type as delivered messages instead of standard 'return' exchange type is a desirable trait. In principle it means "use the same kind of data to route the message to consumer and to the rejected-message-handler". Consider the case of headers exchange. All the data relevant to the routing of the message are stored in headers field. Why shouldn't we use the same data to route the message back instead of introducing some obscure return-key concept ?
5. Maybe there are no multi-level failures used so far, but the use case for that is clear enough: Allow the sender to get rejected message back and process it in its own way, but - if that is impossible (sender died or whatever) - send the messge to dead-letter queue.
First while I haven't had much exposure to the use of AMQP from a clients point of view, as an implementor I do like the idea of multilevel failures.
To follow on from Gordon about not wishing the message content back. Could the binding of a return queue have an additional argument that requests the content be striped. The message ids etc should be sufficient to identify the message in the client. This may be especially useful if the messges are several hundered megabytes.
We already had a bit of discussion of this issue. There are three possible places where the 'strip-content' flag can be placed:
1. Message itself
2. Binding
3. Consumer
1. allows sender of the message to specify how it is delivered. Not good as different rejected-message handlers may require different approach. For example sender may not require content to be delivered back, whereas dead-letter queue would need whole message to be delivered.
2. would require that exchange to modify the message. Exchange is a routing device and should not modify messages routed through it.
3. Seems to be the best option so far.
Exchanges can modify messages but it is problematic for implementations that use message reference counting, so that a single message sent to N queues is not actually duplicated, simply refcounted. Also 'strip content' can be useful in other contexts too, e.g. for an application that simply records the headers of messages and does not care about their contents.
Portfolio
Modifying messages on refcounting exchanges can be done efficiently using some variation of the copy-on-write pattern. "Efficiently" in the sense that it doesn't force additional copies in no-modify cases. It is more complicated than simple refcounts but that's an implementor's problem. It shouldn't affect decisions on how to do things in the spec.
Just thinking about the CopyTo scenario that Steve mentioned.
What if we took a similar approach to SMTP. When a message cannot be immediately delivered it is queued for delivery but a notification is returned to the sender detailing the delay.
If we took this approach we would have to specify a Queue to return notifications to the Publisher. The publisher could then read these notifications and decide what to do.
- Leave it on queue and send else where.
- Remove it from the queue
The same mechanism could be used to notify the publisher that the message they just sent was unroutable and being dropped. Of course the suggestion above could be used to route the message away from the floor. This re-routing could be relayed to the Publisher.
This is a use case we haven't seen yet.
Message is sent. If it can be delivered immediately, it is delivered. If it cannot be delivered immediately, notification is sent to the sender and the message is queued and delivered once there is a consumer for it.
I can imagine the wiring that would implement above scenario, but it is quite complicated. (Explained below.) Does this use case appear so frequently, that it would justify adding of new 'notification' concept to the protocol?
Implementation:
1. In instalation phase exchange E is created.
2. In instalation phase durable queue 'Qqueued' is created and bound to amq.return exchange.
3. On sender application startup temporary queue 'Qnotify' is created, bound to amq amq.reutrn exchange and consumed by the sender.
4. When sender sends a message, it sends it to exchange E.
5. On receiver startup, receiver starts to consume from 'Qqueued'
6. It also creates temporary queue 'Qimmediate', bounds it to exchange E and starts consuming from it.
I realised that 'binding-priority' proposal and 'return' exchange proposal aren't mutually exclusive. They are in fact two solutions, each for one side of message rejection problem.
'binding-priority' is a solution for "What is the algorithm for messages getting rejected within an exchange?"
'return' exchange is a solution for "What's the most common message-returning scenario?"
In other words, 'binding-priority' allows us to use any message-returning scenario, whereas 'return' exchange defines most common one, one that should be made default in the same way as each queue is bound to 'default' exchange when created.