Allow setting remote.NAME.threads=0 for persist-only mode Setting 'remote.NAME.threads' to 0 now puts a remote into a persist-only mode on the current node. No thread pool is created and the node performs no pushes, project deletions or HEAD updates for that remote. Replication tasks are still written to the task storage. This supports multi-primary deployments where a user-facing primary only records replication work, while a separate non user-facing primary (configured with threads > 0 and distributor enabled) picks the tasks up from the shared storage and performs the actual pushes. Also, when a node has no push-enabled remotes, both recovery and the startup replay are skipped entirely. A caveat to setting threads to '0' is that project deletions and HEAD updates originating on a persist-only node are not replicated by other nodes, as they are not persisted to the task storage. If needed, support for persisting them can be added in future changes. Change-Id: I022865cbd7105563b5760ca05ae85847e0d7b97b

diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/Destination.java b/src/main/java/com/googlesource/gerrit/plugins/replication/Destination.java
index f83e1df..b35a5d1 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/Destination.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/Destination.java

@@ -287,9 +287,21 @@
     return queue;
   }
 
+  /**
+   * Whether this destination performs pushes.
+   *
+   * <p>When {@code remote.NAME.threads} is set to 0, no pushes will be done, but the tasks are
+   * still persisted to the {@link ReplicationTasksStorage}.
+   */
+  public boolean isPushEnabled() {
+    return config.getPoolThreads() > 0;
+  }
+
   public void start(WorkQueue workQueue) {
-    String poolName = "ReplicateTo-" + config.getRemoteConfig().getName();
-    pool = workQueue.createQueue(config.getPoolThreads(), poolName);
+    if (isPushEnabled()) {
+      String poolName = "ReplicateTo-" + config.getRemoteConfig().getName();
+      pool = workQueue.createQueue(config.getPoolThreads(), poolName);
+    }
   }
 
   public int shutdown() {
@@ -455,6 +467,9 @@
       ReplicationState state,
       boolean now,
       boolean fromStorage) {
+    if (!isPushEnabled()) {
+      return;
+    }
     ImmutableSet.Builder<String> toSchedule = ImmutableSet.builder();
     for (String ref : refs) {
       if (!shouldReplicate(project, ref, state)) {
@@ -544,17 +559,21 @@
   }
 
   void scheduleDeleteProject(URIish uri, Project.NameKey project, ProjectDeletionState state) {
-    repLog.atFine().log("scheduling deletion of project %s at %s", project, uri);
-    @SuppressWarnings("unused")
-    ScheduledFuture<?> ignored =
-        pool.schedule(deleteProjectFactory.create(uri, project, state), 0, TimeUnit.SECONDS);
-    state.setScheduled(uri);
+    if (isPushEnabled()) {
+      repLog.atFine().log("scheduling deletion of project %s at %s", project, uri);
+      @SuppressWarnings("unused")
+      ScheduledFuture<?> ignored =
+          pool.schedule(deleteProjectFactory.create(uri, project, state), 0, TimeUnit.SECONDS);
+      state.setScheduled(uri);
+    }
   }
 
   void scheduleUpdateHead(URIish uri, Project.NameKey project, String newHead) {
-    @SuppressWarnings("unused")
-    ScheduledFuture<?> ignored =
-        pool.schedule(updateHeadFactory.create(uri, project, newHead), 0, TimeUnit.SECONDS);
+    if (isPushEnabled()) {
+      @SuppressWarnings("unused")
+      ScheduledFuture<?> ignored =
+          pool.schedule(updateHeadFactory.create(uri, project, newHead), 0, TimeUnit.SECONDS);
+    }
   }
 
   /**
@@ -579,6 +598,9 @@
    * @param pushOp The PushOp instance to be scheduled.
    */
   void reschedule(PushOne pushOp, RetryReason reason) {
+    if (!isPushEnabled()) {
+      return;
+    }
     RescheduleStatus status = new RescheduleStatus();
     stateLock.withWriteLock(
         pushOp.getURI(),

diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationQueue.java b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationQueue.java
index 59acb20..9abf407 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationQueue.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationQueue.java

@@ -102,8 +102,11 @@
     if (!running) {
       destinations.get().startup(workQueue);
       running = true;
-      replicationTasksStorage.recoverAll();
-      synchronizePendingEvents(Prune.FALSE);
+      Set<String> pushEnabledRemoteNames = getPushEnabledRemoteNames();
+      if (!pushEnabledRemoteNames.isEmpty()) {
+        replicationTasksStorage.recoverAll(r -> pushEnabledRemoteNames.contains(r.remote()));
+        synchronizePendingEvents(Prune.FALSE);
+      }
       fireBeforeStartupEvents();
       distributor = new Distributor(workQueue);
     }
@@ -150,6 +153,13 @@
     fire(event.getProjectName(), event.getUpdatedRefs());
   }
 
+  private Set<String> getPushEnabledRemoteNames() {
+    return destinations.get().getAll(FilterType.ALL).stream()
+        .filter(Destination::isPushEnabled)
+        .map(Destination::getRemoteConfigName)
+        .collect(Collectors.toSet());
+  }
+
   private void fire(String projectName, Set<UpdatedRef> updatedRefs) {
     ReplicationState state = new ReplicationState(new GitUpdateProcessing(dispatcher));
     fire(Project.nameKey(projectName), null, updatedRefs, state, false);

diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorage.java b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorage.java
index 7fc4acb..bc0d900 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorage.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorage.java

@@ -44,6 +44,7 @@
 import java.util.HashSet;
 import java.util.Optional;
 import java.util.Set;
+import java.util.function.Predicate;
 import java.util.stream.Stream;
 import org.eclipse.jgit.lib.ObjectId;
 import org.eclipse.jgit.transport.URIish;
@@ -189,8 +190,13 @@
     }
   }
 
+  @VisibleForTesting
   public void recoverAll() {
-    streamRunning().forEach(r -> new Task(r).recover());
+    recoverAll(r -> true);
+  }
+
+  public void recoverAll(Predicate<ReplicateRefUpdate> shouldRecover) {
+    streamRunning().filter(shouldRecover).forEach(r -> new Task(r).recover());
   }
 
   public boolean isWaiting(UriUpdates uriUpdates) {

diff --git a/src/main/resources/Documentation/config.md b/src/main/resources/Documentation/config.md
index 305bb28..eb96d29 100644
--- a/src/main/resources/Documentation/config.md
+++ b/src/main/resources/Documentation/config.md

@@ -536,6 +536,17 @@
 	remote block describes 4 URLs, allocating 4 threads in the
 	pool will permit some level of parallel pushing.
 
+	A value of `0` puts this remote into a persist-only mode. No thread
+	pool is created and this node performs no pushes. Replication tasks
+	are still written to the replication task storage, so that other
+	nodes in the cluster sharing the same storage (with `threads` > 0 and
+	`replication.distributionInterval` enabled) pick them up and perform
+	the actual pushes.
+
+	> **NOTE**: In persist-only mode, project deletions and HEAD updates
+	> originating on this node cannot be replicated by other nodes, as
+	> they are not persisted to the task storage.
+
 	By default, 1 thread.
 
 remote.NAME.authGroup

diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationStorageIT.java b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationStorageIT.java
index d0affe2..bfcf5be 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationStorageIT.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationStorageIT.java

@@ -67,6 +67,30 @@
   }
 
   @Test
+  public void shouldPersistTaskButNotPushWhenRemoteHasZeroThreads() throws Exception {
+    String remote = "persistOnly";
+    Project.NameKey target = createTestProject(project + "replica");
+    setReplicationDestination(remote, "replica", ALL_PROJECTS);
+    config.setInt("remote", remote, "threads", 0);
+    config.save();
+    reloadConfig();
+
+    String changeRef = createChange().getPatchSet().refName();
+
+    assertThat(waitingChangeReplicationTasksForRemote(changeRef, remote).count()).isEqualTo(1);
+
+    Destination destination =
+        destinationCollection.getAll(FilterType.ALL).stream()
+            .filter(dest -> remote.equals(dest.getRemoteConfigName()))
+            .findFirst()
+            .get();
+
+    assertThat(destination.isPushEnabled()).isFalse();
+    assertThat(destination.getQueue().pending).isEmpty();
+    assertThat(isPushCompleted(target, changeRef, TEST_PUSH_TIMEOUT)).isFalse();
+  }
+
+  @Test
   public void shouldCreateOneReplicationTaskWhenSchedulingRepoFullSync() throws Exception {
     createTestProject(project + "replica");
 
@@ -76,7 +100,8 @@
     plugin
         .getSysInjector()
         .getInstance(ReplicationQueue.class)
-        .scheduleFullSync(project, null, PushOne.ALL_REFS, Set.of(), new ReplicationState(NO_OP), false);
+        .scheduleFullSync(
+            project, null, PushOne.ALL_REFS, Set.of(), new ReplicationState(NO_OP), false);
 
     assertThat(listWaitingReplicationTasks(Pattern.quote(PushOne.ALL_REFS))).hasSize(1);
   }
@@ -221,7 +246,8 @@
     plugin
         .getSysInjector()
         .getInstance(ReplicationQueue.class)
-        .scheduleFullSync(project, urlMatch, PushOne.ALL_REFS, Set.of(), new ReplicationState(NO_OP), false);
+        .scheduleFullSync(
+            project, urlMatch, PushOne.ALL_REFS, Set.of(), new ReplicationState(NO_OP), false);
 
     assertThat(listWaiting()).hasSize(1);
     tasksStorage
@@ -246,7 +272,8 @@
     plugin
         .getSysInjector()
         .getInstance(ReplicationQueue.class)
-        .scheduleFullSync(project, urlMatch, PushOne.ALL_REFS, Set.of(), new ReplicationState(NO_OP), false);
+        .scheduleFullSync(
+            project, urlMatch, PushOne.ALL_REFS, Set.of(), new ReplicationState(NO_OP), false);
 
     assertThat(listWaiting()).hasSize(1);
     tasksStorage

diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorageTest.java b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorageTest.java
index 6d334fa..cb1d541 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorageTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorageTest.java

@@ -358,6 +358,23 @@
   }
 
   @Test
+  public void recoverAllRecoversOnlyUpdatesMatchingFilter() throws Exception {
+    ReplicateRefUpdate otherRefUpdate =
+        ReplicateRefUpdate.create(PROJECT, Set.of(REF), URISH, "otherRemote");
+    UriUpdates otherUriUpdates = new TestUriUpdates(otherRefUpdate);
+    storage.create(REF_UPDATE);
+    storage.create(otherRefUpdate);
+    storage.start(uriUpdates);
+    storage.start(otherUriUpdates);
+
+    storage.recoverAll(r -> REMOTE.equals(r.remote()));
+
+    assertThatStream(storage.streamWaiting()).containsExactly(STORED_REF_UPDATE);
+    assertThatStream(storage.streamRunning())
+        .containsExactly(ReplicateRefUpdate.create(otherRefUpdate, otherRefUpdate.sha1()));
+  }
+
+  @Test
   public void canCompleteMultipleRecoveredUpdates() throws Exception {
     ReplicateRefUpdate updateB =
         ReplicateRefUpdate.create(