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/05/14 02:31:15 UTC
(james-project) branch 3.8.x updated (6b89b97761 -> 0c5196a5a7)
This is an automated email from the ASF dual-hosted git repository.
rcordier pushed a change to branch 3.8.x
in repository https://gitbox.apache.org/repos/asf/james-project.git
from 6b89b97761 JAMES-4023 Avoid blocking the SMTP Netty event loop (#2230)
new 13d7f3e7d4 JAMES-3955 Check for queues without consumer and resume consumption as needed
new 21adf2146b JAMES-3955 Applicative timeouts before Rabbit timeouts
new 040601a767 JAMES-3955 WARNING logs upon closing channels
new 87d4f0d1af JAMES-3955 Increase consumer timeout for TaskManagerWorkQueue
new 0c5196a5a7 [FIX] Solve weave/rest-smtp-sink: Docker image manifest v2 schema 1 deprecation issue (#2152)
The 5 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../james/backends/rabbitmq/QueueArguments.java | 5 ++
.../backends/rabbitmq/RabbitMQConfiguration.java | 33 +++++++-
.../rabbitmq/ReactorRabbitMQChannelPool.java | 8 ++
.../rabbitmq/RabbitMQConfigurationTest.java | 55 ++++++++++++++
.../org/apache/james/events/ListenerExecutor.java | 9 ++-
.../events/RabbitEventBusConsumerHealthCheck.java | 85 +++++++++++++++++++++
.../modules/ROOT/pages/configure/rabbitmq.adoc | 15 ++++
.../sample-configuration/rabbitmq.properties | 7 +-
.../modules/DistributedTaskManagerModule.java | 6 ++
.../james/modules/event/JMAPEventBusModule.java | 8 ++
.../modules/event/RabbitMQEventBusModule.java | 7 ++
.../modules/queue/rabbitmq/RabbitMQModule.java | 2 +
...itMQWebAdminServerIntegrationImmutableTest.java | 3 +-
.../org/apache/james/queue/rabbitmq/Dequeuer.java | 6 +-
.../RabbitMQMailQueueConsumerHealthCheck.java | 87 ++++++++++++++++++++++
.../DistributedTaskManagerHealthCheck.java} | 55 ++++++++------
.../distributed/RabbitMQWorkQueue.java | 14 ++--
.../RabbitMQWorkQueueReconnectionHandler.java | 29 +-------
.../distributed/RabbitMQWorkQueueTest.java | 12 ++-
.../java/org/apache/james/util/docker/Images.java | 2 +-
src/site/xdoc/server/config-rabbitmq.xml | 10 +++
upgrade-instructions.md | 16 ++++
22 files changed, 408 insertions(+), 66 deletions(-)
create mode 100644 event-bus/distributed/src/main/java/org/apache/james/events/RabbitEventBusConsumerHealthCheck.java
create mode 100644 server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueConsumerHealthCheck.java
copy server/{queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueDeadLetterQueueHealthCheck.java => task/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/DistributedTaskManagerHealthCheck.java} (51%)
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
(james-project) 01/05: JAMES-3955 Check for queues without consumer and resume consumption as needed
Posted by rc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rcordier pushed a commit to branch 3.8.x
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 13d7f3e7d4bb4dc6ed9b048e38fef93c2513b2af
Author: Benoit TELLIER <bt...@linagora.com>
AuthorDate: Fri Nov 3 18:17:22 2023 +0100
JAMES-3955 Check for queues without consumer and resume consumption as needed
---
.../events/RabbitEventBusConsumerHealthCheck.java | 85 +++++++++++++++++++++
.../sample-configuration/rabbitmq.properties | 2 +-
.../modules/DistributedTaskManagerModule.java | 6 ++
.../james/modules/event/JMAPEventBusModule.java | 8 ++
.../modules/event/RabbitMQEventBusModule.java | 7 ++
.../modules/queue/rabbitmq/RabbitMQModule.java | 2 +
...itMQWebAdminServerIntegrationImmutableTest.java | 3 +-
.../RabbitMQMailQueueConsumerHealthCheck.java | 87 ++++++++++++++++++++++
.../DistributedTaskManagerHealthCheck.java | 77 +++++++++++++++++++
.../distributed/RabbitMQWorkQueue.java | 13 ++--
.../RabbitMQWorkQueueReconnectionHandler.java | 29 +-------
11 files changed, 283 insertions(+), 36 deletions(-)
diff --git a/event-bus/distributed/src/main/java/org/apache/james/events/RabbitEventBusConsumerHealthCheck.java b/event-bus/distributed/src/main/java/org/apache/james/events/RabbitEventBusConsumerHealthCheck.java
new file mode 100644
index 0000000000..ceb3c8b106
--- /dev/null
+++ b/event-bus/distributed/src/main/java/org/apache/james/events/RabbitEventBusConsumerHealthCheck.java
@@ -0,0 +1,85 @@
+/****************************************************************
+ * 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.events;
+
+import java.util.Optional;
+import java.util.stream.Stream;
+
+import org.apache.james.backends.rabbitmq.SimpleConnectionPool;
+import org.apache.james.core.healthcheck.ComponentName;
+import org.apache.james.core.healthcheck.HealthCheck;
+import org.apache.james.core.healthcheck.Result;
+import org.apache.james.util.ReactorUtils;
+
+import com.github.fge.lambdas.Throwing;
+import com.rabbitmq.client.Channel;
+
+import reactor.core.publisher.Mono;
+
+public class RabbitEventBusConsumerHealthCheck implements HealthCheck {
+ public static final ComponentName COMPONENT_NAME = new ComponentName("EventbusConsumersHealthCheck");
+ public static final String COMPONENT = "EventbusConsumers";
+
+ private final RabbitMQEventBus eventBus;
+ private final NamingStrategy namingStrategy;
+ private final SimpleConnectionPool connectionPool;
+
+ public RabbitEventBusConsumerHealthCheck(RabbitMQEventBus eventBus, NamingStrategy namingStrategy,
+ SimpleConnectionPool connectionPool) {
+ this.eventBus = eventBus;
+ this.namingStrategy = namingStrategy;
+ this.connectionPool = connectionPool;
+ }
+
+ @Override
+ public ComponentName componentName() {
+ return new ComponentName(COMPONENT + "-" + namingStrategy.getEventBusName().value());
+ }
+
+ @Override
+ public Mono<Result> check() {
+ return connectionPool.getResilientConnection()
+ .map(Throwing.function(connection -> {
+
+ try (Channel channel = connection.createChannel()) {
+ return check(channel);
+ }
+ })).subscribeOn(ReactorUtils.BLOCKING_CALL_WRAPPER);
+ }
+
+ private Result check(Channel channel) {
+ Stream<Group> groups = Stream.concat(
+ eventBus.listRegisteredGroups().stream(),
+ Stream.of(new GroupRegistrationHandler.GroupRegistrationHandlerGroup()));
+
+ Optional<String> queueWithoutConsumers = groups
+ .map(namingStrategy::workQueue)
+ .map(GroupRegistration.WorkQueueName::asString)
+ .filter(Throwing.predicate(queue -> channel.consumerCount(queue) == 0))
+ .findAny();
+
+ if (queueWithoutConsumers.isPresent()) {
+ eventBus.restart();
+ return Result.degraded(componentName(), "No consumers on " + queueWithoutConsumers.get());
+ } else {
+ return Result.healthy(componentName());
+ }
+ }
+}
diff --git a/server/apps/distributed-app/sample-configuration/rabbitmq.properties b/server/apps/distributed-app/sample-configuration/rabbitmq.properties
index c2b3387ccf..1ad27e7aa1 100644
--- a/server/apps/distributed-app/sample-configuration/rabbitmq.properties
+++ b/server/apps/distributed-app/sample-configuration/rabbitmq.properties
@@ -87,4 +87,4 @@ task.consumption.enabled=true
# Configure queue ttl (in ms). References: https://www.rabbitmq.com/ttl.html#queue-ttl.
# 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
\ No newline at end of file
+#notification.queue.ttl=3600000
diff --git a/server/container/guice/distributed/src/main/java/org/apache/james/modules/DistributedTaskManagerModule.java b/server/container/guice/distributed/src/main/java/org/apache/james/modules/DistributedTaskManagerModule.java
index fadcbf2530..fae6ec304f 100644
--- a/server/container/guice/distributed/src/main/java/org/apache/james/modules/DistributedTaskManagerModule.java
+++ b/server/container/guice/distributed/src/main/java/org/apache/james/modules/DistributedTaskManagerModule.java
@@ -30,6 +30,7 @@ import org.apache.commons.configuration2.Configuration;
import org.apache.commons.configuration2.ex.ConfigurationException;
import org.apache.james.backends.cassandra.components.CassandraModule;
import org.apache.james.backends.rabbitmq.SimpleConnectionPool;
+import org.apache.james.core.healthcheck.HealthCheck;
import org.apache.james.modules.server.HostnameModule;
import org.apache.james.modules.server.TaskSerializationModule;
import org.apache.james.task.TaskManager;
@@ -40,6 +41,7 @@ import org.apache.james.task.eventsourcing.WorkQueueSupplier;
import org.apache.james.task.eventsourcing.cassandra.CassandraTaskExecutionDetailsProjection;
import org.apache.james.task.eventsourcing.cassandra.CassandraTaskExecutionDetailsProjectionModule;
import org.apache.james.task.eventsourcing.distributed.CancelRequestQueueName;
+import org.apache.james.task.eventsourcing.distributed.DistributedTaskManagerHealthCheck;
import org.apache.james.task.eventsourcing.distributed.RabbitMQTerminationSubscriber;
import org.apache.james.task.eventsourcing.distributed.RabbitMQWorkQueue;
import org.apache.james.task.eventsourcing.distributed.RabbitMQWorkQueueConfiguration;
@@ -82,6 +84,10 @@ public class DistributedTaskManagerModule extends AbstractModule {
Multibinder<SimpleConnectionPool.ReconnectionHandler> reconnectionHandlerMultibinder = Multibinder.newSetBinder(binder(), SimpleConnectionPool.ReconnectionHandler.class);
reconnectionHandlerMultibinder.addBinding().to(RabbitMQWorkQueueReconnectionHandler.class);
reconnectionHandlerMultibinder.addBinding().to(TerminationReconnectionHandler.class);
+
+ Multibinder.newSetBinder(binder(), HealthCheck.class)
+ .addBinding()
+ .to(DistributedTaskManagerHealthCheck.class);
}
@Provides
diff --git a/server/container/guice/distributed/src/main/java/org/apache/james/modules/event/JMAPEventBusModule.java b/server/container/guice/distributed/src/main/java/org/apache/james/modules/event/JMAPEventBusModule.java
index 25fa79f7e9..429a23ed5d 100644
--- a/server/container/guice/distributed/src/main/java/org/apache/james/modules/event/JMAPEventBusModule.java
+++ b/server/container/guice/distributed/src/main/java/org/apache/james/modules/event/JMAPEventBusModule.java
@@ -27,11 +27,13 @@ import org.apache.james.backends.rabbitmq.RabbitMQConfiguration;
import org.apache.james.backends.rabbitmq.ReactorRabbitMQChannelPool;
import org.apache.james.backends.rabbitmq.ReceiverProvider;
import org.apache.james.backends.rabbitmq.SimpleConnectionPool;
+import org.apache.james.core.healthcheck.HealthCheck;
import org.apache.james.events.EventBus;
import org.apache.james.events.EventBusId;
import org.apache.james.events.EventBusReconnectionHandler;
import org.apache.james.events.EventDeadLetters;
import org.apache.james.events.KeyReconnectionHandler;
+import org.apache.james.events.RabbitEventBusConsumerHealthCheck;
import org.apache.james.events.RabbitMQEventBus;
import org.apache.james.events.RetryBackoffConfiguration;
import org.apache.james.events.RoutingKeyConverter;
@@ -79,6 +81,12 @@ public class JMAPEventBusModule extends AbstractModule {
return new KeyReconnectionHandler(JMAP_NAMING_STRATEGY, eventBusId, configuration);
}
+ @ProvidesIntoSet
+ HealthCheck healthCheck(@Named(InjectionKeys.JMAP) RabbitMQEventBus eventBus,
+ SimpleConnectionPool connectionPool) {
+ return new RabbitEventBusConsumerHealthCheck(eventBus, JMAP_NAMING_STRATEGY, connectionPool);
+ }
+
@Provides
@Singleton
@Named(InjectionKeys.JMAP)
diff --git a/server/container/guice/distributed/src/main/java/org/apache/james/modules/event/RabbitMQEventBusModule.java b/server/container/guice/distributed/src/main/java/org/apache/james/modules/event/RabbitMQEventBusModule.java
index 0e2a652475..9ac96c4619 100644
--- a/server/container/guice/distributed/src/main/java/org/apache/james/modules/event/RabbitMQEventBusModule.java
+++ b/server/container/guice/distributed/src/main/java/org/apache/james/modules/event/RabbitMQEventBusModule.java
@@ -30,6 +30,7 @@ import org.apache.james.events.EventBusReconnectionHandler;
import org.apache.james.events.EventSerializer;
import org.apache.james.events.KeyReconnectionHandler;
import org.apache.james.events.NamingStrategy;
+import org.apache.james.events.RabbitEventBusConsumerHealthCheck;
import org.apache.james.events.RabbitMQEventBus;
import org.apache.james.events.RabbitMQEventBusDeadLetterQueueHealthCheck;
import org.apache.james.events.RegistrationKey;
@@ -72,6 +73,12 @@ public class RabbitMQEventBusModule extends AbstractModule {
.addBinding().to(RabbitMQEventBusDeadLetterQueueHealthCheck.class);
}
+ @ProvidesIntoSet
+ HealthCheck healthCheck(RabbitMQEventBus eventBus, NamingStrategy namingStrategy,
+ SimpleConnectionPool connectionPool) {
+ return new RabbitEventBusConsumerHealthCheck(eventBus, namingStrategy, connectionPool);
+ }
+
@ProvidesIntoSet
InitializationOperation workQueue(RabbitMQEventBus instance) {
return InitilizationOperationBuilder
diff --git a/server/container/guice/queue/rabbitmq/src/main/java/org/apache/james/modules/queue/rabbitmq/RabbitMQModule.java b/server/container/guice/queue/rabbitmq/src/main/java/org/apache/james/modules/queue/rabbitmq/RabbitMQModule.java
index 7d041c2db8..317a9f6943 100644
--- a/server/container/guice/queue/rabbitmq/src/main/java/org/apache/james/modules/queue/rabbitmq/RabbitMQModule.java
+++ b/server/container/guice/queue/rabbitmq/src/main/java/org/apache/james/modules/queue/rabbitmq/RabbitMQModule.java
@@ -37,6 +37,7 @@ import org.apache.james.queue.api.MailQueue;
import org.apache.james.queue.api.MailQueueFactory;
import org.apache.james.queue.api.ManageableMailQueue;
import org.apache.james.queue.rabbitmq.RabbitMQMailQueue;
+import org.apache.james.queue.rabbitmq.RabbitMQMailQueueConsumerHealthCheck;
import org.apache.james.queue.rabbitmq.RabbitMQMailQueueDeadLetterQueueHealthCheck;
import org.apache.james.queue.rabbitmq.RabbitMQMailQueueFactory;
import org.apache.james.queue.rabbitmq.view.RabbitMQMailQueueConfiguration;
@@ -69,6 +70,7 @@ public class RabbitMQModule extends AbstractModule {
Multibinder<HealthCheck> healthCheckMultiBinder = Multibinder.newSetBinder(binder(), HealthCheck.class);
healthCheckMultiBinder.addBinding().to(RabbitMQHealthCheck.class);
healthCheckMultiBinder.addBinding().to(RabbitMQMailQueueDeadLetterQueueHealthCheck.class);
+ healthCheckMultiBinder.addBinding().to(RabbitMQMailQueueConsumerHealthCheck.class);
Multibinder<SimpleConnectionPool.ReconnectionHandler> reconnectionHandlerMultibinder = Multibinder.newSetBinder(binder(), SimpleConnectionPool.ReconnectionHandler.class);
reconnectionHandlerMultibinder.addBinding().to(SpoolerReconnectionHandler.class);
diff --git a/server/protocols/webadmin-integration-test/distributed-webadmin-integration-test/src/test/java/org/apache/james/webadmin/integration/rabbitmq/RabbitMQWebAdminServerIntegrationImmutableTest.java b/server/protocols/webadmin-integration-test/distributed-webadmin-integration-test/src/test/java/org/apache/james/webadmin/integration/rabbitmq/RabbitMQWebAdminServerIntegrationImmutableTest.java
index 8563add768..6b66946691 100644
--- a/server/protocols/webadmin-integration-test/distributed-webadmin-integration-test/src/test/java/org/apache/james/webadmin/integration/rabbitmq/RabbitMQWebAdminServerIntegrationImmutableTest.java
+++ b/server/protocols/webadmin-integration-test/distributed-webadmin-integration-test/src/test/java/org/apache/james/webadmin/integration/rabbitmq/RabbitMQWebAdminServerIntegrationImmutableTest.java
@@ -138,6 +138,7 @@ class RabbitMQWebAdminServerIntegrationImmutableTest extends WebAdminServerInteg
"RabbitMQ backend", "RabbitMQMailQueueDeadLetterQueueHealthCheck",
"RabbitMQEventBusDeadLetterQueueHealthCheck", "MailReceptionCheck",
"Cassandra backend", "EventDeadLettersHealthCheck", "MessageFastViewProjection",
- "RabbitMQMailQueue BrowseStart", "OpenSearch Backend");
+ "RabbitMQMailQueue BrowseStart", "OpenSearch Backend", "DistributedTaskManagerConsumers",
+ "EventbusConsumers-jmapEvent", "MailQueueConsumers", "EventbusConsumers-mailboxEvent");
}
}
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueConsumerHealthCheck.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueConsumerHealthCheck.java
new file mode 100644
index 0000000000..c3b3a44a18
--- /dev/null
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueConsumerHealthCheck.java
@@ -0,0 +1,87 @@
+/****************************************************************
+ * 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.queue.rabbitmq;
+
+import java.util.Set;
+
+import javax.inject.Inject;
+
+import org.apache.james.backends.rabbitmq.SimpleConnectionPool;
+import org.apache.james.core.healthcheck.ComponentName;
+import org.apache.james.core.healthcheck.HealthCheck;
+import org.apache.james.core.healthcheck.Result;
+import org.apache.james.util.ReactorUtils;
+
+import com.github.fge.lambdas.Throwing;
+import com.rabbitmq.client.Channel;
+
+import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
+
+public class RabbitMQMailQueueConsumerHealthCheck implements HealthCheck {
+ public static final ComponentName COMPONENT_NAME = new ComponentName("RabbitMQMailQueueConsumersHealthCheck");
+ public static final ComponentName COMPONENT = new ComponentName("MailQueueConsumers");
+
+ private final RabbitMQMailQueueFactory queueFactory;
+ private final Set<SimpleConnectionPool.ReconnectionHandler> reconnectionHandlers;
+ private final SimpleConnectionPool connectionPool;
+
+ @Inject
+ public RabbitMQMailQueueConsumerHealthCheck(RabbitMQMailQueueFactory queueFactory, Set<SimpleConnectionPool.ReconnectionHandler> reconnectionHandlers, SimpleConnectionPool connectionPool) {
+ this.queueFactory = queueFactory;
+ this.reconnectionHandlers = reconnectionHandlers;
+ this.connectionPool = connectionPool;
+ }
+
+ @Override
+ public ComponentName componentName() {
+ return COMPONENT_NAME;
+ }
+
+ @Override
+ public Mono<Result> check() {
+ return connectionPool.getResilientConnection()
+ .map(Throwing.function(connection -> {
+ try (Channel channel = connection.createChannel()) {
+ return check(channel);
+ }
+ })).subscribeOn(ReactorUtils.BLOCKING_CALL_WRAPPER);
+ }
+
+ private Result check(Channel channel) {
+ boolean queueWithoutConsumers = queueFactory.listCreatedMailQueues()
+ .stream()
+ .map(org.apache.james.queue.api.MailQueueName::asString)
+ .map(MailQueueName::fromString)
+ .map(m -> m.toWorkQueueName().asString())
+ .anyMatch(Throwing.predicate(queue -> channel.consumerCount(queue) == 0));
+
+ if (queueWithoutConsumers) {
+ reconnectionHandlers.forEach(r -> connectionPool.getResilientConnection()
+ .flatMap(c -> Mono.from(r.handleReconnection(c)))
+ .subscribeOn(Schedulers.boundedElastic())
+ .block());
+
+ return Result.degraded(COMPONENT, "No consumers");
+ } else {
+ return Result.healthy(COMPONENT);
+ }
+ }
+}
diff --git a/server/task/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/DistributedTaskManagerHealthCheck.java b/server/task/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/DistributedTaskManagerHealthCheck.java
new file mode 100644
index 0000000000..80abde679f
--- /dev/null
+++ b/server/task/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/DistributedTaskManagerHealthCheck.java
@@ -0,0 +1,77 @@
+/****************************************************************
+ * 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.task.eventsourcing.distributed;
+
+import static org.apache.james.task.eventsourcing.distributed.RabbitMQWorkQueue.QUEUE_NAME;
+
+import java.io.IOException;
+
+import javax.inject.Inject;
+
+import org.apache.james.backends.rabbitmq.SimpleConnectionPool;
+import org.apache.james.core.healthcheck.ComponentName;
+import org.apache.james.core.healthcheck.HealthCheck;
+import org.apache.james.core.healthcheck.Result;
+import org.apache.james.task.eventsourcing.EventSourcingTaskManager;
+import org.apache.james.util.ReactorUtils;
+
+import com.github.fge.lambdas.Throwing;
+import com.rabbitmq.client.Channel;
+
+import reactor.core.publisher.Mono;
+
+public class DistributedTaskManagerHealthCheck implements HealthCheck {
+ public static final ComponentName COMPONENT_NAME = new ComponentName("DistributedTaskManagerConsumersHealthCheck");
+ public static final ComponentName COMPONENT = new ComponentName("DistributedTaskManagerConsumers");
+
+ private final EventSourcingTaskManager taskManager;
+ private final SimpleConnectionPool connectionPool;
+
+ @Inject
+ public DistributedTaskManagerHealthCheck(EventSourcingTaskManager taskManager, SimpleConnectionPool connectionPool) {
+ this.taskManager = taskManager;
+ this.connectionPool = connectionPool;
+ }
+
+ @Override
+ public ComponentName componentName() {
+ return COMPONENT_NAME;
+ }
+
+ @Override
+ public Mono<Result> check() {
+ return connectionPool.getResilientConnection()
+ .map(Throwing.function(connection -> {
+ try (Channel channel = connection.createChannel()) {
+ return check(channel);
+ }
+ })).subscribeOn(ReactorUtils.BLOCKING_CALL_WRAPPER);
+ }
+
+ private Result check(Channel channel) throws IOException {
+ if (channel.consumerCount(QUEUE_NAME) == 0) {
+ taskManager.restart();
+
+ return Result.degraded(COMPONENT, "No consumers");
+ } else {
+ return Result.healthy(COMPONENT);
+ }
+ }
+}
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 8ae4d8d0e5..77582260cb 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
@@ -143,13 +143,8 @@ public class RabbitMQWorkQueue implements WorkQueue {
@Override
public void restart() {
- Disposable previousWorkQueueHandler = receiverHandle;
- consumeWorkqueue();
- previousWorkQueueHandler.dispose();
-
- Disposable previousCancelHandler = this.cancelRequestListenerHandle;
- registerCancelRequestsListener(cancelRequestQueueName.asString());
- previousCancelHandler.dispose();
+ closeRabbitResources();
+ start();
}
private void consumeWorkqueue() {
@@ -271,6 +266,10 @@ public class RabbitMQWorkQueue implements WorkQueue {
} catch (IOException e) {
throw new RuntimeException(e);
}
+ closeRabbitResources();
+ }
+
+ private void closeRabbitResources() {
Optional.ofNullable(receiverHandle).ifPresent(Disposable::dispose);
Optional.ofNullable(sendCancelRequestsQueueHandle).ifPresent(Disposable::dispose);
Optional.ofNullable(cancelRequestListenerHandle).ifPresent(Disposable::dispose);
diff --git a/server/task/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/RabbitMQWorkQueueReconnectionHandler.java b/server/task/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/RabbitMQWorkQueueReconnectionHandler.java
index 6ad3f1def9..6a7648e646 100644
--- a/server/task/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/RabbitMQWorkQueueReconnectionHandler.java
+++ b/server/task/task-distributed/src/main/java/org/apache/james/task/eventsourcing/distributed/RabbitMQWorkQueueReconnectionHandler.java
@@ -20,51 +20,26 @@
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 javax.inject.Inject;
-import org.apache.james.backends.rabbitmq.QueueArguments;
-import org.apache.james.backends.rabbitmq.RabbitMQConfiguration;
import org.apache.james.backends.rabbitmq.SimpleConnectionPool;
import org.apache.james.task.eventsourcing.EventSourcingTaskManager;
import org.reactivestreams.Publisher;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import reactor.core.publisher.Mono;
public class RabbitMQWorkQueueReconnectionHandler implements SimpleConnectionPool.ReconnectionHandler {
- private static final Logger LOGGER = LoggerFactory.getLogger(RabbitMQWorkQueueReconnectionHandler.class);
- private final CancelRequestQueueName cancelRequestQueueName;
private final EventSourcingTaskManager taskManager;
- private final RabbitMQConfiguration configuration;
@Inject
- public RabbitMQWorkQueueReconnectionHandler(CancelRequestQueueName cancelRequestQueueName, EventSourcingTaskManager taskManager, RabbitMQConfiguration configuration) {
- this.cancelRequestQueueName = cancelRequestQueueName;
+ public RabbitMQWorkQueueReconnectionHandler(EventSourcingTaskManager taskManager) {
this.taskManager = taskManager;
- this.configuration = configuration;
}
@Override
public Publisher<Void> handleReconnection(Connection connection) {
- return Mono.fromRunnable(() -> createCancelQueue(connection))
- .then(Mono.fromRunnable(taskManager::restart));
- }
-
- private void createCancelQueue(Connection connection) {
- try (Channel channel = connection.createChannel()) {
- QueueArguments.Builder builder = QueueArguments.builder();
- configuration.getQueueTTL().ifPresent(builder::queueTTL);
- channel.queueDeclare(cancelRequestQueueName.asString(), !DURABLE, !EXCLUSIVE, !AUTO_DELETE, builder.build());
- } catch (Exception e) {
- LOGGER.error("Error recovering connection", e);
- }
+ return Mono.fromRunnable(taskManager::restart);
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
(james-project) 04/05: JAMES-3955 Increase consumer timeout for TaskManagerWorkQueue
Posted by rc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rcordier pushed a commit to branch 3.8.x
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 87d4f0d1afa2138f2b637c4f4739fbdc644a4211
Author: Quan Tran <hq...@linagora.com>
AuthorDate: Mon Nov 20 16:59:52 2023 +0700
JAMES-3955 Increase consumer timeout for TaskManagerWorkQueue
---
.../james/backends/rabbitmq/QueueArguments.java | 5 ++
.../backends/rabbitmq/RabbitMQConfiguration.java | 33 +++++++++++--
.../rabbitmq/RabbitMQConfigurationTest.java | 55 ++++++++++++++++++++++
.../modules/ROOT/pages/configure/rabbitmq.adoc | 15 ++++++
.../sample-configuration/rabbitmq.properties | 5 ++
.../distributed/RabbitMQWorkQueue.java | 1 +
.../distributed/RabbitMQWorkQueueTest.java | 12 ++++-
src/site/xdoc/server/config-rabbitmq.xml | 10 ++++
upgrade-instructions.md | 16 +++++++
9 files changed, 147 insertions(+), 5 deletions(-)
diff --git a/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/QueueArguments.java b/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/QueueArguments.java
index 8a1cb2d47a..7961e0c6bd 100644
--- a/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/QueueArguments.java
+++ b/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/QueueArguments.java
@@ -42,6 +42,11 @@ public class QueueArguments {
return this;
}
+ public Builder consumerTimeout(long consumerTimeoutInMillisecond) {
+ arguments.put("x-consumer-timeout", consumerTimeoutInMillisecond);
+ return this;
+ }
+
public ImmutableMap<String, Object> build() {
return arguments.build();
}
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 a47e17bece..c7963e27d0 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
@@ -30,6 +30,8 @@ import java.io.File;
import java.net.URI;
import java.nio.file.Files;
import java.nio.file.Paths;
+import java.time.Duration;
+import java.time.temporal.ChronoUnit;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
@@ -38,6 +40,7 @@ import java.util.Optional;
import java.util.stream.Stream;
import org.apache.commons.configuration2.Configuration;
+import org.apache.james.util.DurationParser;
import org.apache.james.util.Host;
import com.google.common.base.Preconditions;
@@ -306,6 +309,7 @@ public class RabbitMQConfiguration {
private static final String QUEUE_TTL = "notification.queue.ttl";
private static final String EVENT_BUS_NOTIFICATION_DURABILITY_ENABLED = "event.bus.notification.durability.enabled";
private static final String EVENT_BUS_PUBLISH_CONFIRM_ENABLED = "event.bus.publish.confirm.enabled";
+ private static final String TASK_QUEUE_CONSUMER_TIMEOUT = "task.queue.consumer.timeout";
private static final String VHOST = "vhost";
public static class ManagementCredentials {
@@ -383,6 +387,7 @@ public class RabbitMQConfiguration {
static final int DEFAULT_SHUTDOWN_TIMEOUT = 10_000;
static final int DEFAULT_NETWORK_RECOVERY_INTERVAL = 5_000;
static final int DEFAULT_PORT = 5672;
+ static final Duration DEFAULT_TASK_QUEUE_CONSUMER_TIMEOUT = Duration.ofDays(1);
private final URI amqpUri;
private final URI managementUri;
@@ -404,6 +409,7 @@ public class RabbitMQConfiguration {
private Optional<Boolean> eventBusPublishConfirmEnabled;
private Optional<Boolean> eventBusNotificationDurabilityEnabled;
private Optional<String> vhost;
+ private Optional<Duration> taskQueueConsumerTimeout;
private Builder(URI amqpUri, URI managementUri, ManagementCredentials managementCredentials) {
this.amqpUri = amqpUri;
@@ -426,6 +432,7 @@ public class RabbitMQConfiguration {
this.eventBusPublishConfirmEnabled = Optional.empty();
this.eventBusNotificationDurabilityEnabled = Optional.empty();
this.vhost = Optional.empty();
+ this.taskQueueConsumerTimeout = Optional.empty();
}
public Builder maxRetries(int maxRetries) {
@@ -521,6 +528,11 @@ public class RabbitMQConfiguration {
return this;
}
+ public Builder taskQueueConsumerTimeout(Optional<Duration> taskQueueConsumerTimeout) {
+ this.taskQueueConsumerTimeout = taskQueueConsumerTimeout;
+ return this;
+ }
+
public RabbitMQConfiguration build() {
Preconditions.checkNotNull(amqpUri, "'amqpUri' should not be null");
Preconditions.checkNotNull(managementUri, "'managementUri' should not be null");
@@ -544,7 +556,8 @@ public class RabbitMQConfiguration {
queueTTL,
eventBusPublishConfirmEnabled.orElse(true),
eventBusNotificationDurabilityEnabled.orElse(true),
- vhost);
+ vhost,
+ taskQueueConsumerTimeout.orElse(DEFAULT_TASK_QUEUE_CONSUMER_TIMEOUT));
}
private List<Host> hostsDefaultingToUri() {
@@ -596,6 +609,10 @@ public class RabbitMQConfiguration {
Optional<String> vhost = Optional.ofNullable(configuration.getString(VHOST, null));
ManagementCredentials managementCredentials = ManagementCredentials.from(configuration);
+
+ Optional<Duration> taskQueueConsumerTimeout = Optional.ofNullable(configuration.getString(TASK_QUEUE_CONSUMER_TIMEOUT, null))
+ .map(value -> DurationParser.parse(value, ChronoUnit.SECONDS));
+
return builder()
.amqpUri(amqpUri)
.managementUri(managementUri)
@@ -610,6 +627,7 @@ public class RabbitMQConfiguration {
.eventBusNotificationDurabilityEnabled(configuration.getBoolean(EVENT_BUS_NOTIFICATION_DURABILITY_ENABLED, null))
.eventBusPublishConfirmEnabled(configuration.getBoolean(EVENT_BUS_PUBLISH_CONFIRM_ENABLED, null))
.vhost(vhost)
+ .taskQueueConsumerTimeout(taskQueueConsumerTimeout)
.build();
}
@@ -681,13 +699,14 @@ public class RabbitMQConfiguration {
private final boolean eventBusPublishConfirmEnabled;
private final boolean eventBusNotificationDurabilityEnabled;
private final Optional<String> vhost;
+ private final Duration taskQueueConsumerTimeout;
private RabbitMQConfiguration(URI uri, URI managementUri, ManagementCredentials managementCredentials, int maxRetries, int minDelayInMs,
int connectionTimeoutInMs, int channelRpcTimeoutInMs, int handshakeTimeoutInMs, int shutdownTimeoutInMs,
int networkRecoveryIntervalInMs, Boolean useSsl, Boolean useSslManagement, SSLConfiguration sslConfiguration,
boolean useQuorumQueues, int quorumQueueReplicationFactor, List<Host> hosts, Optional<Long> queueTTL,
boolean eventBusPublishConfirmEnabled, boolean eventBusNotificationDurabilityEnabled,
- Optional<String> vhost) {
+ Optional<String> vhost, Duration taskQueueConsumerTimeout) {
this.uri = uri;
this.managementUri = managementUri;
this.managementCredentials = managementCredentials;
@@ -708,6 +727,7 @@ public class RabbitMQConfiguration {
this.eventBusPublishConfirmEnabled = eventBusPublishConfirmEnabled;
this.eventBusNotificationDurabilityEnabled = eventBusNotificationDurabilityEnabled;
this.vhost = vhost;
+ this.taskQueueConsumerTimeout = taskQueueConsumerTimeout;
}
public URI getUri() {
@@ -799,6 +819,10 @@ public class RabbitMQConfiguration {
return Optional.empty();
}
+ public Duration getTaskQueueConsumerTimeout() {
+ return taskQueueConsumerTimeout;
+ }
+
@Override
public final boolean equals(Object o) {
if (o instanceof RabbitMQConfiguration) {
@@ -823,7 +847,8 @@ public class RabbitMQConfiguration {
&& Objects.equals(this.queueTTL, that.queueTTL)
&& Objects.equals(this.eventBusPublishConfirmEnabled, that.eventBusPublishConfirmEnabled)
&& Objects.equals(this.eventBusNotificationDurabilityEnabled, that.eventBusNotificationDurabilityEnabled)
- && Objects.equals(this.vhost, that.vhost);
+ && Objects.equals(this.vhost, that.vhost)
+ && Objects.equals(this.taskQueueConsumerTimeout, that.taskQueueConsumerTimeout);
}
return false;
}
@@ -832,6 +857,6 @@ public class RabbitMQConfiguration {
public final int hashCode() {
return Objects.hash(uri, managementUri, maxRetries, minDelayInMs, connectionTimeoutInMs, quorumQueueReplicationFactor, useQuorumQueues, hosts,
channelRpcTimeoutInMs, handshakeTimeoutInMs, shutdownTimeoutInMs, networkRecoveryIntervalInMs, managementCredentials, useSsl, useSslManagement,
- sslConfiguration, queueTTL, eventBusPublishConfirmEnabled, eventBusNotificationDurabilityEnabled, vhost);
+ sslConfiguration, queueTTL, eventBusPublishConfirmEnabled, eventBusNotificationDurabilityEnabled, vhost, taskQueueConsumerTimeout);
}
}
diff --git a/backends-common/rabbitmq/src/test/java/org/apache/james/backends/rabbitmq/RabbitMQConfigurationTest.java b/backends-common/rabbitmq/src/test/java/org/apache/james/backends/rabbitmq/RabbitMQConfigurationTest.java
index 9ae0c55c0e..4a2d6735b5 100644
--- a/backends-common/rabbitmq/src/test/java/org/apache/james/backends/rabbitmq/RabbitMQConfigurationTest.java
+++ b/backends-common/rabbitmq/src/test/java/org/apache/james/backends/rabbitmq/RabbitMQConfigurationTest.java
@@ -28,6 +28,7 @@ import java.io.File;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
+import java.time.Duration;
import java.util.Arrays;
import java.util.Optional;
@@ -455,6 +456,60 @@ class RabbitMQConfigurationTest {
.isInstanceOf(ConversionException.class);
}
+ @Test
+ void emptyTaskQueueConsumerTimeoutShouldDefaultToOneDay() {
+ PropertiesConfiguration configuration = new PropertiesConfiguration();
+ configuration.addProperty("uri", "amqp://james:james@rabbitmqhost:5672");
+ configuration.addProperty("management.uri", "http://james:james@rabbitmqhost:15672/api/");
+ configuration.addProperty("management.user", DEFAULT_USER);
+ configuration.addProperty("management.password", DEFAULT_PASSWORD_STRING);
+
+ assertThat(RabbitMQConfiguration.from(configuration).getTaskQueueConsumerTimeout())
+ .isEqualTo(Duration.ofDays(1));
+ }
+
+ @Test
+ void parseValidTaskQueueConsumerTimeoutShouldSucceed() {
+ PropertiesConfiguration configuration = new PropertiesConfiguration();
+ configuration.addProperty("uri", "amqp://james:james@rabbitmqhost:5672");
+ configuration.addProperty("management.uri", "http://james:james@rabbitmqhost:15672/api/");
+ configuration.addProperty("management.user", DEFAULT_USER);
+ configuration.addProperty("management.password", DEFAULT_PASSWORD_STRING);
+
+ configuration.addProperty("task.queue.consumer.timeout", "2day");
+
+ assertThat(RabbitMQConfiguration.from(configuration).getTaskQueueConsumerTimeout())
+ .isEqualTo(Duration.ofDays(2));
+ }
+
+ @Test
+ void parseTaskQueueConsumerTimeoutWithoutTimeUnitShouldDefaultToSecond() {
+ PropertiesConfiguration configuration = new PropertiesConfiguration();
+ configuration.addProperty("uri", "amqp://james:james@rabbitmqhost:5672");
+ configuration.addProperty("management.uri", "http://james:james@rabbitmqhost:15672/api/");
+ configuration.addProperty("management.user", DEFAULT_USER);
+ configuration.addProperty("management.password", DEFAULT_PASSWORD_STRING);
+
+ configuration.addProperty("task.queue.consumer.timeout", "3600");
+
+ assertThat(RabbitMQConfiguration.from(configuration).getTaskQueueConsumerTimeout())
+ .isEqualTo(Duration.ofSeconds(3600));
+ }
+
+ @Test
+ void parseInvalidTaskQueueConsumerTimeoutShouldFail() {
+ PropertiesConfiguration configuration = new PropertiesConfiguration();
+ configuration.addProperty("uri", "amqp://james:james@rabbitmqhost:5672");
+ configuration.addProperty("management.uri", "http://james:james@rabbitmqhost:15672/api/");
+ configuration.addProperty("management.user", DEFAULT_USER);
+ configuration.addProperty("management.password", DEFAULT_PASSWORD_STRING);
+
+ configuration.addProperty("task.queue.consumer.timeout", "invalid");
+
+ assertThatThrownBy(() -> RabbitMQConfiguration.from(configuration))
+ .isInstanceOf(NumberFormatException.class);
+ }
+
@Test
void fromShouldReturnEmptyVhostValueByDefault() {
PropertiesConfiguration configuration = new PropertiesConfiguration();
diff --git a/server/apps/distributed-app/docs/modules/ROOT/pages/configure/rabbitmq.adoc b/server/apps/distributed-app/docs/modules/ROOT/pages/configure/rabbitmq.adoc
index c12a2e91bc..0910b2d61e 100644
--- a/server/apps/distributed-app/docs/modules/ROOT/pages/configure/rabbitmq.adoc
+++ b/server/apps/distributed-app/docs/modules/ROOT/pages/configure/rabbitmq.adoc
@@ -170,4 +170,19 @@ with an exclusive consumer.
Disable with caution (this only makes sense in a distributed setup where other nodes consume tasks).
Defaults to true.
+Limitation: Sometimes, some tasks running on James can be very heavy and take a couple of hours to complete.
+If other tasks are being triggered meanwhile on WebAdmin, they go on the TaskManagerWorkQueue and James unack them,
+telling RabbitMQ it will consume them later. If they don't get consumed before the consumer timeout setup in
+RabbitMQ (default being 30 minutes), RabbitMQ closes the channel on an exception. It is thus advised to declare a
+longer timeout in rabbitmq.conf. More https://www.rabbitmq.com/consumers.html#acknowledgement-timeout[here].
+
+| task.queue.consumer.timeout
+| Task queue consumer timeout.
+
+Optional. Duration (support multiple time units cf `DurationParser`), defaults to 1 day.
+
+Required at least RabbitMQ version 3.12 to have effect.
+This is used to avoid the task queue consumer (which could run very long tasks) being disconnected by RabbitMQ after the default acknowledgement timeout 30 minutes.
+References: https://www.rabbitmq.com/consumers.html#acknowledgement-timeout.
+
|===
\ No newline at end of file
diff --git a/server/apps/distributed-app/sample-configuration/rabbitmq.properties b/server/apps/distributed-app/sample-configuration/rabbitmq.properties
index 1ad27e7aa1..75a562af60 100644
--- a/server/apps/distributed-app/sample-configuration/rabbitmq.properties
+++ b/server/apps/distributed-app/sample-configuration/rabbitmq.properties
@@ -84,6 +84,11 @@ mailqueue.size.metricsEnabled=false
# Defaults to true.
task.consumption.enabled=true
+# Configure task queue consumer timeout. References: https://www.rabbitmq.com/consumers.html#acknowledgement-timeout. Required at least RabbitMQ version 3.12 to have effect.
+# This is used to avoid the task queue consumer (which could run very long tasks) being disconnected by RabbitMQ after the default acknowledgement timeout 30 minutes.
+# Optional. Duration (support multiple time units cf `DurationParser`), defaults to 1 day.
+#task.queue.consumer.timeout=1day
+
# Configure queue ttl (in ms). References: https://www.rabbitmq.com/ttl.html#queue-ttl.
# 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.
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 77582260cb..d1565f67b1 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
@@ -129,6 +129,7 @@ public class RabbitMQWorkQueue implements WorkQueue {
.durable(true)
.arguments(rabbitMQConfiguration.workQueueArgumentsBuilder(ALLOW_QUORUM)
.singleActiveConsumer()
+ .consumerTimeout(rabbitMQConfiguration.getTaskQueueConsumerTimeout().toMillis())
.build()))
.retryWhen(Retry.backoff(NUM_RETRIES, FIRST_BACKOFF));
Mono<AMQP.Queue.BindOk> bindQueueToExchange = sender
diff --git a/server/task/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/RabbitMQWorkQueueTest.java b/server/task/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/RabbitMQWorkQueueTest.java
index 45265689b0..7e4709ff9a 100644
--- a/server/task/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/RabbitMQWorkQueueTest.java
+++ b/server/task/task-distributed/src/test/java/org/apache/james/task/eventsourcing/distributed/RabbitMQWorkQueueTest.java
@@ -30,6 +30,7 @@ import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.IntStream;
import org.apache.james.backends.rabbitmq.RabbitMQExtension;
+import org.apache.james.backends.rabbitmq.RabbitMQManagementAPI;
import org.apache.james.server.task.json.JsonTaskSerializer;
import org.apache.james.server.task.json.TestTask;
import org.apache.james.server.task.json.dto.MemoryReferenceTaskStore;
@@ -63,7 +64,7 @@ class RabbitMQWorkQueueTest {
private RabbitMQWorkQueue testee;
private ImmediateWorker worker;
private JsonTaskSerializer serializer;
-
+ private RabbitMQManagementAPI managementAPI;
@BeforeEach
void setUp() throws Exception {
@@ -71,6 +72,7 @@ class RabbitMQWorkQueueTest {
serializer = JsonTaskSerializer.of(TestTaskDTOModules.COMPLETED_TASK_MODULE, TestTaskDTOModules.MEMORY_REFERENCE_TASK_MODULE.apply(new MemoryReferenceTaskStore()));
testee = new RabbitMQWorkQueue(worker, rabbitMQExtension.getSender(), rabbitMQExtension.getReceiverProvider(), serializer, RabbitMQWorkQueueConfiguration$.MODULE$.enabled(), CancelRequestQueueName.generate(), rabbitMQExtension.getRabbitMQ().getConfiguration());
testee.start();
+ managementAPI = rabbitMQExtension.managementAPI();
}
@AfterEach
@@ -78,6 +80,14 @@ class RabbitMQWorkQueueTest {
testee.close();
}
+ @Test
+ void shouldSetConsumerTimeoutArgumentOnTaskQueue() {
+ assertThat(managementAPI.queueDetails("/", "taskManagerWorkQueue")
+ .getArguments()
+ .get("x-consumer-timeout"))
+ .isEqualTo("86400000");
+ }
+
@Test
void workQueueShouldConsumeSubmittedTask() {
testee.submit(TASK_WITH_ID);
diff --git a/src/site/xdoc/server/config-rabbitmq.xml b/src/site/xdoc/server/config-rabbitmq.xml
index b684a7abc3..f346ad20fe 100644
--- a/src/site/xdoc/server/config-rabbitmq.xml
+++ b/src/site/xdoc/server/config-rabbitmq.xml
@@ -210,6 +210,16 @@
Disable with caution (this only makes sense in a distributed setup where other nodes consume tasks).
Defaults to true.
</dd>
+ <dt><strong>task.queue.consumer.timeout</strong></dt>
+ <dd>
+ Task queue consumer timeout.
+
+ Optional. Duration (support multiple time units cf `DurationParser`), defaults to 1 day.
+
+ Required at least RabbitMQ version 3.12 to have effect.
+ This is used to avoid the task queue consumer (which could run very long tasks) being disconnected by RabbitMQ after the default acknowledgement timeout 30 minutes.
+ References: https://www.rabbitmq.com/consumers.html#acknowledgement-timeout.
+ </dd>
</dl>
</section>
diff --git a/upgrade-instructions.md b/upgrade-instructions.md
index 378839a95c..fb0ad3fd3c 100644
--- a/upgrade-instructions.md
+++ b/upgrade-instructions.md
@@ -28,6 +28,22 @@ Change list:
- [Adding delegatedUser column to user_table](#adding-delegatedusers-column-to-user-table)
- [DeletedMessageVaultHook should not be used on Cassandra based products](#deleted-message-vault-is-now-deactivated-by-default)
+### Increase RabbitMQ consumer timeout on the task queue
+
+Date: 11/20/2023
+
+JIRA: https://issues.apache.org/jira/browse/JAMES-3955
+
+Before, the RabbitMQ default acknowledgement timeout on the `taskManagerWorkQueue` was 30 minutes.
+This is not enough for long-running tasks, therefore we advise to increase the acknowledgement timeout and now enforce it to 1 day by default.
+
+rf: https://www.rabbitmq.com/consumers.html#acknowledgement-timeout
+
+To make the per queue argument feature work, update your RabbitMQ to version 3.12 at least.
+
+If you want the old behavior where consumer timeout defaults to 30 minutes, you can either not upgrade your RabbitMQ or set `task.queue.consumer.timeout=30minutes` in `rabbitmq.properties`.
+
+
### JMAP filters event sourcing increments and snapshots
Date: 19/04/2023
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
(james-project) 03/05: JAMES-3955 WARNING logs upon closing channels
Posted by rc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rcordier pushed a commit to branch 3.8.x
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 040601a767cd96e5ec72a285be108754d6f019e6
Author: Benoit TELLIER <bt...@linagora.com>
AuthorDate: Fri Nov 10 09:19:04 2023 +0100
JAMES-3955 WARNING logs upon closing channels
---
.../james/backends/rabbitmq/ReactorRabbitMQChannelPool.java | 8 ++++++++
1 file changed, 8 insertions(+)
diff --git a/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/ReactorRabbitMQChannelPool.java b/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/ReactorRabbitMQChannelPool.java
index fc65f6b1d6..f465eb0262 100644
--- a/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/ReactorRabbitMQChannelPool.java
+++ b/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/ReactorRabbitMQChannelPool.java
@@ -110,6 +110,10 @@ public class ReactorRabbitMQChannelPool implements ChannelPool, Startable {
@Override
public void close(int closeCode, String closeMessage) throws IOException, TimeoutException {
+ // https://www.rabbitmq.com/amqp-0-9-1-reference.html#domain.reply-code
+ if (closeCode >= 300) {
+ LOGGER.warn("Closing channel {} code:{} message:'{}'", getChannelNumber(), closeCode, closeMessage);
+ }
delegate.close(closeCode, closeMessage);
}
@@ -120,6 +124,10 @@ public class ReactorRabbitMQChannelPool implements ChannelPool, Startable {
@Override
public void abort(int closeCode, String closeMessage) throws IOException {
+ // https://www.rabbitmq.com/amqp-0-9-1-reference.html#domain.reply-code
+ if (closeCode >= 300) {
+ LOGGER.warn("Closing channel {} code:{} message:'{}'", getChannelNumber(), closeCode, closeMessage);
+ }
delegate.abort(closeCode, closeMessage);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
(james-project) 02/05: JAMES-3955 Applicative timeouts before Rabbit timeouts
Posted by rc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rcordier pushed a commit to branch 3.8.x
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 21adf2146b7b6cfb2128535b256f413d4ca368f6
Author: Benoit TELLIER <bt...@linagora.com>
AuthorDate: Fri Nov 10 09:13:53 2023 +0100
JAMES-3955 Applicative timeouts before Rabbit timeouts
---
.../src/main/java/org/apache/james/events/ListenerExecutor.java | 9 ++++++++-
.../src/main/java/org/apache/james/queue/rabbitmq/Dequeuer.java | 6 +++++-
2 files changed, 13 insertions(+), 2 deletions(-)
diff --git a/event-bus/distributed/src/main/java/org/apache/james/events/ListenerExecutor.java b/event-bus/distributed/src/main/java/org/apache/james/events/ListenerExecutor.java
index 39cae87d07..2eb14cc97a 100644
--- a/event-bus/distributed/src/main/java/org/apache/james/events/ListenerExecutor.java
+++ b/event-bus/distributed/src/main/java/org/apache/james/events/ListenerExecutor.java
@@ -21,6 +21,8 @@ package org.apache.james.events;
import static org.apache.james.events.EventBus.Metrics.timerName;
+import java.time.Duration;
+
import org.apache.james.metrics.api.MetricFactory;
import org.apache.james.util.MDCBuilder;
import org.apache.james.util.ReactorUtils;
@@ -28,6 +30,10 @@ import org.apache.james.util.ReactorUtils;
import reactor.core.publisher.Mono;
class ListenerExecutor {
+ // JAMES-3955 RabbitMQ handles timeout by closing channels thus
+ // causing event consumption to halt. We thus need to handle timeout beforehand.
+ private static final Duration TIMEOUT = Duration.ofMinutes(10);
+
private final MetricFactory metricFactory;
ListenerExecutor(MetricFactory metricFactory) {
@@ -38,7 +44,8 @@ class ListenerExecutor {
if (listener.isHandling(event)) {
return Mono.from(metricFactory.decoratePublisherWithTimerMetric(timerName(listener),
Mono.from(listener.reactiveEvent(event))
- .contextWrite(ReactorUtils.context("ListenerExecutor", mdc(listener, mdcBuilder, event)))));
+ .contextWrite(ReactorUtils.context("ListenerExecutor", mdc(listener, mdcBuilder, event)))
+ .timeout(TIMEOUT)));
}
return Mono.empty();
}
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/Dequeuer.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/Dequeuer.java
index 98ed284354..b5eee6ba93 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/Dequeuer.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/Dequeuer.java
@@ -21,6 +21,7 @@ package org.apache.james.queue.rabbitmq;
import static org.apache.james.queue.api.MailQueue.DEQUEUED_METRIC_NAME_PREFIX;
+import java.time.Duration;
import java.util.function.Consumer;
import org.apache.james.backends.rabbitmq.ReceiverProvider;
@@ -50,6 +51,9 @@ import reactor.rabbitmq.Receiver;
class Dequeuer {
private static final Logger LOGGER = LoggerFactory.getLogger(Dequeuer.class);
private static final boolean REQUEUE = true;
+ // JAMES-3955 RabbitMQ handles timeout by closing channels thus
+ // causing event consumption to halt. We thus need to handle timeout beforehand.
+ private static final Duration TIMEOUT = Duration.ofMinutes(15);
private static class RabbitMQMailQueueItem implements MailQueue.MailQueueItem {
@@ -133,7 +137,7 @@ class Dequeuer {
LOGGER.error("Failed to load email, requeue corresponding message", e);
response.nack(REQUEUE);
return Mono.empty();
- });
+ }).timeout(TIMEOUT);
}
private ThrowingConsumer<MailQueue.MailQueueItem.CompletionStatus> ack(AcknowledgableDelivery response, MailWithEnqueueId mailWithEnqueueId) {
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
(james-project) 05/05: [FIX] Solve weave/rest-smtp-sink: Docker image manifest v2 schema 1 deprecation issue (#2152)
Posted by rc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rcordier pushed a commit to branch 3.8.x
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 0c5196a5a7618b2601e64a257b593bf05f1b909a
Author: Trần Hồng Quân <55...@users.noreply.github.com>
AuthorDate: Mon Mar 25 21:20:42 2024 +0700
[FIX] Solve weave/rest-smtp-sink: Docker image manifest v2 schema 1 deprecation issue (#2152)
cf: https://docs.docker.com/engine/deprecated/#pushing-and-pulling-with-image-manifest-v2-schema-1
Use a rebuilt image would be the trick.
---
server/testing/src/main/java/org/apache/james/util/docker/Images.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/server/testing/src/main/java/org/apache/james/util/docker/Images.java b/server/testing/src/main/java/org/apache/james/util/docker/Images.java
index 13c11ad521..740e9e41eb 100644
--- a/server/testing/src/main/java/org/apache/james/util/docker/Images.java
+++ b/server/testing/src/main/java/org/apache/james/util/docker/Images.java
@@ -20,7 +20,7 @@
package org.apache.james.util.docker;
public interface Images {
- String FAKE_SMTP = "weave/rest-smtp-sink:latest";
+ String FAKE_SMTP = "quanth99/rest-smtp-sink:1.0"; // Original Dockerfile: https://github.com/ambled/rest-smtp-sink/blob/master/Dockerfile
String RABBITMQ = "rabbitmq:3.9.18-management";
String ELASTICSEARCH_2 = "elasticsearch:2.4.6";
String ELASTICSEARCH_6 = "docker.elastic.co/elasticsearch/elasticsearch:6.3.2";
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org