diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/AutoReloadConfigDecorator.java b/src/main/java/com/googlesource/gerrit/plugins/replication/AutoReloadConfigDecorator.java
index 2c049fc..5e2c758 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/AutoReloadConfigDecorator.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/AutoReloadConfigDecorator.java
@@ -128,4 +128,9 @@
   public Config getConfig() {
     return currentConfig.getConfig();
   }
+
+  @Override
+  public boolean useLegacyCredentials() {
+    return currentConfig.useLegacyCredentials();
+  }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/AutoReloadSecureCredentialsFactoryDecorator.java b/src/main/java/com/googlesource/gerrit/plugins/replication/AutoReloadSecureCredentialsFactoryDecorator.java
index e9409a6..0bf9026 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/AutoReloadSecureCredentialsFactoryDecorator.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/AutoReloadSecureCredentialsFactoryDecorator.java
@@ -14,49 +14,61 @@
 
 package com.googlesource.gerrit.plugins.replication;
 
-import static com.google.gerrit.common.FileUtil.lastModified;
-
 import com.google.common.flogger.FluentLogger;
+import com.google.gerrit.common.UsedAt;
+import com.google.gerrit.common.UsedAt.Project;
 import com.google.gerrit.server.config.SitePaths;
+import com.google.gerrit.server.securestore.SecureStore;
 import com.google.inject.Inject;
 import com.googlesource.gerrit.plugins.replication.api.ReplicationConfig;
 import java.io.IOException;
-import java.nio.file.Files;
 import java.util.concurrent.atomic.AtomicReference;
 import org.eclipse.jgit.errors.ConfigInvalidException;
 import org.eclipse.jgit.transport.CredentialsProvider;
 
+@UsedAt(Project.PLUGIN_PULL_REPLICATION)
 public class AutoReloadSecureCredentialsFactoryDecorator implements CredentialsFactory {
   private static final FluentLogger logger = FluentLogger.forEnclosingClass();
 
-  private final AtomicReference<SecureCredentialsFactory> secureCredentialsFactory;
-  private volatile long secureCredentialsFactoryLoadTs;
+  private final AtomicReference<CredentialsFactory> secureCredentialsFactory;
+  private final SecureStore secureStore;
   private final SitePaths site;
-  private ReplicationConfig config;
+  private final ReplicationConfig config;
 
   @Inject
-  public AutoReloadSecureCredentialsFactoryDecorator(SitePaths site, ReplicationConfig config)
+  public AutoReloadSecureCredentialsFactoryDecorator(
+      SitePaths site, SecureStore secureStore, ReplicationConfig config)
       throws ConfigInvalidException, IOException {
     this.site = site;
+    this.secureStore = secureStore;
     this.config = config;
-    this.secureCredentialsFactory = new AtomicReference<>(new SecureCredentialsFactory(site));
-    this.secureCredentialsFactoryLoadTs = getSecureConfigLastEditTs();
+    this.secureCredentialsFactory =
+        new AtomicReference<>(newSecureCredentialsFactory(site, secureStore, config));
+    if (config.useLegacyCredentials()) {
+      logger.atWarning().log(
+          "Using legacy credentials in clear text in secure.config. Please encrypt your credentials"
+              + " using 'java -jar gerrit.war passwd' for each remote, remove the"
+              + " gerrit.useLegacyCredentials in replication.config and then reload the replication"
+              + " plugin.");
+    }
   }
 
-  private long getSecureConfigLastEditTs() {
-    if (!Files.exists(site.secure_config)) {
-      return 0L;
+  private static CredentialsFactory newSecureCredentialsFactory(
+      SitePaths site, SecureStore secureStore, ReplicationConfig config)
+      throws ConfigInvalidException, IOException {
+    if (config.useLegacyCredentials()) {
+      return new LegacyCredentialsFactory(site);
     }
-    return lastModified(site.secure_config);
+    return new SecureCredentialsFactory(secureStore);
   }
 
   @Override
   public CredentialsProvider create(String remoteName) {
     try {
       if (needsReload()) {
+        secureStore.reload();
         secureCredentialsFactory.compareAndSet(
-            secureCredentialsFactory.get(), new SecureCredentialsFactory(site));
-        secureCredentialsFactoryLoadTs = getSecureConfigLastEditTs();
+            secureCredentialsFactory.get(), newSecureCredentialsFactory(site, secureStore, config));
         logger.atInfo().log("secure.config reloaded as it was updated on the file system");
       }
     } catch (Exception e) {
@@ -69,7 +81,11 @@
   }
 
   private boolean needsReload() {
-    return config.getConfig().getBoolean("gerrit", "autoReload", false)
-        && getSecureConfigLastEditTs() != secureCredentialsFactoryLoadTs;
+    return config.getConfig().getBoolean("gerrit", "autoReload", false) && secureStore.isOutdated();
+  }
+
+  @Override
+  public boolean validate(String remoteConfigName) {
+    return secureCredentialsFactory.get().validate(remoteConfigName);
   }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/CredentialsFactory.java b/src/main/java/com/googlesource/gerrit/plugins/replication/CredentialsFactory.java
index 3bb64ab..141fb2d 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/CredentialsFactory.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/CredentialsFactory.java
@@ -18,4 +18,8 @@
 public interface CredentialsFactory {
 
   CredentialsProvider create(String remoteName);
+
+  default boolean validate(String remoteConfigName) {
+    return true;
+  }
 }
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 ceaf073..1c2bc94 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/Destination.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/Destination.java
@@ -22,11 +22,11 @@
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Throwables;
 import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Lists;
 import com.google.common.io.Files;
 import com.google.common.net.UrlEscapers;
+import com.google.common.util.concurrent.Striped;
 import com.google.gerrit.common.Nullable;
 import com.google.gerrit.entities.AccountGroup;
 import com.google.gerrit.entities.BranchNameKey;
@@ -71,7 +71,6 @@
 import java.io.IOException;
 import java.net.URISyntaxException;
 import java.util.Collection;
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -82,7 +81,10 @@
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReadWriteLock;
 import java.util.function.Function;
+import java.util.function.Supplier;
 import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 import org.eclipse.jgit.lib.Constants;
@@ -95,19 +97,55 @@
 
 public class Destination {
   private static final NamedFluentLogger repLog = ReplicationQueue.repLog;
+  private static final int MAX_STRIPES = 16;
 
   private static final String PROJECT_NOT_AVAILABLE = "source project %s not available";
 
+  private final CredentialsFactory credentialsFactory;
+
   public interface Factory {
     Destination create(DestinationConfiguration config);
   }
 
+  private static class StateLock {
+    private final Striped<ReadWriteLock> stateLock;
+
+    StateLock(int numStripes) {
+      stateLock = Striped.readWriteLock(numStripes);
+    }
+
+    <V> V withWriteLock(URIish uri, Supplier<V> task) {
+      Lock lock = stateLock.get(uri).writeLock();
+      lock.lock();
+      try {
+        return task.get();
+      } finally {
+        lock.unlock();
+      }
+    }
+
+    <V> V withReadLock(URIish uri, Supplier<V> task) {
+      Lock lock = stateLock.get(uri).readLock();
+      lock.lock();
+      try {
+        return task.get();
+      } finally {
+        lock.unlock();
+      }
+    }
+  }
+
+  private static class RescheduleStatus {
+    boolean isRescheduled;
+    boolean isFailed;
+    RemoteRefUpdate.Status failedStatus;
+  }
+
   private final ReplicationStateListener stateLog;
-  private final Object stateLock = new Object();
+  private final StateLock stateLock;
   // writes are covered by the stateLock, but some reads are still
   // allowed without the lock
-  private final ConcurrentMap<URIish, PushOne> pending = new ConcurrentHashMap<>();
-  private final Map<URIish, PushOne> inFlight = new HashMap<>();
+  private final Queue queue;
   private final PushOne.Factory opFactory;
   private final DeleteProjectTask.Factory deleteProjectFactory;
   private final UpdateHeadTask.Factory updateHeadFactory;
@@ -120,6 +158,7 @@
   private final DestinationConfiguration config;
   private final DynamicItem<EventDispatcher> eventDispatcher;
   private final Provider<ReplicationTasksStorage> replicationTasksStorage;
+  private final UrlDistributionStrategy.Instance urlDistributor;
 
   protected enum RetryReason {
     TRANSPORT_ERROR,
@@ -127,13 +166,13 @@
     REPOSITORY_MISSING;
   }
 
-  public static class QueueInfo {
-    public final ImmutableMap<URIish, PushOne> pending;
-    public final ImmutableMap<URIish, PushOne> inFlight;
+  public static class Queue {
+    public final ConcurrentMap<URIish, PushOne> pending;
+    public final ConcurrentMap<URIish, PushOne> inFlight;
 
-    public QueueInfo(Map<URIish, PushOne> pending, Map<URIish, PushOne> inFlight) {
-      this.pending = ImmutableMap.copyOf(pending);
-      this.inFlight = ImmutableMap.copyOf(inFlight);
+    public Queue() {
+      this.pending = new ConcurrentHashMap<>();
+      this.inFlight = new ConcurrentHashMap<>();
     }
   }
 
@@ -150,15 +189,24 @@
       GroupIncludeCache groupIncludeCache,
       DynamicItem<EventDispatcher> eventDispatcher,
       Provider<ReplicationTasksStorage> rts,
+      CredentialsFactory credentialsFactory,
       @Assisted DestinationConfiguration cfg) {
     this.eventDispatcher = eventDispatcher;
     gitManager = gitRepositoryManager;
+    this.queue = new Queue();
     this.permissionBackend = permissionBackend;
     this.userProvider = userProvider;
     this.projectCache = projectCache;
     this.stateLog = stateLog;
     this.replicationTasksStorage = rts;
+    this.credentialsFactory = credentialsFactory;
     config = cfg;
+    urlDistributor = cfg.getUrlDistributionStrategy().newInstance();
+
+    ImmutableList<String> projects = cfg.getProjects();
+    int numStripes = projects.isEmpty() ? MAX_STRIPES : Math.min(projects.size(), MAX_STRIPES);
+    stateLock = new StateLock(numStripes);
+
     CurrentUser remoteUser;
     if (!cfg.getAuthGroupNames().isEmpty()) {
       ImmutableSet.Builder<AccountGroup.UUID> builder = ImmutableSet.builder();
@@ -217,6 +265,10 @@
     threadScoper = child.getInstance(PerThreadRequestScope.Scoper.class);
   }
 
+  public boolean validate() {
+    return (credentialsFactory == null || credentialsFactory.validate(getRemoteConfigName()));
+  }
+
   private void addRecursiveParents(
       AccountGroup.UUID g,
       ImmutableSet.Builder<AccountGroup.UUID> builder,
@@ -230,10 +282,8 @@
     }
   }
 
-  public QueueInfo getQueueInfo() {
-    synchronized (stateLock) {
-      return new QueueInfo(pending, inFlight);
-    }
+  public Queue getQueue() {
+    return queue;
   }
 
   public void start(WorkQueue workQueue) {
@@ -244,33 +294,31 @@
   public int shutdown() {
     int cnt = 0;
     if (pool != null) {
-      synchronized (stateLock) {
-        int numPending = pending.size();
-        int numInFlight = inFlight.size();
+      int numPending = queue.pending.size();
+      int numInFlight = queue.inFlight.size();
 
-        if (numPending > 0 || numInFlight > 0) {
-          repLog.atWarning().log(
-              "Cancelling replication events (pending=%d, inFlight=%d) for destination %s",
-              numPending, numInFlight, getRemoteConfigName());
+      if (numPending > 0 || numInFlight > 0) {
+        repLog.atWarning().log(
+            "Cancelling replication events (pending=%d, inFlight=%d) for destination %s",
+            numPending, numInFlight, getRemoteConfigName());
 
-          foreachPushOp(
-              pending,
-              push -> {
-                push.cancel();
-                return null;
-              });
-          pending.clear();
-          foreachPushOp(
-              inFlight,
-              push -> {
-                push.setCanceledWhileRunning();
-                return null;
-              });
-          inFlight.clear();
-        }
-        cnt = pool.shutdownNow().size();
-        pool = null;
+        foreachPushOp(
+            queue.pending,
+            push -> {
+              push.cancel();
+              return null;
+            });
+        queue.pending.clear();
+        foreachPushOp(
+            queue.inFlight,
+            push -> {
+              push.setCanceledWhileRunning();
+              return null;
+            });
+        queue.inFlight.clear();
       }
+      cnt = pool.shutdownNow().size();
+      pool = null;
     }
     return cnt;
   }
@@ -279,7 +327,7 @@
     // Callers may modify the provided opsMap concurrently, hence make a defensive copy of the
     // values to loop over them.
     for (PushOne pushOne : ImmutableList.copyOf(opsMap.values())) {
-      pushOneFunction.apply(pushOne);
+      stateLock.withWriteLock(pushOne.getURI(), () -> pushOneFunction.apply(pushOne));
     }
   }
 
@@ -421,10 +469,7 @@
     repLog.atInfo().log("scheduling replication %s:%s => %s", project, refs, uri);
 
     if (!config.replicatePermissions()) {
-      PushOne e;
-      synchronized (stateLock) {
-        e = getPendingPush(uri);
-      }
+      PushOne e = stateLock.withReadLock(uri, () -> getPendingPush(uri));
       if (e == null) {
         try (Repository git = gitManager.openRepository(project)) {
           try {
@@ -446,40 +491,43 @@
     }
 
     ImmutableSet<String> refsToSchedule = toSchedule.build();
-    PushOne task;
-    synchronized (stateLock) {
-      task = getPendingPush(uri);
-      if (task == null) {
-        task = opFactory.create(project, uri);
-        task.addRefBatch(refsToSchedule);
-        task.addState(refsToSchedule, state);
-        @SuppressWarnings("unused")
-        ScheduledFuture<?> ignored =
-            pool.schedule(task, now ? 0 : config.getDelay(), TimeUnit.SECONDS);
-        pending.put(uri, task);
-        repLog.atInfo().log(
-            "scheduled %s:%s => %s to run %s",
-            project, refsToSchedule, task, now ? "now" : "after " + config.getDelay() + "s");
-      } else {
-        boolean added = task.addRefBatch(refsToSchedule);
-        task.addState(refsToSchedule, state);
-        String message = "consolidated %s:%s => %s with an existing pending push";
-        if (added || !fromStorage) {
-          repLog.atInfo().log(message, project, refsToSchedule, task);
-        } else {
-          repLog.atFine().log(message, project, refsToSchedule, task);
-        }
-      }
-      for (String ref : refsToSchedule) {
-        state.increasePushTaskCount(project.get(), ref);
-      }
-    }
+    PushOne task =
+        stateLock.withWriteLock(
+            uri,
+            () -> {
+              PushOne t = getPendingPush(uri);
+              if (t == null) {
+                t = opFactory.create(project, uri);
+                t.addRefBatch(refsToSchedule);
+                t.addState(refsToSchedule, state);
+                @SuppressWarnings("unused")
+                ScheduledFuture<?> ignored =
+                    pool.schedule(t, now ? 0 : config.getDelay(), TimeUnit.SECONDS);
+                queue.pending.put(uri, t);
+                repLog.atInfo().log(
+                    "scheduled %s:%s => %s to run %s",
+                    project, refsToSchedule, t, now ? "now" : "after " + config.getDelay() + "s");
+              } else {
+                boolean added = t.addRefBatch(refsToSchedule);
+                t.addState(refsToSchedule, state);
+                String message = "consolidated %s:%s => %s with an existing pending push";
+                if (added || !fromStorage) {
+                  repLog.atInfo().log(message, project, refsToSchedule, t);
+                } else {
+                  repLog.atFine().log(message, project, refsToSchedule, t);
+                }
+              }
+              for (String ref : refsToSchedule) {
+                state.increasePushTaskCount(project.get(), ref);
+              }
+              return t;
+            });
     postReplicationScheduledEvent(task, refsToSchedule);
   }
 
   @Nullable
   private PushOne getPendingPush(URIish uri) {
-    PushOne e = pending.get(uri);
+    PushOne e = queue.pending.get(uri);
     if (e != null && !e.wasCanceled()) {
       return e;
     }
@@ -487,12 +535,14 @@
   }
 
   void pushWasCanceled(PushOne pushOp) {
-    Set<ImmutableSet<String>> notAttemptedRefs = Collections.emptySet();
-    synchronized (stateLock) {
-      URIish uri = pushOp.getURI();
-      pending.remove(uri);
-      notAttemptedRefs = pushOp.getRefs();
-    }
+    Set<ImmutableSet<String>> notAttemptedRefs =
+        stateLock.withWriteLock(
+            pushOp.getURI(),
+            () -> {
+              URIish uri = pushOp.getURI();
+              queue.pending.remove(uri);
+              return pushOp.getRefs();
+            });
     pushOp.notifyNotAttempted(notAttemptedRefs);
   }
 
@@ -532,124 +582,130 @@
    * @param pushOp The PushOp instance to be scheduled.
    */
   void reschedule(PushOne pushOp, RetryReason reason) {
-    boolean isRescheduled = false;
-    boolean isFailed = false;
-    RemoteRefUpdate.Status failedStatus = null;
+    RescheduleStatus status = new RescheduleStatus();
+    stateLock.withWriteLock(
+        pushOp.getURI(),
+        () -> {
+          URIish uri = pushOp.getURI();
+          PushOne pendingPushOp = getPendingPush(uri);
 
-    synchronized (stateLock) {
-      URIish uri = pushOp.getURI();
-      PushOne pendingPushOp = getPendingPush(uri);
+          if (pendingPushOp != null) {
+            // There is one PushOp instance already pending to same URI.
 
-      if (pendingPushOp != null) {
-        // There is one PushOp instance already pending to same URI.
+            if (pendingPushOp.isRetrying()) {
+              // The one pending is one already retrying, so it should
+              // maintain it and add to it the refs of the one passed
+              // as parameter to the method.
 
-        if (pendingPushOp.isRetrying()) {
-          // The one pending is one already retrying, so it should
-          // maintain it and add to it the refs of the one passed
-          // as parameter to the method.
+              // This scenario would happen if a PushOp has started running
+              // and then before it failed due transport exception, another
+              // one to same URI started. The first one would fail and would
+              // be rescheduled, being present in pending list. When the
+              // second one fails, it will also be rescheduled and then,
+              // here, find out replication to its URI is already pending
+              // for retry (blocking).
+              pendingPushOp.addRefBatches(pushOp.getRefs());
+              pendingPushOp.addStates(pushOp.getStates());
+              pushOp.removeStates();
 
-          // This scenario would happen if a PushOp has started running
-          // and then before it failed due transport exception, another
-          // one to same URI started. The first one would fail and would
-          // be rescheduled, being present in pending list. When the
-          // second one fails, it will also be rescheduled and then,
-          // here, find out replication to its URI is already pending
-          // for retry (blocking).
-          pendingPushOp.addRefBatches(pushOp.getRefs());
-          pendingPushOp.addStates(pushOp.getStates());
-          pushOp.removeStates();
-
-        } else {
-          // The one pending is one that is NOT retrying, it was just
-          // scheduled believing no problem would happen. The one pending
-          // should be canceled, and this is done by setting its canceled
-          // flag, removing it from pending list, and adding its refs to
-          // the pushOp instance that should then, later, in this method,
-          // be scheduled for retry.
-
-          // Notice that the PushOp found pending will start running and,
-          // when notifying it is starting (with pending lock protection),
-          // it will see it was canceled and then it will do nothing with
-          // pending list and it will not execute its run implementation.
-          pendingPushOp.canceledByReplication();
-          pending.remove(uri);
-
-          pushOp.addRefBatches(pendingPushOp.getRefs());
-          pushOp.addStates(pendingPushOp.getStates());
-          pendingPushOp.removeStates();
-        }
-      }
-
-      if (pendingPushOp == null || !pendingPushOp.isRetrying()) {
-        pending.put(uri, pushOp);
-        switch (reason) {
-          case COLLISION:
-            @SuppressWarnings("unused")
-            ScheduledFuture<?> ignored =
-                pool.schedule(pushOp, config.getRescheduleDelay(), TimeUnit.SECONDS);
-            break;
-          case TRANSPORT_ERROR:
-          case REPOSITORY_MISSING:
-          default:
-            failedStatus =
-                RetryReason.REPOSITORY_MISSING.equals(reason)
-                    ? NON_EXISTING
-                    : REJECTED_OTHER_REASON;
-            isFailed = true;
-            if (pushOp.setToRetry()) {
-              isRescheduled = true;
-              replicationTasksStorage.get().reset(pushOp);
-              @SuppressWarnings("unused")
-              ScheduledFuture<?> ignored2 =
-                  pool.schedule(pushOp, config.getRetryDelay(), TimeUnit.MINUTES);
             } else {
-              pushOp.canceledByReplication();
-              pushOp.retryDone();
-              pending.remove(uri);
-              stateLog.error(
-                  "Push to " + pushOp.getURI() + " cancelled after maximum number of retries",
-                  pushOp.getStatesAsArray());
+              // The one pending is one that is NOT retrying, it was just
+              // scheduled believing no problem would happen. The one pending
+              // should be canceled, and this is done by setting its canceled
+              // flag, removing it from pending list, and adding its refs to
+              // the pushOp instance that should then, later, in this method,
+              // be scheduled for retry.
+
+              // Notice that the PushOp found pending will start running and,
+              // when notifying it is starting (with pending lock protection),
+              // it will see it was canceled and then it will do nothing with
+              // pending list and it will not execute its run implementation.
+              pendingPushOp.canceledByReplication();
+              queue.pending.remove(uri);
+
+              pushOp.addRefBatches(pendingPushOp.getRefs());
+              pushOp.addStates(pendingPushOp.getStates());
+              pendingPushOp.removeStates();
             }
-            break;
-        }
-      }
+          }
+
+          if (pendingPushOp == null || !pendingPushOp.isRetrying()) {
+            queue.pending.put(uri, pushOp);
+            switch (reason) {
+              case COLLISION:
+                @SuppressWarnings("unused")
+                ScheduledFuture<?> ignored =
+                    pool.schedule(pushOp, config.getRescheduleDelay(), TimeUnit.SECONDS);
+                break;
+              case TRANSPORT_ERROR:
+              case REPOSITORY_MISSING:
+              default:
+                status.failedStatus =
+                    RetryReason.REPOSITORY_MISSING.equals(reason)
+                        ? NON_EXISTING
+                        : REJECTED_OTHER_REASON;
+                status.isFailed = true;
+                if (pushOp.setToRetry()) {
+                  status.isRescheduled = true;
+                  replicationTasksStorage.get().reset(pushOp);
+                  @SuppressWarnings("unused")
+                  ScheduledFuture<?> ignored2 =
+                      pool.schedule(pushOp, config.getRetryDelay(), TimeUnit.MINUTES);
+                } else {
+                  pushOp.canceledByReplication();
+                  pushOp.retryDone();
+                  queue.pending.remove(uri);
+                  stateLog.error(
+                      "Push to " + pushOp.getURI() + " cancelled after maximum number of retries",
+                      pushOp.getStatesAsArray());
+                }
+                break;
+            }
+          }
+          return status;
+        });
+
+    if (status.isFailed) {
+      postReplicationFailedEvent(pushOp, status.failedStatus);
     }
-    if (isFailed) {
-      postReplicationFailedEvent(pushOp, failedStatus);
-    }
-    if (isRescheduled) {
+    if (status.isRescheduled) {
       postReplicationScheduledEvent(pushOp);
     }
   }
 
   RunwayStatus requestRunway(PushOne op) {
-    synchronized (stateLock) {
-      if (op.wasCanceled()) {
-        return RunwayStatus.canceled();
-      }
-      pending.remove(op.getURI());
-      PushOne inFlightOp = inFlight.get(op.getURI());
-      if (inFlightOp != null) {
-        return RunwayStatus.denied(inFlightOp.getId());
-      }
-      op.notifyNotAttempted(op.setStartedRefs(replicationTasksStorage.get().start(op)));
-      inFlight.put(op.getURI(), op);
-    }
-    return RunwayStatus.allowed();
+    return stateLock.withWriteLock(
+        op.getURI(),
+        () -> {
+          if (op.wasCanceled()) {
+            return RunwayStatus.canceled();
+          }
+          queue.pending.remove(op.getURI());
+          PushOne inFlightOp = queue.inFlight.get(op.getURI());
+          if (inFlightOp != null) {
+            return RunwayStatus.denied(inFlightOp.getId());
+          }
+          op.notifyNotAttempted(op.setStartedRefs(replicationTasksStorage.get().start(op)));
+          queue.inFlight.put(op.getURI(), op);
+          return RunwayStatus.allowed();
+        });
   }
 
   void notifyFinished(PushOne op) {
-    synchronized (stateLock) {
-      if (!op.isRetrying()) {
-        replicationTasksStorage.get().finish(op);
-      }
-      inFlight.remove(op.getURI());
-    }
+    stateLock.withWriteLock(
+        op.getURI(),
+        () -> {
+          if (!op.isRetrying()) {
+            replicationTasksStorage.get().finish(op);
+          }
+          queue.inFlight.remove(op.getURI());
+          return null;
+        });
   }
 
   public Map<ReplicateRefUpdate, String> getTaskNamesByReplicateRefUpdate() {
     Map<ReplicateRefUpdate, String> taskNameByReplicateRefUpdate = new HashMap<>();
-    for (PushOne push : pending.values()) {
+    for (PushOne push : queue.pending.values()) {
       String taskName = push.toString();
       for (ReplicateRefUpdate refUpdate : push.getReplicateRefUpdates()) {
         taskNameByReplicateRefUpdate.put(refUpdate, taskName);
@@ -746,6 +802,14 @@
     return r;
   }
 
+  List<URIish> getDistributedUris(Project.NameKey project, String urlMatch) {
+    return getDistributedUris(getURIs(project, urlMatch));
+  }
+
+  List<URIish> getDistributedUris(List<URIish> candidates) {
+    return urlDistributor.select(candidates);
+  }
+
   URIish getURI(URIish template, Project.NameKey project) throws URISyntaxException {
     return getURI(template, project, config.getRemoteNameStyle(), config.isSingleProjectMatch());
   }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/DestinationConfiguration.java b/src/main/java/com/googlesource/gerrit/plugins/replication/DestinationConfiguration.java
index 41ab4c4..e77b3cf 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/DestinationConfiguration.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/DestinationConfiguration.java
@@ -56,6 +56,7 @@
   private final int slowLatencyThreshold;
   private final Supplier<Integer> pushBatchSize;
   private final ImmutableList<Pattern> excludedRefsPattern;
+  private final UrlDistributionStrategy urlDistributionStrategy;
 
   protected DestinationConfiguration(RemoteConfig remoteConfig, Config cfg) {
     this.remoteConfig = remoteConfig;
@@ -122,6 +123,9 @@
               return 0;
             });
     excludedRefsPattern = getExcludedRefsPattern(cfg, name);
+    urlDistributionStrategy =
+        UrlDistributionStrategy.fromConfig(
+            cfg.getString("remote", name, "urlDistributionStrategy"));
   }
 
   @Override
@@ -227,6 +231,11 @@
     return excludedRefsPattern;
   }
 
+  @Override
+  public UrlDistributionStrategy getUrlDistributionStrategy() {
+    return urlDistributionStrategy;
+  }
+
   private ImmutableList<Pattern> getExcludedRefsPattern(Config cfg, String name) {
     List<Pattern> patterns = new ArrayList<>();
     for (String regex : cfg.getStringList("remote", name, "excludedRefsPattern")) {
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/DestinationsCollection.java b/src/main/java/com/googlesource/gerrit/plugins/replication/DestinationsCollection.java
index 8a41945..82f33d7 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/DestinationsCollection.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/DestinationsCollection.java
@@ -35,6 +35,7 @@
 import com.google.inject.Inject;
 import com.google.inject.Provider;
 import com.google.inject.Singleton;
+import com.googlesource.gerrit.plugins.replication.Destination.Queue;
 import com.googlesource.gerrit.plugins.replication.api.ReplicationConfig;
 import com.googlesource.gerrit.plugins.replication.api.ReplicationConfig.FilterType;
 import java.net.URISyntaxException;
@@ -98,6 +99,7 @@
       }
 
       boolean adminURLUsed = false;
+      List<URIish> validUris = new ArrayList<>();
 
       for (String url : config.getAdminUrls()) {
         if (Strings.isNullOrEmpty(url)) {
@@ -127,15 +129,16 @@
             continue;
           }
         }
-        uris.put(config, uri);
+        validUris.add(uri);
         adminURLUsed = true;
       }
 
       if (!adminURLUsed) {
         for (URIish uri : config.getURIs(projectName, "*")) {
-          uris.put(config, uri);
+          validUris.add(uri);
         }
       }
+      config.getDistributedUris(validUris).forEach(uri -> uris.put(config, uri));
     }
     return uris;
   }
@@ -180,6 +183,12 @@
   public synchronized void startup(WorkQueue workQueue) {
     shuttingDown = false;
     for (Destination cfg : destinations) {
+      if (!cfg.validate()) {
+        throw new IllegalStateException(
+            "Unable to start replication plugin because remote "
+                + cfg.getRemoteConfigName()
+                + " is not valid");
+      }
       cfg.start(workQueue);
     }
   }
@@ -221,8 +230,9 @@
     if (drainQueueAttempts == 0) {
       return;
     }
-    int pending = destination.getQueueInfo().pending.size();
-    int inFlight = destination.getQueueInfo().inFlight.size();
+    Queue queue = destination.getQueue();
+    int pending = queue.pending.size();
+    int inFlight = queue.inFlight.size();
     while ((inFlight > 0 || pending > 0) && drainQueueAttempts > 0) {
       try {
         logger.atInfo().log(
@@ -233,8 +243,8 @@
         logger.atWarning().withCause(ie).log(
             "Wait for replication events to drain has been interrupted");
       }
-      pending = destination.getQueueInfo().pending.size();
-      inFlight = destination.getQueueInfo().inFlight.size();
+      pending = queue.pending.size();
+      inFlight = queue.inFlight.size();
       drainQueueAttempts--;
     }
     if (pending > 0 || inFlight > 0) {
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/FanoutConfigResource.java b/src/main/java/com/googlesource/gerrit/plugins/replication/FanoutConfigResource.java
index 6932c3c..b8392c7 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/FanoutConfigResource.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/FanoutConfigResource.java
@@ -17,17 +17,20 @@
 import static com.google.common.io.Files.getNameWithoutExtension;
 import static java.nio.charset.StandardCharsets.UTF_8;
 
-import com.google.common.collect.Lists;
 import com.google.common.flogger.FluentLogger;
 import com.google.common.hash.Hasher;
 import com.google.common.hash.Hashing;
 import com.google.gerrit.server.config.SitePaths;
 import com.google.inject.Inject;
+import java.io.File;
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
-import java.util.Optional;
+import java.util.List;
 import java.util.Set;
+import java.util.function.Function;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import org.eclipse.jgit.errors.ConfigInvalidException;
 import org.eclipse.jgit.internal.storage.file.FileSnapshot;
@@ -40,6 +43,7 @@
   private static final FluentLogger logger = FluentLogger.forEnclosingClass();
 
   private final Path remoteConfigsDirPath;
+  private final Config fanoutConfig;
 
   @Inject
   FanoutConfigResource(SitePaths site) throws IOException, ConfigInvalidException {
@@ -48,16 +52,59 @@
     removeRemotes(config);
 
     try (Stream<Path> files = Files.list(remoteConfigsDirPath)) {
-      files
-          .filter(Files::isRegularFile)
-          .filter(FanoutConfigResource::isConfig)
-          .map(FanoutConfigResource::loadConfig)
-          .filter(Optional::isPresent)
-          .map(Optional::get)
-          .filter(FanoutConfigResource::isValid)
-          .forEach(cfg -> addRemoteConfig(cfg, config));
-    } catch (IllegalStateException e) {
-      throw new ConfigInvalidException(e.getMessage());
+      Stream<String> mergedConfigs =
+          files
+              .filter(Files::isRegularFile)
+              .filter(FanoutConfigResource::isConfig)
+              .flatMap(path -> getConfigLines(path, line -> addRemoteSubsection(line, path)));
+
+      fanoutConfig = new Config(config);
+      fanoutConfig.fromText(mergedConfigs.collect(Collectors.joining("\n")));
+    }
+  }
+
+  private static String addRemoteSubsection(String line, Path path) {
+    return line.contains("[remote]")
+        ? line.replace(
+            "[remote]", "[remote \"" + getNameWithoutExtension(path.toFile().getName()) + "\"]")
+        : line;
+  }
+
+  private static Stream<String> getConfigLines(
+      Path path, Function<String, String> configLineMapping) {
+    try {
+      List<String> configLines = Files.readAllLines(path);
+      if (!isValid(path, configLines)) {
+        return Stream.empty();
+      }
+      return configLines.stream().map(configLineMapping);
+    } catch (IOException e) {
+      logger.atSevere().withCause(e).log("Unable to access replication config %s", path);
+      return Stream.empty();
+    }
+  }
+
+  @Override
+  public void update(Config updates) throws IOException {
+    super.update(filterOutRemotes(updates));
+
+    Set<String> remotes = updates.getSubsections("remote");
+    for (String remote : remotes) {
+      File remoteFile = remoteConfigsDirPath.resolve(remote + ".config").toFile();
+      FileBasedConfig remoteConfig = new FileBasedConfig(remoteFile, FS.DETECTED);
+      try {
+        remoteConfig.load();
+      } catch (ConfigInvalidException e) {
+        throw new IOException(
+            String.format("Cannot parse configuration file: %s", remoteFile.getAbsoluteFile()), e);
+      }
+      Set<String> options = updates.getNames("remote", remote);
+      for (String option : options) {
+        List<String> values = List.of(updates.getStringList("remote", remote, option));
+        remoteConfig.setStringList("remote", null, option, values);
+        config.setStringList("remote", remote, option, values);
+      }
+      remoteConfig.save();
     }
   }
 
@@ -75,42 +122,67 @@
     }
   }
 
-  private static void addRemoteConfig(FileBasedConfig source, Config destination) {
-    String remoteName = getNameWithoutExtension(source.getFile().getName());
-    for (String name : source.getNames("remote")) {
-      destination.setStringList(
-          "remote",
-          remoteName,
-          name,
-          Lists.newArrayList(source.getStringList("remote", null, name)));
-    }
+  private static Config filterOutRemotes(Config config) {
+    Config filteredConfig = new Config();
+    Set<String> sections = config.getSections();
+    sections.stream()
+        .filter(Predicate.not("remote"::equalsIgnoreCase))
+        .forEach(
+            section -> {
+              config
+                  .getNames(section)
+                  .forEach(
+                      sectionName ->
+                          filteredConfig.setStringList(
+                              section,
+                              null,
+                              sectionName,
+                              List.of(config.getStringList(section, null, sectionName))));
+              config
+                  .getSubsections(section)
+                  .forEach(
+                      subsection ->
+                          config
+                              .getNames(section, subsection)
+                              .forEach(
+                                  name ->
+                                      filteredConfig.setStringList(
+                                          section,
+                                          subsection,
+                                          name,
+                                          List.of(
+                                              config.getStringList(section, subsection, name)))));
+            });
+    return filteredConfig;
   }
 
-  private static boolean isValid(Config cfg) {
-    if (cfg.getSections().size() != 1 || !cfg.getSections().contains("remote")) {
-      logger.atSevere().log(
-          "Remote replication configuration file %s must contain only one remote section.", cfg);
-      return false;
-    }
-    if (cfg.getSubsections("remote").size() > 0) {
-      logger.atSevere().log(
-          "Remote replication configuration file %s cannot contain remote subsections.", cfg);
-      return false;
+  private static boolean isValid(Path path, List<String> remoteConfigLines) {
+    int numRemoteSections = 0;
+    int numRemoteSectionsWithName = 0;
+    boolean valid = true;
+    for (String configLine : remoteConfigLines) {
+      if (configLine.contains("[remote]")) {
+        numRemoteSections++;
+      }
+
+      if (configLine.contains("[remote \"")) {
+        numRemoteSectionsWithName++;
+      }
     }
 
-    return true;
-  }
-
-  private static Optional<FileBasedConfig> loadConfig(Path path) {
-    FileBasedConfig cfg = new FileBasedConfig(path.toFile(), FS.DETECTED);
-    try {
-      cfg.load();
-    } catch (IOException | ConfigInvalidException e) {
-      logger.atSevere().withCause(e).log(
-          "Cannot load remote replication configuration file %s.", path);
-      return Optional.empty();
+    if (numRemoteSectionsWithName > 0) {
+      logger.atSevere().log(
+          "Remote replication configuration file %s cannot contain remote subsections.", path);
+      valid = false;
     }
-    return Optional.of(cfg);
+
+    if (numRemoteSections != 1) {
+      logger.atSevere().log(
+          "Remote replication configuration file %s must contain only one remote section.", path);
+      valid = false;
+    }
+
+    return valid;
   }
 
   private static boolean isConfig(Path p) {
@@ -142,4 +214,9 @@
       return parentVersion;
     }
   }
+
+  @Override
+  public Config getConfig() {
+    return fanoutConfig;
+  }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/FileConfigResource.java b/src/main/java/com/googlesource/gerrit/plugins/replication/FileConfigResource.java
index baccb83..7a14cd1 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/FileConfigResource.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/FileConfigResource.java
@@ -17,6 +17,7 @@
 import static com.googlesource.gerrit.plugins.replication.ReplicationQueue.repLog;
 
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.Lists;
 import com.google.gerrit.common.UsedAt;
 import com.google.gerrit.common.UsedAt.Project;
 import com.google.gerrit.server.config.SitePaths;
@@ -24,6 +25,7 @@
 import com.googlesource.gerrit.plugins.replication.api.ConfigResource;
 import java.io.IOException;
 import java.nio.file.Path;
+import java.util.List;
 import org.eclipse.jgit.errors.ConfigInvalidException;
 import org.eclipse.jgit.lib.Config;
 import org.eclipse.jgit.storage.file.FileBasedConfig;
@@ -54,6 +56,25 @@
   }
 
   @Override
+  public void update(Config updates) throws IOException {
+    for (String section : updates.getSections()) {
+      for (String subsection : updates.getSubsections(section)) {
+        for (String name : updates.getNames(section, subsection, true)) {
+          List<String> values =
+              Lists.newArrayList(updates.getStringList(section, subsection, name));
+          config.setStringList(section, subsection, name, values);
+        }
+      }
+
+      for (String name : updates.getNames(section, true)) {
+        List<String> values = Lists.newArrayList(updates.getStringList(section, null, name));
+        config.setStringList(section, null, name, values);
+      }
+    }
+    config.save();
+  }
+
+  @Override
   public String getVersion() {
     return Long.toString(config.getFile().lastModified());
   }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/LegacyCredentialsFactory.java b/src/main/java/com/googlesource/gerrit/plugins/replication/LegacyCredentialsFactory.java
new file mode 100644
index 0000000..a89c936
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/LegacyCredentialsFactory.java
@@ -0,0 +1,66 @@
+// Copyright (C) 2011 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.googlesource.gerrit.plugins.replication;
+
+import com.google.gerrit.server.config.SitePaths;
+import com.google.inject.Inject;
+import java.io.IOException;
+import java.util.Objects;
+import org.eclipse.jgit.errors.ConfigInvalidException;
+import org.eclipse.jgit.lib.Config;
+import org.eclipse.jgit.storage.file.FileBasedConfig;
+import org.eclipse.jgit.transport.CredentialsProvider;
+import org.eclipse.jgit.transport.UsernamePasswordCredentialsProvider;
+import org.eclipse.jgit.util.FS;
+
+/**
+ * Looks up a remote's password in secure.config.
+ *
+ * @deprecated This class is not secure and should no longer be used; it was allowing to record and
+ *     read credentials in clear text in secure.config even though the Gerrit administrator
+ *     installed a proper SecureStore implementation such as the secure-config.jar libModule.
+ */
+@Deprecated(forRemoval = true)
+class LegacyCredentialsFactory implements CredentialsFactory {
+  private final Config config;
+
+  @Inject
+  public LegacyCredentialsFactory(SitePaths site) throws ConfigInvalidException, IOException {
+    config = load(site);
+  }
+
+  private static Config load(SitePaths site) throws ConfigInvalidException, IOException {
+    FileBasedConfig cfg = new FileBasedConfig(site.secure_config.toFile(), FS.DETECTED);
+    if (cfg.getFile().exists() && cfg.getFile().length() > 0) {
+      try {
+        cfg.load();
+      } catch (ConfigInvalidException e) {
+        throw new ConfigInvalidException(
+            String.format("Config file %s is invalid: %s", cfg.getFile(), e.getMessage()), e);
+      } catch (IOException e) {
+        throw new IOException(
+            String.format("Cannot read %s: %s", cfg.getFile(), e.getMessage()), e);
+      }
+    }
+    return cfg;
+  }
+
+  @Override
+  public CredentialsProvider create(String remoteName) {
+    String user = Objects.toString(config.getString("remote", remoteName, "username"), "");
+    String pass = Objects.toString(config.getString("remote", remoteName, "password"), "");
+    return new UsernamePasswordCredentialsProvider(user, pass);
+  }
+}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/ListCommand.java b/src/main/java/com/googlesource/gerrit/plugins/replication/ListCommand.java
index 8f4e0a1..213d0de 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/ListCommand.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/ListCommand.java
@@ -90,7 +90,7 @@
         addProperty(obj, "AdminUrl", d.getAdminUrls());
         addProperty(obj, "AuthGroup", d.getAuthGroupNames());
         addProperty(obj, "Project", d.getProjects());
-        Destination.QueueInfo q = d.getQueueInfo();
+        Destination.Queue q = d.getQueue();
         addQueueDetails(obj, "InFlight", q.inFlight.values());
         addQueueDetails(obj, "Pending", q.pending.values());
       }
@@ -115,7 +115,7 @@
           out.append("Project: ").append(project).append("\n");
         }
 
-        Destination.QueueInfo q = d.getQueueInfo();
+        Destination.Queue q = d.getQueue();
         out.append("In Flight: ").append(q.inFlight.size()).append("\n");
         addQueueDetails(out, q.inFlight.values());
         out.append("Pending: ").append(q.pending.size()).append("\n");
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/MergedConfigResource.java b/src/main/java/com/googlesource/gerrit/plugins/replication/MergedConfigResource.java
index 43d90a5..58a10f0 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/MergedConfigResource.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/MergedConfigResource.java
@@ -26,6 +26,7 @@
 import com.google.inject.util.Providers;
 import com.googlesource.gerrit.plugins.replication.api.ConfigResource;
 import com.googlesource.gerrit.plugins.replication.api.ReplicationConfigOverrides;
+import java.io.IOException;
 import java.util.function.Supplier;
 import org.eclipse.jgit.errors.ConfigInvalidException;
 import org.eclipse.jgit.lib.Config;
@@ -77,7 +78,15 @@
     return baseVersion + overrides.get().getVersion();
   }
 
-  private boolean noOverrides() {
+  boolean noOverrides() {
     return overrides == null || overrides.get() == null;
   }
+
+  void update(Config remotesConfig) throws IOException {
+    if (noOverrides()) {
+      base.get().update(remotesConfig);
+    } else {
+      overrides.get().update(remotesConfig);
+    }
+  }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/OnStartStop.java b/src/main/java/com/googlesource/gerrit/plugins/replication/OnStartStop.java
index fc80781..c779857 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/OnStartStop.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/OnStartStop.java
@@ -22,6 +22,7 @@
 import com.google.inject.Inject;
 import com.googlesource.gerrit.plugins.replication.PushResultProcessing.GitUpdateProcessing;
 import com.googlesource.gerrit.plugins.replication.api.ReplicationConfig;
+import java.util.Set;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
@@ -53,7 +54,7 @@
       ReplicationState state = new ReplicationState(new GitUpdateProcessing(eventDispatcher.get()));
       pushAllFuture.set(
           pushAll
-              .create(null, ReplicationFilter.all(), state, false)
+              .create(null, Set.of(), ReplicationFilter.all(), state, false)
               .schedule(30, TimeUnit.SECONDS));
     }
   }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/PushAll.java b/src/main/java/com/googlesource/gerrit/plugins/replication/PushAll.java
index 4f60319..8958a2a 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/PushAll.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/PushAll.java
@@ -20,6 +20,7 @@
 import com.google.gerrit.server.project.ProjectCache;
 import com.google.inject.Inject;
 import com.google.inject.assistedinject.Assisted;
+import java.util.Set;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 
@@ -27,13 +28,19 @@
   private final ReplicationStateListener stateLog;
 
   public interface Factory {
-    PushAll create(String urlMatch, ReplicationFilter filter, ReplicationState state, boolean now);
+    PushAll create(
+        String urlMatch,
+        Set<String> remotesToConsider,
+        ReplicationFilter filter,
+        ReplicationState state,
+        boolean now);
   }
 
   private final WorkQueue workQueue;
   private final ProjectCache projectCache;
   private final ReplicationQueue replication;
   private final String urlMatch;
+  private final Set<String> remotesToConsider;
   private final ReplicationFilter filter;
   private final ReplicationState state;
   private final boolean now;
@@ -45,6 +52,7 @@
       ReplicationQueue rq,
       ReplicationStateListeners stateLog,
       @Assisted @Nullable String urlMatch,
+      @Assisted Set<String> remotesToConsider,
       @Assisted ReplicationFilter filter,
       @Assisted ReplicationState state,
       @Assisted boolean now) {
@@ -53,6 +61,7 @@
     this.replication = rq;
     this.stateLog = stateLog;
     this.urlMatch = urlMatch;
+    this.remotesToConsider = remotesToConsider;
     this.filter = filter;
     this.state = state;
     this.now = now;
@@ -67,7 +76,7 @@
     try {
       for (Project.NameKey nameKey : projectCache.all()) {
         if (filter.matches(nameKey)) {
-          replication.scheduleFullSync(nameKey, urlMatch, state, now);
+          replication.scheduleFullSync(nameKey, urlMatch, remotesToConsider, state, now);
         }
       }
     } catch (Exception e) {
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/PushOne.java b/src/main/java/com/googlesource/gerrit/plugins/replication/PushOne.java
index c49890a..f9e1081 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/PushOne.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/PushOne.java
@@ -69,6 +69,7 @@
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 import org.eclipse.jgit.errors.NoRemoteRepositoryException;
 import org.eclipse.jgit.errors.NotSupportedException;
 import org.eclipse.jgit.errors.RemoteRepositoryException;
@@ -79,7 +80,6 @@
 import org.eclipse.jgit.lib.ObjectId;
 import org.eclipse.jgit.lib.Ref;
 import org.eclipse.jgit.lib.Repository;
-import org.eclipse.jgit.transport.CredentialsProvider;
 import org.eclipse.jgit.transport.FetchConnection;
 import org.eclipse.jgit.transport.PushResult;
 import org.eclipse.jgit.transport.RefSpec;
@@ -119,7 +119,7 @@
   private final Destination pool;
   private final RemoteConfig config;
   private final ReplicationConfig replConfig;
-  private final CredentialsProvider credentialsProvider;
+  private final CredentialsFactory credentialsFactory;
   private final PerThreadRequestScope.Scoper threadScoper;
 
   private final Project.NameKey projectName;
@@ -166,7 +166,7 @@
     pool = p;
     config = c;
     replConfig = rc;
-    credentialsProvider = cpFactory.create(c.getName());
+    credentialsFactory = cpFactory;
     threadScoper = ts;
     projectName = d;
     uri = u;
@@ -247,18 +247,14 @@
   protected String getLimitedRefs() {
     Set<ImmutableSet<String>> refs = getRefs();
     int maxRefsToShow = replConfig.getMaxRefsToShow();
-    if (maxRefsToShow == 0) {
-      maxRefsToShow = refs.size();
-    }
-    String refsString =
-        refs.stream()
-            .flatMap(Collection::stream)
-            .limit(maxRefsToShow)
-            .collect(Collectors.joining(" "));
-    int hiddenRefs = refs.size() - maxRefsToShow;
-    if (hiddenRefs > 0) {
-      refsString += " (+" + hiddenRefs + ")";
-    }
+
+    Stream<String> refsStream = refs.stream().flatMap(Collection::stream);
+
+    Stream<String> refsFiltered =
+        (maxRefsToShow == 0) ? refsStream : refsStream.limit(maxRefsToShow);
+
+    String refsString = refsFiltered.collect(Collectors.joining(" "));
+
     return "[" + refsString + "]";
   }
 
@@ -563,7 +559,7 @@
 
   private PushResult pushVia(Transport tn) throws IOException, PermissionBackendException {
     tn.applyConfig(config);
-    tn.setCredentialsProvider(credentialsProvider);
+    tn.setCredentialsProvider(credentialsFactory.create(config.getName()));
 
     List<RemoteRefUpdate> todo = generateUpdates(tn);
     if (todo.isEmpty()) {
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/RemoteConfiguration.java b/src/main/java/com/googlesource/gerrit/plugins/replication/RemoteConfiguration.java
index 67d23b0..c05058b 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/RemoteConfiguration.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/RemoteConfiguration.java
@@ -152,4 +152,9 @@
   default ImmutableList<Pattern> excludedRefsPattern() {
     return ImmutableList.of();
   }
+
+  /** Returns the URL distribution strategy configured for this remote. */
+  default UrlDistributionStrategy getUrlDistributionStrategy() {
+    return UrlDistributionStrategy.ALL;
+  }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationConfigImpl.java b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationConfigImpl.java
index 1223db0..8f9b805 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationConfigImpl.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationConfigImpl.java
@@ -31,6 +31,7 @@
 
   private final SitePaths site;
   private final MergedConfigResource configResource;
+  private final boolean useLegacyCredentials;
   private boolean replicateAllOnPluginStart;
   private boolean defaultForceUpdate;
   private int maxRefsToLog;
@@ -62,6 +63,7 @@
                 DEFAULT_SSH_CONNECTION_TIMEOUT_MS,
                 MILLISECONDS);
     this.pluginDataDir = pluginDataDir;
+    this.useLegacyCredentials = config.getBoolean("gerrit", "useLegacyCredentials", false);
   }
 
   @Nullable
@@ -125,6 +127,11 @@
   }
 
   @Override
+  public boolean useLegacyCredentials() {
+    return useLegacyCredentials;
+  }
+
+  @Override
   public String getVersion() {
     return configResource.getVersion();
   }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationModule.java b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationModule.java
index ac27280..e3e4021 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationModule.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationModule.java
@@ -23,6 +23,7 @@
 import com.google.gerrit.extensions.events.HeadUpdatedListener;
 import com.google.gerrit.extensions.events.LifecycleListener;
 import com.google.gerrit.extensions.events.ProjectDeletedListener;
+import com.google.gerrit.extensions.registration.DynamicItem;
 import com.google.gerrit.extensions.registration.DynamicSet;
 import com.google.gerrit.server.events.EventTypes;
 import com.google.inject.AbstractModule;
@@ -30,6 +31,7 @@
 import com.google.inject.Scopes;
 import com.google.inject.assistedinject.FactoryModuleBuilder;
 import com.google.inject.internal.UniqueAnnotations;
+import com.googlesource.gerrit.plugins.replication.api.ReplicationRemotesApi;
 import com.googlesource.gerrit.plugins.replication.events.ProjectDeletionReplicationDoneEvent;
 import com.googlesource.gerrit.plugins.replication.events.ProjectDeletionReplicationFailedEvent;
 import com.googlesource.gerrit.plugins.replication.events.ProjectDeletionReplicationScheduledEvent;
@@ -52,9 +54,7 @@
 
   @Override
   protected void configure() {
-    install(new FactoryModuleBuilder().build(Destination.Factory.class));
     install(configModule);
-    bind(ReplicationQueue.class).in(Scopes.SINGLETON);
     bind(ObservableQueue.class).to(ReplicationQueue.class);
     bind(LifecycleListener.class)
         .annotatedWith(UniqueAnnotations.create())
@@ -63,6 +63,7 @@
     DynamicSet.bind(binder(), GitBatchRefUpdateListener.class).to(ReplicationQueue.class);
     DynamicSet.bind(binder(), ProjectDeletedListener.class).to(ReplicationQueue.class);
     DynamicSet.bind(binder(), HeadUpdatedListener.class).to(ReplicationQueue.class);
+    DynamicItem.bind(binder(), ReplicationRemotesApi.class).to(ReplicationRemotesApiImpl.class);
 
     bind(OnStartStop.class).in(Scopes.SINGLETON);
     bind(LifecycleListener.class).annotatedWith(UniqueAnnotations.create()).to(OnStartStop.class);
@@ -77,10 +78,8 @@
         .to(StartReplicationCapability.class);
 
     install(new FactoryModuleBuilder().build(PushAll.Factory.class));
-    install(new FactoryModuleBuilder().build(ProjectDeletionState.Factory.class));
 
     bind(EventBus.class).in(Scopes.SINGLETON);
-    bind(ReplicationDestinations.class).to(DestinationsCollection.class);
     bind(ConfigParser.class).to(DestinationConfigParser.class).in(Scopes.SINGLETON);
 
     DynamicSet.setOf(binder(), ReplicationStateListener.class);
@@ -103,5 +102,11 @@
 
     bind(TransportFactory.class).to(TransportFactoryImpl.class).in(Scopes.SINGLETON);
     bind(CloseableHttpClient.class).toProvider(HttpClientProvider.class).in(Scopes.SINGLETON);
+
+    bind(ReplicationQueue.class).in(Scopes.SINGLETON);
+    bind(ReplicationDestinations.class).to(DestinationsCollection.class);
+
+    install(new FactoryModuleBuilder().build(Destination.Factory.class));
+    install(new FactoryModuleBuilder().build(ProjectDeletionState.Factory.class));
   }
 }
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 6a36add..5d7fd19 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationQueue.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationQueue.java
@@ -132,9 +132,19 @@
 
   public void scheduleFullSync(
       Project.NameKey project, String urlMatch, ReplicationState state, boolean now) {
+    scheduleFullSync(project, urlMatch, Set.of(), state, now);
+  }
+
+  public void scheduleFullSync(
+      Project.NameKey project,
+      String urlMatch,
+      Set<String> remotesToConsider,
+      ReplicationState state,
+      boolean now) {
     fire(
         project,
         urlMatch,
+        remotesToConsider,
         Set.of(new GitReferenceUpdated.UpdatedRef(PushOne.ALL_REFS, null, null, null)),
         state,
         now);
@@ -157,6 +167,16 @@
       Set<UpdatedRef> updatedRefs,
       ReplicationState state,
       boolean now) {
+    fire(project, urlMatch, Set.of(), updatedRefs, state, now);
+  }
+
+  private void fire(
+      Project.NameKey project,
+      String urlMatch,
+      Set<String> remotesToConsider,
+      Set<UpdatedRef> updatedRefs,
+      ReplicationState state,
+      boolean now) {
     if (!running) {
       stateLog.warn(
           "Replication plugin did not finish startup before event, event replication is postponed",
@@ -166,13 +186,15 @@
     }
 
     for (Destination cfg : destinations.get().getAll(FilterType.ALL)) {
-      pushReferences(
-          cfg,
-          project,
-          urlMatch,
-          updatedRefs.stream().map(UpdatedRef::getRefName).collect(Collectors.toSet()),
-          state,
-          now);
+      if (remotesToConsider.isEmpty() || remotesToConsider.contains(cfg.getRemoteConfigName())) {
+        pushReferences(
+            cfg,
+            project,
+            urlMatch,
+            updatedRefs.stream().map(UpdatedRef::getRefName).collect(Collectors.toSet()),
+            state,
+            now);
+      }
     }
   }
 
@@ -209,7 +231,7 @@
       }
     }
     if (!refNamesToPush.isEmpty()) {
-      for (URIish uri : cfg.getURIs(project, urlMatch)) {
+      for (URIish uri : cfg.getDistributedUris(project, urlMatch)) {
         replicationTasksStorage.create(
             ReplicateRefUpdate.create(
                 project.get(), refNamesToPush, uri, cfg.getRemoteConfigName()));
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationRemotesApiImpl.java b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationRemotesApiImpl.java
new file mode 100644
index 0000000..cc6bcad
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationRemotesApiImpl.java
@@ -0,0 +1,100 @@
+// Copyright (C) 2024 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.googlesource.gerrit.plugins.replication;
+
+import com.google.gerrit.server.securestore.SecureStore;
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+import com.googlesource.gerrit.plugins.replication.api.ReplicationRemotesApi;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Set;
+import org.eclipse.jgit.lib.Config;
+
+@Singleton
+class ReplicationRemotesApiImpl implements ReplicationRemotesApi {
+
+  private final SecureStore secureStore;
+  private final MergedConfigResource mergedConfigResource;
+
+  @Inject
+  ReplicationRemotesApiImpl(SecureStore secureStore, MergedConfigResource mergedConfigResource) {
+    this.secureStore = secureStore;
+    this.mergedConfigResource = mergedConfigResource;
+  }
+
+  @Override
+  public Config get(String... remoteNames) {
+    Config replicationConfig = mergedConfigResource.getConfig();
+    Config remoteConfig = new Config();
+    for (String remoteName : remoteNames) {
+      Set<String> configNames = replicationConfig.getNames("remote", remoteName);
+      for (String configName : configNames) {
+        String[] values = replicationConfig.getStringList("remote", remoteName, configName);
+        if (values.length > 0) {
+          remoteConfig.setStringList("remote", remoteName, configName, Arrays.asList(values));
+        }
+      }
+    }
+    return remoteConfig;
+  }
+
+  @Override
+  public void update(Config remoteConfig) throws IOException {
+    if (remoteConfig.getSubsections("remote").isEmpty()) {
+      throw new IllegalArgumentException(
+          "configuration update must have at least one 'remote' section");
+    }
+
+    SeparatedRemoteConfigs configs = onlyRemoteSectionsWithSeparatedCredentials(remoteConfig);
+    persistRemotesCredentials(configs);
+
+    mergedConfigResource.update(configs.remotes);
+  }
+
+  private SeparatedRemoteConfigs onlyRemoteSectionsWithSeparatedCredentials(Config configUpdates) {
+    SeparatedRemoteConfigs configs = new SeparatedRemoteConfigs();
+    for (String subSection : configUpdates.getSubsections("remote")) {
+      for (String name : configUpdates.getNames("remote", subSection)) {
+        List<String> values = List.of(configUpdates.getStringList("remote", subSection, name));
+        if ("password".equals(name) || "username".equals(name)) {
+          configs.credentials.setStringList("remote", subSection, name, values);
+        } else {
+          configs.remotes.setStringList("remote", subSection, name, values);
+        }
+      }
+    }
+
+    return configs;
+  }
+
+  private void persistRemotesCredentials(SeparatedRemoteConfigs configs) {
+    for (String subSection : configs.credentials.getSubsections("remote")) {
+      copyRemoteStringList(configs.credentials, subSection, "username");
+      copyRemoteStringList(configs.credentials, subSection, "password");
+    }
+  }
+
+  private void copyRemoteStringList(Config config, String subSection, String key) {
+    secureStore.setList(
+        "remote", subSection, key, List.of(config.getStringList("remote", subSection, key)));
+  }
+
+  private static class SeparatedRemoteConfigs {
+    private final Config remotes = new Config();
+    private final Config credentials = new Config();
+  }
+}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/SecureCredentialsFactory.java b/src/main/java/com/googlesource/gerrit/plugins/replication/SecureCredentialsFactory.java
index ed15b92..3e00089 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/SecureCredentialsFactory.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/SecureCredentialsFactory.java
@@ -1,4 +1,4 @@
-// Copyright (C) 2011 The Android Open Source Project
+// Copyright (C) 2024 The Android Open Source Project
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -14,46 +14,43 @@
 
 package com.googlesource.gerrit.plugins.replication;
 
-import com.google.gerrit.server.config.SitePaths;
+import com.google.common.flogger.FluentLogger;
+import com.google.gerrit.common.UsedAt;
+import com.google.gerrit.common.UsedAt.Project;
+import com.google.gerrit.server.securestore.SecureStore;
 import com.google.inject.Inject;
-import java.io.IOException;
 import java.util.Objects;
-import org.eclipse.jgit.errors.ConfigInvalidException;
-import org.eclipse.jgit.lib.Config;
-import org.eclipse.jgit.storage.file.FileBasedConfig;
 import org.eclipse.jgit.transport.CredentialsProvider;
 import org.eclipse.jgit.transport.UsernamePasswordCredentialsProvider;
-import org.eclipse.jgit.util.FS;
 
-/** Looks up a remote's password in secure.config. */
+/** Looks up a remote's password in SecureStore */
+@UsedAt(Project.PLUGIN_PULL_REPLICATION)
 public class SecureCredentialsFactory implements CredentialsFactory {
-  private final Config config;
+  private static final FluentLogger log = FluentLogger.forEnclosingClass();
+  private final SecureStore secureStore;
 
   @Inject
-  public SecureCredentialsFactory(SitePaths site) throws ConfigInvalidException, IOException {
-    config = load(site);
-  }
-
-  private static Config load(SitePaths site) throws ConfigInvalidException, IOException {
-    FileBasedConfig cfg = new FileBasedConfig(site.secure_config.toFile(), FS.DETECTED);
-    if (cfg.getFile().exists() && cfg.getFile().length() > 0) {
-      try {
-        cfg.load();
-      } catch (ConfigInvalidException e) {
-        throw new ConfigInvalidException(
-            String.format("Config file %s is invalid: %s", cfg.getFile(), e.getMessage()), e);
-      } catch (IOException e) {
-        throw new IOException(
-            String.format("Cannot read %s: %s", cfg.getFile(), e.getMessage()), e);
-      }
-    }
-    return cfg;
+  public SecureCredentialsFactory(SecureStore secureStore) {
+    this.secureStore = secureStore;
   }
 
   @Override
   public CredentialsProvider create(String remoteName) {
-    String user = Objects.toString(config.getString("remote", remoteName, "username"), "");
-    String pass = Objects.toString(config.getString("remote", remoteName, "password"), "");
+    String user = Objects.toString(secureStore.get("remote", remoteName, "username"), "");
+    String pass = Objects.toString(secureStore.get("remote", remoteName, "password"), "");
     return new UsernamePasswordCredentialsProvider(user, pass);
   }
+
+  @Override
+  public boolean validate(String remoteName) {
+    try {
+      String unused = secureStore.get("remote", remoteName, "username");
+      unused = secureStore.get("remote", remoteName, "password");
+      return true;
+    } catch (Throwable t) {
+      log.atSevere().withCause(t).log(
+          "Credentials for replication remote %s are invalid", remoteName);
+      return false;
+    }
+  }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/StartCommand.java b/src/main/java/com/googlesource/gerrit/plugins/replication/StartCommand.java
index 800245e..74d46e1 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/StartCommand.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/StartCommand.java
@@ -20,7 +20,9 @@
 import com.google.inject.Inject;
 import com.googlesource.gerrit.plugins.replication.PushResultProcessing.CommandProcessing;
 import java.util.ArrayList;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Set;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
@@ -40,6 +42,9 @@
   @Option(name = "--url", metaVar = "SUBSTRING", usage = "substring URL must match (or * to match everything)")
   private String urlMatch;
 
+  @Option(name = "--remote", metaVar = "REMOTE", usage = "name of remote to replicate to")
+  private Set<String> remotesToConsider = new HashSet<>();
+
   @Option(name = "--wait", usage = "wait for replication to finish before exiting")
   private boolean wait;
 
@@ -70,7 +75,9 @@
     }
 
     Future<?> future =
-        pushFactory.create(urlMatch, projectFilter, state, now).schedule(0, TimeUnit.SECONDS);
+        pushFactory
+            .create(urlMatch, remotesToConsider, projectFilter, state, now)
+            .schedule(0, TimeUnit.SECONDS);
 
     if (wait) {
       if (future != null) {
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/UrlDistributionStrategy.java b/src/main/java/com/googlesource/gerrit/plugins/replication/UrlDistributionStrategy.java
new file mode 100644
index 0000000..46183eb
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/UrlDistributionStrategy.java
@@ -0,0 +1,84 @@
+// Copyright (C) 2026 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.googlesource.gerrit.plugins.replication;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.eclipse.jgit.transport.URIish;
+
+/**
+ * URL distribution strategy used when a remote has multiple configured URLs.
+ *
+ * <p>Each enum constant acts as a factory: call {@link #newInstance()} to obtain a stateful
+ * executor. Callers (e.g. {@link Destination}) hold the {@link Instance}, while the enum constant
+ * itself remains stateless and safe to use in equality checks.
+ *
+ * <p>Configured via {@code remote.NAME.urlDistribution} in {@code replication.config}.
+ */
+public enum UrlDistributionStrategy {
+  /** Push to all configured URLs. */
+  ALL("all") {
+    @Override
+    public Instance newInstance() {
+      return candidates -> candidates;
+    }
+  },
+
+  /**
+   * Push to one URL at a time, rotating through the list on each push event. Particularly useful
+   * when multiple replica hosts share a single backend (likely via NFS): pushing to all URLs would
+   * cause redundant writes to the same underlying storage, while round-robin distributes load
+   * evenly and ensures each push is executed exactly once.
+   */
+  ROUND_ROBIN("roundRobin") {
+    @Override
+    public Instance newInstance() {
+      final AtomicInteger index = new AtomicInteger();
+      return candidates -> {
+        if (candidates.isEmpty()) {
+          return List.of();
+        }
+        return List.of(candidates.get(Math.floorMod(index.getAndIncrement(), candidates.size())));
+      };
+    }
+  };
+
+  public final String configKey;
+
+  UrlDistributionStrategy(String key) {
+    configKey = key;
+  }
+
+  /** Creates a new stateful executor for this distribution strategy. */
+  public abstract Instance newInstance();
+
+  /**
+   * Returns the distribution strategy for the given config value, or {@link #ALL} if the value is
+   * unrecognized or absent.
+   */
+  public static UrlDistributionStrategy fromConfig(String value) {
+    return Arrays.stream(values())
+        .filter(candidate -> candidate.configKey.equals(value))
+        .findFirst()
+        .orElse(ALL);
+  }
+
+  /** A stateful executor for a {@link UrlDistributionStrategy} strategy. */
+  @FunctionalInterface
+  public interface Instance {
+    List<URIish> select(List<URIish> candidates);
+  }
+}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/api/ApiModule.java b/src/main/java/com/googlesource/gerrit/plugins/replication/api/ApiModule.java
index 3764ede..44ec88c 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/api/ApiModule.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/api/ApiModule.java
@@ -22,5 +22,6 @@
   protected void configure() {
     DynamicItem.itemOf(binder(), ReplicationPushFilter.class);
     DynamicItem.itemOf(binder(), ReplicationConfigOverrides.class);
+    DynamicItem.itemOf(binder(), ReplicationRemotesApi.class);
   }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/api/ConfigResource.java b/src/main/java/com/googlesource/gerrit/plugins/replication/api/ConfigResource.java
index 43733bc..0052fc2 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/api/ConfigResource.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/api/ConfigResource.java
@@ -14,6 +14,7 @@
 
 package com.googlesource.gerrit.plugins.replication.api;
 
+import java.io.IOException;
 import org.eclipse.jgit.lib.Config;
 
 /**
@@ -37,6 +38,16 @@
   Config getConfig();
 
   /**
+   * Update the configuration resource.
+   *
+   * <p>Allows to persist changes to the configuration resource.
+   *
+   * @param config updated configuration
+   * @throws IOException when configuration cannot be persisted
+   */
+  void update(Config config) throws IOException;
+
+  /**
    * Current logical version string of the current configuration on the persistent storage.
    *
    * @return latest logical version number on the persistent storage
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/api/ReplicationConfig.java b/src/main/java/com/googlesource/gerrit/plugins/replication/api/ReplicationConfig.java
index 8ef4e2b..d5ccb02 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/api/ReplicationConfig.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/api/ReplicationConfig.java
@@ -102,4 +102,16 @@
    * @return the config.
    */
   Config getConfig();
+
+  /**
+   * Use legacy credentials for migrating existing sites.
+   *
+   * <p>Existing installations may have used a mix of encrypted and clear text credentials in
+   * secure.config, leveraging the replication plugin bug that was not accessing it using the
+   * correct API. The legacy feature flag 'gerrit.useLegacyCredentials' allows Gerrit to still use
+   * direct access to secure.config without decrypting its values.
+   *
+   * @return true if the secure.config should be read always directly and without decryption
+   */
+  boolean useLegacyCredentials();
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/api/ReplicationRemotesApi.java b/src/main/java/com/googlesource/gerrit/plugins/replication/api/ReplicationRemotesApi.java
new file mode 100644
index 0000000..76d6d35
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/api/ReplicationRemotesApi.java
@@ -0,0 +1,55 @@
+// Copyright (C) 2024 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.googlesource.gerrit.plugins.replication.api;
+
+import static com.google.gerrit.common.UsedAt.Project.PLUGIN_GITHUB;
+
+import com.google.gerrit.common.UsedAt;
+import com.google.gerrit.extensions.registration.DynamicItem;
+import com.google.gerrit.server.securestore.SecureStore;
+import java.io.IOException;
+import org.eclipse.jgit.lib.Config;
+
+/** Public API to update replication plugin remotes configurations programmatically. */
+@UsedAt(PLUGIN_GITHUB)
+@DynamicItem.Final(implementedByPlugin = "replication")
+public interface ReplicationRemotesApi {
+
+  /**
+   * Retrieves the configuration for a remote by name.
+   *
+   * <p>Builds a JGit {@link Config} with the remote's configuration obtained by parsing the
+   * replication configuration sources.
+   *
+   * <p>NOTE: The remotes secrets are excluded for security reasons.
+   *
+   * @param remoteNames the remote names to retrieve.
+   * @return {@link Config} associated with the remoteName(s)
+   */
+  Config get(String... remoteNames);
+
+  /**
+   * Adds or updates the remote configuration for the replication plugin.
+   *
+   * <p>Provided JGit {@link Config} object should contain at least one named <em>remote</em>
+   * section. All other configurations will be ignored.
+   *
+   * <p>NOTE: The {@code remote.$name.password} will be stored using {@link SecureStore}.
+   *
+   * @param remoteConfig remotes to add or update
+   * @throws IOException when persisting fails
+   */
+  void update(Config remoteConfig) throws IOException;
+}
diff --git a/src/main/resources/Documentation/cmd-start.md b/src/main/resources/Documentation/cmd-start.md
index e1b7341..e6d9fbd 100644
--- a/src/main/resources/Documentation/cmd-start.md
+++ b/src/main/resources/Documentation/cmd-start.md
@@ -12,6 +12,7 @@
 ssh -p @SSH_PORT@ @SSH_HOST@ @PLUGIN@ start
   [--now]
   [--wait]
+  [--remote <REMOTE> ...]
   {--url <PATTERN> | [--url <PATTERN>] --all | [--url <PATTERN>] <PROJECT PATTERN> ...}
 ```
 
@@ -102,6 +103,12 @@
 URL contains `PATTERN`. This can be useful to replicate only to
 a previously down node, which has been brought back online.
 
+`--remote <REMOTE>`
+: Replicate only to the destination whose remote name exactly matches
+`REMOTE`. May be specified multiple times to target several remotes.
+Remote names correspond to the `[remote "name"]` sections in
+`replication.config`.
+
 EXAMPLES
 --------
 Replicate every project, to every configured remote:
@@ -142,6 +149,18 @@
   $ ssh -p @SSH_PORT@ @SSH_HOST@ @PLUGIN@ start --url https://example.com/tools/gerrit.git
 ```
 
+Replicate all projects to a specific named remote:
+
+```console
+  $ ssh -p @SSH_PORT@ @SSH_HOST@ @PLUGIN@ start --remote replica1 --all
+```
+
+Replicate all projects to two named remotes:
+
+```console
+  $ ssh -p @SSH_PORT@ @SSH_HOST@ @PLUGIN@ start --remote replica1 --remote replica2 --all
+```
+
 SEE ALSO
 --------
 
diff --git a/src/main/resources/Documentation/config.md b/src/main/resources/Documentation/config.md
index 0243ea1..73b3805 100644
--- a/src/main/resources/Documentation/config.md
+++ b/src/main/resources/Documentation/config.md
@@ -170,6 +170,14 @@
 :	Timeout for SSH connections. If 0, there is no timeout and
         the client waits indefinitely. By default, 2 minutes.
 
+gerrit.useLegacyCredentials
+:	Use legacy credentials on secure.config which did not allow encryption
+	through a SecureStore provider. Legacy credentials should not be used in
+	production and are provided purely for backward compatibility with existing
+	secure.config files. When enabling the SecureStore in Gerrit, for instance,
+	the secure-config.jar libModule, make sure that all existing replication
+	credentials are encrypted. By default, false.
+
 replication.distributionInterval
 :	Interval in seconds for running the replication distributor. When
 	run, the replication distributor will add all persisted waiting tasks
@@ -604,6 +612,24 @@
 
   Do not exclude any refs pattern by default.
 
+remote.NAME.urlDistributionStrategy
+: URL distribution strategy to use when a remote has multiple configured URLs.
+  Applies to both push URLs (`url`) and admin URLs (`adminUrl`).
+
+  `all` (default)
+  : Push to all configured URLs simultaneously. This is the standard
+    replication behaviour.
+
+  `roundRobin`
+  : Push to one URL at a time, rotating through the list on each push event.
+    Particularly useful in high-availability setups where multiple replica
+    hosts share a single NFS backend. Pushing to all URLs simultaneously
+    would cause redundant writes to the same underlying storage; round-robin
+    distributes load evenly and ensures each push is written exactly once.
+    Has no effect if only one URL is configured.
+
+  Defaults to `all`.
+
 Directory `replication`
 --------------------
 The optional directory `$site_path/etc/replication` contains Git-style
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/AbstractConfigTest.java b/src/test/java/com/googlesource/gerrit/plugins/replication/AbstractConfigTest.java
index b8bbc08..935fdb4 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/AbstractConfigTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/AbstractConfigTest.java
@@ -57,7 +57,8 @@
     public final DestinationConfiguration config;
 
     protected FakeDestination(DestinationConfiguration config) {
-      super(injectorMock(), null, null, null, null, null, null, null, null, null, null, config);
+      super(
+          injectorMock(), null, null, null, null, null, null, null, null, null, null, null, config);
       this.config = config;
     }
 
@@ -78,7 +79,7 @@
   }
 
   @Before
-  public void setup() {
+  public void setup() throws IOException {
     when(destinationFactoryMock.create(any(DestinationConfiguration.class)))
         .thenAnswer(
             new Answer<Destination>() {
@@ -103,6 +104,10 @@
     return newReplicationConfig("replication.config");
   }
 
+  protected FileBasedConfig newSecureConfig() {
+    return newReplicationConfig("secure.config");
+  }
+
   protected FileBasedConfig newReplicationConfig(String path) {
     FileBasedConfig replicationConfig =
         new FileBasedConfig(sitePaths.etc_dir.resolve(path).toFile(), FS.DETECTED);
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/AutoReloadSecureCredentialsFactoryDecoratorTest.java b/src/test/java/com/googlesource/gerrit/plugins/replication/AutoReloadSecureCredentialsFactoryDecoratorTest.java
new file mode 100644
index 0000000..c959cc9
--- /dev/null
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/AutoReloadSecureCredentialsFactoryDecoratorTest.java
@@ -0,0 +1,97 @@
+// Copyright (C) 2024 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.googlesource.gerrit.plugins.replication;
+
+import static com.google.common.truth.Truth.assertThat;
+import static org.mockito.Mockito.when;
+
+import com.google.gerrit.server.securestore.SecureStore;
+import com.googlesource.gerrit.plugins.replication.api.ReplicationConfig;
+import java.io.IOException;
+import org.eclipse.jgit.errors.ConfigInvalidException;
+import org.eclipse.jgit.lib.Config;
+import org.eclipse.jgit.storage.file.FileBasedConfig;
+import org.eclipse.jgit.transport.CredentialItem;
+import org.eclipse.jgit.transport.CredentialsProvider;
+import org.eclipse.jgit.transport.URIish;
+import org.eclipse.jgit.transport.UsernamePasswordCredentialsProvider;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+@RunWith(MockitoJUnitRunner.class)
+public class AutoReloadSecureCredentialsFactoryDecoratorTest extends AbstractConfigTest {
+  private static final String REMOTE_TEST = "remote-test";
+  private static final String USERNAME_CLEARTEXT = "username-clear";
+  private static final String PASSWORD_CLEARTEXT = "password-clear";
+  private static final String USERNAME_CIPHERTEXT = "username-encrypted";
+  private static final String PASSWORD_CIPHERTEXT = "password-encrypted";
+
+  @Mock private SecureStore secureStoreMock;
+  @Mock private ReplicationConfig replicationConfigMock;
+
+  public AutoReloadSecureCredentialsFactoryDecoratorTest() throws IOException {
+    super();
+  }
+
+  @Override
+  @Before
+  public void setup() throws IOException {
+    when(replicationConfigMock.getConfig()).thenReturn(new Config());
+
+    when(secureStoreMock.get("remote", REMOTE_TEST, "username")).thenReturn(USERNAME_CIPHERTEXT);
+    when(secureStoreMock.get("remote", REMOTE_TEST, "password")).thenReturn(PASSWORD_CIPHERTEXT);
+
+    FileBasedConfig secureConfig = newSecureConfig();
+    secureConfig.setString("remote", REMOTE_TEST, "username", USERNAME_CLEARTEXT);
+    secureConfig.setString("remote", REMOTE_TEST, "password", PASSWORD_CLEARTEXT);
+    secureConfig.save();
+  }
+
+  @Test
+  public void shouldCreateLegacyCredentials() throws Exception {
+    when(replicationConfigMock.useLegacyCredentials()).thenReturn(true);
+    assertUsernamePasswordCredentials(
+        getCredentialsProvider(), USERNAME_CLEARTEXT, PASSWORD_CLEARTEXT);
+  }
+
+  @Test
+  public void shouldCreateEncryptedCredentialsByDefault() throws Exception {
+    assertUsernamePasswordCredentials(
+        getCredentialsProvider(), USERNAME_CIPHERTEXT, PASSWORD_CIPHERTEXT);
+  }
+
+  private UsernamePasswordCredentialsProvider getCredentialsProvider()
+      throws ConfigInvalidException, IOException {
+    AutoReloadSecureCredentialsFactoryDecorator credFactory =
+        new AutoReloadSecureCredentialsFactoryDecorator(
+            sitePaths, secureStoreMock, replicationConfigMock);
+    CredentialsProvider legacyCredentials = credFactory.create(REMOTE_TEST);
+    assertThat(legacyCredentials).isInstanceOf(UsernamePasswordCredentialsProvider.class);
+    return (UsernamePasswordCredentialsProvider) legacyCredentials;
+  }
+
+  private static void assertUsernamePasswordCredentials(
+      UsernamePasswordCredentialsProvider credentials, String username, String password) {
+    CredentialItem.Username usernameItem = new CredentialItem.Username();
+    CredentialItem.Password passwordItem = new CredentialItem.Password();
+    assertThat(credentials.get(new URIish(), usernameItem, passwordItem)).isTrue();
+
+    assertThat(usernameItem.getValue()).isEqualTo(username);
+    assertThat(new String(passwordItem.getValue())).isEqualTo(password);
+  }
+}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/DestinationConfigurationTest.java b/src/test/java/com/googlesource/gerrit/plugins/replication/DestinationConfigurationTest.java
index 646f915..c7b4174 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/DestinationConfigurationTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/DestinationConfigurationTest.java
@@ -98,4 +98,20 @@
     // then
     assertThat(actual).isEqualTo(globalPushBatchSize);
   }
+
+  @Test
+  public void shouldDefaultUrlDistributionToAll() {
+    assertThat(objectUnderTest.getUrlDistributionStrategy()).isEqualTo(UrlDistributionStrategy.ALL);
+  }
+
+  @Test
+  public void shouldSetUrlDistributionToRoundRobinWhenConfigured() {
+    // given
+    when(cfgMock.getString("remote", REMOTE, "urlDistributionStrategy")).thenReturn("roundRobin");
+    objectUnderTest = new DestinationConfiguration(remoteConfigMock, cfgMock);
+
+    // when / then
+    assertThat(objectUnderTest.getUrlDistributionStrategy())
+        .isEqualTo(UrlDistributionStrategy.ROUND_ROBIN);
+  }
 }
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/FanoutConfigResourceTest.java b/src/test/java/com/googlesource/gerrit/plugins/replication/FanoutConfigResourceTest.java
index 9147ea1..a3d811f 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/FanoutConfigResourceTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/FanoutConfigResourceTest.java
@@ -24,6 +24,7 @@
 import java.io.IOException;
 import java.util.List;
 import org.eclipse.jgit.errors.ConfigInvalidException;
+import org.eclipse.jgit.lib.Config;
 import org.eclipse.jgit.storage.file.FileBasedConfig;
 import org.eclipse.jgit.util.FS;
 import org.junit.Before;
@@ -285,9 +286,73 @@
     assertThat(objectUnderTest.getVersion()).isEqualTo(replicationConfigVersion);
   }
 
+  @Test
+  public void shouldAddConfigOptionToMainConfig() throws Exception {
+    FanoutConfigResource objectUnderTest = new FanoutConfigResource(sitePaths);
+    Config update = new Config();
+    update.setString("new", null, "value", "set");
+
+    objectUnderTest.update(update);
+    Config updatedConfig = objectUnderTest.getConfig();
+
+    assertThat(updatedConfig.getString("new", null, "value")).isEqualTo("set");
+  }
+
+  @Test
+  public void shouldUpdateConfigOptionInMainConfig() throws Exception {
+    FileBasedConfig config = newReplicationConfig();
+    config.setString("updatable", null, "value", "orig");
+    config.save();
+    FanoutConfigResource objectUnderTest = new FanoutConfigResource(sitePaths);
+    Config update = new Config();
+    update.setString("updatable", null, "value", "updated");
+
+    objectUnderTest.update(update);
+    Config updatedConfig = objectUnderTest.getConfig();
+
+    assertThat(updatedConfig.getString("updatable", null, "value")).isEqualTo("updated");
+  }
+
+  @Test
+  public void shouldAddNewRemoteFile() throws Exception {
+    FanoutConfigResource objectUnderTest = new FanoutConfigResource(sitePaths);
+    Config update = new Config();
+    update.setString("remote", remoteName1, "url", remoteUrl1);
+
+    objectUnderTest.update(update);
+
+    assertThat(objectUnderTest.getConfig().getString("remote", remoteName1, "url"))
+        .isEqualTo(remoteUrl1);
+    Config actual = loadRemoteConfig(remoteName1);
+    assertThat(actual.getString("remote", null, "url")).isEqualTo(remoteUrl1);
+  }
+
+  @Test
+  public void shouldUpdateExistingRemote() throws Exception {
+    FileBasedConfig rawRemoteConfig = newRemoteConfig(remoteName1);
+    rawRemoteConfig.setString("remote", remoteName1, "url", remoteUrl1);
+    rawRemoteConfig.save();
+    FanoutConfigResource objectUnderTest = new FanoutConfigResource(sitePaths);
+    Config update = new Config();
+    update.setString("remote", remoteName1, "url", remoteUrl2);
+
+    objectUnderTest.update(update);
+
+    assertThat(objectUnderTest.getConfig().getString("remote", remoteName1, "url"))
+        .isEqualTo(remoteUrl2);
+    Config actual = loadRemoteConfig(remoteName1);
+    assertThat(actual.getString("remote", null, "url")).isEqualTo(remoteUrl2);
+  }
+
   protected FileBasedConfig newRemoteConfig(String configFileName) {
     return new FileBasedConfig(
         sitePaths.etc_dir.resolve("replication/" + configFileName + ".config").toFile(),
         FS.DETECTED);
   }
+
+  private Config loadRemoteConfig(String siteName) throws Exception {
+    FileBasedConfig config = newRemoteConfig(siteName);
+    config.load();
+    return config;
+  }
 }
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/FileConfigResourceTest.java b/src/test/java/com/googlesource/gerrit/plugins/replication/FileConfigResourceTest.java
new file mode 100644
index 0000000..92bf58d
--- /dev/null
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/FileConfigResourceTest.java
@@ -0,0 +1,121 @@
+// Copyright (C) 2024 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.googlesource.gerrit.plugins.replication;
+
+import static com.google.common.io.RecursiveDeleteOption.ALLOW_INSECURE;
+import static com.google.common.truth.Truth.assertThat;
+
+import com.google.common.io.MoreFiles;
+import com.google.gerrit.server.config.SitePaths;
+import com.googlesource.gerrit.plugins.replication.api.ConfigResource;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import org.eclipse.jgit.lib.Config;
+import org.eclipse.jgit.storage.file.FileBasedConfig;
+import org.eclipse.jgit.util.FS;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+public class FileConfigResourceTest {
+  private static final String VALUE_KEY = "value";
+  private static final String INITIAL_KEY = "initial";
+  private static final String UPDATABLE_KEY = "updatable";
+
+  private Path testDir;
+  private SitePaths sitePaths;
+  private ConfigResource configResource;
+
+  @Before
+  public void setUp() throws Exception {
+    testDir = Files.createTempDirectory("fileConfigResourceTest");
+    sitePaths = new SitePaths(testDir);
+    configResource = newFileConfigResource();
+  }
+
+  @After
+  public void tearDown() throws Exception {
+    MoreFiles.deleteRecursively(testDir, ALLOW_INSECURE);
+  }
+
+  @Test
+  public void updateEmptyFile() throws Exception {
+    Config configUpdate = newConfigUpdate();
+
+    Config beforeUpdate = configResource.getConfig();
+    assertThat(beforeUpdate.getSections()).isEmpty();
+
+    configResource.update(configUpdate);
+    Config updatedConfig = newFileConfigResource().getConfig();
+
+    assertConfigUpdate(updatedConfig);
+  }
+
+  @Test
+  public void appendOptionToConfig() throws Exception {
+    FileBasedConfig rawConfig = getRawReplicationConfig();
+    rawConfig.setInt(INITIAL_KEY, null, VALUE_KEY, 10);
+    rawConfig.save();
+    configResource = newFileConfigResource();
+    Config configUpdate = newConfigUpdate();
+
+    configResource.update(configUpdate);
+    Config updatedConfig = configResource.getConfig();
+
+    assertConfigUpdate(updatedConfig);
+    assertThat(updatedConfig.getInt(INITIAL_KEY, null, VALUE_KEY, -1)).isEqualTo(10);
+  }
+
+  @Test
+  public void updateExistingOption() throws Exception {
+    int expectedValue = 20;
+    Config configUpdate = new Config();
+    configUpdate.setInt(UPDATABLE_KEY, null, VALUE_KEY, expectedValue);
+    FileBasedConfig rawConfig = getRawReplicationConfig(newConfigUpdate());
+    rawConfig.save();
+
+    configResource.update(configUpdate);
+    Config updatedConfig = configResource.getConfig();
+
+    assertConfigUpdate(updatedConfig, expectedValue);
+  }
+
+  private FileConfigResource newFileConfigResource() {
+    return new FileConfigResource(sitePaths);
+  }
+
+  private Config newConfigUpdate() {
+    Config configUpdate = new Config();
+    configUpdate.setInt(UPDATABLE_KEY, null, VALUE_KEY, 1);
+    return configUpdate;
+  }
+
+  private void assertConfigUpdate(Config config) {
+    assertConfigUpdate(config, 1);
+  }
+
+  private void assertConfigUpdate(Config config, int expectedValue) {
+    assertThat(config.getInt(UPDATABLE_KEY, null, VALUE_KEY, -1)).isEqualTo(expectedValue);
+  }
+
+  private FileBasedConfig getRawReplicationConfig() {
+    return getRawReplicationConfig(new Config());
+  }
+
+  private FileBasedConfig getRawReplicationConfig(Config base) {
+    Path configPath = sitePaths.etc_dir.resolve(FileConfigResource.CONFIG_NAME);
+    return new FileBasedConfig(base, configPath.toFile(), FS.DETECTED);
+  }
+}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/MergedConfigResourceTest.java b/src/test/java/com/googlesource/gerrit/plugins/replication/MergedConfigResourceTest.java
index 18bb603..cae78e3 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/MergedConfigResourceTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/MergedConfigResourceTest.java
@@ -22,6 +22,7 @@
 import com.googlesource.gerrit.plugins.replication.api.ApiModule;
 import com.googlesource.gerrit.plugins.replication.api.ConfigResource;
 import com.googlesource.gerrit.plugins.replication.api.ReplicationConfigOverrides;
+import org.apache.commons.lang3.NotImplementedException;
 import org.eclipse.jgit.lib.Config;
 import org.junit.Test;
 
@@ -78,6 +79,11 @@
     }
 
     @Override
+    public void update(Config config) {
+      throw new NotImplementedException("not implemented");
+    }
+
+    @Override
     public String getVersion() {
       return "base";
     }
@@ -93,6 +99,11 @@
     }
 
     @Override
+    public void update(Config config) {
+      throw new NotImplementedException("not implemented");
+    }
+
+    @Override
     public String getVersion() {
       return "override";
     }
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationDaemon.java b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationDaemon.java
index ba4a958..062c84a 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationDaemon.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationDaemon.java
@@ -309,4 +309,10 @@
     config.setBoolean("remote", remoteName, "replicateProjectDeletions", replicateProjectDeletion);
     config.save();
   }
+
+  protected void setUrlDistribution(String remoteName, UrlDistributionStrategy distribution)
+      throws IOException {
+    config.setString("remote", remoteName, "urlDistributionStrategy", distribution.configKey);
+    config.save();
+  }
 }
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationIT.java b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationIT.java
index 4414cec..3b3686f 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationIT.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationIT.java
@@ -33,6 +33,7 @@
 import java.util.Arrays;
 import java.util.List;
 import java.util.Optional;
+import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executor;
 import java.util.concurrent.Executors;
@@ -242,7 +243,8 @@
         plugin
             .getSysInjector()
             .getInstance(PushAll.Factory.class)
-            .create(null, new ReplicationFilter(Arrays.asList(project.get())), state, false)
+            .create(
+                null, Set.of(), new ReplicationFilter(Arrays.asList(project.get())), state, false)
             .schedule(0, TimeUnit.SECONDS);
 
     future.get();
@@ -262,7 +264,8 @@
         plugin
             .getSysInjector()
             .getInstance(PushAll.Factory.class)
-            .create(null, new ReplicationFilter(Arrays.asList(project.get())), state, false)
+            .create(
+                null, Set.of(), new ReplicationFilter(Arrays.asList(project.get())), state, false)
             .schedule(0, TimeUnit.SECONDS);
 
     CountDownLatch latch = new CountDownLatch(1);
@@ -451,6 +454,147 @@
     }
   }
 
+  @Test
+  public void shouldReplicateToAllUrlsByDefault() throws Exception {
+    Project.NameKey replica1Project = createTestProject(project + "replica1");
+    Project.NameKey replica2Project = createTestProject(project + "replica2");
+
+    setReplicationDestination(
+        "foo", List.of("replica1", "replica2"), ALL_PROJECTS, TEST_REPLICATION_DELAY_SECONDS);
+    reloadConfig();
+
+    String newRef = "refs/heads/newForTest";
+    createNewBranchWithoutPush("refs/heads/master", newRef);
+
+    plugin
+        .getSysInjector()
+        .getInstance(ReplicationQueue.class)
+        .scheduleFullSync(project, null, new ReplicationState(NO_OP), true);
+
+    // Wait for the push to land on both the refs
+    try (Repository r1 = repoManager.openRepository(replica1Project);
+        Repository r2 = repoManager.openRepository(replica2Project)) {
+      waitUntil(() -> checkedGetRef(r1, newRef) != null && checkedGetRef(r2, newRef) != null);
+    }
+  }
+
+  @Test
+  public void shouldReplicateToOnlyOneUrlWhenRoundRobinEnabled() throws Exception {
+    Project.NameKey replica1Project = createTestProject(project + "replica1");
+    Project.NameKey replica2Project = createTestProject(project + "replica2");
+
+    setReplicationDestination(
+        "foo", List.of("replica1", "replica2"), ALL_PROJECTS, TEST_REPLICATION_DELAY_SECONDS);
+    setUrlDistribution("foo", UrlDistributionStrategy.ROUND_ROBIN);
+    reloadConfig();
+
+    String newRef = "refs/heads/newForTest";
+    createNewBranchWithoutPush("refs/heads/master", newRef);
+
+    plugin
+        .getSysInjector()
+        .getInstance(ReplicationQueue.class)
+        .scheduleFullSync(project, null, new ReplicationState(NO_OP), true);
+
+    // Wait for the push to land in at least one replica
+    try (Repository r1 = repoManager.openRepository(replica1Project);
+        Repository r2 = repoManager.openRepository(replica2Project)) {
+      waitUntil(() -> checkedGetRef(r1, newRef) != null || checkedGetRef(r2, newRef) != null);
+
+      // Exactly one replica should have received the push
+      boolean r1HasRef = checkedGetRef(r1, newRef) != null;
+      boolean r2HasRef = checkedGetRef(r2, newRef) != null;
+      assertThat(r1HasRef ^ r2HasRef).isTrue();
+    }
+  }
+
+  @Test
+  public void shouldRotateAcrossUrlsOnConsecutivePushesWhenRoundRobinEnabled() throws Exception {
+    Project.NameKey replica1Project = createTestProject(project + "replica1");
+    Project.NameKey replica2Project = createTestProject(project + "replica2");
+
+    setReplicationDestination(
+        "foo", List.of("replica1", "replica2"), ALL_PROJECTS, TEST_REPLICATION_DELAY_SECONDS);
+    setUrlDistribution("foo", UrlDistributionStrategy.ROUND_ROBIN);
+    reloadConfig();
+
+    String branch1 = "refs/heads/branch1";
+    String branch2 = "refs/heads/branch2";
+    createNewBranchWithoutPush("refs/heads/master", branch1);
+
+    ReplicationQueue queue = plugin.getSysInjector().getInstance(ReplicationQueue.class);
+
+    // First sync - goes to replica1 (index 0)
+    queue.scheduleFullSync(project, null, new ReplicationState(NO_OP), true);
+
+    try (Repository r1 = repoManager.openRepository(replica1Project)) {
+      waitUntil(() -> checkedGetRef(r1, branch1) != null);
+    }
+
+    // replica2 should not have received branch1 yet
+    try (Repository r2 = repoManager.openRepository(replica2Project)) {
+      assertThat(checkedGetRef(r2, branch1)).isNull();
+    }
+
+    // Second sync - goes to replica2 (index 1), includes branch1 and branch2
+    createNewBranchWithoutPush("refs/heads/master", branch2);
+    queue.scheduleFullSync(project, null, new ReplicationState(NO_OP), true);
+
+    try (Repository r2 = repoManager.openRepository(replica2Project)) {
+      waitUntil(() -> checkedGetRef(r2, branch1) != null && checkedGetRef(r2, branch2) != null);
+    }
+
+    // replica1 should not have received branch2 (it was only in the second sync)
+    try (Repository r1 = repoManager.openRepository(replica1Project)) {
+      assertThat(checkedGetRef(r1, branch2)).isNull();
+    }
+  }
+
+  @Test
+  public void shouldReplicateToMatchingRemote() throws Exception {
+    Project.NameKey targetProject = createTestProject(project + "replica");
+
+    setReplicationDestination("foo", "replica", ALL_PROJECTS);
+    reloadConfig();
+
+    String newRef = "refs/heads/newForTest";
+    ObjectId newRefTip = createNewBranchWithoutPush("refs/heads/master", newRef);
+
+    plugin
+        .getSysInjector()
+        .getInstance(ReplicationQueue.class)
+        .scheduleFullSync(project, null, Set.of("foo"), new ReplicationState(NO_OP), true);
+
+    try (Repository repo = repoManager.openRepository(targetProject)) {
+      waitUntil(() -> checkedGetRef(repo, newRef) != null);
+
+      Ref targetBranchRef = getRef(repo, newRef);
+      assertThat(targetBranchRef).isNotNull();
+      assertThat(targetBranchRef.getObjectId()).isEqualTo(newRefTip);
+    }
+  }
+
+  @Test
+  public void shouldNotReplicateToNonMatchingRemote() throws Exception {
+    Project.NameKey targetProject = createTestProject(project + "replica");
+
+    setReplicationDestination("foo", "replica", ALL_PROJECTS);
+    reloadConfig();
+
+    String newRef = "refs/heads/newForTest";
+    createNewBranchWithoutPush("refs/heads/master", newRef);
+
+    plugin
+        .getSysInjector()
+        .getInstance(ReplicationQueue.class)
+        .scheduleFullSync(project, null, Set.of("bar"), new ReplicationState(NO_OP), true);
+
+    try (Repository repo = repoManager.openRepository(targetProject)) {
+      assertThrows(
+          InterruptedException.class, () -> waitUntil(() -> checkedGetRef(repo, newRef) != null));
+    }
+  }
+
   private void waitUntil(Supplier<Boolean> waitCondition) throws InterruptedException {
     WaitUtil.waitUntil(waitCondition, TEST_TIMEOUT);
   }
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationRemotesApiTest.java b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationRemotesApiTest.java
new file mode 100644
index 0000000..40d9846
--- /dev/null
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationRemotesApiTest.java
@@ -0,0 +1,212 @@
+// Copyright (C) 2024 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.googlesource.gerrit.plugins.replication;
+
+import static com.google.common.io.RecursiveDeleteOption.ALLOW_INSECURE;
+import static com.google.common.truth.Truth.assertThat;
+import static com.google.gerrit.testing.GerritJUnit.assertThrows;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+import com.google.common.io.MoreFiles;
+import com.google.common.truth.StringSubject;
+import com.google.gerrit.extensions.registration.DynamicItem;
+import com.google.gerrit.server.config.SitePaths;
+import com.google.gerrit.server.securestore.SecureStore;
+import com.google.inject.AbstractModule;
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+import com.googlesource.gerrit.plugins.replication.api.ConfigResource;
+import com.googlesource.gerrit.plugins.replication.api.ReplicationConfigOverrides;
+import com.googlesource.gerrit.plugins.replication.api.ReplicationRemotesApi;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicReference;
+import org.eclipse.jgit.lib.Config;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+public class ReplicationRemotesApiTest {
+
+  private Path testSite;
+  private SecureStore secureStoreMock;
+  private FileConfigResource baseConfig;
+  private Injector testInjector;
+  private AtomicReference<ReplicationConfigOverrides> testOverrides;
+  private String url1;
+  private String remoteName1;
+  private String url2;
+  private String remoteName2;
+
+  @Before
+  public void setUp() throws Exception {
+    testSite = Files.createTempDirectory("replicationRemotesUpdateTest");
+    secureStoreMock = mock(SecureStore.class);
+    baseConfig = new FileConfigResource(new SitePaths(testSite));
+    testOverrides = new AtomicReference<>(new TestReplicationConfigOverrides());
+    testInjector =
+        Guice.createInjector(
+            new AbstractModule() {
+              @Override
+              protected void configure() {
+                bind(ConfigResource.class).toInstance(baseConfig);
+                bind(SecureStore.class).toInstance(secureStoreMock);
+                bind(ReplicationRemotesApi.class).to(ReplicationRemotesApiImpl.class);
+                DynamicItem.itemOf(binder(), ReplicationConfigOverrides.class);
+                DynamicItem.bind(binder(), ReplicationConfigOverrides.class)
+                    .toProvider(testOverrides::get);
+              }
+            });
+    url1 = "fake_url1";
+    remoteName1 = "site1";
+    url2 = "fake_url2";
+    remoteName2 = "site2";
+  }
+
+  @After
+  public void tearDown() throws Exception {
+    MoreFiles.deleteRecursively(testSite, ALLOW_INSECURE);
+  }
+
+  @Test
+  public void shouldThrowWhenNoRemotesInTheUpdate() {
+    Config update = new Config();
+    ReplicationRemotesApi objectUnderTest = getReplicationRemotesApi();
+
+    assertThrows(IllegalArgumentException.class, () -> objectUnderTest.update(update));
+
+    update.setString("non-remote", null, "value", "one");
+
+    assertThrows(IllegalArgumentException.class, () -> objectUnderTest.update(update));
+  }
+
+  @Test
+  public void shouldReturnEmptyConfigWhenNoRemotes() {
+    ReplicationRemotesApi objectUnderTest = getReplicationRemotesApi();
+    assertThat(objectUnderTest.get("site").getSections()).isEmpty();
+  }
+
+  @Test
+  public void addRemoteSectionsToBaseConfigWhenNoOverrides() throws Exception {
+    testOverrides.set(null);
+
+    Config update = new Config();
+    setRemoteSite(update, remoteName1, "url", url1);
+    setRemoteSite(update, remoteName2, "url", url2);
+    ReplicationRemotesApi objectUnderTest = getReplicationRemotesApi();
+
+    objectUnderTest.update(update);
+
+    assertRemoteSite(baseConfig.getConfig(), remoteName1, "url").isEqualTo(url1);
+    assertRemoteSite(baseConfig.getConfig(), remoteName2, "url").isEqualTo(url2);
+  }
+
+  @Test
+  public void shouldReturnRemotesFromBaseConfigWhenNoOverrides() {
+    ReplicationRemotesApi objectUnderTest = getReplicationRemotesApi();
+
+    baseConfig.getConfig().setString("remote", remoteName1, "url", url1);
+    baseConfig.getConfig().setString("remote", remoteName2, "url", url2);
+
+    Config remotesConfig = objectUnderTest.get(remoteName1, remoteName2);
+    assertRemoteSite(remotesConfig, remoteName1, "url").isEqualTo(url1);
+    assertRemoteSite(remotesConfig, remoteName2, "url").isEqualTo(url2);
+  }
+
+  @Test
+  public void addRemotesSectionToBaseOverridesConfig() throws Exception {
+    Config update = new Config();
+    setRemoteSite(update, remoteName1, "url", url1);
+    setRemoteSite(update, remoteName2, "url", url2);
+    ReplicationRemotesApi objectUnderTest = getReplicationRemotesApi();
+
+    objectUnderTest.update(update);
+
+    assertRemoteSite(testOverrides.get().getConfig(), remoteName1, "url").isEqualTo(url1);
+    assertRemoteSite(testOverrides.get().getConfig(), remoteName2, "url").isEqualTo(url2);
+    assertRemoteSite(baseConfig.getConfig(), remoteName1, "url").isNull();
+    assertRemoteSite(baseConfig.getConfig(), remoteName2, "url").isNull();
+
+    Config remotesConfig = objectUnderTest.get(remoteName1, remoteName2);
+    assertRemoteSite(remotesConfig, remoteName1, "url").isEqualTo(url1);
+    assertRemoteSite(remotesConfig, remoteName2, "url").isEqualTo(url2);
+  }
+
+  @Test
+  public void shouldSetPasswordViaSecureStoreButNotStoreInConfig() throws Exception {
+    Config update = new Config();
+    String password = "encrypted-password";
+    String remoteName = "site";
+    setRemoteSite(update, remoteName, "password", password);
+    ReplicationRemotesApi objectUnderTest = getReplicationRemotesApi();
+
+    objectUnderTest.update(update);
+
+    verify(secureStoreMock).setList("remote", remoteName, "password", List.of(password));
+    assertRemoteSite(baseConfig.getConfig(), remoteName, "password").isNull();
+    assertRemoteSite(testOverrides.get().getConfig(), remoteName, "password").isNull();
+    assertRemoteSite(objectUnderTest.get(remoteName), remoteName, "password").isNull();
+  }
+
+  @Test
+  public void shouldSetUsernameViaSecureStoreButNotStoreInConfig() throws Exception {
+    Config update = new Config();
+    String username = "encrypted-username";
+    String remoteName = "site";
+    setRemoteSite(update, remoteName, "username", username);
+    ReplicationRemotesApi objectUnderTest = getReplicationRemotesApi();
+
+    objectUnderTest.update(update);
+
+    verify(secureStoreMock).setList("remote", remoteName, "username", List.of(username));
+    assertRemoteSite(baseConfig.getConfig(), remoteName, "username").isNull();
+    assertRemoteSite(testOverrides.get().getConfig(), remoteName, "username").isNull();
+    assertRemoteSite(objectUnderTest.get(remoteName), remoteName, "username").isNull();
+  }
+
+  private void setRemoteSite(Config config, String remoteName, String name, String value) {
+    config.setString("remote", remoteName, name, value);
+  }
+
+  private StringSubject assertRemoteSite(Config config, String remoteName, String name) {
+    return assertThat(config.getString("remote", remoteName, name));
+  }
+
+  private ReplicationRemotesApi getReplicationRemotesApi() {
+    return testInjector.getInstance(ReplicationRemotesApi.class);
+  }
+
+  static class TestReplicationConfigOverrides implements ReplicationConfigOverrides {
+    private Config config = new Config();
+
+    @Override
+    public Config getConfig() {
+      return config;
+    }
+
+    @Override
+    public void update(Config update) throws IOException {
+      config = update;
+    }
+
+    @Override
+    public String getVersion() {
+      return "none";
+    }
+  }
+}
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 e800cff..94646c3 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationStorageIT.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationStorageIT.java
@@ -22,7 +22,7 @@
 import com.google.gerrit.acceptance.WaitUtil;
 import com.google.gerrit.entities.Project;
 import com.google.gerrit.extensions.api.projects.BranchInput;
-import com.googlesource.gerrit.plugins.replication.Destination.QueueInfo;
+import com.googlesource.gerrit.plugins.replication.Destination.Queue;
 import com.googlesource.gerrit.plugins.replication.ReplicationTasksStorage.ReplicateRefUpdate;
 import com.googlesource.gerrit.plugins.replication.api.ReplicationConfig.FilterType;
 import java.io.IOException;
@@ -303,11 +303,11 @@
     createTestProject(projectName);
 
     WaitUtil.waitUntil(
-        () -> isTaskRescheduled(destination.getQueueInfo(), urish), TEST_NEW_PROJECT_TIMEOUT);
+        () -> isTaskRescheduled(destination.getQueue(), urish), TEST_NEW_PROJECT_TIMEOUT);
     // replicationRetry is set to 1 minute which is the minimum value. That's why
     // should be safe to get the pushOne object from pending because it should be
     // here for one minute
-    PushOne pushOp = destination.getQueueInfo().pending.get(urish);
+    PushOne pushOp = destination.getQueue().pending.get(urish);
 
     WaitUtil.waitUntil(() -> pushOp.wasCanceled(), MAX_RETRY_WITH_TOLERANCE_TIMEOUT);
     WaitUtil.waitUntil(() -> isTaskCleanedUp(), TEST_TASK_FINISH_TIMEOUT);
@@ -333,7 +333,7 @@
     assertThat(listWaitingReplicationTasks(branchToDelete)).hasSize(1);
   }
 
-  private boolean isTaskRescheduled(QueueInfo queue, URIish uri) {
+  private boolean isTaskRescheduled(Queue queue, URIish uri) {
     PushOne pushOne = queue.pending.get(uri);
     return pushOne == null ? false : pushOne.isRetrying();
   }