You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@james.apache.org by rc...@apache.org on 2024/04/22 02:26:02 UTC
(james-project) 04/10: JAMES-4027 Adapt some queue configuration following quorum queue usage
This is an automated email from the ASF dual-hosted git repository.
rcordier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit c322810669674360544f40ca0ace6d1d02c13119
Author: Quan Tran <hq...@linagora.com>
AuthorDate: Mon Apr 8 16:32:28 2024 +0700
JAMES-4027 Adapt some queue configuration following quorum queue usage
- Quorum queue does not support `autoDelete` queue
- Quorum queue does not support `non-durable` queue
- Quorum queue does not support `exclusive` queue
Upon quorum queue enabled, we would enforce the above config to be aligned with quorum queue constraint
---
.../apache/james/backends/rabbitmq/Constants.java | 21 ++++
.../backends/rabbitmq/RabbitMQConfiguration.java | 4 +
.../james/backends/rabbitmq/ConstantsTest.java | 114 +++++++++++++++++++++
.../org/apache/james/events/EventDispatcher.java | 9 +-
.../apache/james/events/GroupConsumerRetry.java | 5 +-
.../org/apache/james/events/GroupRegistration.java | 11 +-
.../james/events/GroupRegistrationHandler.java | 9 +-
.../james/events/KeyReconnectionHandler.java | 9 +-
.../james/events/KeyRegistrationHandler.java | 9 +-
.../queue/rabbitmq/RabbitMQMailQueueFactory.java | 15 +--
.../distributed/RabbitMQTerminationSubscriber.java | 8 +-
.../distributed/RabbitMQWorkQueue.java | 8 +-
.../TerminationReconnectionHandler.java | 9 +-
13 files changed, 205 insertions(+), 26 deletions(-)
diff --git a/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/Constants.java b/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/Constants.java
index d56b8846b3..8787451ee8 100644
--- a/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/Constants.java
+++ b/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/Constants.java
@@ -40,4 +40,25 @@ public interface Constants {
String DIRECT_EXCHANGE = "direct";
AMQP.BasicProperties NO_PROPERTIES = new AMQP.BasicProperties();
+
+ static boolean evaluateAutoDelete(boolean autoDeleteWanted, boolean useQuorumQueue) {
+ if (autoDeleteWanted && useQuorumQueue) {
+ return !AUTO_DELETE; // quorum queue does not support autoDelete
+ }
+ return autoDeleteWanted;
+ }
+
+ static boolean evaluateDurable(boolean durableWanted, boolean useQuorumQueue) {
+ if (!durableWanted && useQuorumQueue) {
+ return DURABLE; // quorum queue only supports durable queue
+ }
+ return durableWanted;
+ }
+
+ static boolean evaluateExclusive(boolean exclusiveWanted, boolean useQuorumQueue) {
+ if (exclusiveWanted && useQuorumQueue) {
+ return !EXCLUSIVE; // quorum queue does not support exclusive
+ }
+ return exclusiveWanted;
+ }
}
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 2cd50d1f23..8416e76d08 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
@@ -810,6 +810,10 @@ public class RabbitMQConfiguration {
return builder;
}
+ public boolean isQuorumQueuesUsed() {
+ return useQuorumQueues;
+ }
+
public List<Host> rabbitMQHosts() {
return hosts;
}
diff --git a/backends-common/rabbitmq/src/test/java/org/apache/james/backends/rabbitmq/ConstantsTest.java b/backends-common/rabbitmq/src/test/java/org/apache/james/backends/rabbitmq/ConstantsTest.java
new file mode 100644
index 0000000000..235710fea0
--- /dev/null
+++ b/backends-common/rabbitmq/src/test/java/org/apache/james/backends/rabbitmq/ConstantsTest.java
@@ -0,0 +1,114 @@
+/****************************************************************
+ * Licensed to the Apache Software Foundation (ASF) under one *
+ * or more contributor license agreements. See the NOTICE file *
+ * distributed with this work for additional information *
+ * regarding copyright ownership. The ASF licenses this file *
+ * to you under the Apache License, Version 2.0 (the *
+ * "License"); you may not use this file except in compliance *
+ * with the License. You may obtain a copy of the License at *
+ * *
+ * http://www.apache.org/licenses/LICENSE-2.0 *
+ * *
+ * Unless required by applicable law or agreed to in writing, *
+ * software distributed under the License is distributed on an *
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY *
+ * KIND, either express or implied. See the License for the *
+ * specific language governing permissions and limitations *
+ * under the License. *
+ ****************************************************************/
+
+package org.apache.james.backends.rabbitmq;
+
+import static org.apache.james.backends.rabbitmq.Constants.AUTO_DELETE;
+import static org.apache.james.backends.rabbitmq.Constants.DURABLE;
+import static org.apache.james.backends.rabbitmq.Constants.EXCLUSIVE;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+
+class ConstantsTest {
+ private static final boolean NO_QUORUM_QUEUES = false;
+ private static final boolean USE_QUORUM_QUEUES = true;
+
+ @Nested
+ class EvaluateAutoDelete {
+ @Test
+ void wantedAutoDeleteAndNotQuorumQueueShouldReturnTrue() {
+ assertThat(Constants.evaluateAutoDelete(AUTO_DELETE, NO_QUORUM_QUEUES))
+ .isTrue();
+ }
+
+ @Test
+ void wantedAutoDeleteAndUseQuorumQueueShouldReturnFalse() {
+ assertThat(Constants.evaluateAutoDelete(AUTO_DELETE, USE_QUORUM_QUEUES))
+ .isFalse();
+ }
+
+ @Test
+ void notWantedAutoDeleteAndNoQuorumQueueShouldReturnFalse() {
+ assertThat(Constants.evaluateAutoDelete(!AUTO_DELETE, NO_QUORUM_QUEUES))
+ .isFalse();
+ }
+
+ @Test
+ void notWantedAutoDeleteAndUseQuorumQueueShouldReturnFalse() {
+ assertThat(Constants.evaluateAutoDelete(!AUTO_DELETE, USE_QUORUM_QUEUES))
+ .isFalse();
+ }
+ }
+
+ @Nested
+ class EvaluateDurable {
+ @Test
+ void wantedDurableAndNotQuorumQueueShouldReturnTrue() {
+ assertThat(Constants.evaluateDurable(DURABLE, NO_QUORUM_QUEUES))
+ .isTrue();
+ }
+
+ @Test
+ void wantedDurableAndUseQuorumQueueShouldReturnTrue() {
+ assertThat(Constants.evaluateDurable(DURABLE, USE_QUORUM_QUEUES))
+ .isTrue();
+ }
+
+ @Test
+ void notWantedDurableAndNoQuorumQueueShouldReturnFalse() {
+ assertThat(Constants.evaluateDurable(!DURABLE, NO_QUORUM_QUEUES))
+ .isFalse();
+ }
+
+ @Test
+ void notWantedDurableAndUseQuorumQueueShouldReturnTrue() {
+ assertThat(Constants.evaluateDurable(!DURABLE, USE_QUORUM_QUEUES))
+ .isTrue();
+ }
+ }
+
+ @Nested
+ class EvaluateExclusive {
+ @Test
+ void wantedExclusiveAndNotQuorumQueueShouldReturnTrue() {
+ assertThat(Constants.evaluateExclusive(EXCLUSIVE, NO_QUORUM_QUEUES))
+ .isTrue();
+ }
+
+ @Test
+ void wantedExclusiveAndUseQuorumQueueShouldReturnFalse() {
+ assertThat(Constants.evaluateExclusive(EXCLUSIVE, USE_QUORUM_QUEUES))
+ .isFalse();
+ }
+
+ @Test
+ void notWantedExclusiveAndNoQuorumQueueShouldReturnFalse() {
+ assertThat(Constants.evaluateExclusive(!EXCLUSIVE, NO_QUORUM_QUEUES))
+ .isFalse();
+ }
+
+ @Test
+ void notWantedExclusiveAndUseQuorumQueueShouldReturnFalse() {
+ assertThat(Constants.evaluateExclusive(!EXCLUSIVE, USE_QUORUM_QUEUES))
+ .isFalse();
+ }
+ }
+}
diff --git a/event-bus/distributed/src/main/java/org/apache/james/events/EventDispatcher.java b/event-bus/distributed/src/main/java/org/apache/james/events/EventDispatcher.java
index fb7dc8a5d7..3282389445 100644
--- a/event-bus/distributed/src/main/java/org/apache/james/events/EventDispatcher.java
+++ b/event-bus/distributed/src/main/java/org/apache/james/events/EventDispatcher.java
@@ -25,6 +25,9 @@ import static org.apache.james.backends.rabbitmq.Constants.DIRECT_EXCHANGE;
import static org.apache.james.backends.rabbitmq.Constants.DURABLE;
import static org.apache.james.backends.rabbitmq.Constants.EMPTY_ROUTING_KEY;
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;
+import static org.apache.james.backends.rabbitmq.Constants.evaluateExclusive;
import static org.apache.james.events.RabbitMQEventBus.EVENT_BUS_ID;
import java.time.Duration;
@@ -97,9 +100,9 @@ public class EventDispatcher {
.durable(DURABLE)
.type(DIRECT_EXCHANGE)),
sender.declareQueue(namingStrategy.deadLetterQueue()
- .durable(DURABLE)
- .exclusive(!EXCLUSIVE)
- .autoDelete(!AUTO_DELETE)
+ .durable(evaluateDurable(DURABLE, configuration.isQuorumQueuesUsed()))
+ .exclusive(evaluateExclusive(!EXCLUSIVE, configuration.isQuorumQueuesUsed()))
+ .autoDelete(evaluateAutoDelete(!AUTO_DELETE, configuration.isQuorumQueuesUsed()))
.arguments(configuration.workQueueArgumentsBuilder()
.build())),
sender.bind(BindingSpecification.binding()
diff --git a/event-bus/distributed/src/main/java/org/apache/james/events/GroupConsumerRetry.java b/event-bus/distributed/src/main/java/org/apache/james/events/GroupConsumerRetry.java
index 545fbe3fe3..ff37624277 100644
--- a/event-bus/distributed/src/main/java/org/apache/james/events/GroupConsumerRetry.java
+++ b/event-bus/distributed/src/main/java/org/apache/james/events/GroupConsumerRetry.java
@@ -25,6 +25,7 @@ import static org.apache.james.backends.rabbitmq.Constants.DURABLE;
import static org.apache.james.backends.rabbitmq.Constants.EMPTY_ROUTING_KEY;
import static org.apache.james.events.GroupRegistration.RETRY_COUNT;
+import org.apache.james.backends.rabbitmq.RabbitMQConfiguration;
import org.apache.james.util.MDCStructuredLogger;
import org.apache.james.util.StructuredLogger;
import org.slf4j.Logger;
@@ -63,10 +64,12 @@ class GroupConsumerRetry {
private final EventDeadLetters eventDeadLetters;
private final Group group;
private final EventSerializer eventSerializer;
+ private final RabbitMQConfiguration rabbitMQConfiguration;
GroupConsumerRetry(NamingStrategy namingStrategy, Sender sender, Group group, RetryBackoffConfiguration retryBackoff,
- EventDeadLetters eventDeadLetters, EventSerializer eventSerializer) {
+ EventDeadLetters eventDeadLetters, EventSerializer eventSerializer, RabbitMQConfiguration rabbitMQConfiguration) {
this.sender = sender;
+ this.rabbitMQConfiguration = rabbitMQConfiguration;
this.retryExchangeName = namingStrategy.retryExchange(group);
this.retryBackoff = retryBackoff;
this.eventDeadLetters = eventDeadLetters;
diff --git a/event-bus/distributed/src/main/java/org/apache/james/events/GroupRegistration.java b/event-bus/distributed/src/main/java/org/apache/james/events/GroupRegistration.java
index 15c561a18e..eb0e55a194 100644
--- a/event-bus/distributed/src/main/java/org/apache/james/events/GroupRegistration.java
+++ b/event-bus/distributed/src/main/java/org/apache/james/events/GroupRegistration.java
@@ -23,6 +23,9 @@ import static org.apache.james.backends.rabbitmq.Constants.AUTO_DELETE;
import static org.apache.james.backends.rabbitmq.Constants.DURABLE;
import static org.apache.james.backends.rabbitmq.Constants.EXCLUSIVE;
import static org.apache.james.backends.rabbitmq.Constants.REQUEUE;
+import static org.apache.james.backends.rabbitmq.Constants.evaluateAutoDelete;
+import static org.apache.james.backends.rabbitmq.Constants.evaluateDurable;
+import static org.apache.james.backends.rabbitmq.Constants.evaluateExclusive;
import java.util.Objects;
import java.util.Optional;
@@ -102,7 +105,7 @@ class GroupRegistration implements Registration {
this.listenerExecutor = listenerExecutor;
this.receiverSubscriber = Optional.empty();
this.unregisterGroup = unregisterGroup;
- this.retryHandler = new GroupConsumerRetry(namingStrategy, sender, group, retryBackoff, eventDeadLetters, eventSerializer);
+ this.retryHandler = new GroupConsumerRetry(namingStrategy, sender, group, retryBackoff, eventDeadLetters, eventSerializer, configuration);
this.delayGenerator = WaitDelayGenerator.of(retryBackoff);
this.group = group;
}
@@ -129,9 +132,9 @@ class GroupRegistration implements Registration {
private Mono<Void> createGroupWorkQueue() {
return channelPool.createWorkQueue(
QueueSpecification.queue(queueName.asString())
- .durable(DURABLE)
- .exclusive(!EXCLUSIVE)
- .autoDelete(!AUTO_DELETE)
+ .durable(evaluateDurable(DURABLE, configuration.isQuorumQueuesUsed()))
+ .exclusive(evaluateExclusive(!EXCLUSIVE, configuration.isQuorumQueuesUsed()))
+ .autoDelete(evaluateAutoDelete(!AUTO_DELETE, configuration.isQuorumQueuesUsed()))
.arguments(configuration.workQueueArgumentsBuilder()
.deadLetter(namingStrategy.deadLetterExchange())
.build()));
diff --git a/event-bus/distributed/src/main/java/org/apache/james/events/GroupRegistrationHandler.java b/event-bus/distributed/src/main/java/org/apache/james/events/GroupRegistrationHandler.java
index 44d19b88f1..e5dae70668 100644
--- a/event-bus/distributed/src/main/java/org/apache/james/events/GroupRegistrationHandler.java
+++ b/event-bus/distributed/src/main/java/org/apache/james/events/GroupRegistrationHandler.java
@@ -24,6 +24,9 @@ import static org.apache.james.backends.rabbitmq.Constants.DURABLE;
import static org.apache.james.backends.rabbitmq.Constants.EMPTY_ROUTING_KEY;
import static org.apache.james.backends.rabbitmq.Constants.EXCLUSIVE;
import static org.apache.james.backends.rabbitmq.Constants.REQUEUE;
+import static org.apache.james.backends.rabbitmq.Constants.evaluateAutoDelete;
+import static org.apache.james.backends.rabbitmq.Constants.evaluateDurable;
+import static org.apache.james.backends.rabbitmq.Constants.evaluateExclusive;
import static org.apache.james.events.GroupRegistration.DEFAULT_RETRY_COUNT;
import java.util.Collection;
@@ -105,9 +108,9 @@ class GroupRegistrationHandler {
public void start() {
channelPool.createWorkQueue(
QueueSpecification.queue(queueName.asString())
- .durable(DURABLE)
- .exclusive(!EXCLUSIVE)
- .autoDelete(!AUTO_DELETE)
+ .durable(evaluateDurable(DURABLE, configuration.isQuorumQueuesUsed()))
+ .exclusive(evaluateExclusive(!EXCLUSIVE, configuration.isQuorumQueuesUsed()))
+ .autoDelete(evaluateAutoDelete(!AUTO_DELETE, configuration.isQuorumQueuesUsed()))
.arguments(configuration.workQueueArgumentsBuilder()
.deadLetter(namingStrategy.deadLetterExchange())
.build()),
diff --git a/event-bus/distributed/src/main/java/org/apache/james/events/KeyReconnectionHandler.java b/event-bus/distributed/src/main/java/org/apache/james/events/KeyReconnectionHandler.java
index 2a09e77bb8..34cf26046e 100644
--- a/event-bus/distributed/src/main/java/org/apache/james/events/KeyReconnectionHandler.java
+++ b/event-bus/distributed/src/main/java/org/apache/james/events/KeyReconnectionHandler.java
@@ -22,6 +22,9 @@ package org.apache.james.events;
import static org.apache.james.backends.rabbitmq.Constants.AUTO_DELETE;
import static org.apache.james.backends.rabbitmq.Constants.DURABLE;
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;
+import static org.apache.james.backends.rabbitmq.Constants.evaluateExclusive;
import jakarta.inject.Inject;
@@ -57,7 +60,11 @@ public class KeyReconnectionHandler implements SimpleConnectionPool.Reconnection
try (Channel channel = connection.createChannel()) {
QueueArguments.Builder builder = configuration.workQueueArgumentsBuilder();
configuration.getQueueTTL().ifPresent(builder::queueTTL);
- channel.queueDeclare(namingStrategy.queueName(eventBusId).asString(), DURABLE, !EXCLUSIVE, AUTO_DELETE, builder.build());
+ channel.queueDeclare(namingStrategy.queueName(eventBusId).asString(),
+ evaluateDurable(DURABLE, configuration.isQuorumQueuesUsed()),
+ evaluateExclusive(!EXCLUSIVE, configuration.isQuorumQueuesUsed()),
+ evaluateAutoDelete(AUTO_DELETE, configuration.isQuorumQueuesUsed()),
+ builder.build());
} catch (Exception e) {
LOGGER.error("Error recovering connection", e);
}
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 8f31c2f0c5..d134238579 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
@@ -21,6 +21,9 @@ 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;
+import static org.apache.james.backends.rabbitmq.Constants.evaluateExclusive;
import static org.apache.james.events.RabbitMQEventBus.EVENT_BUS_ID;
import java.time.Duration;
@@ -126,9 +129,9 @@ class KeyRegistrationHandler {
configuration.getQueueTTL().ifPresent(builder::queueTTL);
sender.declareQueue(
QueueSpecification.queue(registrationQueue.asString())
- .durable(configuration.isEventBusNotificationDurabilityEnabled())
- .exclusive(!EXCLUSIVE)
- .autoDelete(AUTO_DELETE)
+ .durable(evaluateDurable(configuration.isEventBusNotificationDurabilityEnabled(), configuration.isQuorumQueuesUsed()))
+ .exclusive(evaluateExclusive(!EXCLUSIVE, configuration.isQuorumQueuesUsed()))
+ .autoDelete(evaluateAutoDelete(AUTO_DELETE, configuration.isQuorumQueuesUsed()))
.arguments(builder.build()))
.timeout(TOPOLOGY_CHANGES_TIMEOUT)
.map(AMQP.Queue.DeclareOk::getQueue)
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueFactory.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueFactory.java
index d55b0296ee..82c3f1afef 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueFactory.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueFactory.java
@@ -23,6 +23,9 @@ import static org.apache.james.backends.rabbitmq.Constants.AUTO_DELETE;
import static org.apache.james.backends.rabbitmq.Constants.DURABLE;
import static org.apache.james.backends.rabbitmq.Constants.EMPTY_ROUTING_KEY;
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;
+import static org.apache.james.backends.rabbitmq.Constants.evaluateExclusive;
import static org.apache.james.queue.api.MailQueue.QUEUE_SIZE_METRIC_NAME_PREFIX;
import java.time.Clock;
@@ -163,16 +166,16 @@ public class RabbitMQMailQueueFactory implements MailQueueFactory<RabbitMQMailQu
.durable(true)
.type("direct")),
sender.declareQueue(QueueSpecification.queue(mailQueueName.toWorkQueueName().asString())
- .durable(DURABLE)
- .exclusive(!EXCLUSIVE)
- .autoDelete(!AUTO_DELETE)
+ .durable(evaluateDurable(DURABLE, configuration.isQuorumQueuesUsed()))
+ .exclusive(evaluateExclusive(!EXCLUSIVE, configuration.isQuorumQueuesUsed()))
+ .autoDelete(evaluateAutoDelete(!AUTO_DELETE, configuration.isQuorumQueuesUsed()))
.arguments(configuration.workQueueArgumentsBuilder()
.deadLetter(mailQueueName.toDeadLetterExchangeName())
.build())),
sender.declareQueue(QueueSpecification.queue(mailQueueName.toDeadLetterQueueName())
- .durable(DURABLE)
- .exclusive(!EXCLUSIVE)
- .autoDelete(!AUTO_DELETE)
+ .durable(evaluateDurable(DURABLE, configuration.isQuorumQueuesUsed()))
+ .exclusive(evaluateExclusive(!EXCLUSIVE, configuration.isQuorumQueuesUsed()))
+ .autoDelete(evaluateAutoDelete(!AUTO_DELETE, configuration.isQuorumQueuesUsed()))
.arguments(configuration.workQueueArgumentsBuilder()
.build())),
sender.bind(BindingSpecification.binding()
diff --git a/server/task/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/RabbitMQTerminationSubscriber.java b/server/task/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/RabbitMQTerminationSubscriber.java
index e7232a7b5e..511c93c2dd 100644
--- a/server/task/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/RabbitMQTerminationSubscriber.java
+++ b/server/task/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/RabbitMQTerminationSubscriber.java
@@ -21,6 +21,8 @@ package org.apache.james.task.eventsourcing.distributed;
import static org.apache.james.backends.rabbitmq.Constants.AUTO_DELETE;
import static org.apache.james.backends.rabbitmq.Constants.DURABLE;
+import static org.apache.james.backends.rabbitmq.Constants.evaluateAutoDelete;
+import static org.apache.james.backends.rabbitmq.Constants.evaluateDurable;
import static org.apache.james.util.ReactorUtils.publishIfPresent;
import static reactor.core.publisher.Sinks.EmitFailureHandler.FAIL_FAST;
@@ -86,7 +88,11 @@ public class RabbitMQTerminationSubscriber implements TerminationSubscriber, Sta
sender.declareExchange(ExchangeSpecification.exchange(EXCHANGE_NAME)).block();
QueueArguments.Builder builder = rabbitMQConfiguration.workQueueArgumentsBuilder();
rabbitMQConfiguration.getQueueTTL().ifPresent(builder::queueTTL);
- sender.declare(QueueSpecification.queue(queueName.asString()).durable(!DURABLE).autoDelete(!AUTO_DELETE).arguments(builder.build())).block();
+ sender.declare(QueueSpecification.queue(queueName.asString())
+ .durable(evaluateDurable(!DURABLE, rabbitMQConfiguration.isQuorumQueuesUsed()))
+ .autoDelete(evaluateAutoDelete(!AUTO_DELETE, rabbitMQConfiguration.isQuorumQueuesUsed()))
+ .arguments(builder.build()))
+ .block();
sender.bind(BindingSpecification.binding(EXCHANGE_NAME, ROUTING_KEY, queueName.asString())).block();
sendQueue = Sinks.many().unicast().onBackpressureBuffer();
sendQueueHandle = sender
diff --git a/server/task/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/RabbitMQWorkQueue.java b/server/task/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/RabbitMQWorkQueue.java
index acdcfbc692..a8402385bc 100644
--- a/server/task/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/RabbitMQWorkQueue.java
+++ b/server/task/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/RabbitMQWorkQueue.java
@@ -23,6 +23,8 @@ import static com.rabbitmq.client.MessageProperties.PERSISTENT_TEXT_PLAIN;
import static org.apache.james.backends.rabbitmq.Constants.AUTO_DELETE;
import static org.apache.james.backends.rabbitmq.Constants.DURABLE;
import static org.apache.james.backends.rabbitmq.Constants.REQUEUE;
+import static org.apache.james.backends.rabbitmq.Constants.evaluateAutoDelete;
+import static org.apache.james.backends.rabbitmq.Constants.evaluateDurable;
import static reactor.core.publisher.Sinks.EmitFailureHandler.FAIL_FAST;
import java.io.IOException;
@@ -124,7 +126,7 @@ public class RabbitMQWorkQueue implements WorkQueue {
.retryWhen(Retry.backoff(NUM_RETRIES, FIRST_BACKOFF));
Mono<AMQP.Queue.DeclareOk> declareQueue = sender
.declare(QueueSpecification.queue(QUEUE_NAME)
- .durable(true)
+ .durable(evaluateDurable(DURABLE, rabbitMQConfiguration.isQuorumQueuesUsed()))
.arguments(rabbitMQConfiguration.workQueueArgumentsBuilder()
.singleActiveConsumer()
.consumerTimeout(rabbitMQConfiguration.getTaskQueueConsumerTimeout().toMillis())
@@ -199,8 +201,8 @@ public class RabbitMQWorkQueue implements WorkQueue {
QueueArguments.Builder builder = rabbitMQConfiguration.workQueueArgumentsBuilder();
rabbitMQConfiguration.getQueueTTL().ifPresent(builder::queueTTL);
QueueSpecification specification = QueueSpecification.queue(cancelRequestQueueName.asString())
- .durable(!DURABLE)
- .autoDelete(AUTO_DELETE)
+ .durable(evaluateDurable(!DURABLE, rabbitMQConfiguration.isQuorumQueuesUsed()))
+ .autoDelete(evaluateAutoDelete(AUTO_DELETE, rabbitMQConfiguration.isQuorumQueuesUsed()))
.arguments(builder.build());
sender.declare(specification).block();
sender.bind(BindingSpecification.binding(CANCEL_REQUESTS_EXCHANGE_NAME, CANCEL_REQUESTS_ROUTING_KEY, cancelRequestQueueName.asString())).block();
diff --git a/server/task/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/TerminationReconnectionHandler.java b/server/task/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/TerminationReconnectionHandler.java
index 006af8af39..f21e4061fa 100644
--- a/server/task/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/TerminationReconnectionHandler.java
+++ b/server/task/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/TerminationReconnectionHandler.java
@@ -22,6 +22,9 @@ package org.apache.james.task.eventsourcing.distributed;
import static org.apache.james.backends.rabbitmq.Constants.AUTO_DELETE;
import static org.apache.james.backends.rabbitmq.Constants.DURABLE;
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;
+import static org.apache.james.backends.rabbitmq.Constants.evaluateExclusive;
import jakarta.inject.Inject;
@@ -61,7 +64,11 @@ public class TerminationReconnectionHandler implements SimpleConnectionPool.Reco
try (Channel channel = connection.createChannel()) {
QueueArguments.Builder builder = QueueArguments.builder();
configuration.getQueueTTL().ifPresent(builder::queueTTL);
- channel.queueDeclare(queueName.asString(), !DURABLE, !EXCLUSIVE, !AUTO_DELETE, builder.build());
+ channel.queueDeclare(queueName.asString(),
+ evaluateDurable(!DURABLE, configuration.isQuorumQueuesUsed()),
+ evaluateExclusive(!EXCLUSIVE, configuration.isQuorumQueuesUsed()),
+ evaluateAutoDelete(!AUTO_DELETE, configuration.isQuorumQueuesUsed()),
+ builder.build());
} catch (Exception e) {
LOGGER.error("Error recovering connection", e);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org