You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@james.apache.org by ka...@apache.org on 2022/09/01 20:07:38 UTC
[james-project] branch 3.7.x updated (9978eb8462 -> b727dec24b)
This is an automated email from the ASF dual-hosted git repository.
kao pushed a change to branch 3.7.x
in repository https://gitbox.apache.org/repos/asf/james-project.git
from 9978eb8462 [maven-release-plugin] prepare for next development iteration
new 903f202552 JAMES-3810 Dequeuer should nack when fails to see if email was deleted
new b727dec24b JAMES-3810 Decrease slice browsing concurrency
The 2 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:
.../main/java/org/apache/james/queue/rabbitmq/Dequeuer.java | 11 +++++++++--
.../rabbitmq/view/cassandra/CassandraMailQueueBrowser.java | 2 +-
2 files changed, 10 insertions(+), 3 deletions(-)
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 02/02: JAMES-3810 Decrease slice browsing concurrency
Posted by ka...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kao pushed a commit to branch 3.7.x
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit b727dec24b4997ac4b353ec789e5ff8bf26e552f
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Mon Aug 22 23:25:29 2022 +0700
JAMES-3810 Decrease slice browsing concurrency
flatMapIterable processes 256 items in parallel which is likely to
be too much when browsing slices of 1h of email traffic.
Decrease this to a more reasonable value.
(cherry picked from commit f463bf9c07196545a506a8229f934721369bf601)
---
.../james/queue/rabbitmq/view/cassandra/CassandraMailQueueBrowser.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueBrowser.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueBrowser.java
index 94314de05d..64f85be0b1 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueBrowser.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueBrowser.java
@@ -134,7 +134,7 @@ public class CassandraMailQueueBrowser {
Flux<EnqueuedItemWithSlicingContext> browseReferences(MailQueueName queueName, Instant browseStart) {
return allSlicesStartingAt(browseStart)
- .flatMapSequential(slice -> browseSlice(queueName, slice));
+ .flatMapSequential(slice -> browseSlice(queueName, slice), 4);
}
private Mono<Pair<EnqueuedItem, Mail>> toMailFuture(EnqueuedItemWithSlicingContext enqueuedItemWithSlicingContext) {
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 01/02: JAMES-3810 Dequeuer should nack when fails to see if email was deleted
Posted by ka...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kao pushed a commit to branch 3.7.x
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 903f2025529e66e099955c370a219f42fabe1c6f
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Mon Aug 22 23:24:05 2022 +0700
JAMES-3810 Dequeuer should nack when fails to see if email was deleted
Failure to do so leads to unackowledge messages and would eventually halt the delivery.
(cherry picked from commit 93eeeb65dfeacfbdca81f01ec7286f919cc50d90)
Backport: Uses elastic scheduler since there is no BLOCKING_CALL_WRAPPER yet
---
.../main/java/org/apache/james/queue/rabbitmq/Dequeuer.java | 11 +++++++++--
1 file changed, 9 insertions(+), 2 deletions(-)
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 92c549d23a..573b7a2497 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
@@ -41,6 +41,7 @@ import com.github.fge.lambdas.consumers.ThrowingConsumer;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
import reactor.rabbitmq.AcknowledgableDelivery;
import reactor.rabbitmq.ConsumeOptions;
import reactor.rabbitmq.Receiver;
@@ -108,14 +109,20 @@ class Dequeuer {
private Mono<RabbitMQMailQueueItem> filterIfDeleted(RabbitMQMailQueueItem item) {
return mailQueueView.isPresent(item.getEnqueueId())
- .handle((isPresent, sink) -> {
+ .<RabbitMQMailQueueItem>handle((isPresent, sink) -> {
if (isPresent) {
sink.next(item);
} else {
item.done(true);
sink.complete();
}
- });
+ })
+ .onErrorResume(e -> Mono.fromRunnable(() -> {
+ LOGGER.error("Failure to see if {} was deleted", item.enqueueId.asUUID(), e);
+ item.done(false);
+ })
+ .subscribeOn(Schedulers.elastic())
+ .then(Mono.error(e)));
}
private Mono<RabbitMQMailQueueItem> loadItem(AcknowledgableDelivery response) {
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org