Comments (10)
@ansd Absolutely. I've made the change and our retries seem to be passing the same tests.
I think this issue is closable from my perspective. Thanks!
from rabbitmq-server.
did you enable the message_containers
feature flag?
from rabbitmq-server.
Not explicitly, but it does show as "Enabled" in the management interface.
from rabbitmq-server.
@ansd and @kjnilsson discussed this internally and have found where the regression was introduced.
from rabbitmq-server.
from rabbitmq-server.
@bmflynn could you possibly explain a bit how you are using this feature? Collaborative editing of x- headers is a bit odd in my view. x- headers "belong" to the broker and using it this way feels somewhat odd.
It is very unlikely we'd provide the same feature for other protocols such as AMQP (1.0)
from rabbitmq-server.
Dead-lettering and the x-death['count']
value are used as part of a retry mechanism where a message with expiration is retried until x-death['count']
is greater than the maximum number of retries.
Essentially, when message handling fails the message properties (including headers) and data are published with a message expiration via an error exchange to a queue with the dead-letter exchange set to the original exchange. The message will remain in this queue to expire at which point it gets routed back to the original exchange and is retried. This mechanism continues until the original message is handled successfully or x-death['count']
reaches the configured maximum number of retries.
There is no editing of the x-headers other than the fact that they are part of the original message properties that are used as part of a new message that is sent to an error exchange.
What is the specific feature you are referring to? Do you mean it's unlikely that a death count, or the number of times a message is dead-lettered, will be available in AMQP 1.0?
from rabbitmq-server.
@bmflynn Starting with "message containers" in 3.13, we would like to treat any x-
headers as belonging to the broker, meaning a client shouldn't publish messages with x-
headers. (If a client does publish with x-
headers, these headers won't be interpreted in any special way anymore as illustrated by your example).
You are right that this is a behavioural change compared to 3.12 and arguably a breaking change belonging to 4.0.
So, we could "fix" / revert to the 3.12 behaviour by having the broker parse incoming headers and therefore interpreting the x-death
header coming from a publisher.
However, we very likely would remove such special interpretation again in RabbitMQ 4.0 (which probably ships end of this year).
Therefore, my question is: Could your client set a custom non x-
header and just increment it itself when it re-publishes the message?
Here is an example based on yours:
Code example with custom header
import os
import time
from collections import namedtuple
from pprint import pprint
from pika import BasicProperties, BlockingConnection, URLParameters
Msg = namedtuple("Msg", ["method", "properties", "body"])
amqpurl = "amqp://localhost:5672"
exchange = "amq.topic"
start_queue = "start_queue"
dlx_exchange = "errors"
error_queue = "dead_letters"
routing_key = "test_retries"
data = "XXX"
conn = BlockingConnection(URLParameters(amqpurl))
ch = conn.channel()
ch.exchange_declare(dlx_exchange, "topic")
ch.queue_declare(error_queue)
ch.queue_bind(error_queue, dlx_exchange, "#")
ch.queue_declare(start_queue, arguments={"x-dead-letter-exchange": dlx_exchange})
ch.queue_bind(start_queue, exchange, routing_key)
print(f"first publish {exchange=} {routing_key=} {data=}")
ch.basic_publish(exchange, routing_key, data, properties=BasicProperties(expiration="500"))
# Wait for message to error out
print("waiting\n")
time.sleep(1)
first = Msg(*ch.basic_get(error_queue))
ch.basic_ack(first.method.delivery_tag)
print("Received:")
pprint(first.properties.headers)
print("Second publish")
Key = "my-retry-count"
# Increment the retry count if it exists
first.properties.headers[Key] = first.properties.headers.get(Key, 0) + 1
first.properties.expiration = "500"
ch.basic_publish(exchange, routing_key, data, properties=first.properties)
# Wait for message to expire again
print("waiting\n")
time.sleep(1)
second = Msg(*ch.basic_get(error_queue))
ch.basic_ack(second.method.delivery_tag)
print("Received:")
pprint(second.properties.headers)
print("Third publish")
first.properties.headers[Key] = first.properties.headers.get(Key, 0) + 1
first.properties.expiration = "500"
ch.basic_publish(exchange, routing_key, data, properties=first.properties)
# Wait for message to expire again
print("waiting\n")
time.sleep(1)
second = Msg(*ch.basic_get(error_queue))
ch.basic_ack(second.method.delivery_tag)
print("Received:")
pprint(second.properties.headers)
ch.close()
conn.close()
from rabbitmq-server.
Related Issues (20)
- OAuth 2: cacertfile is still a mandatory field on Erlang 26 HOT 1
- After upgrading to 3.13, management interface shows 500 for /api/deprecated-features/used HOT 6
- Path prefix not working on v3.13
- Support to define event_exchange.queue_type to be able to use quorum queue for event_exchange plugin HOT 1
- `rabbitmq_auth_mechanism_ssl` usage docs seems invalid HOT 2
- auth_oauth2.jwks_url is always verified HOT 1
- Add more useful data into rabbitmq_cluster_vhost_status metric
- 4.x: investigate if management plugin's TLS options key can be renamed to ssl_options for consistency HOT 6
- Deprecate `queue_master_locator` HOT 1
- Prevent excessive logging in certain failure scenarios HOT 3
- RabbitMQ 3.13.0 nodes with peer discovery enabled fails to start in an IPv6-only environment HOT 24
- AMQP 1.0 Shovels: expose additional capabilities needed for successful connection to some AMQP 1.0 brokers HOT 3
- Khepri: timeouts when one of the nodes stops responding
- RabbitMQ 3.13.0 nodes with Consul peer discovery enabled fails to form a cluster HOT 2
- Odd characters in example file deps/rabbit/docs/rabbitmq.conf.example l.801 HOT 1
- Popup does not close in Managemet UI
- A `rabbitmq-queues grow` equivalent is missing for streams HOT 2
- Channel Metrics Cardinality - /metrics stops working HOT 3
- Restarting crashed queue with function_clause reason after update to RabbitMQ 3.13
Recommend Projects
-
React
A declarative, efficient, and flexible JavaScript library for building user interfaces.
-
Vue.js
🖖 Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
-
Typescript
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
-
TensorFlow
An Open Source Machine Learning Framework for Everyone
-
Django
The Web framework for perfectionists with deadlines.
-
Laravel
A PHP framework for web artisans
-
D3
Bring data to life with SVG, Canvas and HTML. 📊📈🎉
-
Recommend Topics
-
javascript
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
-
web
Some thing interesting about web. New door for the world.
-
server
A server is a program made to process requests and deliver data to clients.
-
Machine learning
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
-
Visualization
Some thing interesting about visualization, use data art
-
Game
Some thing interesting about game, make everyone happy.
Recommend Org
-
Facebook
We are working to build community through open source technology. NB: members must have two-factor auth.
-
Microsoft
Open source projects and samples from Microsoft.
-
Google
Google ❤️ Open Source for everyone.
-
Alibaba
Alibaba Open Source for everyone
-
D3
Data-Driven Documents codes.
-
Tencent
China tencent open source team.
from rabbitmq-server.