Google Git
Sign in
gerrit/plugins/pull-replication/becf2f8b18c581e7c4403274911e3e6a25e9c603^!/.
commit
becf2f8b18c581e7c4403274911e3e6a25e9c603
[log]
author
Luca Milanesio <luca.milanesio@gmail.com>
Mon Oct 09 23:32:27 2023 +0100
committer
Luca Milanesio <luca.milanesio@gmail.com>
Mon Oct 16 15:12:37 2023 +0100
tree
02c953f97afa8b1dac8c816534ec8e82babb3c8f
parent
e490c2778bde2de9de8f0ef67b16f1d5440f0d93 [diff]
Do not rely on async/wait for synchronous fetch replication

Before this change, the request for a sync FetchAction was triggering
an async schedule of a replication task, with a replication delay of
zero.

The consequence was the blocking of the client thread for an async
replication that, potentially, could have been stuck in the queue
waiting for other async replication to complete.
Scheduling a replication task does not guarantee that the fetch will
start immediately because it is influenced by the whole logic behind the
replication queue mechanism:
- waiting for the runway
- being aggregated with an existing pending task
- appended to the bottom of the replication queue

When the client requests a sync replication, it typically has other
client resources associated and locked that won't tolerate the async
execution. Faking a sync execution with async/wait for a Future to
complete would not allow the client resources to be released.

Skip the replication queue altogether and rely on the direct
FetchOp execution when the client REST-API FetchAction call has
requested synchronous execution.

Bug: Issue 304123378
Change-Id: Ia9171dc9525f2543ba16b1eb78308b0580839cbf

diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/FetchAll.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/FetchAll.java
index 42310ff..5a7362a 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/FetchAll.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/FetchAll.java

@@ -30,11 +30,7 @@
   private final ReplicationStateListener stateLog;
 
   public interface Factory {
-    FetchAll create(
-        String urlMatch,
-        ReplicationFilter filter,
-        ReplicationState state,
-        ReplicationType replicationType);
+    FetchAll create(String urlMatch, ReplicationFilter filter, ReplicationState state, boolean now);
   }
 
   private final WorkQueue workQueue;
@@ -42,8 +38,8 @@
   private final String urlMatch;
   private final ReplicationFilter filter;
   private final ReplicationState state;
-  private final ReplicationType replicationType;
   private final SourcesCollection sources;
+  private final boolean now;
 
   @Inject
   protected FetchAll(
@@ -54,7 +50,7 @@
       @Assisted @Nullable String urlMatch,
       @Assisted ReplicationFilter filter,
       @Assisted ReplicationState state,
-      @Assisted ReplicationType replicationType) {
+      @Assisted boolean now) {
     this.workQueue = wq;
     this.projectCache = projectCache;
     this.stateLog = stateLog;
@@ -62,7 +58,7 @@
     this.urlMatch = urlMatch;
     this.filter = filter;
     this.state = state;
-    this.replicationType = replicationType;
+    this.now = now;
   }
 
   Future<?> schedule(long delay, TimeUnit unit) {
@@ -74,7 +70,7 @@
     try {
       for (Project.NameKey nameKey : projectCache.all()) {
         if (filter.matches(nameKey)) {
-          scheduleFullSync(nameKey, urlMatch, state, replicationType);
+          scheduleFullSync(nameKey, urlMatch, state);
         }
       }
     } catch (Exception e) {
@@ -83,16 +79,16 @@
     state.markAllFetchTasksScheduled();
   }
 
-  private void scheduleFullSync(
-      Project.NameKey project,
-      String urlMatch,
-      ReplicationState state,
-      ReplicationType replicationType) {
+  private void scheduleFullSync(Project.NameKey project, String urlMatch, ReplicationState state) {
 
     for (Source cfg : sources.getAll()) {
       if (cfg.wouldFetchProject(project)) {
         for (URIish uri : cfg.getURIs(project, urlMatch)) {
-          cfg.schedule(project, FetchOne.ALL_REFS, uri, state, replicationType, Optional.empty());
+          if (now) {
+            cfg.scheduleNow(project, FetchOne.ALL_REFS, uri, state, Optional.empty());
+          } else {
+            cfg.schedule(project, FetchOne.ALL_REFS, uri, state, Optional.empty());
+          }
         }
       }
     }

diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/FetchOne.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/FetchOne.java
index 3cd1271..370f2fb 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/FetchOne.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/FetchOne.java

@@ -293,22 +293,28 @@
     }
   }
 
+  public void runSync() {
+    try (TraceContext ctx = TraceContext.open().addTag(ID_KEY, HexFormat.fromInt(id))) {
+      doRunFetchOperation(ReplicationType.SYNC);
+    }
+  }
+
   public Set<TransportException> getFetchFailures() {
     return fetchFailures;
   }
 
   private void runFetchOperation() {
     try (TraceContext ctx = TraceContext.open().addTag(ID_KEY, HexFormat.fromInt(id))) {
-      doRunFetchOperation();
+      doRunFetchOperation(ReplicationType.ASYNC);
     }
   }
 
-  private void doRunFetchOperation() {
+  private void doRunFetchOperation(ReplicationType replicationType) {
     // Lock the queue, and remove ourselves, so we can't be modified once
     // we start replication (instead a new instance, with the same URI, is
     // created and scheduled for a future point in time.)
     //
-    if (!pool.requestRunway(this)) {
+    if (replicationType == ReplicationType.ASYNC && !pool.requestRunway(this)) {
       if (!canceled) {
         repLog.info(
             "[{}] Rescheduling replication from {} to avoid collision with an in-flight fetch task [{}].",
@@ -321,8 +327,9 @@
     }
 
     repLog.info(
-        "[{}] Replication from {} started for refs [{}] ...",
+        "[{}] {} replication from {} started for refs [{}] ...",
         taskIdHex,
+        replicationType,
         uri,
         String.join(",", getRefs()));
     Timer1.Context<String> context = metrics.start(config.getName());
@@ -338,8 +345,9 @@
               .flatMap(metrics -> metrics.stop(config.getName()))
               .map(NANOSECONDS::toMillis);
       repLog.info(
-          "[{}] Replication from {} completed in {}ms, {}ms delay, {} retries{}",
+          "[{}] {} replication from {} completed in {}ms, {}ms delay, {} retries{}",
           taskIdHex,
+          replicationType,
           uri,
           elapsed,
           delay,
@@ -368,11 +376,11 @@
       repLog.error(
           String.format("Terminal failure. Cannot replicate [%s] from %s", taskIdHex, uri), e);
     } catch (TransportException e) {
-      if (e instanceof LockFailureException) {
+      repLog.error("[{}] Cannot replicate from {}: {}", taskIdHex, uri, e.getMessage());
+      if (replicationType == ReplicationType.ASYNC && e instanceof LockFailureException) {
         lockRetryCount++;
         // The LockFailureException message contains both URI and reason
         // for this failure.
-        repLog.error("[{}] Cannot replicate from {}: {}", taskIdHex, uri, e.getMessage());
 
         // The remote fetch operation should be retried.
         if (lockRetryCount <= maxLockRetries) {
@@ -390,11 +398,10 @@
               taskIdHex,
               uri);
         }
-      } else {
+      } else if (replicationType == ReplicationType.ASYNC) {
         if (canceledWhileRunning.get()) {
           logCanceledWhileRunningException(e);
         } else {
-          repLog.error("Cannot replicate [{}] from {}", taskIdHex, uri, e);
           // The remote fetch operation should be retried.
           pool.reschedule(this, Source.RetryReason.TRANSPORT_ERROR);
         }
@@ -410,7 +417,10 @@
       if (git != null) {
         git.close();
       }
-      pool.notifyFinished(this);
+
+      if (replicationType == ReplicationType.ASYNC) {
+        pool.notifyFinished(this);
+      }
     }
   }
 
@@ -445,7 +455,7 @@
     }
   }
 
-  private List<RefSpec> getFetchRefSpecs() {
+  public List<RefSpec> getFetchRefSpecs() {
     List<RefSpec> configRefSpecs = config.getFetchRefSpecs();
     if (delta.isEmpty()) {
       return configRefSpecs;

diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/OnStartStop.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/OnStartStop.java
index 5cf8bb6..d8c4a8d 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/OnStartStop.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/OnStartStop.java

@@ -14,8 +14,6 @@
 
 package com.googlesource.gerrit.plugins.replication.pull;
 
-import static com.googlesource.gerrit.plugins.replication.pull.ReplicationType.ASYNC;
-
 import com.google.common.util.concurrent.Atomics;
 import com.google.gerrit.extensions.events.LifecycleListener;
 import com.google.gerrit.extensions.registration.DynamicItem;
@@ -67,7 +65,7 @@
               new FetchResultProcessing.GitUpdateProcessing(eventDispatcher.get()));
       fetchAllFuture.set(
           fetchAll
-              .create(null, ReplicationFilter.all(), state, ASYNC)
+              .create(null, ReplicationFilter.all(), state, false)
               .schedule(30, TimeUnit.SECONDS));
     }
 

diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationStateLogger.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationStateLogger.java
index a62f369..6fa96b4 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationStateLogger.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationStateLogger.java

@@ -47,6 +47,10 @@
   }
 
   private void stateWriteErr(String msg, ReplicationState[] states) {
+    if (states == null) {
+      return;
+    }
+
     for (ReplicationState rs : states) {
       if (rs != null) {
         rs.writeStdErr(msg);

diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/Source.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/Source.java
index 5e4314d..f37b77c 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/Source.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/Source.java

@@ -16,7 +16,6 @@
 
 import static com.googlesource.gerrit.plugins.replication.ReplicationFileBasedConfig.replaceName;
 import static com.googlesource.gerrit.plugins.replication.pull.FetchResultProcessing.resolveNodeName;
-import static com.googlesource.gerrit.plugins.replication.pull.ReplicationType.SYNC;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static java.util.concurrent.TimeUnit.SECONDS;
 
@@ -446,10 +445,9 @@
       Project.NameKey project,
       String ref,
       ReplicationState state,
-      ReplicationType replicationType,
       Optional<PullReplicationApiRequestMetrics> apiRequestMetrics) {
     URIish uri = getURI(project);
-    return schedule(project, ref, uri, state, replicationType, apiRequestMetrics);
+    return schedule(project, ref, uri, state, apiRequestMetrics, false);
   }
 
   public Future<?> schedule(
@@ -457,8 +455,26 @@
       String ref,
       URIish uri,
       ReplicationState state,
-      ReplicationType replicationType,
       Optional<PullReplicationApiRequestMetrics> apiRequestMetrics) {
+    return schedule(project, ref, uri, state, apiRequestMetrics, false);
+  }
+
+  public Future<?> scheduleNow(
+      Project.NameKey project,
+      String ref,
+      URIish uri,
+      ReplicationState state,
+      Optional<PullReplicationApiRequestMetrics> apiRequestMetrics) {
+    return schedule(project, ref, uri, state, apiRequestMetrics, true);
+  }
+
+  private Future<?> schedule(
+      Project.NameKey project,
+      String ref,
+      URIish uri,
+      ReplicationState state,
+      Optional<PullReplicationApiRequestMetrics> apiRequestMetrics,
+      boolean now) {
 
     repLog.info("scheduling replication {}:{} => {}", uri, ref, project);
     if (!shouldReplicate(project, ref, state)) {
@@ -505,7 +521,7 @@
         f =
             pool.schedule(
                 queueMetrics.runWithMetrics(this, e),
-                isSyncCall(replicationType) ? 0 : config.getDelay(),
+                now ? 0 : config.getDelay(),
                 TimeUnit.SECONDS);
         queueMetrics.incrementTaskScheduled(this);
       } else if (!e.getRefs().contains(ref)) {
@@ -521,6 +537,25 @@
     }
   }
 
+  public Optional<FetchOne> fetchSync(
+      Project.NameKey project,
+      String ref,
+      URIish uri,
+      ReplicationState state,
+      Optional<PullReplicationApiRequestMetrics> apiRequestMetrics) {
+    if (shouldReplicate(project, ref)
+        && (config.replicatePermissions() || !ref.equals(RefNames.REFS_CONFIG))) {
+
+      FetchOne e = opFactory.create(project, uri, apiRequestMetrics);
+      e.addRef(ref);
+      e.addState(ref, state);
+      e.runSync();
+      return Optional.of(e);
+    }
+
+    return Optional.empty();
+  }
+
   void scheduleDeleteProject(String uri, Project.NameKey project) {
     @SuppressWarnings("unused")
     ScheduledFuture<?> ignored =
@@ -544,10 +579,6 @@
     postReplicationScheduledEvent(e, ref);
   }
 
-  private boolean isSyncCall(ReplicationType replicationType) {
-    return SYNC.equals(replicationType);
-  }
-
   /**
    * It schedules again a FetchOp instance.
    *

diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/StartFetchCommand.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/StartFetchCommand.java
index 97f8e9e..fed33d7 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/StartFetchCommand.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/StartFetchCommand.java

@@ -14,8 +14,6 @@
 
 package com.googlesource.gerrit.plugins.replication.pull;
 
-import static com.googlesource.gerrit.plugins.replication.pull.ReplicationType.ASYNC;
-import static com.googlesource.gerrit.plugins.replication.pull.ReplicationType.SYNC;
 
 import com.google.gerrit.extensions.annotations.RequiresCapability;
 import com.google.gerrit.extensions.registration.DynamicItem;
@@ -80,10 +78,7 @@
       projectFilter = new ReplicationFilter(projectPatterns);
     }
 
-    future =
-        fetchFactory
-            .create(urlMatch, projectFilter, state, replicationType(now))
-            .schedule(0, TimeUnit.SECONDS);
+    future = fetchFactory.create(urlMatch, projectFilter, state, now).schedule(0, TimeUnit.SECONDS);
 
     if (wait) {
       if (future != null) {
@@ -111,10 +106,6 @@
     }
   }
 
-  private ReplicationType replicationType(Boolean now) {
-    return now ? SYNC : ASYNC;
-  }
-
   @Override
   public void writeStdOutSync(String message) {
     if (wait) {

diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchAction.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchAction.java
index fdb4f8f..04797bd 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchAction.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchAction.java

@@ -36,6 +36,7 @@
 import java.util.Optional;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeoutException;
+import org.eclipse.jgit.errors.TransportException;
 
 public class FetchAction implements RestModifyView<ProjectResource, Input> {
   private final FetchCommand command;
@@ -86,7 +87,8 @@
     } catch (InterruptedException
         | ExecutionException
         | IllegalStateException
-        | TimeoutException e) {
+        | TimeoutException
+        | TransportException e) {
       throw RestApiException.wrap(e.getMessage(), e);
     } catch (RemoteConfigurationMissingException e) {
       throw new UnprocessableEntityException(e.getMessage());
@@ -95,7 +97,7 @@
 
   private Response<?> applySync(Project.NameKey project, Input input)
       throws InterruptedException, ExecutionException, RemoteConfigurationMissingException,
-          TimeoutException {
+          TimeoutException, TransportException {
     command.fetchSync(project, input.label, input.refName);
     return Response.created(input);
   }

diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchCommand.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchCommand.java
index 3a502ef..1e11987 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchCommand.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchCommand.java

@@ -22,6 +22,7 @@
 import com.google.gerrit.server.events.EventDispatcher;
 import com.google.inject.Inject;
 import com.googlesource.gerrit.plugins.replication.pull.Command;
+import com.googlesource.gerrit.plugins.replication.pull.FetchOne;
 import com.googlesource.gerrit.plugins.replication.pull.FetchResultProcessing;
 import com.googlesource.gerrit.plugins.replication.pull.PullReplicationStateLogger;
 import com.googlesource.gerrit.plugins.replication.pull.ReplicationState;
@@ -29,11 +30,15 @@
 import com.googlesource.gerrit.plugins.replication.pull.Source;
 import com.googlesource.gerrit.plugins.replication.pull.SourcesCollection;
 import com.googlesource.gerrit.plugins.replication.pull.api.exception.RemoteConfigurationMissingException;
+import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import java.util.stream.Collectors;
+import org.eclipse.jgit.errors.TransportException;
+import org.eclipse.jgit.transport.RefSpec;
 
 public class FetchCommand implements Command {
 
@@ -60,13 +65,13 @@
       String refName,
       PullReplicationApiRequestMetrics apiRequestMetrics)
       throws InterruptedException, ExecutionException, RemoteConfigurationMissingException,
-          TimeoutException {
+          TimeoutException, TransportException {
     fetch(name, label, refName, ASYNC, Optional.of(apiRequestMetrics));
   }
 
   public void fetchSync(Project.NameKey name, String label, String refName)
       throws InterruptedException, ExecutionException, RemoteConfigurationMissingException,
-          TimeoutException {
+          TimeoutException, TransportException {
     fetch(name, label, refName, SYNC, Optional.empty());
   }
 
@@ -77,10 +82,11 @@
       ReplicationType fetchType,
       Optional<PullReplicationApiRequestMetrics> apiRequestMetrics)
       throws InterruptedException, ExecutionException, RemoteConfigurationMissingException,
-          TimeoutException {
+          TimeoutException, TransportException {
     ReplicationState state =
         fetchReplicationStateFactory.create(
             new FetchResultProcessing.CommandProcessing(this, eventDispatcher.get()));
+
     Optional<Source> source =
         sources.getAll().stream().filter(s -> s.getRemoteConfigName().equals(label)).findFirst();
     if (!source.isPresent()) {
@@ -90,9 +96,23 @@
     }
 
     try {
-      state.markAllFetchTasksScheduled();
-      Future<?> future = source.get().schedule(name, refName, state, fetchType, apiRequestMetrics);
-      future.get(source.get().getTimeout(), TimeUnit.SECONDS);
+      if (fetchType == ReplicationType.ASYNC) {
+        state.markAllFetchTasksScheduled();
+        Future<?> future = source.get().schedule(name, refName, state, apiRequestMetrics);
+        future.get(source.get().getTimeout(), TimeUnit.SECONDS);
+      } else {
+        Optional<FetchOne> maybeFetch =
+            source
+                .get()
+                .fetchSync(name, refName, source.get().getURI(name), state, apiRequestMetrics);
+        if (maybeFetch.map(FetchOne::getFetchRefSpecs).filter(List::isEmpty).isPresent()) {
+          fetchStateLog.warn(
+              String.format(
+                  "[%s] Nothing to fetch, ref-specs is empty", maybeFetch.get().getTaskIdHex()));
+        } else if (maybeFetch.map(fetch -> !fetch.hasSucceeded()).orElse(false)) {
+          throw newTransportException(maybeFetch.get());
+        }
+      }
     } catch (ExecutionException
         | IllegalStateException
         | TimeoutException
@@ -102,13 +122,27 @@
     }
 
     try {
-      state.waitForReplication(source.get().getTimeout());
+      if (fetchType == ReplicationType.ASYNC) {
+        state.waitForReplication(source.get().getTimeout());
+      }
     } catch (InterruptedException e) {
       writeStdErrSync("We are interrupted while waiting replication to complete");
       throw e;
     }
   }
 
+  private TransportException newTransportException(FetchOne fetchOne) {
+    List<RefSpec> fetchRefSpecs = fetchOne.getFetchRefSpecs();
+    String combinedErrorMessage =
+        fetchOne.getFetchFailures().stream()
+            .map(TransportException::getMessage)
+            .collect(Collectors.joining("\n"));
+    return new TransportException(
+        String.format(
+            "[%s] %s trying to fetch %s",
+            fetchOne.getTaskIdHex(), combinedErrorMessage, fetchRefSpecs));
+  }
+
   @Override
   public void writeStdOutSync(String message) {}
 

diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchJob.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchJob.java
index e15dd68..a613c0e 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchJob.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchJob.java

@@ -21,6 +21,7 @@
 import com.googlesource.gerrit.plugins.replication.pull.api.exception.RemoteConfigurationMissingException;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeoutException;
+import org.eclipse.jgit.errors.TransportException;
 
 public class FetchJob implements Runnable {
   private static final FluentLogger log = FluentLogger.forEnclosingClass();
@@ -54,7 +55,8 @@
     } catch (InterruptedException
         | ExecutionException
         | RemoteConfigurationMissingException
-        | TimeoutException e) {
+        | TimeoutException
+        | TransportException e) {
       log.atSevere().withCause(e).log(
           "Exception during the async fetch call for project %s, label %s and ref name %s",
           project.get(), input.label, input.refName);

diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationITAbstract.java b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationITAbstract.java
index da7be9f..8160304 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationITAbstract.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationITAbstract.java

@@ -21,6 +21,7 @@
 import static com.google.gerrit.acceptance.testsuite.project.TestProjectUpdate.allow;
 import static com.google.gerrit.server.group.SystemGroupBackend.REGISTERED_USERS;
 
+import com.google.common.base.Strings;
 import com.google.gerrit.acceptance.PushOneCommit.Result;
 import com.google.gerrit.acceptance.UseLocalDisk;
 import com.google.gerrit.acceptance.config.GerritConfig;
@@ -47,15 +48,18 @@
 import java.util.Optional;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
+import org.eclipse.jgit.errors.ConfigInvalidException;
 import org.eclipse.jgit.internal.storage.dfs.InMemoryRepository;
 import org.eclipse.jgit.junit.TestRepository;
 import org.eclipse.jgit.lib.ObjectId;
 import org.eclipse.jgit.lib.Ref;
 import org.eclipse.jgit.lib.Repository;
 import org.eclipse.jgit.revwalk.RevCommit;
+import org.eclipse.jgit.storage.file.FileBasedConfig;
 import org.eclipse.jgit.transport.PushResult;
 import org.eclipse.jgit.transport.RemoteRefUpdate;
 import org.eclipse.jgit.transport.RemoteRefUpdate.Status;
+import org.eclipse.jgit.util.FS;
 import org.junit.Ignore;
 import org.junit.Test;
 
@@ -119,6 +123,17 @@
     eventListener = plugin.getSysInjector().getInstance(BufferedEventListener.class);
   }
 
+  protected boolean isAsyncReplication() {
+    FileBasedConfig config =
+        new FileBasedConfig(sitePaths.etc_dir.resolve("replication.config").toFile(), FS.DETECTED);
+    try {
+      config.load();
+    } catch (IOException | ConfigInvalidException e) {
+      throw new IllegalStateException(e);
+    }
+    return !Strings.isNullOrEmpty(config.getString("replication", null, "syncRefs"));
+  }
+
   @Override
   protected void setReplicationSource(
       String remoteName, List<String> replicaSuffixes, Optional<String> project)
@@ -175,10 +190,12 @@
   }
 
   private void assertTasksMetricScheduledAndCompleted(int numTasks) {
-    assertTasksMetric("scheduled", numTasks);
-    assertTasksMetric("started", numTasks);
-    assertTasksMetric("completed", numTasks);
-    assertEmptyTasksMetric("failed");
+    if (isAsyncReplication()) {
+      assertTasksMetric("scheduled", numTasks);
+      assertTasksMetric("started", numTasks);
+      assertTasksMetric("completed", numTasks);
+      assertEmptyTasksMetric("failed");
+    }
   }
 
   @Test
@@ -559,12 +576,14 @@
   }
 
   private void waitUntilReplicationTask(String status, int expected) throws Exception {
-    waitUntil(
-        () ->
-            inMemoryMetrics()
-                .counterValue("tasks/" + status, TEST_REPLICATION_REMOTE)
-                .filter(counter -> counter == expected)
-                .isPresent());
+    if (isAsyncReplication()) {
+      waitUntil(
+          () ->
+              inMemoryMetrics()
+                  .counterValue("tasks/" + status, TEST_REPLICATION_REMOTE)
+                  .filter(counter -> counter == expected)
+                  .isPresent());
+    }
   }
 
   private InMemoryMetricMaker inMemoryMetrics() {

diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchCommandTest.java b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchCommandTest.java
index 156481b..1299905 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchCommandTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchCommandTest.java

@@ -15,10 +15,10 @@
 package com.googlesource.gerrit.plugins.replication.pull.api;
 
 import static com.google.gerrit.testing.GerritJUnit.assertThrows;
-import static com.googlesource.gerrit.plugins.replication.pull.ReplicationType.ASYNC;
 import static org.mockito.Mockito.any;
 import static org.mockito.Mockito.anyString;
 import static org.mockito.Mockito.eq;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
@@ -72,7 +72,7 @@
     when(fetchReplicationStateFactory.create(any())).thenReturn(state);
     when(source.getRemoteConfigName()).thenReturn(label);
     when(sources.getAll()).thenReturn(Lists.newArrayList(source));
-    when(source.schedule(eq(projectName), eq(REF_NAME_TO_FETCH), eq(state), any(), any()))
+    when(source.schedule(eq(projectName), eq(REF_NAME_TO_FETCH), eq(state), any()))
         .thenReturn(CompletableFuture.completedFuture(null));
     objectUnderTest =
         new FetchCommand(fetchReplicationStateFactory, fetchStateLog, sources, eventDispatcher);
@@ -83,7 +83,17 @@
     objectUnderTest.fetchAsync(projectName, label, REF_NAME_TO_FETCH, apiRequestMetrics);
 
     verify(source, times(1))
-        .schedule(projectName, REF_NAME_TO_FETCH, state, ASYNC, Optional.of(apiRequestMetrics));
+        .schedule(
+            eq(projectName), eq(REF_NAME_TO_FETCH), eq(state), eq(Optional.of(apiRequestMetrics)));
+  }
+
+  @Test
+  public void shouldNotScheduleAsyncTaskWhenFetchSync() throws Exception {
+    objectUnderTest.fetchSync(projectName, label, REF_NAME_TO_FETCH);
+
+    verify(source, never())
+        .schedule(
+            eq(projectName), eq(REF_NAME_TO_FETCH), eq(state), eq(Optional.of(apiRequestMetrics)));
   }
 
   @Test