This is an automated email from the ASF dual-hosted git repository.
chibenwa pushed a commit to branch 3.9.x
in repository https://gitbox.apache.org/repos/asf/james-project.git
The following commit(s) were added to refs/heads/3.9.x by this push:
new 663fff40bc [FIX] RabbitMQ Allow disabling notification queue auto
delete in favor of x-expires (#3076)
663fff40bc is described below
commit 663fff40bcba8f6056b5d9ab81a0d872dd405ec0
Author: Benoit TELLIER <[email protected]>
AuthorDate: Tue Jun 30 15:49:50 2026 +0200
[FIX] RabbitMQ Allow disabling notification queue auto delete in favor of
x-expires (#3076)
---
.../backends/rabbitmq/RabbitMQConfiguration.java | 20 +++++++++++++++++++-
.../pages/distributed/configure/rabbitmq.adoc | 8 ++++++++
.../modules/servers/partials/configure/rabbitmq.adoc | 8 ++++++++
.../apache/james/events/KeyRegistrationHandler.java | 3 +--
.../sample-configuration/rabbitmq.properties | 10 ++++++++++
.../rabbitmq.properties | 10 ++++++++++
src/site/xdoc/server/config-rabbitmq.xml | 8 ++++++++
7 files changed, 64 insertions(+), 3 deletions(-)
diff --git
a/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/RabbitMQConfiguration.java
b/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/RabbitMQConfiguration.java
index 0535a96a6f..645ea32648 100644
---
a/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/RabbitMQConfiguration.java
+++
b/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/RabbitMQConfiguration.java
@@ -310,6 +310,7 @@ public class RabbitMQConfiguration {
private static final String QUEUE_TTL = "notification.queue.ttl";
private static final String QUEUE_DELIVERY_LIMIT =
"quorum.queues.delivery.limit";
private static final String EVENT_BUS_NOTIFICATION_DURABILITY_ENABLED =
"event.bus.notification.durability.enabled";
+ private static final String EVENT_BUS_NOTIFICATION_QUEUE_AUTO_DELETE =
"notification.queue.autoDelete";
private static final String EVENT_BUS_PUBLISH_CONFIRM_ENABLED =
"event.bus.publish.confirm.enabled";
private static final String EVENT_BUS_PROPAGATE_DISPATCH_ERROR =
"event.bus.propagate.dispatch.error";
private static final String TASK_QUEUE_CONSUMER_TIMEOUT =
"task.queue.consumer.timeout";
@@ -412,6 +413,7 @@ public class RabbitMQConfiguration {
private Optional<Long> queueTTL;
private Optional<Boolean> eventBusPublishConfirmEnabled;
private Optional<Boolean> eventBusNotificationDurabilityEnabled;
+ private Optional<Boolean> eventBusNotificationQueueAutoDelete;
private Optional<Boolean> eventBusPropagateDispatchError;
private Optional<String> vhost;
private Optional<Duration> taskQueueConsumerTimeout;
@@ -437,6 +439,7 @@ public class RabbitMQConfiguration {
this.queueTTL = Optional.empty();
this.eventBusPublishConfirmEnabled = Optional.empty();
this.eventBusNotificationDurabilityEnabled = Optional.empty();
+ this.eventBusNotificationQueueAutoDelete = Optional.empty();
this.vhost = Optional.empty();
this.taskQueueConsumerTimeout = Optional.empty();
this.eventBusPropagateDispatchError = Optional.empty();
@@ -522,6 +525,11 @@ public class RabbitMQConfiguration {
return this;
}
+ public Builder eventBusNotificationQueueAutoDelete(Boolean
eventBusNotificationQueueAutoDelete) {
+ this.eventBusNotificationQueueAutoDelete =
Optional.ofNullable(eventBusNotificationQueueAutoDelete);
+ return this;
+ }
+
public Builder useQuorumQueues(Boolean useQuorumQueues) {
this.useQuorumQueues = Optional.ofNullable(useQuorumQueues);
return this;
@@ -587,6 +595,7 @@ public class RabbitMQConfiguration {
queueTTL,
eventBusPublishConfirmEnabled.orElse(true),
eventBusNotificationDurabilityEnabled.orElse(true),
+
eventBusNotificationQueueAutoDelete.orElse(Constants.AUTO_DELETE),
vhost,
taskQueueConsumerTimeout.orElse(DEFAULT_TASK_QUEUE_CONSUMER_TIMEOUT),
eventBusPropagateDispatchError.orElse(true));
@@ -663,6 +672,7 @@ public class RabbitMQConfiguration {
.hosts(hosts)
.queueTTL(queueTTL)
.eventBusNotificationDurabilityEnabled(configuration.getBoolean(EVENT_BUS_NOTIFICATION_DURABILITY_ENABLED,
null))
+
.eventBusNotificationQueueAutoDelete(configuration.getBoolean(EVENT_BUS_NOTIFICATION_QUEUE_AUTO_DELETE,
null))
.eventBusPublishConfirmEnabled(configuration.getBoolean(EVENT_BUS_PUBLISH_CONFIRM_ENABLED,
null))
.eventBusPropagateDispatchError(configuration.getBoolean(EVENT_BUS_PROPAGATE_DISPATCH_ERROR,
null))
.vhost(vhost)
@@ -738,6 +748,7 @@ public class RabbitMQConfiguration {
private final Optional<Long> queueTTL;
private final boolean eventBusPublishConfirmEnabled;
private final boolean eventBusNotificationDurabilityEnabled;
+ private final boolean eventBusNotificationQueueAutoDelete;
private final Optional<String> vhost;
private final Duration taskQueueConsumerTimeout;
private final boolean eventBusPropagateDispatchError;
@@ -747,6 +758,7 @@ public class RabbitMQConfiguration {
int networkRecoveryIntervalInMs, Boolean
useSsl, Boolean useSslManagement, SSLConfiguration sslConfiguration,
boolean useQuorumQueues, Optional<Integer>
quorumQueueDeliveryLimit, int quorumQueueReplicationFactor, List<Host> hosts,
Optional<Long> queueTTL,
boolean eventBusPublishConfirmEnabled,
boolean eventBusNotificationDurabilityEnabled,
+ boolean eventBusNotificationQueueAutoDelete,
Optional<String> vhost, Duration
taskQueueConsumerTimeout, boolean eventBusPropagateDispatchError) {
this.uri = uri;
this.managementUri = managementUri;
@@ -768,6 +780,7 @@ public class RabbitMQConfiguration {
this.queueTTL = queueTTL;
this.eventBusPublishConfirmEnabled = eventBusPublishConfirmEnabled;
this.eventBusNotificationDurabilityEnabled =
eventBusNotificationDurabilityEnabled;
+ this.eventBusNotificationQueueAutoDelete =
eventBusNotificationQueueAutoDelete;
this.vhost = vhost;
this.taskQueueConsumerTimeout = taskQueueConsumerTimeout;
this.eventBusPropagateDispatchError = eventBusPropagateDispatchError;
@@ -862,6 +875,10 @@ public class RabbitMQConfiguration {
return eventBusNotificationDurabilityEnabled;
}
+ public boolean isEventBusNotificationQueueAutoDeleteEnabled() {
+ return eventBusNotificationQueueAutoDelete;
+ }
+
public Optional<String> getVhost() {
return vhost.or(this::getVhostFromPath);
}
@@ -916,6 +933,7 @@ public class RabbitMQConfiguration {
&& Objects.equals(this.queueTTL, that.queueTTL)
&& Objects.equals(this.eventBusPublishConfirmEnabled,
that.eventBusPublishConfirmEnabled)
&& Objects.equals(this.eventBusNotificationDurabilityEnabled,
that.eventBusNotificationDurabilityEnabled)
+ && Objects.equals(this.eventBusNotificationQueueAutoDelete,
that.eventBusNotificationQueueAutoDelete)
&& Objects.equals(this.vhost, that.vhost)
&& Objects.equals(this.taskQueueConsumerTimeout,
that.taskQueueConsumerTimeout)
&& Objects.equals(this.eventBusPropagateDispatchError,
that.eventBusPropagateDispatchError);
@@ -927,6 +945,6 @@ public class RabbitMQConfiguration {
public final int hashCode() {
return Objects.hash(uri, managementUri, maxRetries, minDelayInMs,
connectionTimeoutInMs, quorumQueueReplicationFactor, quorumQueueDeliveryLimit,
useQuorumQueues, hosts,
channelRpcTimeoutInMs, handshakeTimeoutInMs, shutdownTimeoutInMs,
networkRecoveryIntervalInMs, managementCredentials, useSsl, useSslManagement,
- sslConfiguration, queueTTL, eventBusPublishConfirmEnabled,
eventBusNotificationDurabilityEnabled, vhost, taskQueueConsumerTimeout,
eventBusPropagateDispatchError);
+ sslConfiguration, queueTTL, eventBusPublishConfirmEnabled,
eventBusNotificationDurabilityEnabled, eventBusNotificationQueueAutoDelete,
vhost, taskQueueConsumerTimeout, eventBusPropagateDispatchError);
}
}
diff --git a/docs/modules/servers/pages/distributed/configure/rabbitmq.adoc
b/docs/modules/servers/pages/distributed/configure/rabbitmq.adoc
index 3f183ed468..bc98d54f6a 100644
--- a/docs/modules/servers/pages/distributed/configure/rabbitmq.adoc
+++ b/docs/modules/servers/pages/distributed/configure/rabbitmq.adoc
@@ -51,4 +51,12 @@ Tips for choosing metricsEnabled are explained in
This is used only on queues used to share notification patterns, are exclusive
to a node. If omitted, it will not add the TTL configure when declaring queues.
Optional integer, defaults is 3600000.
+| notification.queue.autoDelete
+| Whether the per-node notification (key registration) queue is declared as
auto-delete. Optional boolean, defaults to true
+(historical behaviour). Forced to false when `quorum.queues.enable` is true,
as quorum queues do not support auto-delete.
+Auto-delete queues are removed by RabbitMQ as soon as their last consumer
disconnects, including on transient connection
+blips, causing key bindings to fail with 404 NOT_FOUND until the queue is
redeclared. Setting this to false, combined with
+`notification.queue.ttl` (x-expires), is more robust: the queue survives short
consumer outages and is only garbage
+collected once it has stayed unused for the TTL, which also cleans up the
queues of nodes that died abruptly.
+
|===
diff --git a/docs/modules/servers/partials/configure/rabbitmq.adoc
b/docs/modules/servers/partials/configure/rabbitmq.adoc
index 689bb17a57..3408aba3a2 100644
--- a/docs/modules/servers/partials/configure/rabbitmq.adoc
+++ b/docs/modules/servers/partials/configure/rabbitmq.adoc
@@ -105,6 +105,14 @@ A coma separated list of hosts, example:
hosts=ip1:5672,ip2:5672
| event.bus.notification.durability.enabled
| Whether or not the queue backing notifications should be durable. Optional
boolean, defaults to true.
+| notification.queue.autoDelete
+| Whether the per-node notification (key registration) queue is declared as
auto-delete. Optional boolean, defaults to true
+(historical behaviour). Forced to false when `quorum.queues.enable` is true,
as quorum queues do not support auto-delete.
+Auto-delete queues are removed by RabbitMQ as soon as their last consumer
disconnects, including on transient connection
+blips, causing key bindings to fail with 404 NOT_FOUND until the queue is
redeclared. Setting this to false, combined with
+`notification.queue.ttl` (x-expires), is more robust: the queue survives short
consumer outages and is only garbage
+collected once it has stayed unused for the TTL, which also cleans up the
queues of nodes that died abruptly.
+
| event.bus.propagate.dispatch.error
| Whether to propagate errors back to the callers when eventbus fails to
dispatch group events to RabbitMQ (then store the failed events in the event
dead letters).
Optional boolean, defaults to true.
diff --git
a/event-bus/distributed/src/main/java/org/apache/james/events/KeyRegistrationHandler.java
b/event-bus/distributed/src/main/java/org/apache/james/events/KeyRegistrationHandler.java
index 84ab3b88d3..cb1fe51742 100644
---
a/event-bus/distributed/src/main/java/org/apache/james/events/KeyRegistrationHandler.java
+++
b/event-bus/distributed/src/main/java/org/apache/james/events/KeyRegistrationHandler.java
@@ -19,7 +19,6 @@
package org.apache.james.events;
-import static org.apache.james.backends.rabbitmq.Constants.AUTO_DELETE;
import static org.apache.james.backends.rabbitmq.Constants.EXCLUSIVE;
import static org.apache.james.backends.rabbitmq.Constants.evaluateAutoDelete;
import static org.apache.james.backends.rabbitmq.Constants.evaluateDurable;
@@ -132,7 +131,7 @@ class KeyRegistrationHandler {
QueueSpecification.queue(registrationQueue.asString())
.durable(evaluateDurable(configuration.isEventBusNotificationDurabilityEnabled(),
configuration.isQuorumQueuesUsed()))
.exclusive(evaluateExclusive(!EXCLUSIVE,
configuration.isQuorumQueuesUsed()))
- .autoDelete(evaluateAutoDelete(AUTO_DELETE,
configuration.isQuorumQueuesUsed()))
+
.autoDelete(evaluateAutoDelete(configuration.isEventBusNotificationQueueAutoDeleteEnabled(),
configuration.isQuorumQueuesUsed()))
.arguments(builder.build()))
.timeout(TOPOLOGY_CHANGES_TIMEOUT)
.map(AMQP.Queue.DeclareOk::getQueue)
diff --git
a/server/apps/distributed-app/sample-configuration/rabbitmq.properties
b/server/apps/distributed-app/sample-configuration/rabbitmq.properties
index 9ad1edc38d..a370aeb381 100644
--- a/server/apps/distributed-app/sample-configuration/rabbitmq.properties
+++ b/server/apps/distributed-app/sample-configuration/rabbitmq.properties
@@ -98,3 +98,13 @@ task.consumption.enabled=true
# This is used only on queues used to share notification patterns, are
exclusive to a node. If omitted, it will not add the TTL configure when
declaring queues.
# Optional integer, defaults is 3600000.
#notification.queue.ttl=3600000
+
+# Boolean. Whether the per-node notification (key registration) queue is
declared as auto-delete.
+# Defaults to true, which preserves the historical behaviour. Forced to false
when quorum.queues.enable=true
+# (quorum queues do not support auto-delete).
+# Auto-delete queues are removed by RabbitMQ as soon as their last consumer
disconnects, including on transient
+# connection blips: the queue then briefly disappears and key bindings fail
with 404 NOT_FOUND until it is redeclared.
+# Setting this to false (combined with notification.queue.ttl above, i.e.
x-expires) is more robust: the queue
+# survives short consumer outages and is only garbage collected once it has
stayed unused for the TTL, which also
+# cleans up the queues of nodes that died abruptly.
+#notification.queue.autoDelete=false
diff --git
a/server/apps/postgres-app/sample-configuration-distributed/rabbitmq.properties
b/server/apps/postgres-app/sample-configuration-distributed/rabbitmq.properties
index 75a562af60..d148e1ee4d 100644
---
a/server/apps/postgres-app/sample-configuration-distributed/rabbitmq.properties
+++
b/server/apps/postgres-app/sample-configuration-distributed/rabbitmq.properties
@@ -93,3 +93,13 @@ task.consumption.enabled=true
# This is used only on queues used to share notification patterns, are
exclusive to a node. If omitted, it will not add the TTL configure when
declaring queues.
# Optional integer, defaults is 3600000.
#notification.queue.ttl=3600000
+
+# Boolean. Whether the per-node notification (key registration) queue is
declared as auto-delete.
+# Defaults to true, which preserves the historical behaviour. Forced to false
when quorum.queues.enable=true
+# (quorum queues do not support auto-delete).
+# Auto-delete queues are removed by RabbitMQ as soon as their last consumer
disconnects, including on transient
+# connection blips: the queue then briefly disappears and key bindings fail
with 404 NOT_FOUND until it is redeclared.
+# Setting this to false (combined with notification.queue.ttl above, i.e.
x-expires) is more robust: the queue
+# survives short consumer outages and is only garbage collected once it has
stayed unused for the TTL, which also
+# cleans up the queues of nodes that died abruptly.
+#notification.queue.autoDelete=false
diff --git a/src/site/xdoc/server/config-rabbitmq.xml
b/src/site/xdoc/server/config-rabbitmq.xml
index b51689e8db..513d3f7ef2 100644
--- a/src/site/xdoc/server/config-rabbitmq.xml
+++ b/src/site/xdoc/server/config-rabbitmq.xml
@@ -148,6 +148,14 @@
<dt><strong>event.bus.notification.durability.enabled</strong></dt>
<dd>Whether or not the queue backing notifications should be
durable. Optional boolean, defaults to true.</dd>
+ <dt><strong>notification.queue.autoDelete</strong></dt>
+ <dd>Whether the per-node notification (key registration) queue is
declared as auto-delete. Optional boolean, defaults to true
+ (historical behaviour). Forced to false when
<code>quorum.queues.enable</code> is true, as quorum queues do not support
auto-delete.
+ Auto-delete queues are removed by RabbitMQ as soon as their last
consumer disconnects, including on transient connection
+ blips, causing key bindings to fail with 404 NOT_FOUND until the
queue is redeclared. Setting this to false, combined with
+ <code>notification.queue.ttl</code> (x-expires), is more robust:
the queue survives short consumer outages and is only garbage
+ collected once it has stayed unused for the TTL, which also
cleans up the queues of nodes that died abruptly.</dd>
+
<dt><strong>event.bus.propagate.dispatch.error</strong></dt>
<dd>Whether to propagate errors back to the callers when eventbus
fails to dispatch group events to RabbitMQ (then store the failed events in the
event dead letters).
Optional boolean, defaults to true.</dd>
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
