This is an automated email from the ASF dual-hosted git repository.
quantranhong1999 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/james-project.git
The following commit(s) were added to refs/heads/master by this push:
new 37216eb0e4 JAMES-4209 CassandraMessagesDAOV3: optionally write
recovery infos (#3053)
37216eb0e4 is described below
commit 37216eb0e4c233f485c2ece18c764b1ab3b7f01d
Author: Benoit TELLIER <[email protected]>
AuthorDate: Fri Jun 5 04:30:09 2026 +0200
JAMES-4209 CassandraMessagesDAOV3: optionally write recovery infos (#3053)
---
.../init/configuration/CassandraConfiguration.java | 45 ++++++++++-
.../cassandra/mail/CassandraMessageDAOV3.java | 30 +++++++-
.../cassandra/CassandraMailboxManagerTest.java | 2 +
.../cassandra/mail/CassandraMessageDAOV3Test.java | 88 +++++++++++++++++++---
.../mailbox/cassandra/mail/utils/GuiceUtils.java | 11 +++
.../org/apache/james/blob/api/BlobStoreDAO.java | 2 +
.../apache/james/blob/file/FileBlobStoreDAO.java | 21 +++++-
.../blob/deduplication/BloomFilterGCAlgorithm.java | 37 ++++++---
.../blob/deduplication/GenerationAwareBlobId.java | 6 ++
.../BloomFilterGCAlgorithmContract.java | 83 ++++++++++++++++++++
.../MetaDataFixInconsistenciesServiceTest.java | 3 +
server/protocols/webadmin/webadmin-pop3/pom.xml | 5 ++
.../Pop3MetaDataFixInconsistenciesRoutesTest.java | 3 +
13 files changed, 310 insertions(+), 26 deletions(-)
diff --git
a/backends-common/cassandra/src/main/java/org/apache/james/backends/cassandra/init/configuration/CassandraConfiguration.java
b/backends-common/cassandra/src/main/java/org/apache/james/backends/cassandra/init/configuration/CassandraConfiguration.java
index 9346821aaa..9c0ffb7e5c 100644
---
a/backends-common/cassandra/src/main/java/org/apache/james/backends/cassandra/init/configuration/CassandraConfiguration.java
+++
b/backends-common/cassandra/src/main/java/org/apache/james/backends/cassandra/init/configuration/CassandraConfiguration.java
@@ -39,6 +39,19 @@ import com.google.common.collect.ImmutableList;
public class CassandraConfiguration {
private static final Logger LOGGER =
LoggerFactory.getLogger(CassandraConfiguration.class);
+ public enum BlobRecoveryMode {
+ NONE, SYNCHRONOUS, ASYNCHRONOUS;
+
+ public static BlobRecoveryMode parse(String value) {
+ return switch (value.toLowerCase()) {
+ case "none" -> NONE;
+ case "synchronous" -> SYNCHRONOUS;
+ case "asynchronous" -> ASYNCHRONOUS;
+ default -> throw new IllegalArgumentException("Unknown blob
recovery mode: '" + value + "'. Expected none, synchronous or asynchronous");
+ };
+ }
+ }
+
public static final float DEFAULT_MAILBOX_READ_REPAIR = 0.1f;
public static final float DEFAULT_MAX_MAILBOX_COUNTERS_READ_REPAIR_CHANCE
= 0.1f;
public static final float
DEFAULT_ONE_HUNDRED_MAILBOX_COUNTERS_READ_REPAIR_CHANCE = 0.01f;
@@ -59,6 +72,7 @@ public class CassandraConfiguration {
public static final boolean DEFAULT_STRONG_CONSISTENCY = true;
public static final boolean DEFAULT_OPTIMISTIC_CONSISTENCY_LEVEL = false;
public static final boolean DEFAULT_MAIL_REPOSITORY_STRONG_CONSISTENCY =
true;
+ public static final BlobRecoveryMode DEFAULT_BLOB_RECOVERY_MODE =
BlobRecoveryMode.NONE;
private static final String MAILBOX_READ_REPAIR =
"mailbox.read.repair.chance";
private static final String MAILBOX_MAX_COUNTERS_READ_REPAIR =
"mailbox.counters.read.repair.chance.max";
@@ -84,6 +98,7 @@ public class CassandraConfiguration {
private static final String MAIL_REPOSITORY_STRONG_CONSISTENCY =
"mailrepository.strong.consistency";
private static final String ACL_ENABLED = "acl.enabled";
private static final String UID_MODSEQ_INCREMENT = "uid.modseq.increment";
+ private static final String MAILBOX_BLOB_RECOVERY_MODE =
"mailbox.blob.recovery.mode";
public static final CassandraConfiguration DEFAULT_CONFIGURATION =
builder().build();
@@ -112,6 +127,7 @@ public class CassandraConfiguration {
private Optional<Boolean> mailRepositoryStrongConsistency =
Optional.empty();
private Optional<Boolean> aclEnabled = Optional.empty();
private Optional<Integer> uidModseqIncrement = Optional.empty();
+ private Optional<BlobRecoveryMode> blobRecoveryMode = Optional.empty();
public Builder mailboxReadStrongConsistency(boolean value) {
this.mailboxReadStrongConsistency = Optional.of(value);
@@ -175,6 +191,16 @@ public class CassandraConfiguration {
return this;
}
+ public Builder blobRecoveryMode(BlobRecoveryMode value) {
+ this.blobRecoveryMode = Optional.of(value);
+ return this;
+ }
+
+ public Builder blobRecoveryMode(Optional<BlobRecoveryMode> value) {
+ this.blobRecoveryMode = value;
+ return this;
+ }
+
public Builder flagsUpdateMessageIdMaxRetry(int value) {
Preconditions.checkArgument(value > 0,
"flagsUpdateMessageIdMaxRetry needs to be strictly positive");
this.flagsUpdateMessageIdMaxRetry = Optional.of(value);
@@ -386,7 +412,8 @@ public class CassandraConfiguration {
uidReadStrongConsistency.orElse(DEFAULT_STRONG_CONSISTENCY),
modseqReadStrongConsistency.orElse(DEFAULT_STRONG_CONSISTENCY),
aclEnabled.orElse(true),
- uidModseqIncrement.orElse(0));
+ uidModseqIncrement.orElse(0),
+ blobRecoveryMode.orElse(DEFAULT_BLOB_RECOVERY_MODE));
}
}
@@ -442,6 +469,8 @@ public class CassandraConfiguration {
propertiesConfiguration.getBoolean(MAIL_REPOSITORY_STRONG_CONSISTENCY, null)))
.aclEnabled(Optional.ofNullable(propertiesConfiguration.getBoolean(ACL_ENABLED,
null)))
.uidModseqIncrement(Optional.ofNullable(propertiesConfiguration.getInteger(UID_MODSEQ_INCREMENT,
null)))
+
.blobRecoveryMode(Optional.ofNullable(propertiesConfiguration.getString(MAILBOX_BLOB_RECOVERY_MODE,
null))
+ .map(BlobRecoveryMode::parse))
.build();
}
@@ -469,6 +498,7 @@ public class CassandraConfiguration {
private final boolean modseqReadStrongConsistency;
private final boolean aclEnabled;
private final int uidModseqIncrement;
+ private final BlobRecoveryMode blobRecoveryMode;
@VisibleForTesting
CassandraConfiguration(int aclMaxRetry, int expungeChunkSize,
@@ -481,7 +511,7 @@ public class CassandraConfiguration {
boolean messageReadStrongConsistency, boolean
messageWriteStrongConsistency,
boolean optimisticConsistencyLevel, boolean
mailRepositoryStrongConsistency,
boolean uidReadStrongConsistency, boolean
modseqReadStrongConsistency, boolean aclEnabled,
- int uidModseqIncrement) {
+ int uidModseqIncrement, BlobRecoveryMode
blobRecoveryMode) {
this.aclMaxRetry = aclMaxRetry;
this.expungeChunkSize = expungeChunkSize;
this.flagsUpdateMessageIdMaxRetry = flagsUpdateMessageIdMaxRetry;
@@ -506,6 +536,7 @@ public class CassandraConfiguration {
this.modseqReadStrongConsistency = modseqReadStrongConsistency;
this.aclEnabled = aclEnabled;
this.uidModseqIncrement = uidModseqIncrement;
+ this.blobRecoveryMode = blobRecoveryMode;
}
public boolean isUidReadStrongConsistency() {
@@ -604,6 +635,10 @@ public class CassandraConfiguration {
return uidModseqIncrement;
}
+ public BlobRecoveryMode getBlobRecoveryMode() {
+ return blobRecoveryMode;
+ }
+
@Override
public final boolean equals(Object o) {
if (o instanceof CassandraConfiguration) {
@@ -632,7 +667,8 @@ public class CassandraConfiguration {
&& Objects.equals(this.modseqReadStrongConsistency,
that.modseqReadStrongConsistency)
&& Objects.equals(this.mailRepositoryStrongConsistency,
that.mailRepositoryStrongConsistency)
&& Objects.equals(this.aclEnabled, that.aclEnabled)
- && Objects.equals(this.uidModseqIncrement,
that.uidModseqIncrement);
+ && Objects.equals(this.uidModseqIncrement,
that.uidModseqIncrement)
+ && Objects.equals(this.blobRecoveryMode,
that.blobRecoveryMode);
}
return false;
}
@@ -646,7 +682,7 @@ public class CassandraConfiguration {
consistencyLevelRegular, consistencyLevelLightweightTransaction,
mailboxReadRepair,
messageReadStrongConsistency, mailboxReadStrongConsistency,
messageWriteStrongConsistency,
optimisticConsistencyLevel, mailRepositoryStrongConsistency,
uidReadStrongConsistency,
- modseqReadStrongConsistency, aclEnabled, uidModseqIncrement);
+ modseqReadStrongConsistency, aclEnabled, uidModseqIncrement,
blobRecoveryMode);
}
@Override
@@ -676,6 +712,7 @@ public class CassandraConfiguration {
.add("uidReadStrongConsistency", uidReadStrongConsistency)
.add("aclEnabled", aclEnabled)
.add("uidModseqIncrement", uidModseqIncrement)
+ .add("blobRecoveryMode", blobRecoveryMode)
.toString();
}
}
diff --git
a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageDAOV3.java
b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageDAOV3.java
index cc986a1bf8..0b2eea3657 100644
---
a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageDAOV3.java
+++
b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageDAOV3.java
@@ -41,6 +41,7 @@ import static
org.apache.james.mailbox.cassandra.table.CassandraMessageV3Table.T
import java.io.IOException;
import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.List;
import java.util.Optional;
@@ -55,6 +56,7 @@ import
org.apache.james.backends.cassandra.utils.CassandraAsyncExecutor;
import org.apache.james.backends.cassandra.utils.ProfileLocator;
import org.apache.james.blob.api.BlobId;
import org.apache.james.blob.api.BlobStore;
+import org.apache.james.blob.api.BlobStoreDAO;
import org.apache.james.mailbox.cassandra.ids.CassandraMessageId;
import
org.apache.james.mailbox.cassandra.table.CassandraMessageV3Table.Attachments;
import org.apache.james.mailbox.model.ByteContent;
@@ -66,6 +68,8 @@ import
org.apache.james.mailbox.model.MessageAttachmentMetadata;
import org.apache.james.mailbox.model.StringBackedAttachmentId;
import org.apache.james.mailbox.store.mail.MessageMapper.FetchType;
import org.apache.james.mailbox.store.mail.model.MailboxMessage;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import com.datastax.oss.driver.api.core.CqlIdentifier;
import com.datastax.oss.driver.api.core.CqlSession;
@@ -88,10 +92,12 @@ import reactor.core.scheduler.Schedulers;
import reactor.util.function.Tuple2;
public class CassandraMessageDAOV3 {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(CassandraMessageDAOV3.class);
private static final byte[] EMPTY_BYTE_ARRAY = {};
private final CassandraAsyncExecutor cassandraAsyncExecutor;
private final BlobStore blobStore;
+ private final BlobStoreDAO blobStoreDAO;
private final BlobId.Factory blobIdFactory;
private final PreparedStatement insert;
private final PreparedStatement delete;
@@ -107,9 +113,11 @@ public class CassandraMessageDAOV3 {
@Inject
public CassandraMessageDAOV3(CqlSession session, CassandraTypesProvider
typesProvider, BlobStore blobStore,
- BlobId.Factory blobIdFactory,
CassandraConfiguration cassandraConfiguration) {
+ BlobStoreDAO blobStoreDAO, BlobId.Factory
blobIdFactory,
+ CassandraConfiguration
cassandraConfiguration) {
this.cassandraAsyncExecutor = new CassandraAsyncExecutor(session);
this.blobStore = blobStore;
+ this.blobStoreDAO = blobStoreDAO;
this.blobIdFactory = blobIdFactory;
this.insert = prepareInsert(session);
@@ -205,10 +213,28 @@ public class CassandraMessageDAOV3 {
Mono<BlobId> headerFuture =
Mono.from(blobStore.save(blobStore.getDefaultBucketName(), headerContent,
SIZE_BASED));
Mono<BlobId> bodyFuture =
Mono.from(blobStore.save(blobStore.getDefaultBucketName(), bodyByteSource,
LOW_COST));
- return headerFuture.zipWith(bodyFuture);
+ return headerFuture.zipWith(bodyFuture)
+ .flatMap(pair -> saveRecovery(pair.getT1(),
pair.getT2()).thenReturn(pair));
});
}
+ private Mono<Void> saveRecovery(BlobId headerId, BlobId bodyId) {
+ return switch (configuration.getBlobRecoveryMode()) {
+ case NONE -> Mono.empty();
+ case SYNCHRONOUS -> writeRecoveryBlob(headerId, bodyId);
+ case ASYNCHRONOUS -> Mono.fromRunnable(() ->
+ writeRecoveryBlob(headerId, bodyId)
+ .subscribeOn(Schedulers.parallel())
+ .subscribe(ignored -> { }, e -> LOGGER.error("Failed to
save recovery blob for header={} body={}", headerId.asString(),
bodyId.asString(), e)));
+ };
+ }
+
+ private Mono<Void> writeRecoveryBlob(BlobId headerId, BlobId bodyId) {
+ BlobId recoveryBlobId =
blobIdFactory.parse(BlobStoreDAO.RECOVERY_BLOB_PREFIX + headerId.asString());
+ BlobStoreDAO.BytesBlob content =
BlobStoreDAO.BytesBlob.of(bodyId.asString().getBytes(StandardCharsets.UTF_8));
+ return Mono.from(blobStoreDAO.save(blobStore.getDefaultBucketName(),
recoveryBlobId, content));
+ }
+
private BoundStatement boundWriteStatement(MailboxMessage message,
Tuple2<BlobId, BlobId> pair) {
CassandraMessageId messageId = (CassandraMessageId)
message.getMessageId();
BoundStatement boundStatement = insert.bind()
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 1a6ffab2f1..a13b5293fb 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
@@ -41,6 +41,7 @@ import
org.apache.james.backends.cassandra.CassandraClusterExtension;
import org.apache.james.backends.cassandra.StatementRecorder;
import
org.apache.james.backends.cassandra.init.configuration.CassandraConfiguration;
import org.apache.james.blob.api.BlobStore;
+import org.apache.james.blob.api.BlobStoreDAO;
import org.apache.james.blob.api.PlainBlobId;
import org.apache.james.blob.cassandra.BlobTables;
import org.apache.james.core.Username;
@@ -832,6 +833,7 @@ public class CassandraMailboxManagerTest extends
MailboxManagerTest<CassandraMai
cassandraCluster.getConf(),
cassandraCluster.getTypesProvider(),
mock(BlobStore.class),
+ mock(BlobStoreDAO.class),
new PlainBlobId.Factory(),
CassandraConfiguration.DEFAULT_CONFIGURATION);
}
diff --git
a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageDAOV3Test.java
b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageDAOV3Test.java
index 7f334b5d0e..9a3067e1fc 100644
---
a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageDAOV3Test.java
+++
b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageDAOV3Test.java
@@ -20,6 +20,7 @@ package org.apache.james.mailbox.cassandra.mail;
import static
org.apache.james.mailbox.store.mail.model.MailboxMessage.EMPTY_SAVE_DATE;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import java.nio.charset.StandardCharsets;
import java.util.Collection;
@@ -35,10 +36,16 @@ import
org.apache.james.backends.cassandra.CassandraClusterExtension;
import org.apache.james.backends.cassandra.components.CassandraDataDefinition;
import
org.apache.james.backends.cassandra.init.configuration.CassandraConfiguration;
import
org.apache.james.backends.cassandra.versions.CassandraSchemaVersionDataDefinition;
+import org.apache.james.blob.api.BlobId;
import org.apache.james.blob.api.BlobStore;
+import org.apache.james.blob.api.BlobStoreDAO;
+import org.apache.james.blob.api.BucketName;
+import org.apache.james.blob.api.ObjectNotFoundException;
import org.apache.james.blob.api.PlainBlobId;
import org.apache.james.blob.cassandra.CassandraBlobDataDefinition;
-import org.apache.james.blob.cassandra.CassandraBlobStoreFactory;
+import org.apache.james.blob.cassandra.CassandraBlobStoreDAO;
+import org.apache.james.blob.cassandra.CassandraBucketDAO;
+import org.apache.james.blob.cassandra.CassandraDefaultBucketDAO;
import org.apache.james.mailbox.MessageUid;
import org.apache.james.mailbox.ModSeq;
import org.apache.james.mailbox.cassandra.ids.CassandraId;
@@ -54,6 +61,7 @@ import org.apache.james.mailbox.store.mail.MessageMapper;
import org.apache.james.mailbox.store.mail.model.MailboxMessage;
import org.apache.james.mailbox.store.mail.model.impl.SimpleMailboxMessage;
import org.apache.james.metrics.tests.RecordingMetricFactory;
+import org.apache.james.server.blob.deduplication.BlobStoreFactory;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
@@ -61,6 +69,7 @@ import org.junit.jupiter.api.extension.RegisterExtension;
import com.google.common.collect.ImmutableList;
import reactor.core.publisher.Mono;
+import reactor.util.function.Tuple2;
class CassandraMessageDAOV3Test {
private static final int BODY_START = 16;
@@ -87,6 +96,9 @@ class CassandraMessageDAOV3Test {
private ThreadId threadId;
private ComposedMessageIdWithMetaData messageIdWithMetadata;
private MessageBlobReferenceSource blobReferenceSource;
+ private BlobStoreDAO blobStoreDAO;
+ private BlobStore blobStore;
+ private PlainBlobId.Factory blobIdFactory;
@BeforeEach
void setUp(CassandraCluster cassandra) {
@@ -94,15 +106,19 @@ class CassandraMessageDAOV3Test {
messageId = messageIdFactory.generate();
messageId2 = messageIdFactory.generate();
threadId = ThreadId.fromBaseMessageId(messageId);
- BlobStore blobStore =
CassandraBlobStoreFactory.forTesting(cassandra.getConf(), new
RecordingMetricFactory())
+
+ blobIdFactory = new PlainBlobId.Factory();
+ RecordingMetricFactory metricFactory = new RecordingMetricFactory();
+ CassandraBucketDAO bucketDAO = new CassandraBucketDAO(blobIdFactory,
cassandra.getConf());
+ CassandraDefaultBucketDAO defaultBucketDAO = new
CassandraDefaultBucketDAO(cassandra.getConf(), blobIdFactory);
+ blobStoreDAO = new CassandraBlobStoreDAO(defaultBucketDAO, bucketDAO,
CassandraConfiguration.DEFAULT_CONFIGURATION, BucketName.DEFAULT,
metricFactory);
+ blobStore = BlobStoreFactory.builder()
+ .blobStoreDAO(blobStoreDAO)
+ .blobIdFactory(blobIdFactory)
+ .defaultBucketName()
.passthrough();
- PlainBlobId.Factory blobIdFactory = new PlainBlobId.Factory();
- testee = new CassandraMessageDAOV3(
- cassandra.getConf(),
- cassandra.getTypesProvider(),
- blobStore,
- blobIdFactory,
- CassandraConfiguration.DEFAULT_CONFIGURATION);
+
+ testee = buildTestee(cassandra,
CassandraConfiguration.DEFAULT_CONFIGURATION);
messageIdWithMetadata = ComposedMessageIdWithMetaData.builder()
.composedMessageId(new ComposedMessageId(MAILBOX_ID,
messageId, messageUid))
@@ -113,6 +129,16 @@ class CassandraMessageDAOV3Test {
blobReferenceSource = new MessageBlobReferenceSource(testee);
}
+ private CassandraMessageDAOV3 buildTestee(CassandraCluster cassandra,
CassandraConfiguration configuration) {
+ return new CassandraMessageDAOV3(
+ cassandra.getConf(),
+ cassandra.getTypesProvider(),
+ blobStore,
+ blobStoreDAO,
+ blobIdFactory,
+ configuration);
+ }
+
@Test
void saveShouldStoreMessageWithFullContent() throws Exception {
message = createMessage(messageId, threadId, CONTENT, BODY_START,
NO_ATTACHMENT, EMPTY_SAVE_DATE);
@@ -145,6 +171,50 @@ class CassandraMessageDAOV3Test {
.isEmpty();
}
+ @Test
+ void saveShouldNotWriteRecoveryBlobByDefault() throws Exception {
+ message = createMessage(messageId, threadId, CONTENT, BODY_START,
NO_ATTACHMENT, EMPTY_SAVE_DATE);
+
+ Tuple2<BlobId, BlobId> blobIds = testee.save(message).block();
+
+ BlobId recoveryBlobId =
blobIdFactory.parse(BlobStoreDAO.RECOVERY_BLOB_PREFIX +
blobIds.getT1().asString());
+ assertThatThrownBy(() ->
Mono.from(blobStoreDAO.readBytes(BucketName.DEFAULT, recoveryBlobId)).block())
+ .isInstanceOf(ObjectNotFoundException.class);
+ }
+
+ @Test
+ void saveShouldWriteRecoveryBlobWhenSynchronousMode(CassandraCluster
cassandra) throws Exception {
+ CassandraConfiguration conf = CassandraConfiguration.builder()
+
.blobRecoveryMode(CassandraConfiguration.BlobRecoveryMode.SYNCHRONOUS)
+ .build();
+ CassandraMessageDAOV3 testeeWithRecovery = buildTestee(cassandra,
conf);
+ message = createMessage(messageId, threadId, CONTENT, BODY_START,
NO_ATTACHMENT, EMPTY_SAVE_DATE);
+
+ Tuple2<BlobId, BlobId> blobIds =
testeeWithRecovery.save(message).block();
+
+ BlobId recoveryBlobId =
blobIdFactory.parse(BlobStoreDAO.RECOVERY_BLOB_PREFIX +
blobIds.getT1().asString());
+ byte[] recoveryContent =
Mono.from(blobStoreDAO.readBytes(BucketName.DEFAULT,
recoveryBlobId)).block().payload();
+ assertThat(new String(recoveryContent,
StandardCharsets.UTF_8)).isEqualTo(blobIds.getT2().asString());
+ }
+
+ @Test
+ void saveShouldWriteRecoveryBlobWhenAsynchronousMode(CassandraCluster
cassandra) throws Exception {
+ CassandraConfiguration conf = CassandraConfiguration.builder()
+
.blobRecoveryMode(CassandraConfiguration.BlobRecoveryMode.ASYNCHRONOUS)
+ .build();
+ CassandraMessageDAOV3 testeeWithRecovery = buildTestee(cassandra,
conf);
+ message = createMessage(messageId, threadId, CONTENT, BODY_START,
NO_ATTACHMENT, EMPTY_SAVE_DATE);
+
+ Tuple2<BlobId, BlobId> blobIds =
testeeWithRecovery.save(message).block();
+
+ BlobId recoveryBlobId =
blobIdFactory.parse(BlobStoreDAO.RECOVERY_BLOB_PREFIX +
blobIds.getT1().asString());
+ // Asynchronous: give the parallel scheduler a moment to complete
+ assertThat(Mono.from(blobStoreDAO.readBytes(BucketName.DEFAULT,
recoveryBlobId))
+ .retryWhen(reactor.util.retry.Retry.fixedDelay(10,
java.time.Duration.ofMillis(100)))
+ .block().payload())
+
.isEqualTo(blobIds.getT2().asString().getBytes(StandardCharsets.UTF_8));
+ }
+
@Test
void blobReferencesShouldReturnAllBlobs() throws Exception {
message = createMessage(messageId, threadId, CONTENT, BODY_START,
NO_ATTACHMENT, EMPTY_SAVE_DATE);
diff --git
a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/utils/GuiceUtils.java
b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/utils/GuiceUtils.java
index 1204119170..bf93e5d4de 100644
---
a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/utils/GuiceUtils.java
+++
b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/utils/GuiceUtils.java
@@ -27,8 +27,13 @@ import
org.apache.james.backends.cassandra.init.CassandraTypesProvider;
import
org.apache.james.backends.cassandra.init.configuration.CassandraConfiguration;
import org.apache.james.blob.api.BlobId;
import org.apache.james.blob.api.BlobStore;
+import org.apache.james.blob.api.BlobStoreDAO;
+import org.apache.james.blob.api.BucketName;
import org.apache.james.blob.api.PlainBlobId;
+import org.apache.james.blob.cassandra.CassandraBlobStoreDAO;
import org.apache.james.blob.cassandra.CassandraBlobStoreFactory;
+import org.apache.james.blob.cassandra.CassandraBucketDAO;
+import org.apache.james.blob.cassandra.CassandraDefaultBucketDAO;
import org.apache.james.eventsourcing.Event;
import org.apache.james.eventsourcing.eventstore.EventNestedTypes;
import org.apache.james.eventsourcing.eventstore.EventStore;
@@ -91,6 +96,12 @@ public class GuiceUtils {
binder ->
binder.bind(ACLMapper.class).to(CassandraACLMapper.class),
binder -> binder.bind(BlobId.Factory.class).toInstance(new
PlainBlobId.Factory()),
binder -> binder.bind(BlobStore.class).toProvider(() ->
CassandraBlobStoreFactory.forTesting(session, new
RecordingMetricFactory()).passthrough()),
+ binder -> binder.bind(BlobStoreDAO.class).toProvider(() -> {
+ PlainBlobId.Factory blobIdFactory = new PlainBlobId.Factory();
+ CassandraBucketDAO bucketDAO = new
CassandraBucketDAO(blobIdFactory, session);
+ CassandraDefaultBucketDAO defaultBucketDAO = new
CassandraDefaultBucketDAO(session, blobIdFactory);
+ return new CassandraBlobStoreDAO(defaultBucketDAO, bucketDAO,
configuration, BucketName.DEFAULT, new RecordingMetricFactory());
+ }),
binder -> binder.bind(CqlSession.class).toInstance(session),
binder -> Multibinder.newSetBinder(binder, new
TypeLiteral<EventDTOModule<? extends Event, ? extends EventDTO>>() {})
.addBinding().toInstance(ACLModule.ACL_UPDATE),
diff --git
a/server/blob/blob-api/src/main/java/org/apache/james/blob/api/BlobStoreDAO.java
b/server/blob/blob-api/src/main/java/org/apache/james/blob/api/BlobStoreDAO.java
index a527dad825..41d2c257d2 100644
---
a/server/blob/blob-api/src/main/java/org/apache/james/blob/api/BlobStoreDAO.java
+++
b/server/blob/blob-api/src/main/java/org/apache/james/blob/api/BlobStoreDAO.java
@@ -52,6 +52,8 @@ import com.google.common.io.FileBackedOutputStream;
* <p>See {@code docs/modules/servers/partials/architecture/blobstore.adoc}
for more details.</p>
*/
public interface BlobStoreDAO {
+ String RECOVERY_BLOB_PREFIX = "recovery/";
+
record BlobMetadataName(String name) {
private static final CharMatcher CHAR_MATCHER =
CharMatcher.inRange('a', 'z')
.or(CharMatcher.inRange('A', 'Z'))
diff --git
a/server/blob/blob-file/src/main/java/org/apache/james/blob/file/FileBlobStoreDAO.java
b/server/blob/blob-file/src/main/java/org/apache/james/blob/file/FileBlobStoreDAO.java
index 0a5d1731c0..c1db4d532a 100644
---
a/server/blob/blob-file/src/main/java/org/apache/james/blob/file/FileBlobStoreDAO.java
+++
b/server/blob/blob-file/src/main/java/org/apache/james/blob/file/FileBlobStoreDAO.java
@@ -39,6 +39,8 @@ import java.nio.file.attribute.UserDefinedFileAttributeView;
import java.time.Duration;
import java.util.Collection;
import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
import jakarta.inject.Inject;
@@ -234,13 +236,23 @@ public class FileBlobStoreDAO implements BlobStoreDAO {
public Publisher<BlobId> listBlobs(BucketName bucketName) {
return Mono.fromCallable(() -> {
File bucketRoot = getBucketRoot(bucketName);
- return Files.list(bucketRoot.toPath());
+ Path rootPath = bucketRoot.toPath();
+ // Blob ids may contain '/' (eg. recovery sidecar keys) and
are then stored in nested
+ // directories, so we walk the tree and rebuild the id from
the bucket-root-relative path.
+ return Files.walk(rootPath)
+ .filter(Files::isRegularFile)
+ .map(path ->
blobIdFactory.parse(toBlobId(rootPath.relativize(path))));
})
.flatMapMany(Flux::fromStream)
- .map(path -> blobIdFactory.parse(path.getFileName().toString()))
.subscribeOn(Schedulers.boundedElastic());
}
+ private String toBlobId(Path relativePath) {
+ return StreamSupport.stream(relativePath.spliterator(), false)
+ .map(Path::toString)
+ .collect(Collectors.joining("/"));
+ }
+
private BlobMetadata readMetadata(Path path) {
UserDefinedFileAttributeView attributeView =
Files.getFileAttributeView(path, UserDefinedFileAttributeView.class);
if (attributeView == null) {
@@ -254,6 +266,8 @@ public class FileBlobStoreDAO implements BlobStoreDAO {
this::asBlobMetadataName,
Throwing.function((String attributeName) -> new
BlobMetadataValue(readFileAttributeValue(attributeView, attributeName)))
.sneakyThrow())));
+ } catch (NoSuchFileException e) {
+ throw new ObjectNotFoundException(String.format("Cannot locate
%s", path), e);
} catch (IOException e) {
throw new ObjectStoreIOException("IOException occurred", e);
}
@@ -299,6 +313,9 @@ public class FileBlobStoreDAO implements BlobStoreDAO {
private File createTempFile(File blob) {
try {
+ // Blob ids may contain '/' (eg. recovery sidecar keys),
introducing nested directories
+ // that must exist before the temp file is created alongside the
target blob.
+ Files.createDirectories(blob.getParentFile().toPath());
return Files.createTempFile(blob.getParentFile().toPath(),
blob.getName(), ".tmp").toFile();
} catch (IOException e) {
throw new ObjectStoreIOException("IOException occurred", e);
diff --git
a/server/blob/blob-storage-strategy/src/main/java/org/apache/james/server/blob/deduplication/BloomFilterGCAlgorithm.java
b/server/blob/blob-storage-strategy/src/main/java/org/apache/james/server/blob/deduplication/BloomFilterGCAlgorithm.java
index 67fa255aec..f9a9ea10fa 100644
---
a/server/blob/blob-storage-strategy/src/main/java/org/apache/james/server/blob/deduplication/BloomFilterGCAlgorithm.java
+++
b/server/blob/blob-storage-strategy/src/main/java/org/apache/james/server/blob/deduplication/BloomFilterGCAlgorithm.java
@@ -25,10 +25,12 @@ import java.nio.charset.StandardCharsets;
import java.time.Clock;
import java.time.Instant;
import java.util.Collection;
+import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.Collectors;
import org.apache.james.blob.api.BlobId;
import org.apache.james.blob.api.BlobReferenceSource;
@@ -39,6 +41,7 @@ import org.apache.james.task.Task.Result;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.MoreObjects;
import com.google.common.hash.BloomFilter;
import com.google.common.hash.Funnel;
@@ -51,6 +54,8 @@ public class BloomFilterGCAlgorithm {
private static final Logger LOGGER =
LoggerFactory.getLogger(BloomFilterGCAlgorithm.class);
private static final Funnel<CharSequence> BLOOM_FILTER_FUNNEL =
Funnels.stringFunnel(StandardCharsets.US_ASCII);
+ @VisibleForTesting
+ static boolean RECOVERY_AWARE =
Boolean.parseBoolean(System.getProperty("james.gc.recover.aware", "true"));
public static class Context {
@@ -273,6 +278,7 @@ public class BloomFilterGCAlgorithm {
private Mono<Result> gc(BloomFilter<CharSequence> bloomFilter, BucketName
bucketName, Context context, int deletionWindowSize) {
return Flux.from(blobStoreDAO.listBlobs(bucketName))
+ .filter(blobId -> !RECOVERY_AWARE ||
!blobId.asString().startsWith(BlobStoreDAO.RECOVERY_BLOB_PREFIX))
.doOnNext(blobId -> context.incrementBlobCount())
.flatMap(blobId -> Mono.fromCallable(() ->
blobIdFactory.parse(blobId.asString())))
.filter(blobId -> {
@@ -290,15 +296,28 @@ public class BloomFilterGCAlgorithm {
private Mono<Result> handlePagedDeletion(BucketName bucketName, Context
context, Flux<BlobId> blobIdFlux) {
return blobIdFlux.collectList()
- .flatMap(orphanBlobIds ->
Mono.from(blobStoreDAO.delete(bucketName, (Collection) orphanBlobIds))
- .then(Mono.fromCallable(() -> {
- context.incrementGCedBlobCount(orphanBlobIds.size());
- return Result.COMPLETED;
- })).onErrorResume(error -> {
- LOGGER.error("Error when gc orphan blob", error);
- context.incrementErrorCount();
- return Mono.just(Result.PARTIAL);
- }));
+ .flatMap(orphanBlobIds -> {
+ Mono<Void> deleteRecoverySidecars = RECOVERY_AWARE
+ ? Mono.from(blobStoreDAO.delete(bucketName,
toRecoveryBlobIds(orphanBlobIds)))
+ : Mono.empty();
+
+ return Mono.from(blobStoreDAO.delete(bucketName, (Collection)
orphanBlobIds))
+ .then(deleteRecoverySidecars)
+ .then(Mono.fromCallable(() -> {
+ context.incrementGCedBlobCount(orphanBlobIds.size());
+ return Result.COMPLETED;
+ })).onErrorResume(error -> {
+ LOGGER.error("Error when gc orphan blob", error);
+ context.incrementErrorCount();
+ return Mono.just(Result.PARTIAL);
+ });
+ });
+ }
+
+ private List<BlobId> toRecoveryBlobIds(List<BlobId> blobIds) {
+ return blobIds.stream()
+ .map(blobId ->
blobIdFactory.parse(BlobStoreDAO.RECOVERY_BLOB_PREFIX + blobId.asString()))
+ .collect(Collectors.toList());
}
private Mono<BloomFilter<CharSequence>> populatedBloomFilter(int
expectedBlobCount, double associatedProbability, Context context) {
diff --git
a/server/blob/blob-storage-strategy/src/main/java/org/apache/james/server/blob/deduplication/GenerationAwareBlobId.java
b/server/blob/blob-storage-strategy/src/main/java/org/apache/james/server/blob/deduplication/GenerationAwareBlobId.java
index 67fd492b16..ddd01b08f6 100644
---
a/server/blob/blob-storage-strategy/src/main/java/org/apache/james/server/blob/deduplication/GenerationAwareBlobId.java
+++
b/server/blob/blob-storage-strategy/src/main/java/org/apache/james/server/blob/deduplication/GenerationAwareBlobId.java
@@ -27,6 +27,7 @@ import java.util.Objects;
import java.util.Optional;
import org.apache.james.blob.api.BlobId;
+import org.apache.james.blob.api.BlobStoreDAO;
import org.apache.james.util.DurationParser;
import com.google.common.annotations.VisibleForTesting;
@@ -128,6 +129,11 @@ public class GenerationAwareBlobId implements BlobId,
GenerationAware {
@Override
public GenerationAwareBlobId parse(String id) {
+ // Recovery sidecar keys (eg. recovery/1_2_blobId) do not follow
the family_generation_blobId
+ // layout: keep them as a plain, non-generation-aware blob id
preserving the original string.
+ if (id.startsWith(BlobStoreDAO.RECOVERY_BLOB_PREFIX)) {
+ return decorateWithoutGeneration(id);
+ }
int separatorIndex1 = id.indexOf('_');
if (separatorIndex1 == -1 || separatorIndex1 == id.length() - 1) {
return decorateWithoutGeneration(id);
diff --git
a/server/blob/blob-storage-strategy/src/test/java/org/apache/james/server/blob/deduplication/BloomFilterGCAlgorithmContract.java
b/server/blob/blob-storage-strategy/src/test/java/org/apache/james/server/blob/deduplication/BloomFilterGCAlgorithmContract.java
index 39397f1f14..46f398611f 100644
---
a/server/blob/blob-storage-strategy/src/test/java/org/apache/james/server/blob/deduplication/BloomFilterGCAlgorithmContract.java
+++
b/server/blob/blob-storage-strategy/src/test/java/org/apache/james/server/blob/deduplication/BloomFilterGCAlgorithmContract.java
@@ -47,6 +47,7 @@ import org.apache.james.task.Task;
import org.apache.james.utils.UpdatableTickingClock;
import org.awaitility.Awaitility;
import org.awaitility.core.ConditionFactory;
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.RepeatedTest;
import org.junit.jupiter.api.Test;
@@ -83,6 +84,11 @@ public interface BloomFilterGCAlgorithmContract {
CLOCK.setInstant(NOW.toInstant());
}
+ @AfterEach
+ default void tearDown() {
+ BloomFilterGCAlgorithm.RECOVERY_AWARE = false;
+ }
+
default BlobStore blobStore() {
return new DeDuplicationBlobStore(blobStoreDAO(), DEFAULT_BUCKET,
GENERATION_AWARE_BLOB_ID_FACTORY);
}
@@ -231,6 +237,83 @@ public interface BloomFilterGCAlgorithmContract {
});
}
+ @Test
+ default void
gcShouldPreserveRecoverySidecarOfReferencedBlobWhenRecoveryAware() {
+ // Without RECOVERY_AWARE the recovery sidecar has NO_FAMILY →
inActiveGeneration()=false
+ // → treated as orphan and deleted even though its parent blob is
alive. This tests the fix.
+ BloomFilterGCAlgorithm.RECOVERY_AWARE = true;
+ BlobStore blobStore = blobStore();
+ BlobId referencedId = Mono.from(blobStore.save(DEFAULT_BUCKET,
UUID.randomUUID().toString(),
BlobStore.StoragePolicy.HIGH_PERFORMANCE)).block();
+ BlobId recoveryBlobId =
GENERATION_AWARE_BLOB_ID_FACTORY.parse(BlobStoreDAO.RECOVERY_BLOB_PREFIX +
referencedId.asString());
+ Mono.from(blobStoreDAO().save(DEFAULT_BUCKET, recoveryBlobId,
BlobStoreDAO.BytesBlob.of("bodyBlobId".getBytes()))).block();
+
+
when(BLOB_REFERENCE_SOURCE.listReferencedBlobs()).thenReturn(Flux.just(referencedId));
+ CLOCK.setInstant(NOW.plusMonths(2).toInstant());
+
+ Context context = new Context(EXPECTED_BLOB_COUNT,
ASSOCIATED_PROBABILITY);
+ Mono.from(bloomFilterGCAlgorithm().gc(EXPECTED_BLOB_COUNT,
DELETION_WINDOW_SIZE, ASSOCIATED_PROBABILITY, DEFAULT_BUCKET, context)).block();
+
+ assertThat(blobStore.read(DEFAULT_BUCKET, referencedId)).isNotNull();
+ assertThat(Mono.from(blobStoreDAO().readBytes(DEFAULT_BUCKET,
recoveryBlobId)).block()).isNotNull();
+ }
+
+ @Test
+ default void
gcShouldDeleteRecoverySidecarOfReferencedBlobWhenNotRecoveryAware() {
+ // Documents the unsafe behavior when the flag is off: recovery
sidecar of a live blob is GC-ed.
+ BlobStore blobStore = blobStore();
+ BlobId referencedId = Mono.from(blobStore.save(DEFAULT_BUCKET,
UUID.randomUUID().toString(),
BlobStore.StoragePolicy.HIGH_PERFORMANCE)).block();
+ BlobId recoveryBlobId =
GENERATION_AWARE_BLOB_ID_FACTORY.parse(BlobStoreDAO.RECOVERY_BLOB_PREFIX +
referencedId.asString());
+ Mono.from(blobStoreDAO().save(DEFAULT_BUCKET, recoveryBlobId,
BlobStoreDAO.BytesBlob.of("bodyBlobId".getBytes()))).block();
+
+
when(BLOB_REFERENCE_SOURCE.listReferencedBlobs()).thenReturn(Flux.just(referencedId));
+ CLOCK.setInstant(NOW.plusMonths(2).toInstant());
+
+ Context context = new Context(EXPECTED_BLOB_COUNT,
ASSOCIATED_PROBABILITY);
+ Mono.from(bloomFilterGCAlgorithm().gc(EXPECTED_BLOB_COUNT,
DELETION_WINDOW_SIZE, ASSOCIATED_PROBABILITY, DEFAULT_BUCKET, context)).block();
+
+ assertThat(blobStore.read(DEFAULT_BUCKET, referencedId)).isNotNull();
+ assertThatThrownBy(() ->
Mono.from(blobStoreDAO().readBytes(DEFAULT_BUCKET, recoveryBlobId)).block())
+ .isInstanceOf(ObjectNotFoundException.class);
+ }
+
+ @Test
+ default void
gcShouldDeleteRecoverySidecarAlongsideOrphanBlobWhenRecoveryAware() {
+ BloomFilterGCAlgorithm.RECOVERY_AWARE = true;
+ BlobStore blobStore = blobStore();
+ BlobId orphanId = Mono.from(blobStore.save(DEFAULT_BUCKET,
UUID.randomUUID().toString(),
BlobStore.StoragePolicy.HIGH_PERFORMANCE)).block();
+ BlobId recoveryBlobId =
GENERATION_AWARE_BLOB_ID_FACTORY.parse(BlobStoreDAO.RECOVERY_BLOB_PREFIX +
orphanId.asString());
+ Mono.from(blobStoreDAO().save(DEFAULT_BUCKET, recoveryBlobId,
BlobStoreDAO.BytesBlob.of("bodyBlobId".getBytes()))).block();
+
+
when(BLOB_REFERENCE_SOURCE.listReferencedBlobs()).thenReturn(Flux.empty());
+ CLOCK.setInstant(NOW.plusMonths(2).toInstant());
+
+ Context context = new Context(EXPECTED_BLOB_COUNT,
ASSOCIATED_PROBABILITY);
+ Mono.from(bloomFilterGCAlgorithm().gc(EXPECTED_BLOB_COUNT,
DELETION_WINDOW_SIZE, ASSOCIATED_PROBABILITY, DEFAULT_BUCKET, context)).block();
+
+ assertThatThrownBy(() -> blobStore.read(DEFAULT_BUCKET, orphanId))
+ .isInstanceOf(ObjectNotFoundException.class);
+ assertThatThrownBy(() ->
Mono.from(blobStoreDAO().readBytes(DEFAULT_BUCKET, recoveryBlobId)).block())
+ .isInstanceOf(ObjectNotFoundException.class);
+ }
+
+ @Test
+ default void gcShouldNotCountRecoveryBlobsInStatsWhenRecoveryAware() {
+ BloomFilterGCAlgorithm.RECOVERY_AWARE = true;
+ BlobStore blobStore = blobStore();
+ BlobId orphanId = Mono.from(blobStore.save(DEFAULT_BUCKET,
UUID.randomUUID().toString(),
BlobStore.StoragePolicy.HIGH_PERFORMANCE)).block();
+ BlobId recoveryBlobId =
GENERATION_AWARE_BLOB_ID_FACTORY.parse(BlobStoreDAO.RECOVERY_BLOB_PREFIX +
orphanId.asString());
+ Mono.from(blobStoreDAO().save(DEFAULT_BUCKET, recoveryBlobId,
BlobStoreDAO.BytesBlob.of("bodyBlobId".getBytes()))).block();
+
+
when(BLOB_REFERENCE_SOURCE.listReferencedBlobs()).thenReturn(Flux.empty());
+ CLOCK.setInstant(NOW.plusMonths(2).toInstant());
+
+ Context context = new Context(EXPECTED_BLOB_COUNT,
ASSOCIATED_PROBABILITY);
+ Mono.from(bloomFilterGCAlgorithm().gc(EXPECTED_BLOB_COUNT,
DELETION_WINDOW_SIZE, ASSOCIATED_PROBABILITY, DEFAULT_BUCKET, context)).block();
+
+ assertThat(context.snapshot().getBlobCount()).isEqualTo(1);
+ assertThat(context.snapshot().getGcedBlobCount()).isEqualTo(1);
+ }
+
@Test
default void gcShouldHandlerErrorWhenException() {
when(BLOB_REFERENCE_SOURCE.listReferencedBlobs()).thenReturn(Flux.empty());
diff --git
a/server/protocols/protocols-pop3-distributed/src/test/java/org/apache/james/pop3server/mailbox/task/MetaDataFixInconsistenciesServiceTest.java
b/server/protocols/protocols-pop3-distributed/src/test/java/org/apache/james/pop3server/mailbox/task/MetaDataFixInconsistenciesServiceTest.java
index 2269094877..88edbd757d 100644
---
a/server/protocols/protocols-pop3-distributed/src/test/java/org/apache/james/pop3server/mailbox/task/MetaDataFixInconsistenciesServiceTest.java
+++
b/server/protocols/protocols-pop3-distributed/src/test/java/org/apache/james/pop3server/mailbox/task/MetaDataFixInconsistenciesServiceTest.java
@@ -31,6 +31,7 @@ import
org.apache.james.backends.cassandra.CassandraClusterExtension;
import org.apache.james.backends.cassandra.components.CassandraDataDefinition;
import
org.apache.james.backends.cassandra.init.configuration.CassandraConfiguration;
import
org.apache.james.backends.cassandra.versions.CassandraSchemaVersionDataDefinition;
+import org.apache.james.blob.api.BlobStoreDAO;
import org.apache.james.blob.api.PlainBlobId;
import org.apache.james.blob.cassandra.CassandraBlobDataDefinition;
import org.apache.james.blob.cassandra.CassandraBlobStoreFactory;
@@ -60,6 +61,7 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Nested;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
+import org.mockito.Mockito;
import com.google.common.collect.ImmutableList;
@@ -166,6 +168,7 @@ public class MetaDataFixInconsistenciesServiceTest {
cassandra.getTypesProvider(),
CassandraBlobStoreFactory.forTesting(cassandra.getConf(), new
RecordingMetricFactory())
.passthrough(),
+ Mockito.mock(BlobStoreDAO.class),
new PlainBlobId.Factory(),
CassandraConfiguration.DEFAULT_CONFIGURATION);
testee = new MetaDataFixInconsistenciesService(imapUidDAO,
pop3MetadataStore, cassandraMessageDAOV3);
diff --git a/server/protocols/webadmin/webadmin-pop3/pom.xml
b/server/protocols/webadmin/webadmin-pop3/pom.xml
index c406d7a901..05f4b7a673 100644
--- a/server/protocols/webadmin/webadmin-pop3/pom.xml
+++ b/server/protocols/webadmin/webadmin-pop3/pom.xml
@@ -82,6 +82,11 @@
<artifactId>rest-assured</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-core</artifactId>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>testcontainers</artifactId>
diff --git
a/server/protocols/webadmin/webadmin-pop3/src/test/java/org/apache/james/pop3/webadmin/Pop3MetaDataFixInconsistenciesRoutesTest.java
b/server/protocols/webadmin/webadmin-pop3/src/test/java/org/apache/james/pop3/webadmin/Pop3MetaDataFixInconsistenciesRoutesTest.java
index cfae5f9bdc..e0e31212b6 100644
---
a/server/protocols/webadmin/webadmin-pop3/src/test/java/org/apache/james/pop3/webadmin/Pop3MetaDataFixInconsistenciesRoutesTest.java
+++
b/server/protocols/webadmin/webadmin-pop3/src/test/java/org/apache/james/pop3/webadmin/Pop3MetaDataFixInconsistenciesRoutesTest.java
@@ -35,6 +35,7 @@ import
org.apache.james.backends.cassandra.CassandraClusterExtension;
import org.apache.james.backends.cassandra.components.CassandraDataDefinition;
import
org.apache.james.backends.cassandra.init.configuration.CassandraConfiguration;
import
org.apache.james.backends.cassandra.versions.CassandraSchemaVersionDataDefinition;
+import org.apache.james.blob.api.BlobStoreDAO;
import org.apache.james.blob.api.PlainBlobId;
import org.apache.james.blob.cassandra.CassandraBlobDataDefinition;
import org.apache.james.blob.cassandra.CassandraBlobStoreFactory;
@@ -74,6 +75,7 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
+import org.mockito.Mockito;
import com.google.common.collect.ImmutableList;
@@ -184,6 +186,7 @@ class Pop3MetaDataFixInconsistenciesRoutesTest {
cassandra.getTypesProvider(),
CassandraBlobStoreFactory.forTesting(cassandra.getConf(), new
RecordingMetricFactory())
.passthrough(),
+ Mockito.mock(BlobStoreDAO.class),
new PlainBlobId.Factory(),
CassandraConfiguration.DEFAULT_CONFIGURATION);
MetaDataFixInconsistenciesService fixInconsistenciesService = new
MetaDataFixInconsistenciesService(imapUidDAO, pop3MetadataStore,
cassandraMessageDAOV3);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
