Merge branch 'stable-2.14' into stable-2.15

Fix conflicting methods in Schema_146 class.

* stable-2.14:
  Parallelize Schema 108
  Parallelize Schema 130 and 131

Change-Id: If1a1006d216e5346e9dbe7ac491e0ef65bea52ca

6 files changed, 201 insertions, 139 deletions

diff --git a/gerrit-server/src/main/java/com/google/gerrit/server/schema/SchemaVersion.java b/gerrit-server/src/main/java/com/google/gerrit/server/schema/SchemaVersion.java
index 5681ecee01..2750f0aeef 100644
--- a/gerrit-server/src/main/java/com/google/gerrit/server/schema/SchemaVersion.java
+++ b/gerrit-server/src/main/java/com/google/gerrit/server/schema/SchemaVersion.java
@@ -26,13 +26,23 @@ import com.google.gwtorm.server.OrmException;
 import com.google.gwtorm.server.SchemaFactory;
 import com.google.gwtorm.server.StatementExecutor;
 import com.google.inject.Provider;
+import java.io.IOException;
 import java.sql.PreparedStatement;
 import java.sql.SQLException;
 import java.sql.Statement;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
 
 /** A version of the database schema. */
 public abstract class SchemaVersion {
@@ -226,4 +236,57 @@ public abstract class SchemaVersion {
   protected static JdbcExecutor newExecutor(ReviewDb db) throws OrmException {
     return new JdbcExecutor(((JdbcSchema) db).getConnection());
   }
+
+  protected int getThreads() {
+    return Runtime.getRuntime().availableProcessors();
+  }
+
+  protected ExecutorService createExecutor(UpdateUI ui) {
+    int threads = getThreads();
+    ui.message(String.format("... using %d threads ...", threads));
+    return Executors.newFixedThreadPool(threads);
+  }
+
+  @FunctionalInterface
+  protected interface ThrowingFunction<I, T> {
+    default T accept(I input) {
+      try {
+        return acceptWithThrow(input);
+      } catch (OrmException | IOException e) {
+        throw new RuntimeException(e);
+      }
+    }
+
+    T acceptWithThrow(I input) throws OrmException, IOException;
+  }
+
+  protected Collection<?> runParallelTasks(
+      ExecutorService executor, Collection<?> lst, ThrowingFunction task, UpdateUI ui) {
+    Collection<Object> returnSet = new HashSet<>();
+    Set<Future> futures =
+        lst.stream()
+            .map(each -> executor.submit(() -> task.accept(each)))
+            .collect(Collectors.toSet());
+    for (Future each : futures) {
+      try {
+        Object rtn = each.get();
+        if (Objects.nonNull(rtn)) {
+          returnSet.add(rtn);
+        }
+      } catch (InterruptedException e) {
+        ui.message(
+            String.format(
+                "Migration step was interrupted. Only %d of %d tasks done.",
+                countDone(futures), lst.size()));
+        throw new RuntimeException(e);
+      } catch (ExecutionException e) {
+        ui.message(e.getCause().getMessage());
+      }
+    }
+    return returnSet;
+  }
+
+  private static long countDone(Collection<Future> futures) {
+    return futures.stream().filter(Future::isDone).count();
+  }
 }
diff --git a/gerrit-server/src/main/java/com/google/gerrit/server/schema/Schema_106.java b/gerrit-server/src/main/java/com/google/gerrit/server/schema/Schema_106.java
index 5bb3669a14..26cf815f9c 100644
--- a/gerrit-server/src/main/java/com/google/gerrit/server/schema/Schema_106.java
+++ b/gerrit-server/src/main/java/com/google/gerrit/server/schema/Schema_106.java
@@ -28,14 +28,7 @@ import com.google.inject.Provider;
 import java.io.File;
 import java.io.IOException;
 import java.io.PrintWriter;
-import java.util.ArrayList;
-import java.util.List;
 import java.util.SortedSet;
-import java.util.concurrent.Callable;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
 import org.eclipse.jgit.lib.ObjectId;
 import org.eclipse.jgit.lib.PersonIdent;
 import org.eclipse.jgit.lib.Repository;
@@ -46,6 +39,7 @@ public class Schema_106 extends SchemaVersion {
   private static final int THREADS_PER_CPU = 4;
   private final GitRepositoryManager repoManager;
   private final PersonIdent serverUser;
+  private int repoCount;
 
   @Inject
   Schema_106(
@@ -65,95 +59,50 @@ public class Schema_106 extends SchemaVersion {
 
     ui.message("listing all repositories ...");
     SortedSet<Project.NameKey> repoList = repoManager.list();
+    repoCount = repoList.size();
     ui.message("done");
 
     ui.message(String.format("creating reflog files for %s branches ...", RefNames.REFS_CONFIG));
 
-    ExecutorService executorPool = createExecutor(ui, repoList.size());
-    List<Future<Void>> futures = new ArrayList<>();
-
-    for (Project.NameKey project : repoList) {
-      Callable<Void> callable = new ReflogCreator(project);
-      futures.add(executorPool.submit(callable));
-    }
-
-    executorPool.shutdown();
-    try {
-      for (Future<Void> future : futures) {
-        try {
-          future.get();
-        } catch (ExecutionException e) {
-          ui.message(e.getCause().getMessage());
-        }
-      }
-      ui.message("done");
-    } catch (InterruptedException ex) {
-      String msg =
-          String.format(
-              "Migration step 106 was interrupted. "
-                  + "Reflog created in %d of %d repositories only.",
-              countDone(futures), repoList.size());
-      ui.message(msg);
-    }
-  }
-
-  private static int countDone(List<Future<Void>> futures) {
-    int count = 0;
-    for (Future<Void> future : futures) {
-      if (future.isDone()) {
-        count++;
-      }
-    }
-
-    return count;
+    runParallelTasks(
+        createExecutor(ui), repoList, (repo) -> createRefLog((Project.NameKey) repo), ui);
   }
 
-  private ExecutorService createExecutor(UpdateUI ui, int repoCount) {
-    int procs = Runtime.getRuntime().availableProcessors();
-    int threads = Math.min(procs * THREADS_PER_CPU, repoCount);
-    ui.message(String.format("... using %d threads ...", threads));
-    return Executors.newFixedThreadPool(threads);
+  @Override
+  protected int getThreads() {
+    return Math.min(Runtime.getRuntime().availableProcessors() * THREADS_PER_CPU, repoCount);
   }
 
-  private class ReflogCreator implements Callable<Void> {
-    private final Project.NameKey project;
-
-    ReflogCreator(Project.NameKey project) {
-      this.project = project;
-    }
-
-    @Override
-    public Void call() throws IOException {
-      try (Repository repo = repoManager.openRepository(project)) {
-        File metaConfigLog = new File(repo.getDirectory(), "logs/" + RefNames.REFS_CONFIG);
-        if (metaConfigLog.exists()) {
-          return null;
-        }
+  public Void createRefLog(Project.NameKey project) throws IOException {
+    try (Repository repo = repoManager.openRepository(project)) {
+      File metaConfigLog = new File(repo.getDirectory(), "logs/" + RefNames.REFS_CONFIG);
+      if (metaConfigLog.exists()) {
+        return null;
+      }
 
-        if (!metaConfigLog.getParentFile().mkdirs() || !metaConfigLog.createNewFile()) {
-          throw new IOException();
-        }
+      if (!metaConfigLog.getParentFile().mkdirs() || !metaConfigLog.createNewFile()) {
+        throw new IOException();
+      }
 
-        ObjectId metaConfigId = repo.resolve(RefNames.REFS_CONFIG);
-        if (metaConfigId != null) {
-          try (PrintWriter writer = new PrintWriter(metaConfigLog, UTF_8.name())) {
-            writer.print(ObjectId.zeroId().name());
-            writer.print(" ");
-            writer.print(metaConfigId.name());
-            writer.print(" ");
-            writer.print(serverUser.toExternalString());
-            writer.print("\t");
-            writer.print("create reflog");
-            writer.println();
-          }
+      ObjectId metaConfigId = repo.resolve(RefNames.REFS_CONFIG);
+      if (metaConfigId != null) {
+        try (PrintWriter writer = new PrintWriter(metaConfigLog, UTF_8.name())) {
+          writer.print(ObjectId.zeroId().name());
+          writer.print(" ");
+          writer.print(metaConfigId.name());
+          writer.print(" ");
+          writer.print(serverUser.toExternalString());
+          writer.print("\t");
+          writer.print("create reflog");
+          writer.println();
         }
-        return null;
-      } catch (IOException e) {
-        throw new IOException(
-            String.format(
-                "ERROR: Failed to create reflog file for the %s branch in repository %s",
-                RefNames.REFS_CONFIG, project.get()));
       }
+      return null;
+    } catch (IOException e) {
+      throw new IOException(
+          String.format(
+              "ERROR: Failed to create reflog file for the %s branch in repository %s",
+              RefNames.REFS_CONFIG, project.get()));
     }
   }
 }
diff --git a/gerrit-server/src/main/java/com/google/gerrit/server/schema/Schema_108.java b/gerrit-server/src/main/java/com/google/gerrit/server/schema/Schema_108.java
index dc88f8de64..66e0d3aeb5 100644
--- a/gerrit-server/src/main/java/com/google/gerrit/server/schema/Schema_108.java
+++ b/gerrit-server/src/main/java/com/google/gerrit/server/schema/Schema_108.java
@@ -27,6 +27,7 @@ import com.google.gerrit.reviewdb.client.Project;
 import com.google.gerrit.reviewdb.client.Project.NameKey;
 import com.google.gerrit.reviewdb.client.RefNames;
 import com.google.gerrit.reviewdb.server.ReviewDb;
+import com.google.gerrit.server.config.GerritServerConfig;
 import com.google.gerrit.server.git.GitRepositoryManager;
 import com.google.gerrit.server.git.GroupCollector;
 import com.google.gerrit.server.project.NoSuchChangeException;
@@ -39,6 +40,7 @@ import java.util.Map;
 import java.util.Set;
 import java.util.SortedSet;
 import org.eclipse.jgit.errors.MissingObjectException;
+import org.eclipse.jgit.lib.Config;
 import org.eclipse.jgit.lib.Constants;
 import org.eclipse.jgit.lib.ObjectId;
 import org.eclipse.jgit.lib.Ref;
@@ -51,37 +53,54 @@ import org.eclipse.jgit.revwalk.RevWalk;
 
 public class Schema_108 extends SchemaVersion {
   private final GitRepositoryManager repoManager;
+  private final Config cfg;
+  private ReviewDb db;
+  private UpdateUI ui;
 
   @Inject
-  Schema_108(Provider<Schema_107> prior, GitRepositoryManager repoManager) {
+  Schema_108(
+      Provider<Schema_107> prior,
+      GitRepositoryManager repoManager,
+      @GerritServerConfig Config cfg) {
     super(prior);
     this.repoManager = repoManager;
+    this.cfg = cfg;
   }
 
   @Override
   protected void migrateData(ReviewDb db, UpdateUI ui) throws OrmException {
+    this.db = db;
+    this.ui = ui;
     ui.message("Listing all changes ...");
     SetMultimap<Project.NameKey, Change.Id> openByProject = getOpenChangesByProject(db, ui);
     ui.message("done");
 
     ui.message("Updating groups for open changes ...");
-    int i = 0;
-    for (Map.Entry<Project.NameKey, Collection<Change.Id>> e : openByProject.asMap().entrySet()) {
-      try (Repository repo = repoManager.openRepository(e.getKey());
-          RevWalk rw = new RevWalk(repo)) {
-        updateProjectGroups(db, repo, rw, (Set<Change.Id>) e.getValue(), ui);
-      } catch (IOException | NoSuchChangeException err) {
-        throw new OrmException(err);
-      }
-      if (++i % 100 == 0) {
-        ui.message("  done " + i + " projects ...");
-      }
-    }
+    runParallelTasks(
+        createExecutor(ui),
+        openByProject.asMap().entrySet(),
+        (batch) -> processProjectBatch((Map.Entry<NameKey, Collection<Change.Id>>) batch),
+        ui);
     ui.message("done");
   }
 
-  private void updateProjectGroups(
-      ReviewDb db, Repository repo, RevWalk rw, Set<Change.Id> changes, UpdateUI ui)
+  private Void processProjectBatch(
+      Map.Entry<Project.NameKey, Collection<Change.Id>> changesByProject) throws OrmException {
+    try (Repository repo = repoManager.openRepository(changesByProject.getKey());
+        RevWalk rw = new RevWalk(repo)) {
+      updateProjectGroups(repo, rw, (Set<Change.Id>) changesByProject.getValue());
+    } catch (IOException | NoSuchChangeException err) {
+      throw new OrmException(err);
+    }
+    return null;
+  }
+
+  @Override
+  protected int getThreads() {
+    return cfg.getInt("cache", "projects", "loadThreads", super.getThreads());
+  }
+
+  private void updateProjectGroups(Repository repo, RevWalk rw, Set<Change.Id> changes)
       throws OrmException, IOException {
     // Match sorting in ReceiveCommits.
     rw.reset();
diff --git a/gerrit-server/src/main/java/com/google/gerrit/server/schema/Schema_130.java b/gerrit-server/src/main/java/com/google/gerrit/server/schema/Schema_130.java
index afb62faf3a..85adb65a77 100644
--- a/gerrit-server/src/main/java/com/google/gerrit/server/schema/Schema_130.java
+++ b/gerrit-server/src/main/java/com/google/gerrit/server/schema/Schema_130.java
@@ -19,14 +19,16 @@ import static java.util.stream.Collectors.joining;
 import com.google.gerrit.reviewdb.client.Project;
 import com.google.gerrit.reviewdb.server.ReviewDb;
 import com.google.gerrit.server.GerritPersonIdent;
+import com.google.gerrit.server.config.GerritServerConfig;
 import com.google.gerrit.server.extensions.events.GitReferenceUpdated;
 import com.google.gerrit.server.git.GitRepositoryManager;
 import com.google.gerrit.server.git.MetaDataUpdate;
 import com.google.gwtorm.server.OrmException;
 import com.google.inject.Inject;
 import com.google.inject.Provider;
+import java.util.Collection;
 import java.util.SortedSet;
-import java.util.TreeSet;
+import org.eclipse.jgit.lib.Config;
 import org.eclipse.jgit.lib.PersonIdent;
 import org.eclipse.jgit.lib.Repository;
 
@@ -39,36 +41,49 @@ public class Schema_130 extends SchemaVersion {
 
   private final GitRepositoryManager repoManager;
   private final PersonIdent serverUser;
+  private final Config cfg;
 
   @Inject
   Schema_130(
       Provider<Schema_129> prior,
       GitRepositoryManager repoManager,
-      @GerritPersonIdent PersonIdent serverUser) {
+      @GerritPersonIdent PersonIdent serverUser,
+      @GerritServerConfig Config cfg) {
     super(prior);
     this.repoManager = repoManager;
     this.serverUser = serverUser;
+    this.cfg = cfg;
   }
 
   @Override
   protected void migrateData(ReviewDb db, UpdateUI ui) throws OrmException {
     SortedSet<Project.NameKey> repoList = repoManager.list();
-    SortedSet<Project.NameKey> repoUpgraded = new TreeSet<>();
     ui.message("\tMigrating " + repoList.size() + " repositories ...");
-    for (Project.NameKey projectName : repoList) {
-      try (Repository git = repoManager.openRepository(projectName);
-          MetaDataUpdate md = new MetaDataUpdate(GitReferenceUpdated.DISABLED, projectName, git)) {
-        ProjectConfigSchemaUpdate cfg = ProjectConfigSchemaUpdate.read(md);
-        cfg.removeForceFromPermission("pushTag");
-        if (cfg.isUpdated()) {
-          repoUpgraded.add(projectName);
-        }
-        cfg.save(serverUser, COMMIT_MSG);
-      } catch (Exception ex) {
-        throw new OrmException("Cannot migrate project " + projectName, ex);
-      }
-    }
+    Collection<Project.NameKey> repoUpgraded =
+        (Collection<Project.NameKey>)
+            runParallelTasks(
+                createExecutor(ui),
+                repoList,
+                (repo) -> removePushTagForcePerms((Project.NameKey) repo),
+                ui);
     ui.message("\tMigration completed:  " + repoUpgraded.size() + " repositories updated:");
     ui.message("\t" + repoUpgraded.stream().map(n -> n.get()).collect(joining(" ")));
   }
+
+  @Override
+  protected int getThreads() {
+    return cfg.getInt("cache", "projects", "loadThreads", super.getThreads());
+  }
+
+  private Project.NameKey removePushTagForcePerms(Project.NameKey project) throws OrmException {
+    try (Repository git = repoManager.openRepository(project);
+        MetaDataUpdate md = new MetaDataUpdate(GitReferenceUpdated.DISABLED, project, git)) {
+      ProjectConfigSchemaUpdate cfg = ProjectConfigSchemaUpdate.read(md);
+      cfg.removeForceFromPermission("pushTag");
+      cfg.save(serverUser, COMMIT_MSG);
+      return cfg.isUpdated() ? project : null;
+    } catch (Exception ex) {
+      throw new OrmException("Cannot migrate project " + project, ex);
+    }
+  }
 }
diff --git a/gerrit-server/src/main/java/com/google/gerrit/server/schema/Schema_131.java b/gerrit-server/src/main/java/com/google/gerrit/server/schema/Schema_131.java
index 0387e357c0..0147669ceb 100644
--- a/gerrit-server/src/main/java/com/google/gerrit/server/schema/Schema_131.java
+++ b/gerrit-server/src/main/java/com/google/gerrit/server/schema/Schema_131.java
@@ -19,6 +19,7 @@ import static java.util.stream.Collectors.joining;
 import com.google.gerrit.reviewdb.client.Project;
 import com.google.gerrit.reviewdb.server.ReviewDb;
 import com.google.gerrit.server.GerritPersonIdent;
+import com.google.gerrit.server.config.GerritServerConfig;
 import com.google.gerrit.server.extensions.events.GitReferenceUpdated;
 import com.google.gerrit.server.git.GitRepositoryManager;
 import com.google.gerrit.server.git.MetaDataUpdate;
@@ -27,9 +28,10 @@ import com.google.gwtorm.server.OrmException;
 import com.google.inject.Inject;
 import com.google.inject.Provider;
 import java.io.IOException;
+import java.util.Collection;
 import java.util.SortedSet;
-import java.util.TreeSet;
 import org.eclipse.jgit.errors.ConfigInvalidException;
+import org.eclipse.jgit.lib.Config;
 import org.eclipse.jgit.lib.PersonIdent;
 import org.eclipse.jgit.lib.Repository;
 
@@ -39,38 +41,54 @@ public class Schema_131 extends SchemaVersion {
 
   private final GitRepositoryManager repoManager;
   private final PersonIdent serverUser;
+  private final Config cfg;
 
   @Inject
   Schema_131(
       Provider<Schema_130> prior,
       GitRepositoryManager repoManager,
-      @GerritPersonIdent PersonIdent serverUser) {
+      @GerritPersonIdent PersonIdent serverUser,
+      @GerritServerConfig Config cfg) {
     super(prior);
     this.repoManager = repoManager;
     this.serverUser = serverUser;
+    this.cfg = cfg;
   }
 
   @Override
   protected void migrateData(ReviewDb db, UpdateUI ui) throws OrmException {
     SortedSet<Project.NameKey> repoList = repoManager.list();
-    SortedSet<Project.NameKey> repoUpgraded = new TreeSet<>();
     ui.message("\tMigrating " + repoList.size() + " repositories ...");
-    for (Project.NameKey projectName : repoList) {
-      try (Repository git = repoManager.openRepository(projectName);
-          MetaDataUpdate md = new MetaDataUpdate(GitReferenceUpdated.DISABLED, projectName, git)) {
-        ProjectConfig config = ProjectConfig.read(md);
-        if (config.hasLegacyPermissions()) {
-          md.getCommitBuilder().setAuthor(serverUser);
-          md.getCommitBuilder().setCommitter(serverUser);
-          md.setMessage(COMMIT_MSG);
-          config.commit(md);
-          repoUpgraded.add(projectName);
-        }
-      } catch (ConfigInvalidException | IOException ex) {
-        throw new OrmException("Cannot migrate project " + projectName, ex);
-      }
-    }
+    Collection<Project.NameKey> repoUpgraded =
+        (Collection<Project.NameKey>)
+            runParallelTasks(
+                createExecutor(ui),
+                repoList,
+                (repo) -> renamePushTagPermissions((Project.NameKey) repo),
+                ui);
     ui.message("\tMigration completed:  " + repoUpgraded.size() + " repositories updated:");
     ui.message("\t" + repoUpgraded.stream().map(n -> n.get()).collect(joining(" ")));
   }
+
+  @Override
+  protected int getThreads() {
+    return cfg.getInt("cache", "projects", "loadThreads", super.getThreads());
+  }
+
+  private Project.NameKey renamePushTagPermissions(Project.NameKey project) throws OrmException {
+    try (Repository git = repoManager.openRepository(project);
+        MetaDataUpdate md = new MetaDataUpdate(GitReferenceUpdated.DISABLED, project, git)) {
+      ProjectConfig config = ProjectConfig.read(md);
+      if (config.hasLegacyPermissions()) {
+        md.getCommitBuilder().setAuthor(serverUser);
+        md.getCommitBuilder().setCommitter(serverUser);
+        md.setMessage(COMMIT_MSG);
+        config.commit(md);
+        return project;
+      }
+    } catch (ConfigInvalidException | IOException ex) {
+      throw new OrmException("Cannot migrate project " + project, ex);
+    }
+    return null;
+  }
 }
diff --git a/gerrit-server/src/main/java/com/google/gerrit/server/schema/Schema_146.java b/gerrit-server/src/main/java/com/google/gerrit/server/schema/Schema_146.java
index 72e8c8a47e..b08b536fc0 100644
--- a/gerrit-server/src/main/java/com/google/gerrit/server/schema/Schema_146.java
+++ b/gerrit-server/src/main/java/com/google/gerrit/server/schema/Schema_146.java
@@ -120,15 +120,13 @@ public class Schema_146 extends SchemaVersion {
     ui.message(String.format("... (%.3f s) full gc completed", elapsed()));
   }
 
-  private ExecutorService createExecutor(UpdateUI ui) {
-    int threads;
+  @Override
+  protected int getThreads() {
     try {
-      threads = Integer.parseInt(System.getProperty("threadcount"));
+      return Integer.parseInt(System.getProperty("threadcount"));
     } catch (NumberFormatException e) {
-      threads = Runtime.getRuntime().availableProcessors();
+      return super.getThreads();
     }
-    ui.message(String.format("... using %d threads ...", threads));
-    return Executors.newFixedThreadPool(threads);
   }
 
   private void processBatch(List<Entry<Account.Id, Timestamp>> batch, UpdateUI ui) {