elangelo commented on code in PR #4023:
URL: https://github.com/apache/solr/pull/4023#discussion_r2917287084
##########
solr/core/src/java/org/apache/solr/handler/RestoreCore.java:
##########
@@ -107,34 +125,140 @@ public boolean doRestore() throws Exception {
DirectoryFactory.DirContext.DEFAULT,
core.getSolrConfig().indexConfig.lockType);
Set<String> indexDirFiles = new
HashSet<>(Arrays.asList(indexDir.listAll()));
- // 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);
+
+ // Capture directories as final for lambda access
+ final Directory finalIndexDir = indexDir;
+ final Directory finalRestoreIndexDir = restoreIndexDir;
+
+ // Only use an executor for parallel downloads when parallelism > 1
+ // When set to 1, run synchronously to avoid thread-local state issues
with CallerRunsPolicy
+ int maxParallelDownloads = DEFAULT_MAX_PARALLEL_DOWNLOADS;
+ ExecutorService executor =
+ maxParallelDownloads > 1
+ ? new ExecutorUtil.MDCAwareThreadPoolExecutor(
+ 0,
+ maxParallelDownloads,
+ 60L,
+ TimeUnit.SECONDS,
+ new SynchronousQueue<>(),
+ new SolrNamedThreadFactory("RestoreCore"),
+ new ThreadPoolExecutor.CallerRunsPolicy())
+ : null;
+
+ List<Future<?>> downloadFutures = new ArrayList<>();
+
+ try {
+ // Move all files from backupDir to restoreIndexDir
+ for (String filename : repository.listAllFiles()) {
+ checkInterrupted();
+
+ // 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);
+ }
+ };
+
+ if (executor != null) {
+ downloadFutures.add(executor.submit(downloadTask));
+ } else {
+ // Run synchronously when parallelism is 1
+ try {
+ downloadTask.run();
+ } catch (RuntimeException e) {
+ if (e.getCause() instanceof IOException) {
+ throw (IOException) e.getCause();
+ }
+ throw e;
Review Comment:
fixed
##########
solr/core/src/java/org/apache/solr/handler/RestoreCore.java:
##########
@@ -107,34 +125,140 @@ public boolean doRestore() throws Exception {
DirectoryFactory.DirContext.DEFAULT,
core.getSolrConfig().indexConfig.lockType);
Set<String> indexDirFiles = new
HashSet<>(Arrays.asList(indexDir.listAll()));
- // 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);
+
+ // Capture directories as final for lambda access
+ final Directory finalIndexDir = indexDir;
+ final Directory finalRestoreIndexDir = restoreIndexDir;
+
+ // Only use an executor for parallel downloads when parallelism > 1
+ // When set to 1, run synchronously to avoid thread-local state issues
with CallerRunsPolicy
+ int maxParallelDownloads = DEFAULT_MAX_PARALLEL_DOWNLOADS;
+ ExecutorService executor =
+ maxParallelDownloads > 1
+ ? new ExecutorUtil.MDCAwareThreadPoolExecutor(
+ 0,
+ maxParallelDownloads,
+ 60L,
+ TimeUnit.SECONDS,
+ new SynchronousQueue<>(),
+ new SolrNamedThreadFactory("RestoreCore"),
+ new ThreadPoolExecutor.CallerRunsPolicy())
+ : null;
+
+ List<Future<?>> downloadFutures = new ArrayList<>();
+
Review Comment:
see above
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
