This is an automated email from the ASF dual-hosted git repository. chibenwa pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/james-project.git
commit bfb69c8f7309ccdc3b10faef8ad94efcd8b80a9f Author: Benoit TELLIER <[email protected]> AuthorDate: Sat Sep 5 01:10:05 2026 +0200 JAMES-4225 Cassandra migration 16: backfill messagev3 property in frontal message tables This allows for a future removal of those very collumns thus making the Cassandra schema slightly lighter --- .../versions/CassandraSchemaVersionManager.java | 2 +- .../cassandra/mail/CassandraMessageIdDAO.java | 24 ++ .../mail/CassandraMessageIdToImapUidDAO.java | 49 +++- .../migration/MessageDenormalizationMigration.java | 130 +++++++++++ .../mail/MessageDenormalizationMigrationTest.java | 247 +++++++++++++++++++++ .../modules/webadmin/CassandraRoutesModule.java | 3 + upgrade-instructions.md | 20 ++ 7 files changed, 469 insertions(+), 6 deletions(-) diff --git a/backends-common/cassandra/src/main/java/org/apache/james/backends/cassandra/versions/CassandraSchemaVersionManager.java b/backends-common/cassandra/src/main/java/org/apache/james/backends/cassandra/versions/CassandraSchemaVersionManager.java index 59d9ff75ae..ee9e1ea7f8 100644 --- a/backends-common/cassandra/src/main/java/org/apache/james/backends/cassandra/versions/CassandraSchemaVersionManager.java +++ b/backends-common/cassandra/src/main/java/org/apache/james/backends/cassandra/versions/CassandraSchemaVersionManager.java @@ -36,7 +36,7 @@ import reactor.core.publisher.Mono; public class CassandraSchemaVersionManager { public static final SchemaVersion MIN_VERSION = new SchemaVersion(12); - public static final SchemaVersion MAX_VERSION = new SchemaVersion(15); + public static final SchemaVersion MAX_VERSION = new SchemaVersion(16); public static final SchemaVersion DEFAULT_VERSION = MIN_VERSION; private static final Logger LOGGER = LoggerFactory.getLogger(CassandraSchemaVersionManager.class); diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdDAO.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdDAO.java index 68c854b96c..3b7dc735f4 100644 --- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdDAO.java +++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdDAO.java @@ -117,6 +117,7 @@ public class CassandraMessageIdDAO { private final BlobId.Factory blobIdFactory; private final PreparedStatement delete; private final PreparedStatement insert; + private final PreparedStatement updateDenormalizedFields; private final PreparedStatement select; private final PreparedStatement selectAll; private final PreparedStatement selectAllUids; @@ -144,6 +145,7 @@ public class CassandraMessageIdDAO { this.delete = prepareDelete(session); this.insert = prepareInsert(session); this.update = prepareUpdate(session); + this.updateDenormalizedFields = prepareUpdateDenormalizedFields(session); this.select = prepareSelect(session); this.selectAll = prepareSelectAll(session); this.selectAllUids = prepareSelectAllUids(session); @@ -191,6 +193,17 @@ public class CassandraMessageIdDAO { .build()); } + private PreparedStatement prepareUpdateDenormalizedFields(CqlSession session) { + return session.prepare(update(TABLE_NAME) + .set(setColumn(INTERNAL_DATE, bindMarker(INTERNAL_DATE)), + setColumn(BODY_START_OCTET, bindMarker(BODY_START_OCTET)), + setColumn(FULL_CONTENT_OCTETS, bindMarker(FULL_CONTENT_OCTETS)), + setColumn(HEADER_CONTENT, bindMarker(HEADER_CONTENT))) + .where(column(MAILBOX_ID).isEqualTo(bindMarker(MAILBOX_ID)), + column(IMAP_UID).isEqualTo(bindMarker(IMAP_UID))) + .build()); + } + private PreparedStatement prepareUpdate(CqlSession session) { return session.prepare(update(TABLE_NAME) .set(setColumn(MOD_SEQ, bindMarker(MOD_SEQ)), @@ -369,6 +382,17 @@ public class CassandraMessageIdDAO { .build()); } + public Mono<Void> updateDenormalizedFields(CassandraId mailboxId, MessageUid uid, Date internalDate, + int bodyStartOctet, long size, BlobId headerContent) { + return cassandraAsyncExecutor.executeVoid(updateDenormalizedFields.bind() + .setUuid(MAILBOX_ID, mailboxId.asUuid()) + .setLong(IMAP_UID, uid.asLong()) + .setInstant(INTERNAL_DATE, internalDate.toInstant()) + .setInt(BODY_START_OCTET, bodyStartOctet) + .setLong(FULL_CONTENT_OCTETS, size) + .setString(HEADER_CONTENT, headerContent.asString())); + } + public Mono<Void> updateMetadata(ComposedMessageId composedMessageId, UpdatedFlags updatedFlags) { return cassandraAsyncExecutor.executeVoid(updateBoundStatement(composedMessageId, updatedFlags)); } diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdToImapUidDAO.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdToImapUidDAO.java index 6f793a0cef..bfb7b7339e 100644 --- a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdToImapUidDAO.java +++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/CassandraMessageIdToImapUidDAO.java @@ -45,6 +45,7 @@ import static org.apache.james.mailbox.cassandra.table.Flag.USER_FLAGS; import static org.apache.james.mailbox.cassandra.table.MessageIdToImapUid.MOD_SEQ; import static org.apache.james.mailbox.cassandra.table.MessageIdToImapUid.TABLE_NAME; import static org.apache.james.mailbox.cassandra.table.MessageIdToImapUid.THREAD_ID; +import static org.apache.james.util.ReactorUtils.publishIfPresent; import java.time.Duration; import java.util.Date; @@ -87,6 +88,7 @@ import com.google.common.collect.Sets; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; +import reactor.core.scheduler.Schedulers; public class CassandraMessageIdToImapUidDAO { private static final String MOD_SEQ_CONDITION = "modSeqCondition"; @@ -98,6 +100,7 @@ public class CassandraMessageIdToImapUidDAO { private final PreparedStatement delete; private final PreparedStatement insert; private final PreparedStatement update; + private final PreparedStatement updateDenormalizedFields; private final PreparedStatement selectAll; private final PreparedStatement select; private final PreparedStatement listStatement; @@ -116,6 +119,7 @@ public class CassandraMessageIdToImapUidDAO { this.cassandraConfiguration = cassandraConfiguration; this.delete = prepareDelete(); this.insert = prepareInsert(); + this.updateDenormalizedFields = prepareUpdateDenormalizedFields(); this.update = prepareUpdate(); this.selectAll = prepareSelectAll(); this.select = prepareSelect(); @@ -178,6 +182,18 @@ public class CassandraMessageIdToImapUidDAO { } } + private PreparedStatement prepareUpdateDenormalizedFields() { + return session.prepare(QueryBuilder.update(TABLE_NAME) + .set(setColumn(INTERNAL_DATE, bindMarker(INTERNAL_DATE)), + setColumn(BODY_START_OCTET, bindMarker(BODY_START_OCTET)), + setColumn(FULL_CONTENT_OCTETS, bindMarker(FULL_CONTENT_OCTETS)), + setColumn(HEADER_CONTENT, bindMarker(HEADER_CONTENT))) + .where(column(MESSAGE_ID).isEqualTo(bindMarker(MESSAGE_ID)), + column(MAILBOX_ID).isEqualTo(bindMarker(MAILBOX_ID)), + column(IMAP_UID).isEqualTo(bindMarker(IMAP_UID))) + .build()); + } + private PreparedStatement prepareUpdate() { Update update = QueryBuilder.update(TABLE_NAME) .set(setColumn(MOD_SEQ, bindMarker(MOD_SEQ)), @@ -261,6 +277,18 @@ public class CassandraMessageIdToImapUidDAO { .build()); } + public Mono<Void> updateDenormalizedFields(CassandraMessageId messageId, CassandraId mailboxId, MessageUid uid, + Date internalDate, int bodyStartOctet, long size, BlobId headerContent) { + return cassandraAsyncExecutor.executeVoid(updateDenormalizedFields.bind() + .setUuid(MESSAGE_ID, messageId.get()) + .setUuid(MAILBOX_ID, mailboxId.asUuid()) + .setLong(IMAP_UID, uid.asLong()) + .setInstant(INTERNAL_DATE, internalDate.toInstant()) + .setInt(BODY_START_OCTET, bodyStartOctet) + .setLong(FULL_CONTENT_OCTETS, size) + .setString(HEADER_CONTENT, headerContent.asString())); + } + public Mono<Boolean> updateMetadata(ComposedMessageId id, UpdatedFlags updatedFlags, ModSeq previousModeq) { if (cassandraConfiguration.isMessageWriteStrongConsistency()) { return cassandraAsyncExecutor.executeReturnApplied(updateBoundStatement(id, updatedFlags, previousModeq)); @@ -335,7 +363,8 @@ public class CassandraMessageIdToImapUidDAO { public Flux<CassandraMessageMetadata> retrieve(CassandraMessageId messageId, Optional<CassandraId> mailboxId, JamesExecutionProfiles.ConsistencyChoice readConsistencyChoice) { return cassandraAsyncExecutor.executeRows(setExecutionProfileIfNeeded(selectStatement(messageId, mailboxId), readConsistencyChoice)) - .map(this::toComposedMessageIdWithMetadata); + .map(this::toComposedMessageIdWithMetadata) + .handle(publishIfPresent()); } @VisibleForTesting @@ -346,12 +375,22 @@ public class CassandraMessageIdToImapUidDAO { public Flux<CassandraMessageMetadata> retrieveAllMessages() { return cassandraAsyncExecutor.executeRows(listStatement.bind() .setTimeout(Duration.ofDays(1))) - .map(this::toComposedMessageIdWithMetadata); + .map(this::toComposedMessageIdWithMetadata) + .handle(publishIfPresent()); } - private CassandraMessageMetadata toComposedMessageIdWithMetadata(Row row) { + private Optional<CassandraMessageMetadata> toComposedMessageIdWithMetadata(Row row) { final CassandraMessageId messageId = CassandraMessageId.Factory.of(row.getUuid(MESSAGE_ID)); - return CassandraMessageMetadata.builder() + if (row.get(MOD_SEQ, Long.class) == null) { + // Out of order updates with concurrent deletes can result in the row being partially deleted + // We filter out such records, and cleanup them. + // TODO Test INTERNAL_DATE instead once schema version 16 is enforced: unlike MOD_SEQ it also catches rows resurrected by a flag update. + delete(messageId, CassandraId.of(row.getUuid(MAILBOX_ID))) + .subscribeOn(Schedulers.parallel()) + .subscribe(); + return Optional.empty(); + } + return Optional.of(CassandraMessageMetadata.builder() .ids(ComposedMessageIdWithMetaData.builder() .composedMessageId(new ComposedMessageId( CassandraId.of(row.getUuid(MAILBOX_ID)), @@ -369,7 +408,7 @@ public class CassandraMessageIdToImapUidDAO { .size(row.get(FULL_CONTENT_OCTETS, Long.class)) .headerContent(Optional.ofNullable(row.getString(HEADER_CONTENT)) .map(blobIdFactory::parse)) - .build(); + .build()); } private ThreadId getThreadIdFromRow(Row row, MessageId messageId) { diff --git a/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/migration/MessageDenormalizationMigration.java b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/migration/MessageDenormalizationMigration.java new file mode 100644 index 0000000000..26d9c5d064 --- /dev/null +++ b/mailbox/cassandra/src/main/java/org/apache/james/mailbox/cassandra/mail/migration/MessageDenormalizationMigration.java @@ -0,0 +1,130 @@ +/**************************************************************** + * Licensed to the Apache Software Foundation (ASF) under one * + * or more contributor license agreements. See the NOTICE file * + * distributed with this work for additional information * + * regarding copyright ownership. The ASF licenses this file * + * to you under the Apache License, Version 2.0 (the * + * "License"); you may not use this file except in compliance * + * with the License. You may obtain a copy of the License at * + * * + * http://www.apache.org/licenses/LICENSE-2.0 * + * * + * Unless required by applicable law or agreed to in writing, * + * software distributed under the License is distributed on an * + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY * + * KIND, either express or implied. See the License for the * + * specific language governing permissions and limitations * + * under the License. * + ****************************************************************/ + +package org.apache.james.mailbox.cassandra.mail.migration; + +import java.util.function.Function; + +import jakarta.inject.Inject; + +import org.apache.james.backends.cassandra.migration.Migration; +import org.apache.james.mailbox.cassandra.ids.CassandraId; +import org.apache.james.mailbox.cassandra.ids.CassandraMessageId; +import org.apache.james.mailbox.cassandra.mail.CassandraMessageDAOV3; +import org.apache.james.mailbox.cassandra.mail.CassandraMessageIdDAO; +import org.apache.james.mailbox.cassandra.mail.CassandraMessageIdToImapUidDAO; +import org.apache.james.mailbox.cassandra.mail.CassandraMessageMetadata; +import org.apache.james.mailbox.cassandra.mail.MessageRepresentation; +import org.apache.james.mailbox.model.ComposedMessageId; +import org.apache.james.mailbox.store.mail.MessageMapper.FetchType; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +/** + * Backfills the fields `messageIdTable` and `imapUidTable` denormalize from `messagev3`. + * + * <p>Those columns were introduced by JAMES-3576 in 3.7.0 and never backfilled: JAMES-3815 chose instead + * to tolerate their absence, which is what {@link CassandraMessageMetadata#isComplete()} tests, falling + * back to `messagev3` when they are missing. Messages written by James 3.6 and earlier therefore still + * carry null there.</p> + * + * <p>Once every row is complete, that fallback becomes dead: metadata and header fetches are answered + * from a single read, and `messagev3` no longer needs to carry the denormalized copies at all.</p> + * + * See JAMES-4225 + */ +public class MessageDenormalizationMigration implements Migration { + private static final Logger LOGGER = LoggerFactory.getLogger(MessageDenormalizationMigration.class); + private static final int CONCURRENCY = 8; + + private final CassandraMessageIdDAO messageIdDAO; + private final CassandraMessageIdToImapUidDAO imapUidDAO; + private final CassandraMessageDAOV3 messageDAO; + + @Inject + public MessageDenormalizationMigration(CassandraMessageIdDAO messageIdDAO, + CassandraMessageIdToImapUidDAO imapUidDAO, + CassandraMessageDAOV3 messageDAO) { + this.messageIdDAO = messageIdDAO; + this.imapUidDAO = imapUidDAO; + this.messageDAO = messageDAO; + } + + @Override + public void apply() { + backfill() + .then(cleanUpPartialRows()) + .block(); + } + + private Mono<Void> backfill() { + return Flux.concat( + backfill(imapUidDAO.retrieveAllMessages(), this::backfillImapUid), + backfill(messageIdDAO.retrieveAllMessages(), this::backfillMessageId)) + .then(); + } + + private Mono<Void> cleanUpPartialRows() { + return Flux.concat(imapUidDAO.retrieveAllMessages(), messageIdDAO.retrieveAllMessages()) + .then(); + } + + private Flux<Void> backfill(Flux<CassandraMessageMetadata> rows, + Function<CassandraMessageMetadata, Mono<Void>> backfill) { + return rows.filter(metadata -> !metadata.isComplete()) + .flatMap(backfill, CONCURRENCY); + } + + private Mono<Void> backfillMessageId(CassandraMessageMetadata metadata) { + ComposedMessageId id = metadata.getComposedMessageId().getComposedMessageId(); + + return representation(metadata) + .flatMap(representation -> messageIdDAO.updateDenormalizedFields( + (CassandraId) id.getMailboxId(), + id.getUid(), + representation.getInternalDate(), + representation.getBodyStartOctet(), + representation.getSize(), + representation.getHeaderId())); + } + + private Mono<Void> backfillImapUid(CassandraMessageMetadata metadata) { + ComposedMessageId id = metadata.getComposedMessageId().getComposedMessageId(); + + return representation(metadata) + .flatMap(representation -> imapUidDAO.updateDenormalizedFields( + (CassandraMessageId) id.getMessageId(), + (CassandraId) id.getMailboxId(), + id.getUid(), + representation.getInternalDate(), + representation.getBodyStartOctet(), + representation.getSize(), + representation.getHeaderId())); + } + + private Mono<MessageRepresentation> representation(CassandraMessageMetadata metadata) { + return messageDAO.retrieveMessage(metadata.getComposedMessageId(), FetchType.METADATA) + .doOnError(e -> LOGGER.error("Failed to read messagev3 for {}", + metadata.getComposedMessageId().getComposedMessageId(), e)) + .onErrorResume(e -> Mono.empty()); + } +} diff --git a/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/MessageDenormalizationMigrationTest.java b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/MessageDenormalizationMigrationTest.java new file mode 100644 index 0000000000..1aa100736d --- /dev/null +++ b/mailbox/cassandra/src/test/java/org/apache/james/mailbox/cassandra/mail/MessageDenormalizationMigrationTest.java @@ -0,0 +1,247 @@ +/**************************************************************** + * Licensed to the Apache Software Foundation (ASF) under one * + * or more contributor license agreements. See the NOTICE file * + * distributed with this work for additional information * + * regarding copyright ownership. The ASF licenses this file * + * to you under the Apache License, Version 2.0 (the * + * "License"); you may not use this file except in compliance * + * with the License. You may obtain a copy of the License at * + * * + * http://www.apache.org/licenses/LICENSE-2.0 * + * * + * Unless required by applicable law or agreed to in writing, * + * software distributed under the License is distributed on an * + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY * + * KIND, either express or implied. See the License for the * + * specific language governing permissions and limitations * + * under the License. * + ****************************************************************/ + +package org.apache.james.mailbox.cassandra.mail; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.nio.charset.StandardCharsets; +import java.util.Date; +import java.util.Optional; + +import jakarta.mail.Flags; + +import org.apache.james.backends.cassandra.CassandraCluster; +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.BlobStore; +import org.apache.james.blob.api.BlobStoreCacheCallback; +import org.apache.james.blob.api.BucketName; +import org.apache.james.blob.api.PlainBlobId; +import org.apache.james.blob.cassandra.CassandraBlobDataDefinition; +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; +import org.apache.james.mailbox.cassandra.ids.CassandraMessageId; +import org.apache.james.mailbox.cassandra.mail.migration.MessageDenormalizationMigration; +import org.apache.james.mailbox.cassandra.modules.CassandraMessageDataDefinition; +import org.apache.james.mailbox.model.ByteContent; +import org.apache.james.mailbox.model.ComposedMessageId; +import org.apache.james.mailbox.model.ComposedMessageIdWithMetaData; +import org.apache.james.mailbox.model.ThreadId; +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.awaitility.Awaitility; +import org.awaitility.Durations; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; + +import com.google.common.collect.ImmutableList; + +class MessageDenormalizationMigrationTest { + private static final CassandraId MAILBOX_ID = CassandraId.timeBased(); + private static final MessageUid MESSAGE_UID = MessageUid.of(1); + private static final String CONTENT = "Subject: test\n\nBody\n"; + private static final int BODY_START = 15; + + public static final CassandraDataDefinition MODULES = CassandraDataDefinition.aggregateModules( + CassandraMessageDataDefinition.MODULE, + CassandraBlobDataDefinition.MODULE, + CassandraSchemaVersionDataDefinition.MODULE); + + @RegisterExtension + static CassandraClusterExtension cassandraCluster = new CassandraClusterExtension(MODULES); + + private CassandraMessageIdDAO messageIdDAO; + private CassandraMessageIdToImapUidDAO imapUidDAO; + private CassandraMessageDAOV3 messageDAO; + private MessageDenormalizationMigration testee; + + private CassandraMessageId messageId; + private ComposedMessageIdWithMetaData ids; + + @BeforeEach + void setUp(CassandraCluster cassandra) { + PlainBlobId.Factory blobIdFactory = new PlainBlobId.Factory(); + CassandraBlobStoreDAO blobStoreDAO = new CassandraBlobStoreDAO( + new CassandraDefaultBucketDAO(cassandra.getConf(), blobIdFactory), + new CassandraBucketDAO(blobIdFactory, cassandra.getConf()), + CassandraConfiguration.DEFAULT_CONFIGURATION, BucketName.DEFAULT, new RecordingMetricFactory()); + BlobStore blobStore = BlobStoreFactory.builder() + .blobStoreDAO(blobStoreDAO) + .blobIdFactory(blobIdFactory) + .defaultBucketName() + .passthrough(); + + messageIdDAO = new CassandraMessageIdDAO(cassandra.getConf(), blobIdFactory); + imapUidDAO = new CassandraMessageIdToImapUidDAO(cassandra.getConf(), blobIdFactory, + CassandraConfiguration.DEFAULT_CONFIGURATION); + messageDAO = new CassandraMessageDAOV3(cassandra.getConf(), cassandra.getTypesProvider(), blobStore, + blobStoreDAO, blobIdFactory, CassandraConfiguration.DEFAULT_CONFIGURATION, BlobStoreCacheCallback.NOOP); + testee = new MessageDenormalizationMigration(messageIdDAO, imapUidDAO, messageDAO); + + messageId = new CassandraMessageId.Factory().generate(); + ids = ComposedMessageIdWithMetaData.builder() + .composedMessageId(new ComposedMessageId(MAILBOX_ID, messageId, MESSAGE_UID)) + .flags(new Flags()) + .modSeq(ModSeq.of(1)) + .threadId(ThreadId.fromBaseMessageId(messageId)) + .build(); + } + + @Test + void migrationShouldCompleteMessageIdRows() throws Exception { + saveMessage(); + givenIncompleteRows(); + + testee.apply(); + + assertThat(retrieveMessageId()).hasValueSatisfying(metadata -> + assertThat(metadata.isComplete()).isTrue()); + } + + @Test + void migrationShouldCompleteImapUidRows() throws Exception { + saveMessage(); + givenIncompleteRows(); + + testee.apply(); + + assertThat(retrieveImapUid()).hasValueSatisfying(metadata -> + assertThat(metadata.isComplete()).isTrue()); + } + + @Test + void migrationShouldCopyTheFieldsOfMessageV3() throws Exception { + SimpleMailboxMessage message = saveMessage(); + givenIncompleteRows(); + + testee.apply(); + + CassandraMessageMetadata metadata = retrieveMessageId().get(); + assertThat(metadata.getInternalDate()).contains(message.getInternalDate()); + assertThat(metadata.getBodyStartOctet()).contains((long) BODY_START); + assertThat(metadata.getSize()).contains((long) CONTENT.length()); + assertThat(metadata.getHeaderContent()).isNotEmpty(); + } + + @Test + void migrationShouldNotAlterCompleteRows() throws Exception { + saveMessage(); + givenCompleteRows(); + CassandraMessageMetadata before = retrieveMessageId().get(); + + testee.apply(); + + assertThat(retrieveMessageId()).contains(before); + } + + @Test + void migrationShouldIgnoreRowsWhoseMessageIsMissing() throws Exception { + givenIncompleteRows(); + + testee.apply(); + + assertThat(retrieveMessageId()).hasValueSatisfying(metadata -> + assertThat(metadata.isComplete()).isFalse()); + } + + /** + * Backfilling is an upsert: a message deleted between the read and the write comes back as a row + * carrying the denormalized columns alone. Writing one directly is the deterministic way to assert + * such a row never surfaces, the race itself not being reproducible. + */ + @Test + void resurrectedMessageIdRowsShouldNotSurface() { + messageIdDAO.updateDenormalizedFields(MAILBOX_ID, MESSAGE_UID, new Date(), BODY_START, + CONTENT.length(), new PlainBlobId.Factory().of("headerBlobId")).block(); + + assertThat(retrieveMessageId()).isEmpty(); + } + + @Test + void resurrectedImapUidRowsShouldNotSurface() { + imapUidDAO.updateDenormalizedFields(messageId, MAILBOX_ID, MESSAGE_UID, new Date(), BODY_START, + CONTENT.length(), new PlainBlobId.Factory().of("headerBlobId")).block(); + + assertThat(retrieveImapUid()).isEmpty(); + } + + @Test + void resurrectedRowsShouldBeCleanedUp() { + messageIdDAO.updateDenormalizedFields(MAILBOX_ID, MESSAGE_UID, new Date(), BODY_START, + CONTENT.length(), new PlainBlobId.Factory().of("headerBlobId")).block(); + + retrieveMessageId(); + + Awaitility.await().atMost(Durations.TEN_SECONDS) + .untilAsserted(() -> assertThat(messageIdDAO.retrieveAllMessages().collectList().block()).isEmpty()); + } + + private SimpleMailboxMessage saveMessage() { + SimpleMailboxMessage message = SimpleMailboxMessage.builder() + .messageId(messageId) + .threadId(ThreadId.fromBaseMessageId(messageId)) + .mailboxId(MAILBOX_ID) + .uid(MESSAGE_UID) + .internalDate(new Date()) + .bodyStartOctet(BODY_START) + .size(CONTENT.length()) + .content(new ByteContent(CONTENT.getBytes(StandardCharsets.UTF_8))) + .flags(new Flags()) + .addAttachments(ImmutableList.of()) + .build(); + messageDAO.save(message).block(); + return message; + } + + private void givenIncompleteRows() { + CassandraMessageMetadata metadata = CassandraMessageMetadata.builder().ids(ids).build(); + messageIdDAO.insertNullInternalDateAndHeaderContent(metadata).block(); + imapUidDAO.insertNullInternalDateAndHeaderContent(metadata).block(); + } + + private void givenCompleteRows() { + CassandraMessageMetadata metadata = CassandraMessageMetadata.builder() + .ids(ids) + .internalDate(new Date()) + .bodyStartOctet((long) BODY_START) + .size((long) CONTENT.length()) + .headerContent(Optional.of(new PlainBlobId.Factory().of("headerBlobId"))) + .build(); + messageIdDAO.insert(metadata).block(); + imapUidDAO.insert(metadata).block(); + } + + private Optional<CassandraMessageMetadata> retrieveMessageId() { + return messageIdDAO.retrieve(MAILBOX_ID, MESSAGE_UID).block(); + } + + private Optional<CassandraMessageMetadata> retrieveImapUid() { + return imapUidDAO.retrieve(messageId, Optional.of(MAILBOX_ID)).collectList().block() + .stream().findFirst(); + } +} diff --git a/server/container/guice/cassandra/src/main/java/org/apache/james/modules/webadmin/CassandraRoutesModule.java b/server/container/guice/cassandra/src/main/java/org/apache/james/modules/webadmin/CassandraRoutesModule.java index b4c17d7cac..07d5e79cfd 100644 --- a/server/container/guice/cassandra/src/main/java/org/apache/james/modules/webadmin/CassandraRoutesModule.java +++ b/server/container/guice/cassandra/src/main/java/org/apache/james/modules/webadmin/CassandraRoutesModule.java @@ -25,6 +25,7 @@ import org.apache.james.backends.cassandra.migration.MigrationTask; import org.apache.james.backends.cassandra.versions.CassandraSchemaVersionManager; import org.apache.james.backends.cassandra.versions.SchemaTransition; import org.apache.james.backends.cassandra.versions.SchemaVersion; +import org.apache.james.mailbox.cassandra.mail.migration.MessageDenormalizationMigration; import org.apache.james.mailbox.cassandra.quota.migration.CassandraCurrentQuotaManagerMigration; import org.apache.james.mailbox.cassandra.quota.migration.CassandraPerUserMaxQuotaManagerMigration; import org.apache.james.sieve.cassandra.migration.SieveQuotaMigration; @@ -42,6 +43,7 @@ public class CassandraRoutesModule extends AbstractModule { private static final SchemaTransition FROM_V12_TO_V13 = SchemaTransition.to(new SchemaVersion(13)); private static final SchemaTransition FROM_V13_TO_V14 = SchemaTransition.to(new SchemaVersion(14)); private static final SchemaTransition FROM_V14_TO_V15 = SchemaTransition.to(new SchemaVersion(15)); + private static final SchemaTransition FROM_V15_TO_V16 = SchemaTransition.to(new SchemaVersion(16)); @Override protected void configure() { @@ -60,6 +62,7 @@ public class CassandraRoutesModule extends AbstractModule { allMigrationClazzBinder.addBinding(FROM_V12_TO_V13).to(CassandraCurrentQuotaManagerMigration.class); allMigrationClazzBinder.addBinding(FROM_V13_TO_V14).to(CassandraPerUserMaxQuotaManagerMigration.class); allMigrationClazzBinder.addBinding(FROM_V14_TO_V15).to(SieveQuotaMigration.class); + allMigrationClazzBinder.addBinding(FROM_V15_TO_V16).to(MessageDenormalizationMigration.class); bind(SchemaVersion.class) .annotatedWith(Names.named(CassandraMigrationService.LATEST_VERSION)) diff --git a/upgrade-instructions.md b/upgrade-instructions.md index b70cc0d643..8f63beb31a 100644 --- a/upgrade-instructions.md +++ b/upgrade-instructions.md @@ -24,6 +24,26 @@ Change list: - [JAMES-4210 ManageSieve SASL adoption](#james-4210-managesieve-sasl-adoption) - [JAMES-4225 Blob ids default to 128 bits of entropy](#james-4225-blob-ids-default-to-128-bits-of-entropy) - [Dropping the bodyOctets column of the Cassandra messagev3 table](#dropping-the-bodyoctets-column-of-the-cassandra-messagev3-table) + - [Cassandra schema version 16: mandatory message denormalization migration](#cassandra-schema-version-16-mandatory-message-denormalization-migration) + +### Cassandra schema version 16: mandatory message denormalization migration + +Date: 05/09/2026 + +Concerned products: James products using Cassandra as mailbox storage + +`messageIdTable` and `imapUidTable` denormalize four fields from `messagev3`: `internalDate`, +`bodyStartOctet`, `fullContentOctets` and `headerContent`. They were introduced in 3.7.0 and never +backfilled: James tolerates their absence instead, and falls back to reading `messagev3`. Messages +written by James 3.6 and earlier therefore still carry null there. + +Schema version 16 backfills them. Run it as any other migration, for instance: + +``` +curl -XPOST 'http://ip:port/cassandra/version/upgrade' -d '16' +``` + +Later releases will drop this backward support and drop soon-to-be-useless collumns. ### JAMES-4225 Blob ids default to 128 bits of entropy --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
