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 2020/06/25 06:50:44 UTC
[james-project] branch master updated (1ebc356 -> 5bfcbe0)
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 1ebc356 JAMES-3187 Added User Model
new a2dd14b JAMES-3265 CassandraMessageMapper should limit modseq allocation upon flags updates
new bc0e45d JAMES-3265 Introduce StatementRecorder.Selector
new a74ca35 JAMES-3265 Reduce statement count upon CassandraMessageMapper::delete & Flags updates
new 5bfcbe0 JAMES-3170 CassandraBlobStoreCache should not propagate failures
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:
.../backends/cassandra/StatementRecorder.java | 30 +++++
.../backends/cassandra/TestingSessionTest.java | 11 +-
.../cassandra/mail/CassandraIndexTableHandler.java | 107 +++++++++++++--
.../cassandra/mail/CassandraMailboxCounterDAO.java | 4 +-
.../cassandra/mail/CassandraMessageMapper.java | 67 +++++-----
.../mail/CassandraIndexTableHandlerTest.java | 2 +-
.../cassandra/mail/CassandraMessageMapperTest.java | 148 ++++++++++++++++++---
.../cassandra/cache/CassandraBlobStoreCache.java | 6 +-
.../blob/cassandra/cache/CachedBlobStoreTest.java | 15 +++
9 files changed, 321 insertions(+), 69 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-3265 CassandraMessageMapper should
limit modseq allocation upon flags updates
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 a2dd14bc57eabeb8780556ca7cd24c74c198f751
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Wed Jun 24 08:51:37 2020 +0700
JAMES-3265 CassandraMessageMapper should limit modseq allocation upon flags updates
---
.../cassandra/mail/CassandraMessageMapper.java | 6 +++---
.../cassandra/mail/CassandraMessageMapperTest.java | 21 +++++++++++++++++++++
2 files changed, 24 insertions(+), 3 deletions(-)
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMapper.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMapper.java
index 82c9e6b..2934984 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMapper.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMapper.java
@@ -344,9 +344,9 @@ public class CassandraMessageMapper implements MessageMapper {
}
private Mono<FlagsUpdateStageResult> runUpdateStage(CassandraId mailboxId, Flux<ComposedMessageIdWithMetaData> toBeUpdated, FlagsUpdateCalculator flagsUpdateCalculator) {
- Mono<ModSeq> newModSeq = computeNewModSeq(mailboxId);
- return toBeUpdated
- .concatMap(metadata -> newModSeq.flatMap(modSeq -> tryFlagsUpdate(flagsUpdateCalculator, modSeq, metadata)))
+ return computeNewModSeq(mailboxId)
+ .flatMapMany(newModSeq -> toBeUpdated
+ .concatMap(metadata -> tryFlagsUpdate(flagsUpdateCalculator, newModSeq, metadata)))
.reduce(FlagsUpdateStageResult.none(), FlagsUpdateStageResult::merge)
.flatMap(result -> updateIndexesForUpdatesResult(mailboxId, result));
}
diff --git a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMapperTest.java b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMapperTest.java
index b97cf66..f115e60 100644
--- a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMapperTest.java
+++ b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMapperTest.java
@@ -25,13 +25,17 @@ import static org.assertj.core.api.Assertions.assertThat;
import java.util.Iterator;
import java.util.Optional;
+import javax.mail.Flags;
+
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.mailbox.MessageManager;
import org.apache.james.mailbox.cassandra.ids.CassandraId;
import org.apache.james.mailbox.cassandra.ids.CassandraMessageId;
import org.apache.james.mailbox.exception.MailboxException;
import org.apache.james.mailbox.model.MessageRange;
+import org.apache.james.mailbox.store.FlagsUpdateCalculator;
import org.apache.james.mailbox.store.mail.MessageMapper.FetchType;
import org.apache.james.mailbox.store.mail.model.MailboxMessage;
import org.apache.james.mailbox.store.mail.model.MapperProvider;
@@ -73,6 +77,23 @@ class CassandraMessageMapperTest extends MessageMapperTest {
.hasSize(limit);
}
+ @Test
+ void updateFlagsShouldLimitModSeqAllocation(CassandraCluster cassandra) throws MailboxException {
+ saveMessages();
+
+ StatementRecorder statementRecorder = new StatementRecorder();
+ cassandra.getConf().recordStatements(statementRecorder);
+
+ messageMapper.updateFlags(benwaInboxMailbox, new FlagsUpdateCalculator(new Flags(Flags.Flag.ANSWERED), MessageManager.FlagsUpdateMode.REPLACE), MessageRange.all());
+
+ assertThat(statementRecorder.listExecutedStatements())
+ .filteredOn(statement -> statement instanceof BoundStatement)
+ .extracting(BoundStatement.class::cast)
+ .extracting(statement -> statement.preparedStatement().getQueryString())
+ .filteredOn(statementString -> statementString.equals("UPDATE modseq SET nextModseq=:nextModseq WHERE mailboxId=:mailboxId IF nextModseq=:modSeqCondition;"))
+ .hasSize(1);
+ }
+
private void consume(Iterator<MailboxMessage> inMailbox) {
ImmutableList.copyOf(inMailbox);
}
---------------------------------------------------------------------
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-3265 Reduce statement count upon
CassandraMessageMapper::delete & Flags updates
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 a74ca3548cb4675052aa7ce1cd04a742cbf72510
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Wed Jun 24 13:18:40 2020 +0700
JAMES-3265 Reduce statement count upon CassandraMessageMapper::delete & Flags updates
This effectively optimizes IMAP EXPUNGE & STORE operations, dramatically limit counter writes,
and should prevent uneeded large tumbstones ranges creations for deletedMessages and recentMessages projections.
---
.../cassandra/mail/CassandraIndexTableHandler.java | 107 +++++++++++++--
.../cassandra/mail/CassandraMailboxCounterDAO.java | 4 +-
.../cassandra/mail/CassandraMessageMapper.java | 61 +++++----
.../mail/CassandraIndexTableHandlerTest.java | 2 +-
.../cassandra/mail/CassandraMessageMapperTest.java | 145 +++++++++++++++++----
5 files changed, 253 insertions(+), 66 deletions(-)
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraIndexTableHandler.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraIndexTableHandler.java
index 8fc3d91..47fd1eb 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraIndexTableHandler.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraIndexTableHandler.java
@@ -19,15 +19,23 @@
package org.apache.james.mailbox.cassandra.mail;
+import java.util.Collection;
+import java.util.List;
+
import javax.inject.Inject;
import javax.mail.Flags;
import org.apache.james.mailbox.MessageUid;
import org.apache.james.mailbox.cassandra.ids.CassandraId;
import org.apache.james.mailbox.model.ComposedMessageIdWithMetaData;
+import org.apache.james.mailbox.model.MailboxCounters;
+import org.apache.james.mailbox.model.MessageMetaData;
import org.apache.james.mailbox.model.UpdatedFlags;
import org.apache.james.mailbox.store.mail.model.MailboxMessage;
+import org.apache.james.util.streams.Iterators;
+import com.github.steveash.guavate.Guavate;
+import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
import reactor.core.publisher.Flux;
@@ -61,11 +69,39 @@ public class CassandraIndexTableHandler {
return Flux.mergeDelayError(Queues.XS_BUFFER_SIZE,
updateFirstUnseenOnDelete(mailboxId, composedMessageIdWithMetaData.getFlags(), composedMessageIdWithMetaData.getComposedMessageId().getUid()),
mailboxRecentDAO.removeFromRecent(mailboxId, composedMessageIdWithMetaData.getComposedMessageId().getUid()),
- deletedMessageDAO.removeDeleted(mailboxId, uid),
+ updateDeletedMessageProjectionOnDelete(mailboxId, uid, composedMessageIdWithMetaData.getFlags()),
decrementCountersOnDelete(mailboxId, composedMessageIdWithMetaData.getFlags()))
.then();
}
+ public Mono<Void> updateIndexOnDelete(CassandraId mailboxId, Collection<MessageMetaData> metaData) {
+ return Flux.mergeDelayError(Queues.XS_BUFFER_SIZE,
+ Flux.fromIterable(metaData)
+ .flatMap(message -> updateFirstUnseenOnDelete(mailboxId, message.getFlags(), message.getUid())),
+ Flux.fromIterable(metaData)
+ .flatMap(message -> updateRecentOnDelete(mailboxId, message.getUid(), message.getFlags())),
+ Flux.fromIterable(metaData)
+ .flatMap(message -> updateDeletedMessageProjectionOnDelete(mailboxId, message.getUid(), message.getFlags())),
+ decrementCountersOnDelete(mailboxId, metaData))
+ .then();
+ }
+
+ private Mono<Void> updateRecentOnDelete(CassandraId mailboxId, MessageUid uid, Flags flags) {
+ if (flags.contains(Flags.Flag.RECENT)) {
+ return mailboxRecentDAO.removeFromRecent(mailboxId, uid);
+ }
+
+ return Mono.empty();
+ }
+
+ private Mono<Void> updateDeletedMessageProjectionOnDelete(CassandraId mailboxId, MessageUid uid, Flags flags) {
+ if (flags.contains(Flags.Flag.DELETED)) {
+ return deletedMessageDAO.removeDeleted(mailboxId, uid);
+ }
+
+ return Mono.empty();
+ }
+
public Mono<Void> updateIndexOnAdd(MailboxMessage message, CassandraId mailboxId) {
Flags flags = message.createFlags();
@@ -79,15 +115,32 @@ public class CassandraIndexTableHandler {
}
public Mono<Void> updateIndexOnFlagsUpdate(CassandraId mailboxId, UpdatedFlags updatedFlags) {
+ return updateIndexOnFlagsUpdate(mailboxId, ImmutableList.of(updatedFlags));
+ }
+
+ public Mono<Void> updateIndexOnFlagsUpdate(CassandraId mailboxId, List<UpdatedFlags> updatedFlags) {
return Flux.mergeDelayError(Queues.XS_BUFFER_SIZE,
manageUnseenMessageCountsOnFlagsUpdate(mailboxId, updatedFlags),
manageRecentOnFlagsUpdate(mailboxId, updatedFlags),
updateFirstUnseenOnFlagsUpdate(mailboxId, updatedFlags),
- applicableFlagDAO.updateApplicableFlags(mailboxId, ImmutableSet.copyOf(updatedFlags.userFlagIterator())),
+ manageApplicableFlagsOnFlagsUpdate(mailboxId, updatedFlags),
updateDeletedOnFlagsUpdate(mailboxId, updatedFlags))
.then();
}
+ private Mono<Void> manageApplicableFlagsOnFlagsUpdate(CassandraId mailboxId, List<UpdatedFlags> updatedFlags) {
+ return applicableFlagDAO.updateApplicableFlags(mailboxId,
+ updatedFlags.stream()
+ .flatMap(flags -> Iterators.toStream(flags.userFlagIterator()))
+ .collect(Guavate.toImmutableSet()));
+ }
+
+ private Mono<Void> updateDeletedOnFlagsUpdate(CassandraId mailboxId, List<UpdatedFlags> updatedFlags) {
+ return Flux.fromIterable(updatedFlags)
+ .concatMap(flags -> updateDeletedOnFlagsUpdate(mailboxId, flags))
+ .then();
+ }
+
private Mono<Void> updateDeletedOnFlagsUpdate(CassandraId mailboxId, UpdatedFlags updatedFlags) {
if (updatedFlags.isModifiedToSet(Flags.Flag.DELETED)) {
return deletedMessageDAO.addDeleted(mailboxId, updatedFlags.getUid());
@@ -105,6 +158,19 @@ public class CassandraIndexTableHandler {
return mailboxCounterDAO.decrementUnseenAndCount(mailboxId);
}
+ private Mono<Void> decrementCountersOnDelete(CassandraId mailboxId, Collection<MessageMetaData> metaData) {
+ long seenCount = metaData.stream()
+ .map(MessageMetaData::getFlags)
+ .filter(flags -> flags.contains(Flags.Flag.SEEN))
+ .count();
+
+ return mailboxCounterDAO.remove(MailboxCounters.builder()
+ .mailboxId(mailboxId)
+ .count(metaData.size())
+ .unseen(seenCount)
+ .build());
+ }
+
private Mono<Void> incrementCountersOnSave(CassandraId mailboxId, Flags flags) {
if (flags.contains(Flags.Flag.SEEN)) {
return mailboxCounterDAO.incrementCount(mailboxId);
@@ -119,16 +185,35 @@ public class CassandraIndexTableHandler {
return Mono.empty();
}
- private Mono<Void> manageUnseenMessageCountsOnFlagsUpdate(CassandraId mailboxId, UpdatedFlags updatedFlags) {
- if (updatedFlags.isModifiedToUnset(Flags.Flag.SEEN)) {
- return mailboxCounterDAO.incrementUnseen(mailboxId);
- }
- if (updatedFlags.isModifiedToSet(Flags.Flag.SEEN)) {
- return mailboxCounterDAO.decrementUnseen(mailboxId);
+ private Mono<Void> manageUnseenMessageCountsOnFlagsUpdate(CassandraId mailboxId, List<UpdatedFlags> updatedFlags) {
+ int sum = updatedFlags.stream()
+ .mapToInt(flags -> {
+ if (flags.isModifiedToUnset(Flags.Flag.SEEN)) {
+ return 1;
+ }
+ if (flags.isModifiedToSet(Flags.Flag.SEEN)) {
+ return -1;
+ }
+ return 0;
+ })
+ .sum();
+
+ if (sum != 0) {
+ return mailboxCounterDAO.add(MailboxCounters.builder()
+ .mailboxId(mailboxId)
+ .count(0)
+ .unseen(sum)
+ .build());
}
return Mono.empty();
}
+ private Mono<Void> manageRecentOnFlagsUpdate(CassandraId mailboxId, List<UpdatedFlags> updatedFlags) {
+ return Flux.fromIterable(updatedFlags)
+ .concatMap(flags -> manageRecentOnFlagsUpdate(mailboxId, flags))
+ .then();
+ }
+
private Mono<Void> manageRecentOnFlagsUpdate(CassandraId mailboxId, UpdatedFlags updatedFlags) {
if (updatedFlags.isModifiedToUnset(Flags.Flag.RECENT)) {
return mailboxRecentDAO.removeFromRecent(mailboxId, updatedFlags.getUid());
@@ -161,6 +246,12 @@ public class CassandraIndexTableHandler {
return firstUnseenDAO.removeUnread(mailboxId, uid);
}
+ private Mono<Void> updateFirstUnseenOnFlagsUpdate(CassandraId mailboxId, List<UpdatedFlags> updatedFlags) {
+ return Flux.fromIterable(updatedFlags)
+ .concatMap(flags -> updateFirstUnseenOnFlagsUpdate(mailboxId, flags))
+ .then();
+ }
+
private Mono<Void> updateFirstUnseenOnFlagsUpdate(CassandraId mailboxId, UpdatedFlags updatedFlags) {
if (updatedFlags.isModifiedToUnset(Flags.Flag.SEEN)) {
return firstUnseenDAO.addUnread(mailboxId, updatedFlags.getUid());
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMailboxCounterDAO.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMailboxCounterDAO.java
index 80ae4ba..460decf 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMailboxCounterDAO.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMailboxCounterDAO.java
@@ -136,7 +136,7 @@ public class CassandraMailboxCounterDAO {
.build();
}
- private Mono<Void> add(MailboxCounters counters) {
+ public Mono<Void> add(MailboxCounters counters) {
CassandraId mailboxId = (CassandraId) counters.getMailboxId();
return cassandraAsyncExecutor.executeVoid(
bindWithMailbox(mailboxId, addToCounters)
@@ -144,7 +144,7 @@ public class CassandraMailboxCounterDAO {
.setLong(UNSEEN, counters.getUnseen()));
}
- private Mono<Void> remove(MailboxCounters counters) {
+ public Mono<Void> remove(MailboxCounters counters) {
CassandraId mailboxId = (CassandraId) counters.getMailboxId();
return cassandraAsyncExecutor.executeVoid(
bindWithMailbox(mailboxId, removeToCounters)
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMapper.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMapper.java
index 2934984..9fbf32c 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMapper.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMapper.java
@@ -64,6 +64,7 @@ import com.google.common.collect.ImmutableList;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
import reactor.util.retry.Retry;
public class CassandraMessageMapper implements MessageMapper {
@@ -139,25 +140,30 @@ public class CassandraMessageMapper implements MessageMapper {
@Override
public void delete(Mailbox mailbox, MailboxMessage message) {
- deleteAsFuture(message)
+ ComposedMessageIdWithMetaData metaData = message.getComposedMessageIdWithMetaData();
+
+ deleteAndHandleIndexUpdates(metaData)
.block();
}
- private Mono<Void> deleteAsFuture(MailboxMessage message) {
- ComposedMessageIdWithMetaData composedMessageIdWithMetaData = message.getComposedMessageIdWithMetaData();
+ private Mono<Void> deleteAndHandleIndexUpdates(ComposedMessageIdWithMetaData composedMessageIdWithMetaData) {
+ ComposedMessageId composedMessageId = composedMessageIdWithMetaData.getComposedMessageId();
+ CassandraId mailboxId = (CassandraId) composedMessageId.getMailboxId();
- return deleteUsingMailboxId(composedMessageIdWithMetaData);
+ return delete(composedMessageIdWithMetaData)
+ .then(indexTableHandler.updateIndexOnDelete(composedMessageIdWithMetaData, mailboxId));
}
- private Mono<Void> deleteUsingMailboxId(ComposedMessageIdWithMetaData composedMessageIdWithMetaData) {
+ private Mono<Void> delete(ComposedMessageIdWithMetaData composedMessageIdWithMetaData) {
ComposedMessageId composedMessageId = composedMessageIdWithMetaData.getComposedMessageId();
CassandraMessageId messageId = (CassandraMessageId) composedMessageId.getMessageId();
CassandraId mailboxId = (CassandraId) composedMessageId.getMailboxId();
MessageUid uid = composedMessageId.getUid();
+
return Flux.merge(
imapUidDAO.delete(messageId, mailboxId),
messageIdDAO.delete(mailboxId, uid))
- .then(indexTableHandler.updateIndexOnDelete(composedMessageIdWithMetaData, mailboxId));
+ .then();
}
@Override
@@ -211,25 +217,20 @@ public class CassandraMessageMapper implements MessageMapper {
public Map<MessageUid, MessageMetaData> deleteMessages(Mailbox mailbox, List<MessageUid> uids) {
CassandraId mailboxId = (CassandraId) mailbox.getMailboxId();
- return Flux.fromStream(uids.stream())
- .flatMap(messageUid -> expungeOne(mailboxId, messageUid), cassandraConfiguration.getExpungeChunkSize())
- .collect(Guavate.<SimpleMailboxMessage, MessageUid, MessageMetaData>toImmutableMap(MailboxMessage::getUid, MailboxMessage::metaData))
+ return Flux.fromIterable(MessageRange.toRanges(uids))
+ .concatMap(range -> messageIdDAO.retrieveMessages(mailboxId, range, Limit.unlimited()))
+ .flatMap(this::expungeOne, cassandraConfiguration.getExpungeChunkSize())
+ .collect(Guavate.toImmutableMap(MailboxMessage::getUid, MailboxMessage::metaData))
+ .flatMap(messageMap -> indexTableHandler.updateIndexOnDelete(mailboxId, messageMap.values())
+ .thenReturn(messageMap))
+ .subscribeOn(Schedulers.elastic())
.block();
}
- private Mono<SimpleMailboxMessage> expungeOne(CassandraId mailboxId, MessageUid messageUid) {
- return retrieveComposedId(mailboxId, messageUid)
- .flatMap(idWithMetadata -> deleteUsingMailboxId(idWithMetadata).thenReturn(idWithMetadata))
- .flatMap(idWithMetadata -> messageDAO.retrieveMessage(idWithMetadata, FetchType.Metadata)
- .map(pair -> pair.toMailboxMessage(idWithMetadata, ImmutableList.of())));
- }
-
- private Mono<ComposedMessageIdWithMetaData> retrieveComposedId(CassandraId mailboxId, MessageUid uid) {
- return messageIdDAO.retrieve(mailboxId, uid)
- .handle((t, sink) ->
- t.ifPresentOrElse(
- sink::next,
- () -> LOGGER.warn("Could not retrieve message {} {}", mailboxId, uid)));
+ private Mono<SimpleMailboxMessage> expungeOne(ComposedMessageIdWithMetaData metaData) {
+ return delete(metaData)
+ .then(messageDAO.retrieveMessage(metaData, FetchType.Metadata)
+ .map(pair -> pair.toMailboxMessage(metaData, ImmutableList.of())));
}
@Override
@@ -237,7 +238,7 @@ public class CassandraMessageMapper implements MessageMapper {
ComposedMessageIdWithMetaData composedMessageIdWithMetaData = original.getComposedMessageIdWithMetaData();
MessageMetaData messageMetaData = copy(destinationMailbox, original);
- deleteUsingMailboxId(composedMessageIdWithMetaData).block();
+ deleteAndHandleIndexUpdates(composedMessageIdWithMetaData).block();
return messageMetaData;
}
@@ -357,14 +358,12 @@ public class CassandraMessageMapper implements MessageMapper {
}
private Mono<FlagsUpdateStageResult> updateIndexesForUpdatesResult(CassandraId mailboxId, FlagsUpdateStageResult result) {
- return Flux.fromIterable(result.getSucceeded())
- .flatMap(Throwing
- .function((UpdatedFlags updatedFlags) -> indexTableHandler.updateIndexOnFlagsUpdate(mailboxId, updatedFlags))
- .fallbackTo(failedIndex -> {
- LOGGER.error("Could not update flag indexes for mailboxId {} UID {}. This will lead to inconsistencies across Cassandra tables", mailboxId, failedIndex.getUid());
- return Mono.empty();
- }))
- .then(Mono.just(result));
+ return indexTableHandler.updateIndexOnFlagsUpdate(mailboxId, result.getSucceeded())
+ .onErrorResume(e -> {
+ LOGGER.error("Could not update flag indexes for mailboxId {}. This will lead to inconsistencies across Cassandra tables", mailboxId, e);
+ return Mono.empty();
+ })
+ .thenReturn(result);
}
@Override
diff --git a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraIndexTableHandlerTest.java b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraIndexTableHandlerTest.java
index ae090d1..beb08d5 100644
--- a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraIndexTableHandlerTest.java
+++ b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraIndexTableHandlerTest.java
@@ -286,7 +286,7 @@ class CassandraIndexTableHandlerTest {
testee.updateIndexOnDelete(new ComposedMessageIdWithMetaData(
new ComposedMessageId(MAILBOX_ID, CASSANDRA_MESSAGE_ID, MESSAGE_UID),
- new Flags(),
+ new Flags(Flags.Flag.DELETED),
MODSEQ),
MAILBOX_ID).block();
diff --git a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMapperTest.java b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMapperTest.java
index c3c8bc3..8581377 100644
--- a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMapperTest.java
+++ b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMapperTest.java
@@ -59,39 +59,136 @@ class CassandraMessageMapperTest extends MessageMapperTest {
return new CassandraMapperProvider(cassandraCluster.getCassandraCluster());
}
- @Test
- void findInMailboxLimitShouldLimitProjectionReadCassandraQueries(CassandraCluster cassandra) throws MailboxException {
- saveMessages();
+ @Nested
+ class StatementLimitationTests {
+ @Test
+ void deleteMessagesShouldGroupMessageReads(CassandraCluster cassandra) throws MailboxException {
+ saveMessages();
- StatementRecorder statementRecorder = new StatementRecorder();
- cassandra.getConf().recordStatements(statementRecorder);
+ StatementRecorder statementRecorder = new StatementRecorder();
+ cassandra.getConf().recordStatements(statementRecorder);
+ cassandra.getConf().printStatements();
- int limit = 2;
- consume(messageMapper.findInMailbox(benwaInboxMailbox, MessageRange.all(), FetchType.Full, limit));
+ messageMapper.deleteMessages(benwaInboxMailbox, ImmutableList.of(message1.getUid(), message2.getUid(), message3.getUid()));
+ assertThat(statementRecorder.listExecutedStatements(Selector.preparedStatementStartingWith(
+ "SELECT messageId,mailboxId,uid,modSeq,flagAnswered,flagDeleted,flagDraft,flagFlagged,flagRecent,flagSeen," +
+ "flagUser,userFlags FROM messageIdTable WHERE mailboxId=:mailboxId AND ")))
+ .hasSize(1);
+ }
- assertThat(statementRecorder.listExecutedStatements(Selector.preparedStatement(
- "SELECT messageId,internalDate,bodyStartOctet,fullContentOctets,bodyOctets,bodyContent,headerContent,textualLineCount,properties,attachments " +
- "FROM messageV2 WHERE messageId=:messageId;")))
- .hasSize(limit);
- }
+ @Test
+ void deleteMessagesShouldGroupCounterUpdates(CassandraCluster cassandra) throws MailboxException {
+ saveMessages();
- @Test
- void updateFlagsShouldLimitModSeqAllocation(CassandraCluster cassandra) throws MailboxException {
- saveMessages();
+ StatementRecorder statementRecorder = new StatementRecorder();
+ cassandra.getConf().recordStatements(statementRecorder);
+ cassandra.getConf().printStatements();
- StatementRecorder statementRecorder = new StatementRecorder();
- cassandra.getConf().recordStatements(statementRecorder);
+ messageMapper.deleteMessages(benwaInboxMailbox, ImmutableList.of(message1.getUid(), message2.getUid(), message3.getUid()));
- messageMapper.updateFlags(benwaInboxMailbox, new FlagsUpdateCalculator(new Flags(Flags.Flag.ANSWERED), MessageManager.FlagsUpdateMode.REPLACE), MessageRange.all());
+ assertThat(statementRecorder.listExecutedStatements(
+ Selector.preparedStatementStartingWith("UPDATE mailboxCounters SET ")))
+ .hasSize(1);
+ }
- assertThat(statementRecorder.listExecutedStatements(Selector.preparedStatement(
- "UPDATE modseq SET nextModseq=:nextModseq WHERE mailboxId=:mailboxId IF nextModseq=:modSeqCondition;")))
- .hasSize(1);
- }
+ @Test
+ void deleteMessagesShouldNotDeleteMessageNotMarkedAsDeletedInDeletedProjection(CassandraCluster cassandra) throws MailboxException {
+ saveMessages();
- private void consume(Iterator<MailboxMessage> inMailbox) {
- ImmutableList.copyOf(inMailbox);
+ StatementRecorder statementRecorder = new StatementRecorder();
+ cassandra.getConf().recordStatements(statementRecorder);
+ cassandra.getConf().printStatements();
+
+ messageMapper.deleteMessages(benwaInboxMailbox, ImmutableList.of(message1.getUid(), message2.getUid(), message3.getUid()));
+
+ assertThat(statementRecorder.listExecutedStatements(
+ Selector.preparedStatement("DELETE FROM messageDeleted WHERE mailboxId=:mailboxId AND uid=:uid;")))
+ .isEmpty();
+ }
+
+ @Test
+ void deleteMessagesShouldNotDeleteMessageNotMarkedAsRecentInRecentProjection(CassandraCluster cassandra) throws MailboxException {
+ saveMessages();
+
+ StatementRecorder statementRecorder = new StatementRecorder();
+ cassandra.getConf().recordStatements(statementRecorder);
+ cassandra.getConf().printStatements();
+
+ messageMapper.deleteMessages(benwaInboxMailbox, ImmutableList.of(message1.getUid(), message2.getUid(), message3.getUid()));
+
+ assertThat(statementRecorder.listExecutedStatements(
+ Selector.preparedStatement("DELETE FROM messageDeleted WHERE mailboxId=:mailboxId AND uid=:uid;")))
+ .isEmpty();
+ }
+
+ @Test
+ void deleteMessagesShouldNotDeleteMessageNotMarkedAsUnSeenInFirstUnseenProjection(CassandraCluster cassandra) throws MailboxException {
+ saveMessages();
+ FlagsUpdateCalculator markAsRead = new FlagsUpdateCalculator(new Flags(Flags.Flag.SEEN), MessageManager.FlagsUpdateMode.ADD);
+ messageMapper.updateFlags(benwaInboxMailbox, markAsRead, MessageRange.all());
+
+ StatementRecorder statementRecorder = new StatementRecorder();
+ cassandra.getConf().recordStatements(statementRecorder);
+ cassandra.getConf().printStatements();
+
+ messageMapper.deleteMessages(benwaInboxMailbox, ImmutableList.of(message1.getUid(), message2.getUid(), message3.getUid()));
+
+ assertThat(statementRecorder.listExecutedStatements(
+ Selector.preparedStatement("DELETE FROM firstUnseen WHERE mailboxId=:mailboxId AND uid=:uid;")))
+ .isEmpty();
+ }
+
+ @Test
+ void updateFlagsShouldUpdateMailboxCountersOnce(CassandraCluster cassandra) throws MailboxException {
+ saveMessages();
+
+ StatementRecorder statementRecorder = new StatementRecorder();
+ cassandra.getConf().recordStatements(statementRecorder);
+ cassandra.getConf().printStatements();
+
+ messageMapper.updateFlags(benwaInboxMailbox, new FlagsUpdateCalculator(new Flags(Flags.Flag.SEEN), MessageManager.FlagsUpdateMode.REPLACE), MessageRange.all());
+
+
+ assertThat(statementRecorder.listExecutedStatements(Selector.preparedStatementStartingWith(
+ "UPDATE mailboxCounters SET ")))
+ .hasSize(1);
+ }
+
+ @Test
+ void findInMailboxLimitShouldLimitProjectionReadCassandraQueries(CassandraCluster cassandra) throws MailboxException {
+ saveMessages();
+
+ StatementRecorder statementRecorder = new StatementRecorder();
+ cassandra.getConf().recordStatements(statementRecorder);
+
+ int limit = 2;
+ consume(messageMapper.findInMailbox(benwaInboxMailbox, MessageRange.all(), FetchType.Full, limit));
+
+
+ assertThat(statementRecorder.listExecutedStatements(Selector.preparedStatement(
+ "SELECT messageId,internalDate,bodyStartOctet,fullContentOctets,bodyOctets,bodyContent,headerContent,textualLineCount,properties,attachments " +
+ "FROM messageV2 WHERE messageId=:messageId;")))
+ .hasSize(limit);
+ }
+
+ @Test
+ void updateFlagsShouldLimitModSeqAllocation(CassandraCluster cassandra) throws MailboxException {
+ saveMessages();
+
+ StatementRecorder statementRecorder = new StatementRecorder();
+ cassandra.getConf().recordStatements(statementRecorder);
+
+ messageMapper.updateFlags(benwaInboxMailbox, new FlagsUpdateCalculator(new Flags(Flags.Flag.ANSWERED), MessageManager.FlagsUpdateMode.REPLACE), MessageRange.all());
+
+ assertThat(statementRecorder.listExecutedStatements(Selector.preparedStatement(
+ "UPDATE modseq SET nextModseq=:nextModseq WHERE mailboxId=:mailboxId IF nextModseq=:modSeqCondition;")))
+ .hasSize(1);
+ }
+
+ private void consume(Iterator<MailboxMessage> inMailbox) {
+ ImmutableList.copyOf(inMailbox);
+ }
}
@Nested
---------------------------------------------------------------------
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-3265 Introduce
StatementRecorder.Selector
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 bc0e45d7a6951d57513ae885dcdf697897da86dd
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Wed Jun 24 12:48:09 2020 +0700
JAMES-3265 Introduce StatementRecorder.Selector
This ease Cassandra statement testing
---
.../backends/cassandra/StatementRecorder.java | 30 ++++++++++++++++++++++
.../backends/cassandra/TestingSessionTest.java | 11 ++++----
.../cassandra/mail/CassandraMessageMapperTest.java | 18 +++++--------
3 files changed, 42 insertions(+), 17 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 e8773f5..ddd595f 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
@@ -22,11 +22,37 @@ package org.apache.james.backends.cassandra;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedDeque;
+import java.util.function.Predicate;
+import com.datastax.driver.core.BoundStatement;
import com.datastax.driver.core.Statement;
+import com.github.steveash.guavate.Guavate;
import com.google.common.collect.ImmutableList;
public class StatementRecorder {
+ @FunctionalInterface
+ public interface Selector {
+ Selector ALL = statements -> statements;
+
+ static Selector preparedStatement(String statementString) {
+ return preparedStatementMatching(statement -> statement.preparedStatement().getQueryString().equals(statementString));
+ }
+
+ static Selector preparedStatementStartingWith(String statementString) {
+ return preparedStatementMatching(statement -> statement.preparedStatement().getQueryString().startsWith(statementString));
+ }
+
+ private static StatementRecorder.Selector preparedStatementMatching(Predicate<BoundStatement> condition) {
+ return statements -> statements.stream()
+ .filter(BoundStatement.class::isInstance)
+ .map(BoundStatement.class::cast)
+ .filter(condition)
+ .collect(Guavate.toImmutableList());
+ }
+
+ List<Statement> select(List<Statement> statements);
+ }
+
private final ConcurrentLinkedDeque statements;
public StatementRecorder() {
@@ -40,4 +66,8 @@ public class StatementRecorder {
public List<Statement> listExecutedStatements() {
return ImmutableList.copyOf(statements);
}
+
+ public List<Statement> listExecutedStatements(Selector selector) {
+ return selector.select(ImmutableList.copyOf(statements));
+ }
}
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 e496e34..6f9987b 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
@@ -33,6 +33,7 @@ import java.util.concurrent.CompletableFuture;
import org.apache.james.backends.cassandra.Scenario.Barrier;
import org.apache.james.backends.cassandra.Scenario.InjectedFailureException;
+import org.apache.james.backends.cassandra.StatementRecorder.Selector;
import org.apache.james.backends.cassandra.utils.CassandraAsyncExecutor;
import org.apache.james.backends.cassandra.versions.CassandraSchemaVersionDAO;
import org.apache.james.backends.cassandra.versions.CassandraSchemaVersionModule;
@@ -81,11 +82,9 @@ class TestingSessionTest {
dao.getCurrentSchemaVersion().block();
- assertThat(statementRecorder.listExecutedStatements())
- .filteredOn(statement -> statement instanceof BoundStatement)
- .extracting(BoundStatement.class::cast)
- .extracting(statement -> statement.preparedStatement().getQueryString())
- .containsExactly("SELECT value FROM schemaVersion;");
+ assertThat(statementRecorder.listExecutedStatements(
+ Selector.preparedStatement("SELECT value FROM schemaVersion;")))
+ .hasSize(1);
}
@Test
@@ -96,7 +95,7 @@ class TestingSessionTest {
dao.updateVersion(new SchemaVersion(36)).block();
dao.getCurrentSchemaVersion().block();
- assertThat(statementRecorder.listExecutedStatements())
+ assertThat(statementRecorder.listExecutedStatements(Selector.ALL))
.filteredOn(statement -> statement instanceof BoundStatement)
.extracting(BoundStatement.class::cast)
.extracting(statement -> statement.preparedStatement().getQueryString())
diff --git a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMapperTest.java b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMapperTest.java
index f115e60..c3c8bc3 100644
--- a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMapperTest.java
+++ b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageMapperTest.java
@@ -30,6 +30,7 @@ import javax.mail.Flags;
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.StatementRecorder.Selector;
import org.apache.james.mailbox.MessageManager;
import org.apache.james.mailbox.cassandra.ids.CassandraId;
import org.apache.james.mailbox.cassandra.ids.CassandraMessageId;
@@ -47,7 +48,6 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
import org.testcontainers.shaded.com.google.common.collect.ImmutableList;
-import com.datastax.driver.core.BoundStatement;
import com.github.fge.lambdas.Throwing;
class CassandraMessageMapperTest extends MessageMapperTest {
@@ -69,11 +69,10 @@ class CassandraMessageMapperTest extends MessageMapperTest {
int limit = 2;
consume(messageMapper.findInMailbox(benwaInboxMailbox, MessageRange.all(), FetchType.Full, limit));
- assertThat(statementRecorder.listExecutedStatements())
- .filteredOn(statement -> statement instanceof BoundStatement)
- .extracting(BoundStatement.class::cast)
- .extracting(statement -> statement.preparedStatement().getQueryString())
- .filteredOn(statementString -> statementString.equals("SELECT messageId,internalDate,bodyStartOctet,fullContentOctets,bodyOctets,bodyContent,headerContent,textualLineCount,properties,attachments FROM messageV2 WHERE messageId=:messageId;"))
+
+ assertThat(statementRecorder.listExecutedStatements(Selector.preparedStatement(
+ "SELECT messageId,internalDate,bodyStartOctet,fullContentOctets,bodyOctets,bodyContent,headerContent,textualLineCount,properties,attachments " +
+ "FROM messageV2 WHERE messageId=:messageId;")))
.hasSize(limit);
}
@@ -86,11 +85,8 @@ class CassandraMessageMapperTest extends MessageMapperTest {
messageMapper.updateFlags(benwaInboxMailbox, new FlagsUpdateCalculator(new Flags(Flags.Flag.ANSWERED), MessageManager.FlagsUpdateMode.REPLACE), MessageRange.all());
- assertThat(statementRecorder.listExecutedStatements())
- .filteredOn(statement -> statement instanceof BoundStatement)
- .extracting(BoundStatement.class::cast)
- .extracting(statement -> statement.preparedStatement().getQueryString())
- .filteredOn(statementString -> statementString.equals("UPDATE modseq SET nextModseq=:nextModseq WHERE mailboxId=:mailboxId IF nextModseq=:modSeqCondition;"))
+ assertThat(statementRecorder.listExecutedStatements(Selector.preparedStatement(
+ "UPDATE modseq SET nextModseq=:nextModseq WHERE mailboxId=:mailboxId IF nextModseq=:modSeqCondition;")))
.hasSize(1);
}
---------------------------------------------------------------------
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: JAMES-3170 CassandraBlobStoreCache should
not propagate failures
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 5bfcbe090c41b37ff3d1f109982d895fce0b953e
Author: Benoit Tellier <bt...@linagora.com>
AuthorDate: Tue Jun 23 11:33:11 2020 +0700
JAMES-3170 CassandraBlobStoreCache should not propagate failures
---
.../blob/cassandra/cache/CassandraBlobStoreCache.java | 6 +++++-
.../james/blob/cassandra/cache/CachedBlobStoreTest.java | 15 +++++++++++++++
2 files changed, 20 insertions(+), 1 deletion(-)
diff --git a/server/blob/blob-cassandra/src/main/java/org/apache/james/blob/cassandra/cache/CassandraBlobStoreCache.java b/server/blob/blob-cassandra/src/main/java/org/apache/james/blob/cassandra/cache/CassandraBlobStoreCache.java
index 5584a6c..fe912bb 100644
--- a/server/blob/blob-cassandra/src/main/java/org/apache/james/blob/cassandra/cache/CassandraBlobStoreCache.java
+++ b/server/blob/blob-cassandra/src/main/java/org/apache/james/blob/cassandra/cache/CassandraBlobStoreCache.java
@@ -87,7 +87,11 @@ public class CassandraBlobStoreCache implements BlobStoreCache {
.setString(ID, blobId.asString())
.setConsistencyLevel(ONE)
.setReadTimeoutMillis(readTimeOutFromDataBase))
- .map(this::toByteArray);
+ .map(this::toByteArray)
+ .onErrorResume(e -> {
+ LOGGER.warn("Fail reading blob store cache", e);
+ return Mono.empty();
+ });
}
@Override
diff --git a/server/blob/blob-cassandra/src/test/java/org/apache/james/blob/cassandra/cache/CachedBlobStoreTest.java b/server/blob/blob-cassandra/src/test/java/org/apache/james/blob/cassandra/cache/CachedBlobStoreTest.java
index 1d2abcd..c8c66a0 100644
--- a/server/blob/blob-cassandra/src/test/java/org/apache/james/blob/cassandra/cache/CachedBlobStoreTest.java
+++ b/server/blob/blob-cassandra/src/test/java/org/apache/james/blob/cassandra/cache/CachedBlobStoreTest.java
@@ -18,6 +18,7 @@
****************************************************************/
package org.apache.james.blob.cassandra.cache;
+import static org.apache.james.backends.cassandra.Scenario.Builder.fail;
import static org.apache.james.blob.api.BlobStore.StoragePolicy.HIGH_PERFORMANCE;
import static org.apache.james.blob.api.BlobStore.StoragePolicy.LOW_COST;
import static org.apache.james.blob.api.BlobStore.StoragePolicy.SIZE_BASED;
@@ -282,6 +283,20 @@ public class CachedBlobStoreTest implements BlobStoreContract {
});
}
+ @Test
+ public void readShouldNotPropagateFailures(CassandraCluster cassandra) {
+ BlobId blobId = Mono.from(testee.save(DEFAULT_BUCKETNAME, APPROXIMATELY_FIVE_KILOBYTES, SIZE_BASED)).block();
+
+ cassandra.getConf().registerScenario(fail()
+ .times(1)
+ .whenQueryStartsWith("SELECT * FROM blob_cache WHERE id=:id"));
+
+ Mono.from(cache.read(blobId)).block();
+
+ assertThat(testee().read(DEFAULT_BUCKETNAME, blobId))
+ .hasSameContentAs(new ByteArrayInputStream(APPROXIMATELY_FIVE_KILOBYTES));
+ }
+
@Nested
class MetricsTest {
@Test
---------------------------------------------------------------------
To unsubscribe, e-mail: server-dev-unsubscribe@james.apache.org
For additional commands, e-mail: server-dev-help@james.apache.org