This is an automated email from the ASF dual-hosted git repository. Arsnael pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/james-project.git
commit 26d0b6d74c82ff1639b54a23b3c07333309fe500 Author: Benoit TELLIER <[email protected]> AuthorDate: Fri Jul 24 09:50:11 2026 +0200 JAMES-4209 Manage concurrency --- .../servers/pages/distributed/operate/backup.adoc | 31 +++++++++++++++++++++ server/apps/distributed-app/README.adoc | 21 ++++++++++++++ .../org/apache/james/RecoveryConfiguration.java | 32 ++++++++++++++++++++-- .../java/org/apache/james/S3RecoveryService.java | 7 ++--- .../apache/james/RecoveryConfigurationTest.java | 24 ++++++++++++++++ 5 files changed, 109 insertions(+), 6 deletions(-) diff --git a/docs/modules/servers/pages/distributed/operate/backup.adoc b/docs/modules/servers/pages/distributed/operate/backup.adoc index 961545a29e..92550ab174 100644 --- a/docs/modules/servers/pages/distributed/operate/backup.adoc +++ b/docs/modules/servers/pages/distributed/operate/backup.adoc @@ -175,6 +175,37 @@ generation instead of scanning the whole bucket: ... org.apache.james.S3RecoveryMain --header-blob-prefix=1_42_ ---- +The dominant cost of a recovery run is not the listing but the per-message work: three blob reads plus +a full re-store of each message through the mailbox. The `--concurrency=<n>` argument (default `8`, also +settable via the `RECOVERY_CONCURRENCY` environment variable or the `recovery.concurrency` system +property) controls how many messages are restored in parallel and is therefore the main lever on +wall-clock recovery time. Raise it to saturate the object store and Cassandra; watch their load and +back off if they become the bottleneck: + +[source,bash] +---- +... org.apache.james.S3RecoveryMain --concurrency=32 +---- + +=== Scaling recovery: shard by generation + +For very large recoveries, a single process is limited by its own concurrency and offers no easy resume +point. Because `--header-blob-prefix` scopes a run to a slice of the key space, you can *shard* the +recovery across several independent processes — typically one per blob generation +(`family_generation_`) — and run them in parallel, each with its own concurrency: + +[source,bash] +---- +# On different hosts / containers, in parallel +... org.apache.james.S3RecoveryMain --header-blob-prefix=1_40_ --concurrency=16 +... org.apache.james.S3RecoveryMain --header-blob-prefix=1_41_ --concurrency=16 +... org.apache.james.S3RecoveryMain --header-blob-prefix=1_42_ --concurrency=16 +---- + +The shards are disjoint (a given header blob belongs to exactly one generation), so they never restore +the same message twice, aggregate throughput scales with the number of shards, and a shard that fails +can be re-run on its own without redoing the others. + Notes: * Restored messages are re-stored (and get a fresh `recovery/` sidecar), so re-running the recovery diff --git a/server/apps/distributed-app/README.adoc b/server/apps/distributed-app/README.adoc index 4acd0061f8..3371787798 100644 --- a/server/apps/distributed-app/README.adoc +++ b/server/apps/distributed-app/README.adoc @@ -163,6 +163,27 @@ Because header blob ids are generation-aware (`family_generation_...`), a clever $ java ... org.apache.james.S3RecoveryMain --header-blob-prefix=1_42_ ---- +The `--concurrency=<n>` argument (default `8`, also settable via the `RECOVERY_CONCURRENCY` environment +variable or the `recovery.concurrency` system property) controls how many messages are restored in +parallel. The per-message work (blob reads plus a full re-store through the mailbox) dominates recovery +time, so this is the main lever on wall-clock duration: + +[source] +---- +$ java ... org.apache.james.S3RecoveryMain --concurrency=32 +---- + +For very large recoveries, shard the run by generation and run several processes in parallel, each on a +disjoint slice of the key space with its own concurrency. The shards never restore the same message +twice, throughput scales with their number, and a failed shard can be re-run on its own: + +[source] +---- +$ java ... org.apache.james.S3RecoveryMain --header-blob-prefix=1_40_ --concurrency=16 +$ java ... org.apache.james.S3RecoveryMain --header-blob-prefix=1_41_ --concurrency=16 +$ java ... org.apache.james.S3RecoveryMain --header-blob-prefix=1_42_ --concurrency=16 +---- + Notes: * Restored messages are re-stored (and get a fresh `recovery/` sidecar), so re-running the recovery diff --git a/server/apps/distributed-app/src/main/java/org/apache/james/RecoveryConfiguration.java b/server/apps/distributed-app/src/main/java/org/apache/james/RecoveryConfiguration.java index 860f314812..75ca00fb58 100644 --- a/server/apps/distributed-app/src/main/java/org/apache/james/RecoveryConfiguration.java +++ b/server/apps/distributed-app/src/main/java/org/apache/james/RecoveryConfiguration.java @@ -24,6 +24,8 @@ import java.time.format.DateTimeParseException; import java.util.Arrays; import java.util.Optional; +import com.google.common.base.Preconditions; + /** * Configuration for the S3 blob store recovery run. * @@ -39,19 +41,34 @@ import java.util.Optional; * recovery sidecars) and can be provided as a {@code --header-blob-prefix=<prefix>} program argument, * the {@code RECOVERY_HEADER_BLOB_PREFIX} environment variable, or the {@code recovery.header.blob.prefix} * system property.</p> + * + * <p>The {@code concurrency} controls how many messages are restored in parallel. Since the dominant + * cost is the per-message work (blob reads plus a full re-store through the mailbox), this is the main + * lever on recovery wall-clock time. It defaults to {@value #DEFAULT_CONCURRENCY} and can be provided + * as a {@code --concurrency=<n>} program argument, the {@code RECOVERY_CONCURRENCY} environment + * variable, or the {@code recovery.concurrency} system property.</p> */ -public record RecoveryConfiguration(Optional<Instant> restoreAfter, String headerBlobPrefix) { +public record RecoveryConfiguration(Optional<Instant> restoreAfter, String headerBlobPrefix, int concurrency) { + public static final int DEFAULT_CONCURRENCY = 8; private static final String RESTORE_AFTER_ARG = "--restore-after="; private static final String RESTORE_AFTER_ENV = "RESTORE_MESSAGES_AFTER"; private static final String RESTORE_AFTER_PROPERTY = "restore.messages.after"; private static final String HEADER_BLOB_PREFIX_ARG = "--header-blob-prefix="; private static final String HEADER_BLOB_PREFIX_ENV = "RECOVERY_HEADER_BLOB_PREFIX"; private static final String HEADER_BLOB_PREFIX_PROPERTY = "recovery.header.blob.prefix"; + private static final String CONCURRENCY_ARG = "--concurrency="; + private static final String CONCURRENCY_ENV = "RECOVERY_CONCURRENCY"; + private static final String CONCURRENCY_PROPERTY = "recovery.concurrency"; + + public RecoveryConfiguration { + Preconditions.checkArgument(concurrency > 0, "'concurrency' must be strictly positive"); + } public static RecoveryConfiguration parse(String[] args) { return new RecoveryConfiguration( option(args, RESTORE_AFTER_ARG, RESTORE_AFTER_ENV, RESTORE_AFTER_PROPERTY).map(RecoveryConfiguration::parseInstant), - option(args, HEADER_BLOB_PREFIX_ARG, HEADER_BLOB_PREFIX_ENV, HEADER_BLOB_PREFIX_PROPERTY).orElse("")); + option(args, HEADER_BLOB_PREFIX_ARG, HEADER_BLOB_PREFIX_ENV, HEADER_BLOB_PREFIX_PROPERTY).orElse(""), + option(args, CONCURRENCY_ARG, CONCURRENCY_ENV, CONCURRENCY_PROPERTY).map(RecoveryConfiguration::parseConcurrency).orElse(DEFAULT_CONCURRENCY)); } private static Optional<String> option(String[] args, String argPrefix, String envName, String propertyName) { @@ -73,4 +90,15 @@ public record RecoveryConfiguration(Optional<Instant> restoreAfter, String heade + "'. Expected an ISO-8601 instant, e.g. 2026-01-01T00:00:00Z", e); } } + + private static int parseConcurrency(String value) { + try { + int concurrency = Integer.parseInt(value); + Preconditions.checkArgument(concurrency > 0); + return concurrency; + } catch (IllegalArgumentException e) { + throw new IllegalArgumentException("Invalid '" + CONCURRENCY_ARG + "' value: '" + value + + "'. Expected a strictly positive integer", e); + } + } } diff --git a/server/apps/distributed-app/src/main/java/org/apache/james/S3RecoveryService.java b/server/apps/distributed-app/src/main/java/org/apache/james/S3RecoveryService.java index d69b88daa4..f693149fca 100644 --- a/server/apps/distributed-app/src/main/java/org/apache/james/S3RecoveryService.java +++ b/server/apps/distributed-app/src/main/java/org/apache/james/S3RecoveryService.java @@ -88,7 +88,6 @@ public class S3RecoveryService { private static final Logger LOGGER = LoggerFactory.getLogger(S3RecoveryService.class); private static final String RESTORE_MAILBOX = "Restored-messages"; - private static final int CONCURRENCY = 8; private static final String DELIVERED_TO = "Delivered-To"; private static final Report RESTORED = new Report(1, 1, 0, 0, 0); private static final Report SKIPPED_BY_DATE = new Report(1, 0, 1, 0, 0); @@ -118,11 +117,11 @@ public class S3RecoveryService { public Mono<Report> run() { BucketName bucket = blobStore.getDefaultBucketName(); String prefix = RECOVERY_BLOB_PREFIX + configuration.headerBlobPrefix(); - LOGGER.info("Starting S3 recovery on bucket {} (prefix: {}, restore after: {})", - bucket.asString(), prefix, configuration.restoreAfter()); + LOGGER.info("Starting S3 recovery on bucket {} (prefix: {}, restore after: {}, concurrency: {})", + bucket.asString(), prefix, configuration.restoreAfter(), configuration.concurrency()); return Flux.from(blobStoreDAO.listBlobs(bucket, prefix)) .map(BlobId::asString) - .flatMap(recoveryKey -> restoreOne(bucket, recoveryKey), CONCURRENCY) + .flatMap(recoveryKey -> restoreOne(bucket, recoveryKey), configuration.concurrency()) .reduce(Report.empty(), Report::merge) .doOnNext(report -> LOGGER.info("S3 recovery finished: {}", report)); } diff --git a/server/apps/distributed-app/src/test/java/org/apache/james/RecoveryConfigurationTest.java b/server/apps/distributed-app/src/test/java/org/apache/james/RecoveryConfigurationTest.java index df3dae23d4..e2205b8e3a 100644 --- a/server/apps/distributed-app/src/test/java/org/apache/james/RecoveryConfigurationTest.java +++ b/server/apps/distributed-app/src/test/java/org/apache/james/RecoveryConfigurationTest.java @@ -54,4 +54,28 @@ class RecoveryConfigurationTest { assertThat(RecoveryConfiguration.parse(new String[] {"--header-blob-prefix=1_42_"}).headerBlobPrefix()) .isEqualTo("1_42_"); } + + @Test + void parseShouldDefaultConcurrency() { + assertThat(RecoveryConfiguration.parse(new String[] {}).concurrency()) + .isEqualTo(RecoveryConfiguration.DEFAULT_CONCURRENCY); + } + + @Test + void parseShouldReadConcurrencyArgument() { + assertThat(RecoveryConfiguration.parse(new String[] {"--concurrency=32"}).concurrency()) + .isEqualTo(32); + } + + @Test + void parseShouldRejectNonPositiveConcurrency() { + assertThatThrownBy(() -> RecoveryConfiguration.parse(new String[] {"--concurrency=0"})) + .isInstanceOf(IllegalArgumentException.class); + } + + @Test + void parseShouldRejectNonNumericConcurrency() { + assertThatThrownBy(() -> RecoveryConfiguration.parse(new String[] {"--concurrency=many"})) + .isInstanceOf(IllegalArgumentException.class); + } } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
