You are viewing a plain text version of this content. The canonical link for it is here.
Posted to server-dev@james.apache.org by bt...@apache.org on 2019/09/23 02:17:13 UTC
[james-project] branch master updated (c89fe64 -> 70f3f27)
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git.
from c89fe64 JAMES-2886 Package apache-mailet-crypto as a JAR with dependency
new da13fb2 JAMES-2885 avoid concurrent nexModSeq during flags update
new e644a6f JAMES-2813 Demonstrate how exclusive consumer works
new c89a8b2 JAMES-2721 Increase parallelism for mailbox/cassandra projects
new 70f3f27 Modify some commiters no longer existing emails
The 4 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:
.../apache/james/backend/rabbitmq/Constants.java | 1 +
.../james/backend/rabbitmq/InMemoryConsumer.java | 6 +-
.../backend/rabbitmq/RabbitMQClusterTest.java | 2 +-
.../james/backend/rabbitmq/RabbitMQTest.java | 79 ++++++++++++++++++++++
mailbox/cassandra/pom.xml | 13 ++++
.../cassandra/mail/CassandraMessageIdMapper.java | 9 ++-
pom.xml | 4 +-
7 files changed, 103 insertions(+), 11 deletions(-)
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 01/04: JAMES-2885 avoid concurrent nexModSeq during
flags update
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit da13fb2739f8bd8bca7a12f624783e1983592d64
Author: Rémi Kowalski <rk...@linagora.com>
AuthorDate: Tue Sep 10 10:55:33 2019 +0200
JAMES-2885 avoid concurrent nexModSeq during flags update
---
.../james/mailbox/cassandra/mail/CassandraMessageIdMapper.java | 9 ++++-----
1 file changed, 4 insertions(+), 5 deletions(-)
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdMapper.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdMapper.java
index 33db97c..a6637b8 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdMapper.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdMapper.java
@@ -210,17 +210,16 @@ public class CassandraMessageIdMapper implements MessageIdMapper {
return Flux.fromIterable(mailboxIds)
.distinct()
.map(mailboxId -> (CassandraId) mailboxId)
- .filter(mailboxId -> haveMetaData(messageId, mailboxId))
- .flatMap(mailboxId -> flagsUpdateWithRetry(newState, updateMode, mailboxId, messageId))
+ .filterWhen(mailboxId -> haveMetaData(messageId, mailboxId))
+ .concatMap(mailboxId -> flagsUpdateWithRetry(newState, updateMode, mailboxId, messageId))
.flatMap(this::updateCounts)
.collect(Guavate.toImmutableMap(Pair::getLeft, Pair::getRight))
.block();
}
- private boolean haveMetaData(MessageId messageId, CassandraId mailboxId) {
+ private Mono<Boolean> haveMetaData(MessageId messageId, CassandraId mailboxId) {
return imapUidDAO.retrieve((CassandraMessageId) messageId, Optional.of(mailboxId))
- .hasElements()
- .block();
+ .hasElements();
}
private Mono<Pair<MailboxId, UpdatedFlags>> flagsUpdateWithRetry(Flags newState, MessageManager.FlagsUpdateMode updateMode, MailboxId mailboxId, MessageId messageId) {
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 02/04: JAMES-2813 Demonstrate how exclusive
consumer works
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit e644a6f591a3fe4423538aa98941d53b048560d0
Author: Matthieu Baechler <ma...@apache.org>
AuthorDate: Wed Aug 21 13:46:10 2019 +0200
JAMES-2813 Demonstrate how exclusive consumer works
---
.../apache/james/backend/rabbitmq/Constants.java | 1 +
.../james/backend/rabbitmq/InMemoryConsumer.java | 6 +-
.../backend/rabbitmq/RabbitMQClusterTest.java | 2 +-
.../james/backend/rabbitmq/RabbitMQTest.java | 79 ++++++++++++++++++++++
4 files changed, 84 insertions(+), 4 deletions(-)
diff --git a/backends-common/rabbitmq/src/main/java/org/apache/james/backend/rabbitmq/Constants.java b/backends-common/rabbitmq/src/main/java/org/apache/james/backend/rabbitmq/Constants.java
index 4fc2410..e6cea0f 100644
--- a/backends-common/rabbitmq/src/main/java/org/apache/james/backend/rabbitmq/Constants.java
+++ b/backends-common/rabbitmq/src/main/java/org/apache/james/backend/rabbitmq/Constants.java
@@ -26,6 +26,7 @@ public interface Constants {
boolean DURABLE = true;
boolean AUTO_DELETE = true;
boolean EXCLUSIVE = true;
+ boolean NO_LOCAL = true;
boolean AUTO_ACK = true;
boolean MULTIPLE = true;
diff --git a/backends-common/rabbitmq/src/test/java/org/apache/james/backend/rabbitmq/InMemoryConsumer.java b/backends-common/rabbitmq/src/test/java/org/apache/james/backend/rabbitmq/InMemoryConsumer.java
index 712c70d..3cc93ae 100644
--- a/backends-common/rabbitmq/src/test/java/org/apache/james/backend/rabbitmq/InMemoryConsumer.java
+++ b/backends-common/rabbitmq/src/test/java/org/apache/james/backend/rabbitmq/InMemoryConsumer.java
@@ -32,14 +32,14 @@ public class InMemoryConsumer extends DefaultConsumer {
@FunctionalInterface
interface Operation {
- void perform();
+ void perform(Envelope envelope, byte[] body) throws IOException;
}
private final ConcurrentLinkedQueue<Integer> messages;
private final Operation operation;
public InMemoryConsumer(Channel channel) {
- this(channel, () -> { });
+ this(channel, (envelope, body) -> { });
}
public InMemoryConsumer(Channel channel, Operation operation) {
@@ -50,7 +50,7 @@ public class InMemoryConsumer extends DefaultConsumer {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
- operation.perform();
+ operation.perform(envelope, body);
Integer payload = Integer.valueOf(new String(body, StandardCharsets.UTF_8));
messages.add(payload);
}
diff --git a/backends-common/rabbitmq/src/test/java/org/apache/james/backend/rabbitmq/RabbitMQClusterTest.java b/backends-common/rabbitmq/src/test/java/org/apache/james/backend/rabbitmq/RabbitMQClusterTest.java
index 348d3ff..b4066a1 100644
--- a/backends-common/rabbitmq/src/test/java/org/apache/james/backend/rabbitmq/RabbitMQClusterTest.java
+++ b/backends-common/rabbitmq/src/test/java/org/apache/james/backend/rabbitmq/RabbitMQClusterTest.java
@@ -278,7 +278,7 @@ class RabbitMQClusterTest {
AtomicInteger counter = new AtomicInteger(0);
InMemoryConsumer consumer = new InMemoryConsumer(resilientChannel,
- () -> stopWhenHalfProcessed(cluster, nbMessages, counter));
+ (envelope, body) -> stopWhenHalfProcessed(cluster, nbMessages, counter));
resilientChannel.basicConsume(QUEUE, consumer);
awaitAtMostOneMinute.until(() -> consumer.getConsumedMessages().size() == nbMessages);
diff --git a/backends-common/rabbitmq/src/test/java/org/apache/james/backend/rabbitmq/RabbitMQTest.java b/backends-common/rabbitmq/src/test/java/org/apache/james/backend/rabbitmq/RabbitMQTest.java
index b6f1b31..d858258 100644
--- a/backends-common/rabbitmq/src/test/java/org/apache/james/backend/rabbitmq/RabbitMQTest.java
+++ b/backends-common/rabbitmq/src/test/java/org/apache/james/backend/rabbitmq/RabbitMQTest.java
@@ -23,17 +23,23 @@ import static org.apache.james.backend.rabbitmq.Constants.AUTO_DELETE;
import static org.apache.james.backend.rabbitmq.Constants.DIRECT_EXCHANGE;
import static org.apache.james.backend.rabbitmq.Constants.DURABLE;
import static org.apache.james.backend.rabbitmq.Constants.EXCLUSIVE;
+import static org.apache.james.backend.rabbitmq.Constants.MULTIPLE;
+import static org.apache.james.backend.rabbitmq.Constants.NO_LOCAL;
import static org.apache.james.backend.rabbitmq.Constants.NO_PROPERTIES;
+import static org.apache.james.backend.rabbitmq.Constants.REQUEUE;
import static org.apache.james.backend.rabbitmq.RabbitMQFixture.EXCHANGE_NAME;
import static org.apache.james.backend.rabbitmq.RabbitMQFixture.ROUTING_KEY;
import static org.apache.james.backend.rabbitmq.RabbitMQFixture.WORK_QUEUE;
import static org.apache.james.backend.rabbitmq.RabbitMQFixture.awaitAtMostOneMinute;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.stream.IntStream;
@@ -48,9 +54,11 @@ import com.github.steveash.guavate.Guavate;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Iterables;
+import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
+import com.rabbitmq.client.DeliverCallback;
class RabbitMQTest {
@@ -248,6 +256,77 @@ class RabbitMQTest {
.containsOnlyElementsOf(expectedResult);
}
+ @Test
+ void rabbitMQShouldRejectSecondConsumerInExclusiveWorkQueueCase() throws Exception {
+ channel1.exchangeDeclare(EXCHANGE_NAME, "direct", DURABLE);
+ channel1.queueDeclare(WORK_QUEUE, DURABLE, !EXCLUSIVE, !AUTO_DELETE, ImmutableMap.of());
+ channel1.queueBind(WORK_QUEUE, EXCHANGE_NAME, ROUTING_KEY);
+
+ IntStream.range(0, 10)
+ .mapToObj(String::valueOf)
+ .map(RabbitMQTest.this::asBytes)
+ .forEach(Throwing.consumer(
+ bytes -> channel1.basicPublish(EXCHANGE_NAME, ROUTING_KEY, NO_PROPERTIES, bytes)));
+
+ ConcurrentLinkedQueue<Integer> receivedMessages = new ConcurrentLinkedQueue<>();
+ String dyingConsumerTag = "dyingConsumer";
+ ImmutableMap<String, Object> arguments = ImmutableMap.of();
+ channel2.basicConsume(WORK_QUEUE, AUTO_ACK, dyingConsumerTag, !NO_LOCAL, EXCLUSIVE, arguments,
+ (consumerTag, message) -> {
+ try {
+ TimeUnit.SECONDS.sleep(1);
+ } catch (InterruptedException e) {
+ //do nothing
+ }
+ },
+ (consumerTag -> { }));
+ assertThatThrownBy(() ->
+ channel3.basicConsume(WORK_QUEUE, AUTO_ACK, "fallbackConsumer", !NO_LOCAL, EXCLUSIVE, arguments,
+ (consumerTag, message) -> { },
+ consumerTag -> { }))
+ .isInstanceOf(IOException.class)
+ .hasStackTraceContaining("ACCESS_REFUSED");
+ }
+
+ @Test
+ void rabbitMQShouldSupportTheExclusiveWorkQueueCase() throws Exception {
+ channel1.exchangeDeclare(EXCHANGE_NAME, "direct", DURABLE);
+ channel1.queueDeclare(WORK_QUEUE, DURABLE, !EXCLUSIVE, !AUTO_DELETE, ImmutableMap.of());
+ channel1.queueBind(WORK_QUEUE, EXCHANGE_NAME, ROUTING_KEY);
+
+ IntStream.range(0, 10)
+ .mapToObj(String::valueOf)
+ .map(RabbitMQTest.this::asBytes)
+ .forEach(Throwing.consumer(
+ bytes -> channel1.basicPublish(EXCHANGE_NAME, ROUTING_KEY, NO_PROPERTIES, bytes)));
+
+ String dyingConsumerTag = "dyingConsumer";
+ ImmutableMap<String, Object> arguments = ImmutableMap.of();
+ ConcurrentLinkedQueue<Integer> receivedMessages = new ConcurrentLinkedQueue<>();
+ CancelCallback doNothingOnCancel = consumerTag -> { };
+ DeliverCallback ackFirstMessageOnly = (consumerTag, message) -> {
+ if (receivedMessages.size() == 0) {
+ receivedMessages.add(Integer.valueOf(new String(message.getBody(), StandardCharsets.UTF_8)));
+ channel2.basicAck(message.getEnvelope().getDeliveryTag(), !MULTIPLE);
+ } else {
+ channel2.basicNack(message.getEnvelope().getDeliveryTag(), !MULTIPLE, REQUEUE);
+ }
+ };
+ channel2.basicConsume(WORK_QUEUE, !AUTO_ACK, dyingConsumerTag, !NO_LOCAL, EXCLUSIVE, arguments, ackFirstMessageOnly, doNothingOnCancel);
+
+ awaitAtMostOneMinute.until(() -> receivedMessages.size() == 1);
+
+ channel2.basicCancel(dyingConsumerTag);
+
+ InMemoryConsumer fallbackConsumer = new InMemoryConsumer(channel3);
+ channel3.basicConsume(WORK_QUEUE, AUTO_ACK, "fallbackConsumer", !NO_LOCAL, EXCLUSIVE, arguments, fallbackConsumer);
+
+ awaitAtMostOneMinute.until(() -> countReceivedMessages(fallbackConsumer) >= 1);
+
+ assertThat(receivedMessages).containsExactly(0);
+ assertThat(fallbackConsumer.getConsumedMessages()).contains(1, 2).doesNotContain(0);
+ }
+
}
@Nested
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 03/04: JAMES-2721 Increase parallelism for
mailbox/cassandra projects
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit c89a8b2fc502879e1faf9c270a78541bd49c26a2
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Tue Sep 10 14:48:59 2019 +0700
JAMES-2721 Increase parallelism for mailbox/cassandra projects
Last build (ci-02) for Mailbox :: Cassandra: [06:04 min]
Running 2 Cassandra in parallel is doable and allow to gain 1 minute 30
---
mailbox/cassandra/pom.xml | 13 +++++++++++++
1 file changed, 13 insertions(+)
diff --git a/mailbox/cassandra/pom.xml b/mailbox/cassandra/pom.xml
index fff5244..d3cb872 100644
--- a/mailbox/cassandra/pom.xml
+++ b/mailbox/cassandra/pom.xml
@@ -160,4 +160,17 @@
<scope>test</scope>
</dependency>
</dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <configuration>
+ <reuseForks>true</reuseForks>
+ <forkCount>2</forkCount>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
</project>
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org
[james-project] 04/04: Modify some commiters no longer existing
emails
Posted by bt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 70f3f277a106b509e99ada5d9af61df603dee21d
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Fri Sep 20 09:36:27 2019 +0700
Modify some commiters no longer existing emails
---
pom.xml | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git a/pom.xml b/pom.xml
index 13db5ba..1e7eed7 100644
--- a/pom.xml
+++ b/pom.xml
@@ -478,11 +478,11 @@
</contributor>
<contributor>
<name>Luc Duzan</name>
- <email>lduzan@linagora.com</email>
+ <email>lduzan@apache.org</email>
</contributor>
<contributor>
<name>Quynh Nguyen</name>
- <email>qnguyen@linagora.com</email>
+ <email>qnguyen@apache.org</email>
</contributor>
</contributors>
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org