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 2025/06/23 06:17:53 UTC
(james-project) 01/01: JAMES-3516 Rely on Cassandra for Thread lookups
This is an automated email from the ASF dual-hosted git repository.
btellier pushed a commit to branch JAMES-3516-bte
in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 2337eb3429df3761b4771a5584cd3c75a830d0d5
Author: Benoit TELLIER <bt...@linagora.com>
AuthorDate: Mon Jun 23 08:16:57 2025 +0200
JAMES-3516 Rely on Cassandra for Thread lookups
- Reorganise the lookup DAO in order to include the ThreadId as
part of the primary key and relegate the messageId as a clustering
key. Acceptable for our usage pattern, this enables listing content of
a thread.
Migration strategy: none. Truncate the old v2 table and start with
an empty v3 table which is acceptable.
- Rely on this method for thread resolution instead of the search index
- Pay an attention that messages in the tread are accessible to
the end user (right resolution)
---
.../CassandraThreadIdGuessingAlgorithm.java | 36 +++++++++---------
.../mailbox/cassandra/DeleteMessageListener.java | 32 +++++++---------
.../mailbox/cassandra/ids/CassandraMessageId.java | 4 ++
.../cassandra/mail/CassandraThreadLookupDAO.java | 35 +++++++++++++++---
.../modules/CassandraThreadDataDefinition.java | 3 +-
.../table/CassandraThreadLookupTable.java | 2 +-
.../cassandra/CassandraMailboxManagerTest.java | 18 ++++-----
.../CassandraThreadIdGuessingAlgorithmTest.java | 7 +++-
.../mail/CassandraThreadLookupDAOTest.java | 43 +++++++++++++++-------
9 files changed, 111 insertions(+), 69 deletions(-)
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/CassandraThreadIdGuessingAlgorithm.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/CassandraThreadIdGuessingAlgorithm.java
index 812191c7cd..2e486fc872 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/CassandraThreadIdGuessingAlgorithm.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/CassandraThreadIdGuessingAlgorithm.java
@@ -25,17 +25,15 @@ import java.util.Optional;
import java.util.Set;
import java.util.stream.Collectors;
+import com.google.common.collect.ImmutableList;
import jakarta.inject.Inject;
import org.apache.commons.lang3.tuple.Pair;
-import org.apache.james.mailbox.MailboxManager;
import org.apache.james.mailbox.MailboxSession;
+import org.apache.james.mailbox.MessageIdManager;
import org.apache.james.mailbox.cassandra.mail.CassandraThreadDAO;
import org.apache.james.mailbox.cassandra.mail.CassandraThreadLookupDAO;
-import org.apache.james.mailbox.exception.ThreadNotFoundException;
import org.apache.james.mailbox.model.MessageId;
-import org.apache.james.mailbox.model.MultimailboxesSearchQuery;
-import org.apache.james.mailbox.model.SearchQuery;
import org.apache.james.mailbox.model.ThreadId;
import org.apache.james.mailbox.store.mail.ThreadIdGuessingAlgorithm;
import org.apache.james.mailbox.store.mail.model.MimeMessageId;
@@ -50,13 +48,13 @@ import reactor.core.publisher.Mono;
public class CassandraThreadIdGuessingAlgorithm implements ThreadIdGuessingAlgorithm {
private static final boolean DISABLE_THREADS = Boolean.valueOf(System.getProperty("james.mailbox.threads.disable", "false"));
- private final MailboxManager mailboxManager;
+ private final MessageIdManager messageIdManager;
private final CassandraThreadDAO threadDAO;
private final CassandraThreadLookupDAO threadLookupDAO;
@Inject
- public CassandraThreadIdGuessingAlgorithm(MailboxManager mailboxManager, CassandraThreadDAO threadDAO, CassandraThreadLookupDAO threadLookupDAO) {
- this.mailboxManager = mailboxManager;
+ public CassandraThreadIdGuessingAlgorithm(MessageIdManager messageIdManager, CassandraThreadDAO threadDAO, CassandraThreadLookupDAO threadLookupDAO) {
+ this.messageIdManager = messageIdManager;
this.threadDAO = threadDAO;
this.threadLookupDAO = threadLookupDAO;
}
@@ -78,7 +76,7 @@ public class CassandraThreadIdGuessingAlgorithm implements ThreadIdGuessingAlgor
.switchIfEmpty(Mono.just(ThreadId.fromBaseMessageId(messageId)))
.flatMap(threadId -> threadDAO
.insertSome(session.getUser(), hashMimeMessageIds, messageId, threadId, hashBaseSubject)
- .then(threadLookupDAO.insert(messageId, session.getUser(), hashMimeMessageIds))
+ .then(threadLookupDAO.insert(messageId, threadId, session.getUser(), hashMimeMessageIds))
.then(Mono.just(threadId)));
}
@@ -88,17 +86,17 @@ public class CassandraThreadIdGuessingAlgorithm implements ThreadIdGuessingAlgor
return Flux.just(threadId.getBaseMessageId());
}
- SearchQuery searchQuery = SearchQuery.builder()
- .andCriteria(SearchQuery.threadId(threadId))
- .sorts(new SearchQuery.Sort(SearchQuery.Sort.SortClause.Arrival, SearchQuery.Sort.Order.NATURAL))
- .build();
-
- MultimailboxesSearchQuery expression = MultimailboxesSearchQuery
- .from(searchQuery)
- .build();
-
- return Flux.from(mailboxManager.search(expression, session, Integer.MAX_VALUE))
- .switchIfEmpty(Mono.error(() -> new ThreadNotFoundException(threadId)));
+ return threadLookupDAO.selectAll(threadId)
+ .collectList()
+ .map(ids -> {
+ // Handle umpopulated lookup dao
+ if (ids.isEmpty()) {
+ return ImmutableList.of(threadId.getBaseMessageId());
+ }
+ return ids;
+ })
+ .flatMapMany(messageIds -> messageIdManager.accessibleMessagesReactive(messageIds, session))
+ .flatMapIterable(x -> x);
}
private Set<MimeMessageId> buildMimeMessageIdSet(Optional<MimeMessageId> mimeMessageId, Optional<MimeMessageId> inReplyTo, Optional<List<MimeMessageId>> references) {
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/DeleteMessageListener.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/DeleteMessageListener.java
index e01cc966d1..4871780646 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/DeleteMessageListener.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/DeleteMessageListener.java
@@ -57,13 +57,12 @@ import org.apache.james.mailbox.cassandra.mail.CassandraUserMailboxRightsDAO;
import org.apache.james.mailbox.cassandra.mail.MessageRepresentation;
import org.apache.james.mailbox.events.MailboxEvents.Expunged;
import org.apache.james.mailbox.events.MailboxEvents.MailboxDeletion;
-import org.apache.james.mailbox.model.ComposedMessageIdWithMetaData;
import org.apache.james.mailbox.model.MailboxACL;
import org.apache.james.mailbox.model.MailboxId;
import org.apache.james.mailbox.model.MailboxPath;
import org.apache.james.mailbox.model.MessageId;
-import org.apache.james.mailbox.model.MessageMetaData;
import org.apache.james.mailbox.model.MessageRange;
+import org.apache.james.mailbox.model.ThreadId;
import org.apache.james.mailbox.store.mail.MessageMapper;
import org.apache.james.util.streams.Limit;
import org.reactivestreams.Publisher;
@@ -232,11 +231,10 @@ public class DeleteMessageListener implements EventListener.ReactiveGroupEventLi
int prefetch = 1;
return Flux.mergeDelayError(prefetch,
messageIdDAO.retrieveMessages(mailboxId, MessageRange.all(), Limit.unlimited())
- .map(CassandraMessageMetadata::getComposedMessageId)
- .map(ComposedMessageIdWithMetaData::getComposedMessageId)
- .concatMap(metadata -> handleMessageDeletionAsPartOfMailboxDeletion((CassandraMessageId) metadata.getMessageId(), mailboxId, path.getUser())
- .then(imapUidDAO.delete((CassandraMessageId) metadata.getMessageId(), mailboxId))
- .then(messageIdDAO.delete(mailboxId, metadata.getUid()))),
+ .concatMap(metadata -> handleMessageDeletionAsPartOfMailboxDeletion((CassandraMessageId) metadata.getComposedMessageId().getComposedMessageId().getMessageId(),
+ metadata.getComposedMessageId().getThreadId(), mailboxId, path.getUser())
+ .then(imapUidDAO.delete((CassandraMessageId) metadata.getComposedMessageId().getComposedMessageId().getMessageId(), mailboxId))
+ .then(messageIdDAO.delete(mailboxId, metadata.getComposedMessageId().getComposedMessageId().getUid()))),
deleteAcl(mailboxId),
applicableFlagDAO.delete(mailboxId),
firstUnseenDAO.removeAll(mailboxId),
@@ -247,11 +245,9 @@ public class DeleteMessageListener implements EventListener.ReactiveGroupEventLi
}
private Mono<Void> handleMessageDeletion(Expunged expunged) {
- return Flux.fromIterable(expunged.getExpunged()
- .values())
- .map(MessageMetaData::getMessageId)
- .map(CassandraMessageId.class::cast)
- .concatMap(messageId -> handleMessageDeletion(messageId, expunged.getMailboxId(), expunged.getMailboxPath().getUser()))
+ return Flux.fromIterable(expunged.getExpunged().values())
+ .concatMap(metaData -> handleMessageDeletion((CassandraMessageId) metaData.getMessageId(),
+ expunged.getMailboxId(), metaData.getThreadId(), expunged.getMailboxPath().getUser()))
.then();
}
@@ -261,7 +257,7 @@ public class DeleteMessageListener implements EventListener.ReactiveGroupEventLi
.then(aclMapper.delete(mailboxId)));
}
- private Mono<Void> handleMessageDeletion(CassandraMessageId messageId, MailboxId mailboxId, Username owner) {
+ private Mono<Void> handleMessageDeletion(CassandraMessageId messageId, MailboxId mailboxId, ThreadId threadId, Username owner) {
return Mono.just(messageId)
.filterWhen(this::isReferenced)
.flatMap(id -> readMessage(id)
@@ -269,13 +265,13 @@ public class DeleteMessageListener implements EventListener.ReactiveGroupEventLi
.flatMap(message -> deleteUnreferencedAttachments(message).thenReturn(message))
.flatMap(this::deleteMessageBlobs)
.then(messageDAOV3.delete(messageId))
- .then(threadLookupDAO.selectOneRow(messageId)
+ .then(threadLookupDAO.selectOneRow(threadId, messageId)
.flatMap(key -> threadDAO.deleteSome(key.getUsername(), key.getMimeMessageIds())
.collectList()))
- .then(threadLookupDAO.deleteOneRow(messageId)));
+ .then(threadLookupDAO.deleteOneRow(threadId, messageId)));
}
- private Mono<Void> handleMessageDeletionAsPartOfMailboxDeletion(CassandraMessageId messageId, CassandraId excludedId, Username owner) {
+ private Mono<Void> handleMessageDeletionAsPartOfMailboxDeletion(CassandraMessageId messageId, ThreadId threadId, CassandraId excludedId, Username owner) {
return Mono.just(messageId)
.filterWhen(id -> isReferenced(id, excludedId))
.flatMap(id -> readMessage(id)
@@ -283,10 +279,10 @@ public class DeleteMessageListener implements EventListener.ReactiveGroupEventLi
.flatMap(message -> deleteUnreferencedAttachments(message).thenReturn(message))
.flatMap(this::deleteMessageBlobs)
.then(messageDAOV3.delete(messageId))
- .then(threadLookupDAO.selectOneRow(messageId)
+ .then(threadLookupDAO.selectOneRow(threadId, messageId)
.flatMap(key -> threadDAO.deleteSome(key.getUsername(), key.getMimeMessageIds())
.collectList()))
- .then(threadLookupDAO.deleteOneRow(messageId)));
+ .then(threadLookupDAO.deleteOneRow(threadId, messageId)));
}
private Mono<MessageRepresentation> deleteMessageBlobs(MessageRepresentation message) {
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/ids/CassandraMessageId.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/ids/CassandraMessageId.java
index cdfafa706a..9c9913305a 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/ids/CassandraMessageId.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/ids/CassandraMessageId.java
@@ -46,6 +46,10 @@ public class CassandraMessageId implements MessageId {
}
}
+ public static CassandraMessageId of(UUID uuid) {
+ return new CassandraMessageId(uuid);
+ }
+
private final UUID uuid;
private CassandraMessageId(UUID uuid) {
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraThreadLookupDAO.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraThreadLookupDAO.java
index f910c928f8..a3ba779a08 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraThreadLookupDAO.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraThreadLookupDAO.java
@@ -30,6 +30,7 @@ import static org.apache.james.mailbox.cassandra.table.CassandraMessageIds.MESSA
import static org.apache.james.mailbox.cassandra.table.CassandraThreadLookupTable.MIME_MESSAGE_IDS;
import static org.apache.james.mailbox.cassandra.table.CassandraThreadLookupTable.TABLE_NAME;
import static org.apache.james.mailbox.cassandra.table.CassandraThreadTable.USERNAME;
+import static org.apache.james.mailbox.cassandra.table.MessageIdToImapUid.THREAD_ID;
import java.util.Set;
@@ -39,6 +40,7 @@ import org.apache.james.backends.cassandra.utils.CassandraAsyncExecutor;
import org.apache.james.core.Username;
import org.apache.james.mailbox.cassandra.ids.CassandraMessageId;
import org.apache.james.mailbox.model.MessageId;
+import org.apache.james.mailbox.model.ThreadId;
import com.datastax.oss.driver.api.core.CqlSession;
import com.datastax.oss.driver.api.core.cql.PreparedStatement;
@@ -48,6 +50,7 @@ import com.datastax.oss.driver.api.core.type.codec.TypeCodecs;
import com.datastax.oss.driver.api.core.type.codec.registry.CodecRegistry;
import com.google.common.collect.ImmutableSet;
+import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
public class CassandraThreadLookupDAO {
@@ -56,6 +59,7 @@ public class CassandraThreadLookupDAO {
private final CassandraAsyncExecutor executor;
private final PreparedStatement insert;
private final PreparedStatement select;
+ private final PreparedStatement selectAll;
private final PreparedStatement delete;
@Inject
@@ -64,35 +68,54 @@ public class CassandraThreadLookupDAO {
insert = session.prepare(insertInto(TABLE_NAME)
.value(MESSAGE_ID, bindMarker(MESSAGE_ID))
+ .value(THREAD_ID, bindMarker(THREAD_ID))
.value(USERNAME, bindMarker(USERNAME))
.value(MIME_MESSAGE_IDS, bindMarker(MIME_MESSAGE_IDS))
.build());
select = session.prepare(selectFrom(TABLE_NAME)
.columns(USERNAME, MIME_MESSAGE_IDS)
- .where(column(MESSAGE_ID).isEqualTo(bindMarker(MESSAGE_ID)))
+ .where(column(MESSAGE_ID).isEqualTo(bindMarker(MESSAGE_ID)),
+ column(THREAD_ID).isEqualTo(bindMarker(THREAD_ID)))
+ .build());
+
+ selectAll = session.prepare(selectFrom(TABLE_NAME)
+ .columns(MESSAGE_ID)
+ .where(column(THREAD_ID).isEqualTo(bindMarker(THREAD_ID)))
.build());
delete = session.prepare(deleteFrom(TABLE_NAME)
- .where(column(MESSAGE_ID).isEqualTo(bindMarker(MESSAGE_ID)))
+ .where(column(MESSAGE_ID).isEqualTo(bindMarker(MESSAGE_ID)),
+ column(THREAD_ID).isEqualTo(bindMarker(THREAD_ID)))
.build());
}
- public Mono<Void> insert(MessageId messageId, Username username, Set<Integer> hashMimeMessageIds) {
+ public Mono<Void> insert(MessageId messageId, ThreadId threadId, Username username, Set<Integer> hashMimeMessageIds) {
return executor.executeVoid(insert.bind()
.set(MESSAGE_ID, ((CassandraMessageId) messageId).get(), TypeCodecs.TIMEUUID)
+ .set(THREAD_ID, ((CassandraMessageId) threadId.getBaseMessageId()).get(), TypeCodecs.TIMEUUID)
.set(USERNAME, username.asString(), TypeCodecs.TEXT)
.set(MIME_MESSAGE_IDS, hashMimeMessageIds, SET_OF_INTS_CODEC));
}
- public Mono<ThreadTablePartitionKey> selectOneRow(MessageId messageId) {
+ public Flux<MessageId> selectAll(ThreadId threadId) {
+ return executor.executeRows(
+ selectAll.bind()
+ .set(THREAD_ID, ((CassandraMessageId) threadId.getBaseMessageId()).get(), TypeCodecs.TIMEUUID))
+ .map(row -> CassandraMessageId.of(row.get(MESSAGE_ID, TypeCodecs.TIMEUUID)));
+ }
+
+ public Mono<ThreadTablePartitionKey> selectOneRow(ThreadId threadId, MessageId messageId) {
return executor.executeSingleRow(
- select.bind().set(MESSAGE_ID, ((CassandraMessageId) messageId).get(), TypeCodecs.TIMEUUID))
+ select.bind()
+ .set(THREAD_ID, ((CassandraMessageId) threadId.getBaseMessageId()).get(), TypeCodecs.TIMEUUID)
+ .set(MESSAGE_ID, ((CassandraMessageId) messageId).get(), TypeCodecs.TIMEUUID))
.map(this::readRow);
}
- public Mono<Void> deleteOneRow(MessageId messageId) {
+ public Mono<Void> deleteOneRow(ThreadId threadId, MessageId messageId) {
return executor.executeVoid(delete.bind()
+ .set(THREAD_ID, ((CassandraMessageId) threadId.getBaseMessageId()).get(), TypeCodecs.TIMEUUID)
.set(MESSAGE_ID, ((CassandraMessageId) messageId).get(), TypeCodecs.TIMEUUID));
}
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/modules/CassandraThreadDataDefinition.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/modules/CassandraThreadDataDefinition.java
index a0efdeae01..48ac6350e6 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/modules/CassandraThreadDataDefinition.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/modules/CassandraThreadDataDefinition.java
@@ -47,7 +47,8 @@ public interface CassandraThreadDataDefinition {
.table(CassandraThreadLookupTable.TABLE_NAME)
.comment("Thread table lookup by messageId, using for deletion thread data")
.statement(statement -> types -> statement
- .withPartitionKey(MESSAGE_ID, TIMEUUID)
+ .withPartitionKey(THREAD_ID, TIMEUUID)
+ .withClusteringColumn(MESSAGE_ID, TIMEUUID)
.withColumn(USERNAME, TEXT)
.withColumn(MIME_MESSAGE_IDS, frozenSetOf(INT)))
.build();
diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/table/CassandraThreadLookupTable.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/table/CassandraThreadLookupTable.java
index 2d8f734161..3389bc1a2c 100644
--- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/table/CassandraThreadLookupTable.java
+++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/table/CassandraThreadLookupTable.java
@@ -22,7 +22,7 @@ package org.apache.james.mailbox.cassandra.table;
import com.datastax.oss.driver.api.core.CqlIdentifier;
public interface CassandraThreadLookupTable {
- String TABLE_NAME = "thread_lookup_2";
+ String TABLE_NAME = "thread_lookup_3";
CqlIdentifier MIME_MESSAGE_IDS = CqlIdentifier.fromCql("mimeMessageIds");
}
diff --git a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/CassandraMailboxManagerTest.java b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/CassandraMailboxManagerTest.java
index ee8bce8a78..34972330f2 100644
--- a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/CassandraMailboxManagerTest.java
+++ b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/CassandraMailboxManagerTest.java
@@ -653,7 +653,7 @@ public class CassandraMailboxManagerTest extends MailboxManagerTest<CassandraMai
saveThreadData(session.getUser(), mimeMessageIds, message.getId().getMessageId(), message.getThreadId(), Optional.of(new Subject("Test"))).block();
CassandraMessageId cassandraMessageId = (CassandraMessageId) message.getId().getMessageId();
ThreadTablePartitionKey partitionKey = threadLookupDAO(cassandraCluster)
- .selectOneRow(cassandraMessageId).block();
+ .selectOneRow(message.getThreadId(), cassandraMessageId).block();
mailboxManager.deleteMailbox(inbox, session);
@@ -663,7 +663,7 @@ public class CassandraMailboxManagerTest extends MailboxManagerTest<CassandraMai
.isEmpty();
softly.assertThat(threadLookupDAO(cassandraCluster)
- .selectOneRow(cassandraMessageId).block())
+ .selectOneRow(message.getThreadId(), cassandraMessageId).block())
.isNull();
});
}
@@ -685,7 +685,7 @@ public class CassandraMailboxManagerTest extends MailboxManagerTest<CassandraMai
saveThreadData(session.getUser(), mimeMessageIds, message.getId().getMessageId(), message.getThreadId(), Optional.of(new Subject("Test"))).block();
CassandraMessageId cassandraMessageId = (CassandraMessageId) message.getId().getMessageId();
ThreadTablePartitionKey partitionKey = threadLookupDAO(cassandraCluster)
- .selectOneRow(cassandraMessageId).block();
+ .selectOneRow(message.getThreadId(), cassandraMessageId).block();
cassandraCluster.getConf().registerScenario(fail()
.times(1)
@@ -702,7 +702,7 @@ public class CassandraMailboxManagerTest extends MailboxManagerTest<CassandraMai
.isEmpty();
softly.assertThat(threadLookupDAO(cassandraCluster)
- .selectOneRow(cassandraMessageId).block())
+ .selectOneRow(message.getThreadId(), cassandraMessageId).block())
.isNull();
});
}
@@ -724,7 +724,7 @@ public class CassandraMailboxManagerTest extends MailboxManagerTest<CassandraMai
saveThreadData(session.getUser(), mimeMessageIds, message.getId().getMessageId(), message.getThreadId(), Optional.of(new Subject("Test"))).block();
CassandraMessageId cassandraMessageId = (CassandraMessageId) message.getId().getMessageId();
ThreadTablePartitionKey partitionKey = threadLookupDAO(cassandraCluster)
- .selectOneRow(cassandraMessageId).block();
+ .selectOneRow(message.getThreadId(), cassandraMessageId).block();
inboxManager.delete(ImmutableList.of(message.getId().getUid()), session);
@@ -734,7 +734,7 @@ public class CassandraMailboxManagerTest extends MailboxManagerTest<CassandraMai
.isEmpty();
softly.assertThat(threadLookupDAO(cassandraCluster)
- .selectOneRow(cassandraMessageId).block())
+ .selectOneRow(message.getThreadId(), cassandraMessageId).block())
.isNull();
});
}
@@ -756,7 +756,7 @@ public class CassandraMailboxManagerTest extends MailboxManagerTest<CassandraMai
saveThreadData(session.getUser(), mimeMessageIds, message.getId().getMessageId(), message.getThreadId(), Optional.of(new Subject("Test"))).block();
CassandraMessageId cassandraMessageId = (CassandraMessageId) message.getId().getMessageId();
ThreadTablePartitionKey partitionKey = threadLookupDAO(cassandraCluster)
- .selectOneRow(cassandraMessageId).block();
+ .selectOneRow(message.getThreadId(), cassandraMessageId).block();
cassandraCluster.getConf().registerScenario(fail()
.times(1)
@@ -773,7 +773,7 @@ public class CassandraMailboxManagerTest extends MailboxManagerTest<CassandraMai
.isEmpty();
softly.assertThat(threadLookupDAO(cassandraCluster)
- .selectOneRow(cassandraMessageId).block())
+ .selectOneRow(message.getThreadId(), cassandraMessageId).block())
.isNull();
});
}
@@ -847,7 +847,7 @@ public class CassandraMailboxManagerTest extends MailboxManagerTest<CassandraMai
return threadDAO(cassandra.getCassandraCluster())
.insertSome(username, hashMimeMessagesIds(mimeMessageIds), messageId, threadId, hashSubject(baseSubject))
.then(threadLookupDAO(cassandra.getCassandraCluster())
- .insert(messageId, username, hashMimeMessagesIds(mimeMessageIds)));
+ .insert(messageId, threadId, username, hashMimeMessagesIds(mimeMessageIds)));
}
private Set<MimeMessageId> buildMimeMessageIdSet(Optional<MimeMessageId> mimeMessageId, Optional<MimeMessageId> inReplyTo, Optional<List<MimeMessageId>> references) {
diff --git a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/CassandraThreadIdGuessingAlgorithmTest.java b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/CassandraThreadIdGuessingAlgorithmTest.java
index 1d7e8dd5a4..1cc9fea409 100644
--- a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/CassandraThreadIdGuessingAlgorithmTest.java
+++ b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/CassandraThreadIdGuessingAlgorithmTest.java
@@ -34,6 +34,7 @@ import org.apache.james.events.InVMEventBus;
import org.apache.james.events.MemoryEventDeadLetters;
import org.apache.james.events.delivery.InVmEventDelivery;
import org.apache.james.mailbox.MailboxSession;
+import org.apache.james.mailbox.MessageIdManager;
import org.apache.james.mailbox.cassandra.ids.CassandraMessageId;
import org.apache.james.mailbox.cassandra.mail.CassandraThreadDAO;
import org.apache.james.mailbox.cassandra.mail.CassandraThreadLookupDAO;
@@ -53,6 +54,7 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
public class CassandraThreadIdGuessingAlgorithmTest extends ThreadIdGuessingAlgorithmContract {
+ private MessageIdManager messageIdManager;
private CassandraMailboxManager mailboxManager;
private CassandraThreadDAO threadDAO;
private CassandraThreadLookupDAO threadLookupDAO;
@@ -65,6 +67,7 @@ public class CassandraThreadIdGuessingAlgorithmTest extends ThreadIdGuessingAlgo
eventBus = new InVMEventBus(new InVmEventDelivery(new RecordingMetricFactory()), EventBusTestFixture.RETRY_BACKOFF_CONFIGURATION, new MemoryEventDeadLetters());
CassandraCombinationManagerTestSystem testSystem = (CassandraCombinationManagerTestSystem) CassandraCombinationManagerTestSystem.createTestingData(cassandraCluster.getCassandraCluster(), new NoQuotaManager(), eventBus);
mailboxManager = (CassandraMailboxManager) testSystem.getMailboxManager();
+ messageIdManager = testSystem.getMessageIdManager();
messageIdFactory = new CassandraMessageId.Factory();
return testSystem;
}
@@ -73,7 +76,7 @@ public class CassandraThreadIdGuessingAlgorithmTest extends ThreadIdGuessingAlgo
protected ThreadIdGuessingAlgorithm initThreadIdGuessingAlgorithm(CombinationManagerTestSystem testingData) {
threadDAO = new CassandraThreadDAO(cassandraCluster.getCassandraCluster().getConf());
threadLookupDAO = new CassandraThreadLookupDAO(cassandraCluster.getCassandraCluster().getConf());
- return new CassandraThreadIdGuessingAlgorithm(mailboxManager, threadDAO, threadLookupDAO);
+ return new CassandraThreadIdGuessingAlgorithm(messageIdManager, threadDAO, threadLookupDAO);
}
@Override
@@ -111,7 +114,7 @@ public class CassandraThreadIdGuessingAlgorithmTest extends ThreadIdGuessingAlgo
Optional.of(new MimeMessageId("someInReplyTo")),
Optional.of(List.of(new MimeMessageId("someReferences"), new MimeMessageId("Message-ID1"))));
- assertThat(threadLookupDAO.selectOneRow(newBasedMessageId).block())
+ assertThat(threadLookupDAO.selectOneRow(ThreadId.fromBaseMessageId(newBasedMessageId), newBasedMessageId).block())
.isEqualTo(new ThreadTablePartitionKey(username, hashMimeMessagesIds(mimeMessageIds)));
}
}
diff --git a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraThreadLookupDAOTest.java b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraThreadLookupDAOTest.java
index 9e8c1f0af3..e0cc486291 100644
--- a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraThreadLookupDAOTest.java
+++ b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraThreadLookupDAOTest.java
@@ -20,8 +20,9 @@
package org.apache.james.mailbox.cassandra.mail;
import static org.apache.james.mailbox.cassandra.mail.CassandraThreadDAOTest.hashMimeMessagesIds;
-import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
+import static org.assertj.core.api.Assertions.assertThat;
+import java.util.List;
import java.util.Set;
import org.apache.james.backends.cassandra.CassandraCluster;
@@ -29,6 +30,8 @@ import org.apache.james.backends.cassandra.CassandraClusterExtension;
import org.apache.james.core.Username;
import org.apache.james.mailbox.cassandra.ids.CassandraMessageId;
import org.apache.james.mailbox.cassandra.modules.CassandraThreadDataDefinition;
+import org.apache.james.mailbox.model.MessageId;
+import org.apache.james.mailbox.model.ThreadId;
import org.apache.james.mailbox.store.mail.model.MimeMessageId;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -44,6 +47,7 @@ class CassandraThreadLookupDAOTest {
private CassandraThreadLookupDAO testee;
private CassandraMessageId messageId1;
private CassandraMessageId messageId2;
+ private CassandraMessageId messageId3;
private MimeMessageId mimeMessageId1;
private MimeMessageId mimeMessageId2;
private MimeMessageId mimeMessageId3;
@@ -56,6 +60,7 @@ class CassandraThreadLookupDAOTest {
CassandraMessageId.Factory messageIdFactory = new CassandraMessageId.Factory();
messageId1 = messageIdFactory.generate();
messageId2 = messageIdFactory.generate();
+ messageId3 = messageIdFactory.generate();
mimeMessageId1 = new MimeMessageId("MimeMessageID1");
mimeMessageId2 = new MimeMessageId("MimeMessageID2");
@@ -65,45 +70,57 @@ class CassandraThreadLookupDAOTest {
@Test
void insertShouldSuccess() {
- testee.insert(messageId1, ALICE, hashMimeMessagesIds(Set.of(mimeMessageId1, mimeMessageId2))).block();
+ testee.insert(messageId1, ThreadId.fromBaseMessageId(messageId1), ALICE, hashMimeMessagesIds(Set.of(mimeMessageId1, mimeMessageId2))).block();
- assertThat(testee.selectOneRow(messageId1).block())
+ assertThat(testee.selectOneRow(ThreadId.fromBaseMessageId(messageId1), messageId1).block())
.isEqualTo(new ThreadTablePartitionKey(ALICE, hashMimeMessagesIds(Set.of(mimeMessageId1, mimeMessageId2))));
}
@Test
void selectShouldReturnNullWhenMessageIdNonExist() {
- assertThat(testee.selectOneRow(messageId1).block())
+ assertThat(testee.selectOneRow(ThreadId.fromBaseMessageId(messageId1), messageId1).block())
.isNull();
}
@Test
void selectShouldReturnOnlyRelatedDataByThatMessageId() {
- testee.insert(messageId1, ALICE, hashMimeMessagesIds(Set.of(mimeMessageId1, mimeMessageId2))).block();
- testee.insert(messageId2, BOB, hashMimeMessagesIds(Set.of(mimeMessageId3, mimeMessageId4))).block();
+ testee.insert(messageId1, ThreadId.fromBaseMessageId(messageId1), ALICE, hashMimeMessagesIds(Set.of(mimeMessageId1, mimeMessageId2))).block();
+ testee.insert(messageId2, ThreadId.fromBaseMessageId(messageId2), BOB, hashMimeMessagesIds(Set.of(mimeMessageId3, mimeMessageId4))).block();
- assertThat(testee.selectOneRow(messageId1).block())
+ assertThat(testee.selectOneRow(ThreadId.fromBaseMessageId(messageId1), messageId1).block())
.isEqualTo(new ThreadTablePartitionKey(ALICE, hashMimeMessagesIds(Set.of(mimeMessageId1, mimeMessageId2))));
}
+ @Test
+ void selectAllShouldReturnThreadContent() {
+ testee.insert(messageId1, ThreadId.fromBaseMessageId(messageId1), ALICE, hashMimeMessagesIds(Set.of(mimeMessageId1, mimeMessageId2))).block();
+ testee.insert(messageId2, ThreadId.fromBaseMessageId(messageId2), BOB, hashMimeMessagesIds(Set.of(mimeMessageId3, mimeMessageId4))).block();
+ testee.insert(messageId3, ThreadId.fromBaseMessageId(messageId1), BOB, hashMimeMessagesIds(Set.of(mimeMessageId3, mimeMessageId4))).block();
+
+ List<MessageId> block = testee.selectAll(ThreadId.fromBaseMessageId(messageId1)).collectList().block();
+ assertThat(block)
+ .containsOnly(messageId1, messageId3)
+ .doesNotContain(messageId2);
+ }
+
@Test
void deletedEntriesShouldNotBeReturned() {
- testee.insert(messageId1, ALICE, hashMimeMessagesIds(Set.of(mimeMessageId1, mimeMessageId2))).block();
+ testee.insert(messageId1, ThreadId.fromBaseMessageId(messageId1), ALICE, hashMimeMessagesIds(Set.of(mimeMessageId1, mimeMessageId2))).block();
- testee.deleteOneRow(messageId1).block();
+ testee.deleteOneRow(ThreadId.fromBaseMessageId(messageId1), messageId1).block();
- assertThat(testee.selectOneRow(messageId1).block())
+ assertThat(testee.selectOneRow(ThreadId.fromBaseMessageId(messageId1), messageId1).block())
.isNull();
}
@Test
void deleteByNonExistMessageIdShouldDeleteNothing() {
- testee.insert(messageId1, ALICE, hashMimeMessagesIds(Set.of(mimeMessageId1, mimeMessageId2))).block();
+ testee.insert(messageId1, ThreadId.fromBaseMessageId(messageId1), ALICE, hashMimeMessagesIds(Set.of(mimeMessageId1, mimeMessageId2))).block();
- testee.deleteOneRow(messageId2).block();
+ testee.deleteOneRow(ThreadId.fromBaseMessageId(messageId2), messageId2).block();
// message1's data should remain
- assertThat(testee.selectOneRow(messageId1).block())
+ assertThat(testee.selectOneRow(ThreadId.fromBaseMessageId(messageId1), messageId1).block())
.isEqualTo(new ThreadTablePartitionKey(ALICE, hashMimeMessagesIds(Set.of(mimeMessageId1, mimeMessageId2))));
}
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@james.apache.org
For additional commands, e-mail: notifications-help@james.apache.org