This is an automated email from the ASF dual-hosted git repository. dsmiley pushed a commit to branch branch_9x in repository https://gitbox.apache.org/repos/asf/solr.git
commit 031e198f6a2283734cf0ebc181ae03a5b77403d0 Author: Samuel Verstraete <[email protected]> AuthorDate: Wed Jun 10 20:12:20 2026 +0200 SOLR-17208: Parallelize backup and restore file transfers (#4023) Co-authored-by: samuel <[email protected]> Co-authored-by: David Smiley <[email protected]> (cherry picked from commit 74974b1be7e5b4a58dbe93468a359aca9621a647) (cherry picked from commit e2edd26b12c9e97d32c5b3383b7da5f9579d2665) --- .../unreleased/SOLR-1092-parallelizebackups.yml | 10 ++ .../solr/handler/IncrementalShardBackup.java | 136 ++++++++++++++++----- .../java/org/apache/solr/handler/RestoreCore.java | 116 ++++++++++++++---- .../apache/solr/gcs/GCSIncrementalBackupTest.java | 3 + .../apache/solr/s3/S3IncrementalBackupTest.java | 3 + .../deployment-guide/pages/backup-restore.adoc | 32 +++++ .../org/apache/solr/common/util/ObjectCache.java | 3 + 7 files changed, 248 insertions(+), 55 deletions(-) diff --git a/changelog/unreleased/SOLR-1092-parallelizebackups.yml b/changelog/unreleased/SOLR-1092-parallelizebackups.yml new file mode 100644 index 00000000000..84853f8bc39 --- /dev/null +++ b/changelog/unreleased/SOLR-1092-parallelizebackups.yml @@ -0,0 +1,10 @@ +title: Parallelize Backup and Restore File Operations +type: changed +authors: + - name: Samuel Verstraete + nick: elangelo + - name: David Smiley + nick: dsmiley +links: + - name: SOLR-1092 + url: https://issues.apache.org/jira/browse/SOLR-1092 diff --git a/solr/core/src/java/org/apache/solr/handler/IncrementalShardBackup.java b/solr/core/src/java/org/apache/solr/handler/IncrementalShardBackup.java index 0e07ac0ca28..792f2217363 100644 --- a/solr/core/src/java/org/apache/solr/handler/IncrementalShardBackup.java +++ b/solr/core/src/java/org/apache/solr/handler/IncrementalShardBackup.java @@ -23,15 +23,25 @@ import java.io.IOException; import java.lang.invoke.MethodHandles; import java.net.URI; import java.time.Instant; +import java.util.ArrayList; import java.util.Collection; +import java.util.List; import java.util.Optional; import java.util.UUID; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; import org.apache.commons.math3.util.Precision; import org.apache.lucene.index.IndexCommit; import org.apache.lucene.store.Directory; import org.apache.solr.client.api.model.SolrJerseyResponse; import org.apache.solr.cloud.CloudDescriptor; import org.apache.solr.common.SolrException; +import org.apache.solr.common.util.EnvUtils; +import org.apache.solr.common.util.ExecutorUtil; +import org.apache.solr.common.util.SolrNamedThreadFactory; import org.apache.solr.core.DirectoryFactory; import org.apache.solr.core.IndexDeletionPolicyWrapper; import org.apache.solr.core.SolrCore; @@ -52,6 +62,15 @@ import org.slf4j.LoggerFactory; */ public class IncrementalShardBackup { private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); + + /** + * Maximum number of files to upload in parallel during backup. Can be configured via the system + * property {@code solr.backup.maxparalleluploads} or environment variable {@code + * SOLR_BACKUP_MAXPARALLELUPLOADS}. + */ + private static final int MAX_PARALLEL_UPLOADS = + EnvUtils.getPropertyAsInteger("solr.backup.maxparalleluploads", 1); + private SolrCore solrCore; private BackupFilePaths incBackupFiles; @@ -154,8 +173,8 @@ public class IncrementalShardBackup { solrCore.getSolrConfig().indexConfig.lockType); try { BackupStats stats = incrementalCopy(files, dir); - details.indexFileCount = stats.fileCount; - details.uploadedIndexFileCount = stats.uploadedFileCount; + details.indexFileCount = stats.fileCount.get(); + details.uploadedIndexFileCount = stats.uploadedFileCount.get(); details.indexSizeMB = stats.getIndexSizeMB(); details.uploadedIndexFileMB = stats.getTotalUploadedMB(); } finally { @@ -191,25 +210,80 @@ public class IncrementalShardBackup { URI indexDir = incBackupFiles.getIndexDir(); BackupStats backupStats = new BackupStats(); + var executor = + solrCore + .getCoreContainer() + .getObjectCache() + .computeIfAbsent( + "BackupUploadExecutor", + ExecutorService.class, + s -> + ExecutorUtil.newMDCAwareCachedThreadPool( + MAX_PARALLEL_UPLOADS, + Integer.MAX_VALUE, + new SolrNamedThreadFactory("BackupUploadExecutor"))); + + List<Future<?>> uploadFutures = new ArrayList<>(); for (String fileName : indexFiles) { - Optional<ShardBackupMetadata.BackedFile> opBackedFile = oldBackupPoint.getFile(fileName); - Checksum originalFileCS = backupRepo.checksum(dir, fileName); - - if (opBackedFile.isPresent()) { - ShardBackupMetadata.BackedFile backedFile = opBackedFile.get(); - Checksum existedFileCS = backedFile.fileChecksum; - if (existedFileCS.equals(originalFileCS)) { - currentBackupPoint.addBackedFile(opBackedFile.get()); - backupStats.skippedUploadingFile(existedFileCS); - continue; - } - } - - String backedFileName = UUID.randomUUID().toString(); - backupRepo.copyIndexFileFrom(dir, fileName, indexDir, backedFileName); + // Capture variable for lambda + final String fileNameFinal = fileName; + + Runnable uploadTask = + () -> { + try { + // Calculate checksum and check if file already exists in previous backup + Optional<ShardBackupMetadata.BackedFile> opBackedFile = + oldBackupPoint.getFile(fileNameFinal); + Checksum originalFileCS = backupRepo.checksum(dir, fileNameFinal); + + if (opBackedFile.isPresent()) { + ShardBackupMetadata.BackedFile backedFile = opBackedFile.get(); + Checksum existedFileCS = backedFile.fileChecksum; + if (existedFileCS.equals(originalFileCS)) { + synchronized (currentBackupPoint) { + currentBackupPoint.addBackedFile(opBackedFile.get()); + } + backupStats.skippedUploadingFile(existedFileCS); + return; + } + } + + // File doesn't exist or has changed - upload it + String backedFileName = UUID.randomUUID().toString(); + backupRepo.copyIndexFileFrom(dir, fileNameFinal, indexDir, backedFileName); + + synchronized (currentBackupPoint) { + currentBackupPoint.addBackedFile(backedFileName, fileNameFinal, originalFileCS); + } + backupStats.uploadedFile(originalFileCS); + } catch (IOException e) { + throw new RuntimeException("Failed to process file: " + fileNameFinal, e); + } + }; + + uploadFutures.add(executor.submit(uploadTask)); + } - currentBackupPoint.addBackedFile(backedFileName, fileName, originalFileCS); - backupStats.uploadedFile(originalFileCS); + try { + for (Future<?> future : uploadFutures) { + future.get(); + } + } catch (ExecutionException e) { + uploadFutures.forEach(f -> f.cancel(true)); + Throwable cause = e.getCause(); + if (cause instanceof Error) { + throw (Error) cause; + } else if (cause instanceof IOException) { + throw (IOException) cause; + } else if (cause instanceof RuntimeException) { + throw (RuntimeException) cause; + } else { + throw new SolrException( + SolrException.ErrorCode.UNKNOWN, "Error during parallel backup upload", cause); + } + } catch (InterruptedException e) { + uploadFutures.forEach(f -> f.cancel(true)); + throw new SolrException(SolrException.ErrorCode.UNKNOWN, "Backup interrupted", e); } currentBackupPoint.store(backupRepo, incBackupFiles.getShardBackupMetadataDir(), shardBackupId); @@ -217,29 +291,29 @@ public class IncrementalShardBackup { } private static class BackupStats { - private int fileCount; - private int uploadedFileCount; - private long indexSize; - private long totalUploadedBytes; + private final AtomicInteger fileCount = new AtomicInteger(); + private final AtomicInteger uploadedFileCount = new AtomicInteger(); + private final AtomicLong indexSize = new AtomicLong(); + private final AtomicLong totalUploadedBytes = new AtomicLong(); public void uploadedFile(Checksum file) { - fileCount++; - uploadedFileCount++; - indexSize += file.size; - totalUploadedBytes += file.size; + fileCount.incrementAndGet(); + uploadedFileCount.incrementAndGet(); + indexSize.addAndGet(file.size); + totalUploadedBytes.addAndGet(file.size); } public void skippedUploadingFile(Checksum existedFile) { - fileCount++; - indexSize += existedFile.size; + fileCount.incrementAndGet(); + indexSize.addAndGet(existedFile.size); } public double getIndexSizeMB() { - return Precision.round(indexSize / (1024.0 * 1024), 3); + return Precision.round(indexSize.get() / (1024.0 * 1024), 3); } public double getTotalUploadedMB() { - return Precision.round(totalUploadedBytes / (1024.0 * 1024), 3); + return Precision.round(totalUploadedBytes.get() / (1024.0 * 1024), 3); } } diff --git a/solr/core/src/java/org/apache/solr/handler/RestoreCore.java b/solr/core/src/java/org/apache/solr/handler/RestoreCore.java index 7f4abc18ffd..5b92cc804b3 100644 --- a/solr/core/src/java/org/apache/solr/handler/RestoreCore.java +++ b/solr/core/src/java/org/apache/solr/handler/RestoreCore.java @@ -21,19 +21,26 @@ import java.lang.invoke.MethodHandles; import java.lang.reflect.Array; import java.net.URI; import java.text.SimpleDateFormat; +import java.util.ArrayList; import java.util.Arrays; import java.util.Date; import java.util.HashSet; +import java.util.List; import java.util.Locale; import java.util.Optional; import java.util.Set; import java.util.concurrent.Callable; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Future; import org.apache.lucene.codecs.CodecUtil; import org.apache.lucene.store.Directory; import org.apache.lucene.store.IOContext; import org.apache.lucene.store.IndexInput; import org.apache.solr.common.SolrException; +import org.apache.solr.common.util.EnvUtils; +import org.apache.solr.common.util.ExecutorUtil; +import org.apache.solr.common.util.SolrNamedThreadFactory; import org.apache.solr.core.DirectoryFactory; import org.apache.solr.core.SolrCore; import org.apache.solr.core.backup.BackupFilePaths; @@ -48,6 +55,14 @@ public class RestoreCore implements Callable<Boolean> { private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); + /** + * Maximum number of files to download in parallel during restore. Can be configured via the + * system property {@code solr.backup.maxparalleldownloads} or environment variable {@code + * SOLR_BACKUP_MAXPARALLELDOWNLOADS}. + */ + private static final int MAX_PARALLEL_DOWNLOADS = + EnvUtils.getPropertyAsInteger("solr.backup.maxparalleldownloads", 1); + private final SolrCore core; private RestoreRepository repository; @@ -107,35 +122,88 @@ public class RestoreCore implements Callable<Boolean> { DirectoryFactory.DirContext.DEFAULT, core.getSolrConfig().indexConfig.lockType); Set<String> indexDirFiles = new HashSet<>(Arrays.asList(indexDir.listAll())); + + // Capture directories as final for lambda access + final Directory finalIndexDir = indexDir; + final Directory finalRestoreIndexDir = restoreIndexDir; + + List<Future<?>> downloadFutures = new ArrayList<>(); + + var executor = + core.getCoreContainer() + .getObjectCache() + .computeIfAbsent( + "RestoreDownloadExecutor", + ExecutorService.class, + s -> + ExecutorUtil.newMDCAwareCachedThreadPool( + MAX_PARALLEL_DOWNLOADS, + Integer.MAX_VALUE, + new SolrNamedThreadFactory("RestoreDownloadExecutor"))); + // Move all files from backupDir to restoreIndexDir for (String filename : repository.listAllFiles()) { checkInterrupted(); - try { - if (indexDirFiles.contains(filename)) { - Checksum cs = repository.checksum(filename); - IndexFetcher.CompareResult compareResult; - if (cs == null) { - compareResult = new IndexFetcher.CompareResult(); - compareResult.equal = false; - } else { - compareResult = IndexFetcher.compareFile(indexDir, filename, cs.size, cs.checksum); - } - if (!compareResult.equal - || (IndexFetcher.filesToAlwaysDownloadIfNoChecksums( - filename, cs.size, compareResult))) { - repository.repoCopy(filename, restoreIndexDir); - } else { - // prefer local copy - repository.localCopy(indexDir, filename, restoreIndexDir); - } - } else { - repository.repoCopy(filename, restoreIndexDir); - } - } catch (Exception e) { - log.warn("Exception while restoring the backup index ", e); + + // Capture variables for lambda + final String filenameFinal = filename; + final boolean fileExistsLocally = indexDirFiles.contains(filename); + + Runnable downloadTask = + () -> { + try { + if (fileExistsLocally) { + Checksum cs = repository.checksum(filenameFinal); + IndexFetcher.CompareResult compareResult; + if (cs == null) { + compareResult = new IndexFetcher.CompareResult(); + compareResult.equal = false; + } else { + compareResult = + IndexFetcher.compareFile( + finalIndexDir, filenameFinal, cs.size, cs.checksum); + } + if (!compareResult.equal + || (IndexFetcher.filesToAlwaysDownloadIfNoChecksums( + filenameFinal, cs.size, compareResult))) { + repository.repoCopy(filenameFinal, finalRestoreIndexDir); + } else { + // prefer local copy + repository.localCopy(finalIndexDir, filenameFinal, finalRestoreIndexDir); + } + } else { + repository.repoCopy(filenameFinal, finalRestoreIndexDir); + } + } catch (Exception e) { + log.warn("Exception while restoring the backup index ", e); + throw new RuntimeException( + "Exception while restoring the backup index for file: " + filenameFinal, e); + } + }; + + downloadFutures.add(executor.submit(downloadTask)); + } + + try { + for (Future<?> future : downloadFutures) { + future.get(); + } + } catch (ExecutionException e) { + downloadFutures.forEach(f -> f.cancel(true)); + Throwable cause = e.getCause(); + if (cause instanceof Error) { + throw (Error) cause; + } else if (cause instanceof IOException) { + throw (IOException) cause; + } else if (cause instanceof RuntimeException) { + throw (RuntimeException) cause; + } else { throw new SolrException( - SolrException.ErrorCode.UNKNOWN, "Exception while restoring the backup index", e); + SolrException.ErrorCode.UNKNOWN, "Error during parallel restore download", cause); } + } catch (InterruptedException e) { + downloadFutures.forEach(f -> f.cancel(true)); + throw new SolrException(SolrException.ErrorCode.UNKNOWN, "Restore interrupted", e); } log.debug("Switching directories"); core.modifyIndexProps(restoreIndexName); diff --git a/solr/modules/gcs-repository/src/test/org/apache/solr/gcs/GCSIncrementalBackupTest.java b/solr/modules/gcs-repository/src/test/org/apache/solr/gcs/GCSIncrementalBackupTest.java index 8dd8f8f4f42..88a8d578953 100644 --- a/solr/modules/gcs-repository/src/test/org/apache/solr/gcs/GCSIncrementalBackupTest.java +++ b/solr/modules/gcs-repository/src/test/org/apache/solr/gcs/GCSIncrementalBackupTest.java @@ -72,6 +72,9 @@ public class GCSIncrementalBackupTest extends AbstractIncrementalBackupTest { @BeforeClass public static void setupClass() throws Exception { + // Enable parallel backup/restore for cloud storage tests + System.setProperty("solr.backup.maxparalleluploads", "2"); + System.setProperty("solr.backup.maxparalleldownloads", "2"); configureCluster(NUM_NODES) // nodes .addConfig("conf1", getFile("conf/solrconfig.xml").getParentFile().toPath()) diff --git a/solr/modules/s3-repository/src/test/org/apache/solr/s3/S3IncrementalBackupTest.java b/solr/modules/s3-repository/src/test/org/apache/solr/s3/S3IncrementalBackupTest.java index 8e033de8c6e..cdab66201af 100644 --- a/solr/modules/s3-repository/src/test/org/apache/solr/s3/S3IncrementalBackupTest.java +++ b/solr/modules/s3-repository/src/test/org/apache/solr/s3/S3IncrementalBackupTest.java @@ -92,6 +92,9 @@ public class S3IncrementalBackupTest extends AbstractIncrementalBackupTest { public static void setupClass() throws Exception { System.setProperty("aws.accessKeyId", "foo"); System.setProperty("aws.secretAccessKey", "bar"); + // Enable parallel backup/restore for cloud storage tests + System.setProperty("solr.backup.maxparalleluploads", "2"); + System.setProperty("solr.backup.maxparalleldownloads", "2"); String retryMode; switch (random().nextInt(3)) { case 0: diff --git a/solr/solr-ref-guide/modules/deployment-guide/pages/backup-restore.adoc b/solr/solr-ref-guide/modules/deployment-guide/pages/backup-restore.adoc index 6b48bc179e1..c6be96086fd 100644 --- a/solr/solr-ref-guide/modules/deployment-guide/pages/backup-restore.adoc +++ b/solr/solr-ref-guide/modules/deployment-guide/pages/backup-restore.adoc @@ -396,6 +396,38 @@ Any children under the `<repository>` tag are passed as additional configuration Information on each of the repository implementations provided with Solr is provided below. +=== Parallel File Transfers + +Backup and restore operations can transfer multiple index files in parallel to improve throughput, especially when using cloud storage repositories like S3 or GCS where latency is higher. +The parallelism is controlled via system properties or environment variables: + +`solr.backup.maxparalleluploads`:: ++ +[%autowidth,frame=none] +|=== +|Optional |Default: `1` +|=== ++ +Maximum number of index files to upload in parallel during backup operations. +Can also be set via the `SOLR_BACKUP_MAXPARALLELUPLOADS` environment variable. + +`solr.backup.maxparalleldownloads`:: ++ +[%autowidth,frame=none] +|=== +|Optional |Default: `1` +|=== ++ +Maximum number of index files to download in parallel during restore operations. +Can also be set via the `SOLR_BACKUP_MAXPARALLELDOWNLOADS` environment variable. + +TIP: These are two independent thread pools — one for backup uploads and one for restore downloads. +Increasing either value can significantly improve throughput when using cloud storage (S3, GCS), but too high a value will increase IOPS and bandwidth pressure on your cluster. +Start small and increase based on observed throughput and available resources. +The configured limit applies across all concurrent backup or restore operations on the node. + +=== Checksum Verification + By default, all the repository implementations verify the integrity of the index files before they are copied to the destination. However, it is possible to disable this integrity check by setting the optional configuration property `verifyChecksum`. `verifyChecksum`:: diff --git a/solr/solrj/src/java/org/apache/solr/common/util/ObjectCache.java b/solr/solrj/src/java/org/apache/solr/common/util/ObjectCache.java index 2bbb6529c0f..045eb7f04f3 100644 --- a/solr/solrj/src/java/org/apache/solr/common/util/ObjectCache.java +++ b/solr/solrj/src/java/org/apache/solr/common/util/ObjectCache.java @@ -19,6 +19,7 @@ package org.apache.solr.common.util; import java.io.Closeable; import java.io.IOException; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Function; import org.apache.solr.common.SolrCloseable; @@ -93,6 +94,8 @@ public class ObjectCache extends MapBackedCache<String, Object> implements SolrC for (Object value : map.values()) { if (value instanceof Closeable) { ((Closeable) value).close(); + } else if (value instanceof ExecutorService) { + ExecutorUtil.shutdownAndAwaitTermination((ExecutorService) value); } } map.clear();
