You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@james.apache.org by bt...@apache.org on 2023/08/30 07:21:47 UTC
[james-project] branch 3.7.x updated (8c36c8f451 -> 7d46378ef4)
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a change to branch 3.7.x
in repository https://gitbox.apache.org/repos/asf/james-project.git
from 8c36c8f451 JAMES-3934 Allow external scheduling for RabbitMQ mailQueue browse st… (#1691)
new b6bdfa0703 JAMES-3924 Allow conditional Cassandra statement recording
new 944e9a9970 JAMES-3924 Test and fix browse start updates
new 3a069194f1 JAMES-3924 Tests for contentStart updates
new 7674d94aa9 JAMES-3924 Improve dequeuer Lifecycle management
new 9f84f10732 JAMES-3924 Review CassandraMailQueueBrowser parallelism
new 952271012d JAMES-3924 Fix checkstyles and some unstable tests
new 7864143c5e JAMES-3924 QoS 1 for browser/content start update tests
new c525c6f4e5 JAMES-3924 Remove repeated tests
new 7d46378ef4 JAMES-3924 Fix rebase issues during 3.7.x backport
The 9 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:
.../backends/cassandra/StatementRecorder.java | 10 +-
.../james/backends/cassandra/TestingSession.java | 14 ++-
.../backends/cassandra/TestingSessionTest.java | 6 +-
.../org/apache/james/queue/rabbitmq/Dequeuer.java | 3 +
.../view/cassandra/CassandraMailQueueBrowser.java | 12 +-
.../cassandra/CassandraMailQueueMailDelete.java | 13 +-
.../queue/rabbitmq/RabbitMQMailQueueTest.java | 135 +++++++++++++++++----
7 files changed, 149 insertions(+), 44 deletions(-)
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 07/09: JAMES-3924 QoS 1 for browser/content start update tests
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch 3.7.x
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 7864143c5ee1d7c46e27502f0786b10bf4ca0095
Author: Quan Tran <hq...@linagora.com>
AuthorDate: Thu Aug 3 13:53:43 2023 +0700
JAMES-3924 QoS 1 for browser/content start update tests
---
.../queue/rabbitmq/RabbitMQMailQueueTest.java | 38 +++++++++++++++-------
1 file changed, 27 insertions(+), 11 deletions(-)
diff --git a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueTest.java b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueTest.java
index 25623493b8..82f9a3176e 100644
--- a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueTest.java
+++ b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueTest.java
@@ -64,7 +64,6 @@ import org.apache.james.blob.cassandra.CassandraBlobStoreFactory;
import org.apache.james.blob.mail.MimeMessageStore;
import org.apache.james.core.builder.MimeMessageBuilder;
import org.apache.james.eventsourcing.eventstore.cassandra.CassandraEventStoreModule;
-import org.apache.james.junit.categories.Unstable;
import org.apache.james.lifecycle.api.LifecycleUtil;
import org.apache.james.metrics.api.Gauge;
import org.apache.james.metrics.tests.RecordingMetricFactory;
@@ -93,7 +92,6 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Nested;
import org.junit.jupiter.api.RepeatedTest;
-import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
import org.mockito.ArgumentCaptor;
@@ -152,7 +150,8 @@ class RabbitMQMailQueueTest {
.sizeMetricsEnabled(true)
.build(),
CassandraBlobStoreFactory.forTesting(cassandra.getConf(), new RecordingMetricFactory())
- .passthrough());
+ .passthrough(),
+ MailQueueFactory.prefetchCount(3));
}
@Override
@@ -201,8 +200,15 @@ class RabbitMQMailQueueTest {
}
@RepeatedTest(50)
- @Tag(Unstable.TAG)
- void browseStartShouldBeUpdated(CassandraCluster cassandraCluster) {
+ void browseStartShouldBeUpdated(CassandraCluster cassandraCluster, MailQueueMetricExtension.MailQueueMetricTestSystem metricTestSystem) throws Exception {
+ setUp(cassandraCluster, metricTestSystem,
+ RabbitMQMailQueueConfiguration.builder()
+ .sizeMetricsEnabled(true)
+ .build(),
+ CassandraBlobStoreFactory.forTesting(cassandraCluster.getConf(), new RecordingMetricFactory())
+ .passthrough(),
+ MailQueueFactory.prefetchCount(1));
+
int emailCount = 250;
StatementRecorder.Selector selector = preparedStatementStartingWith("UPDATE browsestart");
@@ -227,8 +233,15 @@ class RabbitMQMailQueueTest {
}
@RepeatedTest(50)
- @Tag(Unstable.TAG)
- void contentStartShouldBeUpdated(CassandraCluster cassandraCluster) {
+ void contentStartShouldBeUpdated(CassandraCluster cassandraCluster, MailQueueMetricExtension.MailQueueMetricTestSystem metricTestSystem) throws Exception {
+ setUp(cassandraCluster, metricTestSystem,
+ RabbitMQMailQueueConfiguration.builder()
+ .sizeMetricsEnabled(true)
+ .build(),
+ CassandraBlobStoreFactory.forTesting(cassandraCluster.getConf(), new RecordingMetricFactory())
+ .passthrough(),
+ MailQueueFactory.prefetchCount(1));
+
int emailCount = 250;
StatementRecorder.Selector selector = preparedStatementStartingWith("UPDATE contentstart");
@@ -940,7 +953,8 @@ class RabbitMQMailQueueTest {
.sizeMetricsEnabled(false)
.build(),
CassandraBlobStoreFactory.forTesting(cassandra.getConf(), new RecordingMetricFactory())
- .passthrough());
+ .passthrough(),
+ MailQueueFactory.prefetchCount(3));
}
@Test
@@ -963,7 +977,8 @@ class RabbitMQMailQueueTest {
.sizeMetricsEnabled(true)
.build(),
CassandraBlobStoreFactory.forTesting(cassandra.getConf(), new RecordingMetricFactory())
- .deduplication());
+ .deduplication(),
+ MailQueueFactory.prefetchCount(3));
}
@Test
@@ -998,7 +1013,8 @@ class RabbitMQMailQueueTest {
}
}
- private void setUp(CassandraCluster cassandra, MailQueueMetricExtension.MailQueueMetricTestSystem metricTestSystem, RabbitMQMailQueueConfiguration configuration, BlobStore blobStore) throws Exception {
+ private void setUp(CassandraCluster cassandra, MailQueueMetricExtension.MailQueueMetricTestSystem metricTestSystem, RabbitMQMailQueueConfiguration configuration,
+ BlobStore blobStore, MailQueueFactory.PrefetchCount prefetchCount) throws Exception {
MimeMessageStore.Factory mimeMessageStoreFactory = MimeMessageStore.factory(blobStore);
clock = new UpdatableTickingClock(IN_SLICE_1);
@@ -1025,6 +1041,6 @@ class RabbitMQMailQueueTest {
configuration);
mqManagementApi = new RabbitMQMailQueueManagement(rabbitMQExtension.managementAPI());
mailQueueFactory = new RabbitMQMailQueueFactory(rabbitMQExtension.getSender(), mqManagementApi, factory, rabbitMQExtension.getRabbitMQ().getConfiguration());
- mailQueue = mailQueueFactory.createQueue(SPOOL, MailQueueFactory.prefetchCount(3));
+ mailQueue = mailQueueFactory.createQueue(SPOOL, prefetchCount);
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 02/09: JAMES-3924 Test and fix browse start updates
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch 3.7.x
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 944e9a9970143247b72b15bd4aed9e0d3114abaa
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Wed May 3 11:25:28 2023 +0700
JAMES-3924 Test and fix browse start updates
-> Propose unit tests for browseStart updates using Cassandra session instrumentation
(Ease testing a behaviour not visible at the API level)
-> The update limitation to not occur more than once er slice was not correctly applied
-> Fix a couple of warnings
---
.../cassandra/CassandraMailQueueMailDelete.java | 13 +++++-----
.../queue/rabbitmq/RabbitMQMailQueueTest.java | 29 +++++++++++++++++++++-
2 files changed, 35 insertions(+), 7 deletions(-)
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueMailDelete.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueMailDelete.java
index 1253dc11d4..b5ef41f2c9 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueMailDelete.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueMailDelete.java
@@ -93,12 +93,13 @@ public class CassandraMailQueueMailDelete {
}
private Mono<Instant> findNewBrowseStart(MailQueueName mailQueueName) {
- Slice currentSlice = Slice.of(clock.instant());
+ Instant now= clock.instant();
return browseStartDao.findBrowseStart(mailQueueName)
- .filter(browseStart -> browseStart.isBefore(currentSlice.getStartSliceInstant()))
- .flatMapMany(browseStart -> cassandraMailQueueBrowser.browseReferences(mailQueueName, browseStart))
- .map(enqueuedItem -> enqueuedItem.getSlicingContext().getTimeRangeStart())
- .next();
+ .filter(browseStart -> browseStart.isBefore(now.minus(configuration.getSliceWindow())))
+ .flatMap(browseStart -> cassandraMailQueueBrowser.browseReferences(mailQueueName, browseStart)
+ .map(enqueuedItem -> enqueuedItem.getSlicingContext().getTimeRangeStart())
+ .next()
+ .filter(newBrowseStart -> newBrowseStart.isAfter(browseStart)));
}
private Mono<Void> updateNewBrowseStart(MailQueueName mailQueueName, Instant newBrowseStartInstant) {
@@ -128,6 +129,6 @@ public class CassandraMailQueueMailDelete {
private boolean shouldUpdateBrowseStart() {
int threshold = configuration.getUpdateBrowseStartPace();
- return Math.abs(ThreadLocalRandom.current().nextInt()) % threshold == 0;
+ return ThreadLocalRandom.current().nextInt(threshold) % threshold == 0;
}
}
diff --git a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueTest.java b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueTest.java
index 8fa34fa0d2..d466ac69d9 100644
--- a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueTest.java
+++ b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueTest.java
@@ -24,6 +24,7 @@ import static java.time.temporal.ChronoUnit.HOURS;
import static org.apache.james.backends.cassandra.Scenario.Builder.executeNormally;
import static org.apache.james.backends.cassandra.Scenario.Builder.fail;
import static org.apache.james.backends.cassandra.Scenario.Builder.returnEmpty;
+import static org.apache.james.backends.cassandra.StatementRecorder.Selector.preparedStatementStartingWith;
import static org.apache.james.backends.rabbitmq.Constants.EMPTY_ROUTING_KEY;
import static org.apache.james.queue.api.Mails.defaultMail;
import static org.apache.james.queue.api.Mails.defaultMailNoRecipient;
@@ -51,6 +52,7 @@ import java.util.stream.Stream;
import org.apache.james.backends.cassandra.CassandraCluster;
import org.apache.james.backends.cassandra.CassandraClusterExtension;
+import org.apache.james.backends.cassandra.StatementRecorder;
import org.apache.james.backends.cassandra.components.CassandraModule;
import org.apache.james.backends.cassandra.versions.CassandraSchemaVersionModule;
import org.apache.james.backends.rabbitmq.RabbitMQExtension;
@@ -87,6 +89,7 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.RepeatedTest;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
import org.mockito.ArgumentCaptor;
@@ -104,7 +107,7 @@ import reactor.rabbitmq.Sender;
class RabbitMQMailQueueTest {
private static final HashBlobId.Factory BLOB_ID_FACTORY = new HashBlobId.Factory();
private static final int THREE_BUCKET_COUNT = 3;
- private static final int UPDATE_BROWSE_START_PACE = 2;
+ private static final int UPDATE_BROWSE_START_PACE = 10;
private static final Duration ONE_HOUR_SLICE_WINDOW = Duration.ofHours(1);
private static final org.apache.james.queue.api.MailQueueName SPOOL = org.apache.james.queue.api.MailQueueName.of("spool");
private static final Instant IN_SLICE_1 = Instant.parse("2007-12-03T10:15:30.00Z");
@@ -193,6 +196,30 @@ class RabbitMQMailQueueTest {
"5-1", "5-2", "5-3", "5-4", "5-5");
}
+ @Test
+ void browseStartShouldBeUpdated(CassandraCluster cassandraCluster) {
+ int emailCount = 100;
+
+ StatementRecorder statementRecorder = new StatementRecorder();
+ cassandraCluster.getConf().recordStatements(statementRecorder);
+
+ clock.setInstant(IN_SLICE_1);
+ enqueueSomeMails(namePatternForSlice(1), emailCount);
+ dequeueMails(emailCount);
+
+ clock.setInstant(IN_SLICE_2);
+ enqueueSomeMails(namePatternForSlice(2), emailCount);
+ dequeueMails(emailCount);
+
+ clock.setInstant(IN_SLICE_3);
+ enqueueSomeMails(namePatternForSlice(3), emailCount);
+ dequeueMails(emailCount);
+
+ // The actual rate of update should actually be lower than the update probability.
+ assertThat(statementRecorder.listExecutedStatements(preparedStatementStartingWith("UPDATE browsestart")))
+ .hasSizeBetween(2, 5);
+ }
+
@Test
void dequeueShouldDeleteBlobs(CassandraCluster cassandra) throws Exception {
String name1 = "myMail1";
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 01/09: JAMES-3924 Allow conditional Cassandra statement recording
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch 3.7.x
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit b6bdfa0703a6c22e6836bf142fef5579763681a6
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Fri Jun 30 21:49:36 2023 +0700
JAMES-3924 Allow conditional Cassandra statement recording
Upon massive operations, recording all statements leads to OOM
---
.../apache/james/backends/cassandra/StatementRecorder.java | 8 ++++++--
.../apache/james/backends/cassandra/TestingSession.java | 14 +++++++++++++-
.../james/backends/cassandra/TestingSessionTest.java | 6 +++---
3 files changed, 22 insertions(+), 6 deletions(-)
diff --git a/backends-common/cassandra/src/test/java/org/apache/james/backends/cassandra/StatementRecorder.java b/backends-common/cassandra/src/test/java/org/apache/james/backends/cassandra/StatementRecorder.java
index 443ff02e07..840f7799bc 100644
--- a/backends-common/cassandra/src/test/java/org/apache/james/backends/cassandra/StatementRecorder.java
+++ b/backends-common/cassandra/src/test/java/org/apache/james/backends/cassandra/StatementRecorder.java
@@ -52,14 +52,18 @@ public class StatementRecorder {
List<Statement> select(List<Statement> statements);
}
+ private final StatementRecorder.Selector selector;
private final ConcurrentLinkedDeque statements;
- public StatementRecorder() {
+ StatementRecorder(Selector selector) {
+ this.selector = selector;
statements = new ConcurrentLinkedDeque();
}
void recordStatement(Statement statement) {
- statements.add(statement);
+ if (statements.addAll(selector.select(ImmutableList.of(statement)))) {
+ System.out.println("recordStatement");
+ }
}
public List<Statement> listExecutedStatements() {
diff --git a/backends-common/cassandra/src/test/java/org/apache/james/backends/cassandra/TestingSession.java b/backends-common/cassandra/src/test/java/org/apache/james/backends/cassandra/TestingSession.java
index 4b9e23545e..7ca271592a 100644
--- a/backends-common/cassandra/src/test/java/org/apache/james/backends/cassandra/TestingSession.java
+++ b/backends-common/cassandra/src/test/java/org/apache/james/backends/cassandra/TestingSession.java
@@ -69,8 +69,20 @@ public class TestingSession implements Session {
this.scenario = Scenario.combine(hooks);
}
- public void recordStatements(StatementRecorder statementRecorder) {
+ public StatementRecorder recordStatements(StatementRecorder statementRecorder) {
this.statementRecorder = Optional.of(statementRecorder);
+ return statementRecorder;
+ }
+
+
+ public StatementRecorder recordStatements() {
+ return recordStatements(StatementRecorder.Selector.ALL);
+ }
+
+ public StatementRecorder recordStatements(StatementRecorder.Selector selector) {
+ StatementRecorder statementRecorder = new StatementRecorder(selector);
+ this.statementRecorder = Optional.of(statementRecorder);
+ return statementRecorder;
}
public void stopRecordingStatements() {
diff --git a/backends-common/cassandra/src/test/java/org/apache/james/backends/cassandra/TestingSessionTest.java b/backends-common/cassandra/src/test/java/org/apache/james/backends/cassandra/TestingSessionTest.java
index e6cd6073cf..4b1e93d999 100644
--- a/backends-common/cassandra/src/test/java/org/apache/james/backends/cassandra/TestingSessionTest.java
+++ b/backends-common/cassandra/src/test/java/org/apache/james/backends/cassandra/TestingSessionTest.java
@@ -89,7 +89,7 @@ class TestingSessionTest {
@Test
void recordStatementsShouldKeepTraceOfExecutedStatement(CassandraCluster cassandra) {
- StatementRecorder statementRecorder = new StatementRecorder();
+ StatementRecorder statementRecorder = new StatementRecorder(Selector.ALL);
cassandra.getConf().recordStatements(statementRecorder);
dao.getCurrentSchemaVersion().block();
@@ -101,7 +101,7 @@ class TestingSessionTest {
@Test
void recordStatementsShouldKeepTraceOfExecutedStatements(CassandraCluster cassandra) {
- StatementRecorder statementRecorder = new StatementRecorder();
+ StatementRecorder statementRecorder = new StatementRecorder(Selector.ALL);
cassandra.getConf().recordStatements(statementRecorder);
dao.updateVersion(new SchemaVersion(36)).block();
@@ -118,7 +118,7 @@ class TestingSessionTest {
void recordStatementsShouldNotKeepTraceOfExecutedStatementsBeforeRecording(CassandraCluster cassandra) {
dao.getCurrentSchemaVersion().block();
- StatementRecorder statementRecorder = new StatementRecorder();
+ StatementRecorder statementRecorder = new StatementRecorder(Selector.ALL);
cassandra.getConf().recordStatements(statementRecorder);
assertThat(statementRecorder.listExecutedStatements())
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 08/09: JAMES-3924 Remove repeated tests
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch 3.7.x
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit c525c6f4e562744bd1603bf29736c08efa5d3730
Author: Benoit TELLIER <bt...@linagora.com>
AuthorDate: Fri Aug 25 12:28:58 2023 +0700
JAMES-3924 Remove repeated tests
2 * 50 * 6s =~ 10 minutes ... too long
---
.../java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueTest.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueTest.java b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueTest.java
index 82f9a3176e..b2aad2c3e9 100644
--- a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueTest.java
+++ b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueTest.java
@@ -199,7 +199,7 @@ class RabbitMQMailQueueTest {
"5-1", "5-2", "5-3", "5-4", "5-5");
}
- @RepeatedTest(50)
+ @Test
void browseStartShouldBeUpdated(CassandraCluster cassandraCluster, MailQueueMetricExtension.MailQueueMetricTestSystem metricTestSystem) throws Exception {
setUp(cassandraCluster, metricTestSystem,
RabbitMQMailQueueConfiguration.builder()
@@ -232,7 +232,7 @@ class RabbitMQMailQueueTest {
.hasSizeBetween(2, 12);
}
- @RepeatedTest(50)
+ @Test
void contentStartShouldBeUpdated(CassandraCluster cassandraCluster, MailQueueMetricExtension.MailQueueMetricTestSystem metricTestSystem) throws Exception {
setUp(cassandraCluster, metricTestSystem,
RabbitMQMailQueueConfiguration.builder()
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 03/09: JAMES-3924 Tests for contentStart updates
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch 3.7.x
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 3a069194f1621da4f0bd624d5f96ed90c37520a3
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Wed May 3 13:14:11 2023 +0700
JAMES-3924 Tests for contentStart updates
---
.../queue/rabbitmq/RabbitMQMailQueueTest.java | 49 ++++++++++++++++++----
1 file changed, 41 insertions(+), 8 deletions(-)
diff --git a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueTest.java b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueTest.java
index d466ac69d9..eee81a381f 100644
--- a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueTest.java
+++ b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueTest.java
@@ -64,6 +64,7 @@ import org.apache.james.blob.cassandra.CassandraBlobStoreFactory;
import org.apache.james.blob.mail.MimeMessageStore;
import org.apache.james.core.builder.MimeMessageBuilder;
import org.apache.james.eventsourcing.eventstore.cassandra.CassandraEventStoreModule;
+import org.apache.james.lifecycle.api.LifecycleUtil;
import org.apache.james.metrics.api.Gauge;
import org.apache.james.metrics.tests.RecordingMetricFactory;
import org.apache.james.queue.api.MailQueue;
@@ -83,6 +84,7 @@ import org.apache.james.queue.rabbitmq.view.cassandra.model.BucketedSlices.Slice
import org.apache.james.util.streams.Iterators;
import org.apache.james.utils.UpdatableTickingClock;
import org.apache.mailet.Mail;
+import org.apache.mailet.base.test.FakeMail;
import org.assertj.core.api.SoftAssertions;
import org.awaitility.Awaitility;
import org.junit.jupiter.api.AfterEach;
@@ -200,8 +202,9 @@ class RabbitMQMailQueueTest {
void browseStartShouldBeUpdated(CassandraCluster cassandraCluster) {
int emailCount = 100;
- StatementRecorder statementRecorder = new StatementRecorder();
- cassandraCluster.getConf().recordStatements(statementRecorder);
+ StatementRecorder.Selector selector = preparedStatementStartingWith("UPDATE browsestart");
+ StatementRecorder statementRecorder = cassandraCluster.getConf()
+ .recordStatements(selector);
clock.setInstant(IN_SLICE_1);
enqueueSomeMails(namePatternForSlice(1), emailCount);
@@ -216,8 +219,32 @@ class RabbitMQMailQueueTest {
dequeueMails(emailCount);
// The actual rate of update should actually be lower than the update probability.
- assertThat(statementRecorder.listExecutedStatements(preparedStatementStartingWith("UPDATE browsestart")))
- .hasSizeBetween(2, 5);
+ assertThat(statementRecorder.listExecutedStatements(selector))
+ .hasSizeBetween(2, 9);
+ }
+
+ @Test
+ void contentStartShouldBeUpdated(CassandraCluster cassandraCluster) {
+ int emailCount = 100;
+
+ StatementRecorder.Selector selector = preparedStatementStartingWith("UPDATE contentstart");
+ StatementRecorder statementRecorder = cassandraCluster.getConf().recordStatements(selector);
+
+ clock.setInstant(IN_SLICE_1);
+ enqueueSomeMails(namePatternForSlice(1), emailCount);
+ dequeueMails(emailCount);
+
+ clock.setInstant(IN_SLICE_2);
+ enqueueSomeMails(namePatternForSlice(2), emailCount);
+ dequeueMails(emailCount);
+
+ clock.setInstant(IN_SLICE_3);
+ enqueueSomeMails(namePatternForSlice(3), emailCount);
+ dequeueMails(emailCount);
+
+ // The actual rate of update should actually be lower than the update probability.
+ assertThat(statementRecorder.listExecutedStatements(selector))
+ .hasSizeBetween(2, 9);
}
@Test
@@ -667,9 +694,13 @@ class RabbitMQMailQueueTest {
private void enqueueSomeMails(Function<Integer, String> namePattern, int emailCount) {
IntStream.rangeClosed(1, emailCount)
- .forEach(Throwing.intConsumer(i -> enQueue(defaultMail()
- .name(namePattern.apply(i))
- .build())));
+ .forEach(Throwing.intConsumer(i -> {
+ FakeMail mail = defaultMail()
+ .name(namePattern.apply(i))
+ .build();
+ enQueue(mail);
+ LifecycleUtil.dispose(mail);
+ }));
}
private void dequeueMails(int times) {
@@ -688,7 +719,9 @@ class RabbitMQMailQueueTest {
.subscribe();
try {
- await().untilAsserted(() -> assertThat(counter.get()).isGreaterThanOrEqualTo(times));
+ await()
+ .atMost(Duration.ofMinutes(10))
+ .untilAsserted(() -> assertThat(counter.get()).isGreaterThanOrEqualTo(times));
} finally {
disposable.dispose();
}
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 09/09: JAMES-3924 Fix rebase issues during 3.7.x backport
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch 3.7.x
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 7d46378ef498039ac3eb37f6aa4319e192a9581f
Author: Benoit TELLIER <bt...@linagora.com>
AuthorDate: Sun Aug 27 01:37:46 2023 +0700
JAMES-3924 Fix rebase issues during 3.7.x backport
---
.../org/apache/james/backends/cassandra/StatementRecorder.java | 4 ++++
.../org/apache/james/queue/rabbitmq/RabbitMQMailQueueTest.java | 10 +++++++---
2 files changed, 11 insertions(+), 3 deletions(-)
diff --git a/backends-common/cassandra/src/test/java/org/apache/james/backends/cassandra/StatementRecorder.java b/backends-common/cassandra/src/test/java/org/apache/james/backends/cassandra/StatementRecorder.java
index 840f7799bc..56d47830dc 100644
--- a/backends-common/cassandra/src/test/java/org/apache/james/backends/cassandra/StatementRecorder.java
+++ b/backends-common/cassandra/src/test/java/org/apache/james/backends/cassandra/StatementRecorder.java
@@ -55,6 +55,10 @@ public class StatementRecorder {
private final StatementRecorder.Selector selector;
private final ConcurrentLinkedDeque statements;
+ public StatementRecorder() {
+ this(Selector.ALL);
+ }
+
StatementRecorder(Selector selector) {
this.selector = selector;
statements = new ConcurrentLinkedDeque();
diff --git a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueTest.java b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueTest.java
index b2aad2c3e9..5f06257f69 100644
--- a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueTest.java
+++ b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueTest.java
@@ -211,7 +211,7 @@ class RabbitMQMailQueueTest {
int emailCount = 250;
- StatementRecorder.Selector selector = preparedStatementStartingWith("UPDATE browsestart");
+ StatementRecorder.Selector selector = preparedStatementStartingWith("UPDATE browseStart");
StatementRecorder statementRecorder = cassandraCluster.getConf()
.recordStatements(selector);
@@ -244,7 +244,7 @@ class RabbitMQMailQueueTest {
int emailCount = 250;
- StatementRecorder.Selector selector = preparedStatementStartingWith("UPDATE contentstart");
+ StatementRecorder.Selector selector = preparedStatementStartingWith("UPDATE contentStart");
StatementRecorder statementRecorder = cassandraCluster.getConf().recordStatements(selector);
clock.setInstant(IN_SLICE_1);
@@ -1005,7 +1005,11 @@ class RabbitMQMailQueueTest {
.concatMap(mailQueueItem ->
Mono.fromCallable(() -> {
assertThat(mailQueueItem.getMail().getMessage().getContent()).isEqualTo(identicalContent);
- mailQueueItem.done(true);
+ try {
+ mailQueueItem.done(true);
+ } catch (Exception e) {
+ // Ignore
+ }
return mailQueueItem;
}).subscribeOn(Schedulers.elastic()))
.collectList()
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 04/09: JAMES-3924 Improve dequeuer Lifecycle management
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch 3.7.x
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 7674d94aa9a581bc0df4725070797cc296e28326
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Fri Jun 30 21:50:53 2023 +0700
JAMES-3924 Improve dequeuer Lifecycle management
---
.../src/main/java/org/apache/james/queue/rabbitmq/Dequeuer.java | 3 +++
1 file changed, 3 insertions(+)
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 671400a06d..214537df36 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
@@ -25,6 +25,7 @@ import java.util.function.Consumer;
import org.apache.james.backends.rabbitmq.ReceiverProvider;
import org.apache.james.blob.api.ObjectNotFoundException;
+import org.apache.james.lifecycle.api.LifecycleUtil;
import org.apache.james.metrics.api.Metric;
import org.apache.james.metrics.api.MetricFactory;
import org.apache.james.queue.api.MailQueue;
@@ -140,9 +141,11 @@ class Dequeuer {
if (success) {
dequeueMetric.increment();
response.ack();
+ LifecycleUtil.dispose(mailWithEnqueueId.getMail());
Mono.from(mailQueueView.delete(DeleteCondition.withEnqueueId(mailWithEnqueueId.getEnqueueId(), mailWithEnqueueId.getBlobIds()))).block();
} else {
response.nack(REQUEUE);
+ LifecycleUtil.dispose(mailWithEnqueueId.getMail());
}
};
}
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 06/09: JAMES-3924 Fix checkstyles and some unstable tests
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch 3.7.x
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 952271012d1e509632c32b21a685d5e538d28090
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Sat Jul 1 19:03:40 2023 +0700
JAMES-3924 Fix checkstyles and some unstable tests
Co-authored-by: Rene Cordier <rc...@linagora.com>
Co-authored-by: Quan Tran <hq...@linagora.com>
---
.../view/cassandra/CassandraMailQueueBrowser.java | 1 -
.../cassandra/CassandraMailQueueMailDelete.java | 2 +-
.../queue/rabbitmq/RabbitMQMailQueueTest.java | 61 ++++++++++++----------
3 files changed, 33 insertions(+), 31 deletions(-)
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 9c1241351d..40ca47a197 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
@@ -21,7 +21,6 @@ package org.apache.james.queue.rabbitmq.view.cassandra;
import static org.apache.james.queue.rabbitmq.view.cassandra.model.BucketedSlices.BucketId;
import static org.apache.james.queue.rabbitmq.view.cassandra.model.BucketedSlices.Slice;
-import static org.apache.james.util.ReactorUtils.DEFAULT_CONCURRENCY;
import java.time.Clock;
import java.time.Instant;
diff --git a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueMailDelete.java b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueMailDelete.java
index b5ef41f2c9..76a599a3bf 100644
--- a/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueMailDelete.java
+++ b/server/queue/queue-rabbitmq/src/main/java/org/apache/james/queue/rabbitmq/view/cassandra/CassandraMailQueueMailDelete.java
@@ -93,7 +93,7 @@ public class CassandraMailQueueMailDelete {
}
private Mono<Instant> findNewBrowseStart(MailQueueName mailQueueName) {
- Instant now= clock.instant();
+ Instant now = clock.instant();
return browseStartDao.findBrowseStart(mailQueueName)
.filter(browseStart -> browseStart.isBefore(now.minus(configuration.getSliceWindow())))
.flatMap(browseStart -> cassandraMailQueueBrowser.browseReferences(mailQueueName, browseStart)
diff --git a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueTest.java b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueTest.java
index eee81a381f..25623493b8 100644
--- a/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueTest.java
+++ b/server/queue/queue-rabbitmq/src/test/java/org/apache/james/queue/rabbitmq/RabbitMQMailQueueTest.java
@@ -64,10 +64,12 @@ import org.apache.james.blob.cassandra.CassandraBlobStoreFactory;
import org.apache.james.blob.mail.MimeMessageStore;
import org.apache.james.core.builder.MimeMessageBuilder;
import org.apache.james.eventsourcing.eventstore.cassandra.CassandraEventStoreModule;
+import org.apache.james.junit.categories.Unstable;
import org.apache.james.lifecycle.api.LifecycleUtil;
import org.apache.james.metrics.api.Gauge;
import org.apache.james.metrics.tests.RecordingMetricFactory;
import org.apache.james.queue.api.MailQueue;
+import org.apache.james.queue.api.MailQueueFactory;
import org.apache.james.queue.api.MailQueueMetricContract;
import org.apache.james.queue.api.MailQueueMetricExtension;
import org.apache.james.queue.api.ManageableMailQueue;
@@ -86,12 +88,12 @@ import org.apache.james.utils.UpdatableTickingClock;
import org.apache.mailet.Mail;
import org.apache.mailet.base.test.FakeMail;
import org.assertj.core.api.SoftAssertions;
-import org.awaitility.Awaitility;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Nested;
import org.junit.jupiter.api.RepeatedTest;
+import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
import org.mockito.ArgumentCaptor;
@@ -109,7 +111,7 @@ import reactor.rabbitmq.Sender;
class RabbitMQMailQueueTest {
private static final HashBlobId.Factory BLOB_ID_FACTORY = new HashBlobId.Factory();
private static final int THREE_BUCKET_COUNT = 3;
- private static final int UPDATE_BROWSE_START_PACE = 10;
+ private static final int UPDATE_BROWSE_START_PACE = 25;
private static final Duration ONE_HOUR_SLICE_WINDOW = Duration.ofHours(1);
private static final org.apache.james.queue.api.MailQueueName SPOOL = org.apache.james.queue.api.MailQueueName.of("spool");
private static final Instant IN_SLICE_1 = Instant.parse("2007-12-03T10:15:30.00Z");
@@ -198,13 +200,14 @@ class RabbitMQMailQueueTest {
"5-1", "5-2", "5-3", "5-4", "5-5");
}
- @Test
+ @RepeatedTest(50)
+ @Tag(Unstable.TAG)
void browseStartShouldBeUpdated(CassandraCluster cassandraCluster) {
- int emailCount = 100;
+ int emailCount = 250;
StatementRecorder.Selector selector = preparedStatementStartingWith("UPDATE browsestart");
StatementRecorder statementRecorder = cassandraCluster.getConf()
- .recordStatements(selector);
+ .recordStatements(selector);
clock.setInstant(IN_SLICE_1);
enqueueSomeMails(namePatternForSlice(1), emailCount);
@@ -220,12 +223,13 @@ class RabbitMQMailQueueTest {
// The actual rate of update should actually be lower than the update probability.
assertThat(statementRecorder.listExecutedStatements(selector))
- .hasSizeBetween(2, 9);
+ .hasSizeBetween(2, 12);
}
- @Test
+ @RepeatedTest(50)
+ @Tag(Unstable.TAG)
void contentStartShouldBeUpdated(CassandraCluster cassandraCluster) {
- int emailCount = 100;
+ int emailCount = 250;
StatementRecorder.Selector selector = preparedStatementStartingWith("UPDATE contentstart");
StatementRecorder statementRecorder = cassandraCluster.getConf().recordStatements(selector);
@@ -244,7 +248,7 @@ class RabbitMQMailQueueTest {
// The actual rate of update should actually be lower than the update probability.
assertThat(statementRecorder.listExecutedStatements(selector))
- .hasSizeBetween(2, 9);
+ .hasSizeBetween(2, 12);
}
@Test
@@ -352,7 +356,7 @@ class RabbitMQMailQueueTest {
@Test
void enqueuedEmailsShouldEventuallyBeCleaned() {
ManageableMailQueue mailQueue = getManageableMailQueue();
- int emailCount = 5;
+ int emailCount = 100;
clock.setInstant(IN_SLICE_1);
enqueueSomeMails(namePatternForSlice(1), emailCount);
@@ -367,10 +371,10 @@ class RabbitMQMailQueueTest {
enqueueSomeMails(namePatternForSlice(5), emailCount);
clock.setInstant(IN_SLICE_7);
- dequeueMails(5);
- dequeueMails(5);
- dequeueMails(5);
- dequeueMails(5);
+ dequeueMails(emailCount);
+ dequeueMails(emailCount);
+ dequeueMails(emailCount);
+ dequeueMails(emailCount);
// ensure slice 1 was cleaned
EnqueuedMailsDAO mailsDAO = new EnqueuedMailsDAO(cassandraCluster.getCassandraCluster().getConf(), new HashBlobId.Factory());
@@ -696,8 +700,8 @@ class RabbitMQMailQueueTest {
IntStream.rangeClosed(1, emailCount)
.forEach(Throwing.intConsumer(i -> {
FakeMail mail = defaultMail()
- .name(namePattern.apply(i))
- .build();
+ .name(namePattern.apply(i))
+ .build();
enQueue(mail);
LifecycleUtil.dispose(mail);
}));
@@ -720,8 +724,8 @@ class RabbitMQMailQueueTest {
try {
await()
- .atMost(Duration.ofMinutes(10))
- .untilAsserted(() -> assertThat(counter.get()).isGreaterThanOrEqualTo(times));
+ .atMost(Duration.ofMinutes(10))
+ .untilAsserted(() -> assertThat(counter.get()).isGreaterThanOrEqualTo(times));
} finally {
disposable.dispose();
}
@@ -980,18 +984,17 @@ class RabbitMQMailQueueTest {
.setSubject(identicalSubject)
.setText(identicalContent))
.build());
-
- List<MailQueue.MailQueueItem> items = dequeueFlux.take(2)
- .concatMap(mailQueueItem -> Mono.fromCallable(() -> {
- mailQueueItem.done(true);
- return mailQueueItem;
- }))
+
+ Flux.from(mailQueue.deQueue())
+ .take(2)
+ .concatMap(mailQueueItem ->
+ Mono.fromCallable(() -> {
+ assertThat(mailQueueItem.getMail().getMessage().getContent()).isEqualTo(identicalContent);
+ mailQueueItem.done(true);
+ return mailQueueItem;
+ }).subscribeOn(Schedulers.elastic()))
.collectList()
.block(Duration.ofSeconds(10));
-
- assertThat(items)
- .allSatisfy(Throwing.consumer(item -> assertThat(item.getMail().getMessage().getContent())
- .isEqualTo(identicalContent)));
}
}
@@ -1022,6 +1025,6 @@ class RabbitMQMailQueueTest {
configuration);
mqManagementApi = new RabbitMQMailQueueManagement(rabbitMQExtension.managementAPI());
mailQueueFactory = new RabbitMQMailQueueFactory(rabbitMQExtension.getSender(), mqManagementApi, factory, rabbitMQExtension.getRabbitMQ().getConfiguration());
- mailQueue = mailQueueFactory.createQueue(SPOOL);
+ mailQueue = mailQueueFactory.createQueue(SPOOL, MailQueueFactory.prefetchCount(3));
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org
[james-project] 05/09: JAMES-3924 Review CassandraMailQueueBrowser parallelism
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch 3.7.x
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 9f84f10732e6d58ca9e41c9481695cd70c287c56
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Fri Jun 30 21:51:29 2023 +0700
JAMES-3924 Review CassandraMailQueueBrowser parallelism
---
.../rabbitmq/view/cassandra/CassandraMailQueueBrowser.java | 11 +++++------
1 file changed, 5 insertions(+), 6 deletions(-)
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 64f85be0b1..9c1241351d 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), 4);
+ .concatMap(slice -> browseSlice(queueName, slice));
}
private Mono<Pair<EnqueuedItem, Mail>> toMailFuture(EnqueuedItemWithSlicingContext enqueuedItemWithSlicingContext) {
@@ -156,15 +156,14 @@ public class CassandraMailQueueBrowser {
}
private Flux<EnqueuedItemWithSlicingContext> browseSlice(MailQueueName queueName, Slice slice) {
- return
- allBucketIds()
- .concatMap(bucketId -> browseBucket(queueName, slice, bucketId), DEFAULT_CONCURRENCY)
- .sort(Comparator.comparing(enqueuedMail -> enqueuedMail.getEnqueuedItem().getEnqueuedTime()));
+ return allBucketIds()
+ .concatMap(bucketId -> browseBucket(queueName, slice, bucketId), 4)
+ .sort(Comparator.comparing(enqueuedMail -> enqueuedMail.getEnqueuedItem().getEnqueuedTime()));
}
private Flux<EnqueuedItemWithSlicingContext> browseBucket(MailQueueName queueName, Slice slice, BucketId bucketId) {
return enqueuedMailsDao.selectEnqueuedMails(queueName, slice, bucketId)
- .filterWhen(mailReference -> deletedMailsDao.isStillEnqueued(queueName, mailReference.getEnqueuedItem().getEnqueueId()), DEFAULT_CONCURRENCY);
+ .filterWhen(mailReference -> deletedMailsDao.isStillEnqueued(queueName, mailReference.getEnqueuedItem().getEnqueueId()), 4);
}
private Flux<Slice> allSlicesStartingAt(Instant browseStart) {
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org