diff --git a/BUILD b/BUILD
index 0983983..972f0d8 100644
--- a/BUILD
+++ b/BUILD
@@ -12,6 +12,7 @@
         "Gerrit-InitStep: com.googlesource.gerrit.plugins.replication.pull.InitPlugin",
         "Gerrit-SshModule: com.googlesource.gerrit.plugins.replication.pull.SshModule",
         "Gerrit-HttpModule: com.googlesource.gerrit.plugins.replication.pull.api.HttpModule",
+        "Gerrit-ReloadMode: restart",
     ],
     resources = glob(["src/main/resources/**/*"]),
     deps = [
@@ -24,6 +25,7 @@
 
 junit_tests(
     name = "pull_replication_tests",
+    size = "large",
     srcs = glob([
         "src/test/java/**/*Test.java",
     ]),
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/Completable.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/Completable.java
new file mode 100644
index 0000000..fa3d53b
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/Completable.java
@@ -0,0 +1,20 @@
+// Copyright (C) 2023 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.pull;
+
+interface Completable {
+
+  boolean hasSucceeded();
+}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/DeleteProjectTask.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/DeleteProjectTask.java
index 2882482..4818ec7 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/DeleteProjectTask.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/DeleteProjectTask.java
@@ -28,7 +28,7 @@
 import java.net.URISyntaxException;
 import org.eclipse.jgit.transport.URIish;
 
-public class DeleteProjectTask implements Runnable {
+public class DeleteProjectTask implements Runnable, Completable {
   private static final FluentLogger logger = FluentLogger.forEnclosingClass();
 
   interface Factory {
@@ -40,6 +40,7 @@
   private final String uri;
   private final Project.NameKey project;
   private final FetchApiClient.Factory fetchClientFactory;
+  private boolean succeeded;
 
   @Inject
   DeleteProjectTask(
@@ -63,6 +64,7 @@
       if (!httpResult.isSuccessful()) {
         throw new IOException(httpResult.getMessage().orElse("Unknown"));
       }
+      succeeded = true;
       logger.atFine().log("Successfully deleted project %s on remote %s", project.get(), uri);
     } catch (URISyntaxException | IOException e) {
       String errorMessage =
@@ -76,4 +78,9 @@
   public String toString() {
     return String.format("[%s] delete-project %s at %s", HexFormat.fromInt(id), project.get(), uri);
   }
+
+  @Override
+  public boolean hasSucceeded() {
+    return succeeded;
+  }
 }
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 39db764..a6b5ab7 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
@@ -67,7 +67,7 @@
  * <p>Instance members are protected by the lock within FetchQueue. Callers must take that lock to
  * ensure they are working with a current view of the object.
  */
-public class FetchOne implements ProjectRunnable, CanceledWhileRunning {
+public class FetchOne implements ProjectRunnable, CanceledWhileRunning, Completable {
   private final ReplicationStateListener stateLog;
   public static final String ALL_REFS = "..all..";
   static final String ID_KEY = "fetchOneId";
@@ -103,6 +103,7 @@
   private final FetchFactory fetchFactory;
   private final Optional<PullReplicationApiRequestMetrics> apiRequestMetrics;
   private DynamicItem<ReplicationFetchFilter> replicationFetchFilter;
+  private boolean succeeded;
 
   @Inject
   FetchOne(
@@ -177,7 +178,7 @@
 
   @Override
   public String toString() {
-    String print = "[" + taskIdHex + "] fetch " + uri;
+    String print = "[" + taskIdHex + "] fetch " + uri + " [" + String.join(",", delta) + "]";
 
     if (retryCount > 0) {
       print = "(retry " + retryCount + ") " + print;
@@ -503,6 +504,7 @@
         case FORCED:
         case RENAMED:
         case FAST_FORWARD:
+          succeeded = true;
           break;
         case NOT_ATTEMPTED:
         case REJECTED:
@@ -580,4 +582,9 @@
   public Optional<PullReplicationApiRequestMetrics> getRequestMetrics() {
     return apiRequestMetrics;
   }
+
+  @Override
+  public boolean hasSucceeded() {
+    return succeeded;
+  }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationModule.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationModule.java
index 03345e2..49a0ca1 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationModule.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationModule.java
@@ -24,6 +24,7 @@
 import com.google.gerrit.extensions.events.LifecycleListener;
 import com.google.gerrit.extensions.events.ProjectDeletedListener;
 import com.google.gerrit.extensions.registration.DynamicSet;
+import com.google.gerrit.metrics.MetricMaker;
 import com.google.gerrit.server.config.SitePaths;
 import com.google.gerrit.server.events.EventListener;
 import com.google.gerrit.server.events.EventTypes;
@@ -33,6 +34,7 @@
 import com.google.inject.Scopes;
 import com.google.inject.assistedinject.FactoryModuleBuilder;
 import com.google.inject.internal.UniqueAnnotations;
+import com.google.inject.name.Names;
 import com.googlesource.gerrit.plugins.replication.AutoReloadConfigDecorator;
 import com.googlesource.gerrit.plugins.replication.AutoReloadSecureCredentialsFactoryDecorator;
 import com.googlesource.gerrit.plugins.replication.ConfigParser;
@@ -66,15 +68,20 @@
 class PullReplicationModule extends AbstractModule {
   private final SitePaths site;
   private final Path cfgPath;
+  private final MetricMaker pluginMetricMaker;
 
   @Inject
-  public PullReplicationModule(SitePaths site) {
+  public PullReplicationModule(SitePaths site, MetricMaker pluginMetricMaker) {
     this.site = site;
     cfgPath = site.etc_dir.resolve("replication.config");
+    this.pluginMetricMaker = pluginMetricMaker;
   }
 
   @Override
   protected void configure() {
+    bind(MetricMaker.class)
+        .annotatedWith(Names.named(ReplicationQueueMetrics.REPLICATION_QUEUE_METRICS))
+        .toInstance(pluginMetricMaker);
 
     bind(CapabilityDefinition.class)
         .annotatedWith(Exports.named(CALL_FETCH_ACTION))
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/ReplicationQueue.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/ReplicationQueue.java
index b88db69..1ba47de 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/ReplicationQueue.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/ReplicationQueue.java
@@ -32,6 +32,7 @@
 import com.google.gerrit.server.git.WorkQueue;
 import com.google.inject.Inject;
 import com.google.inject.Provider;
+import com.google.inject.Singleton;
 import com.googlesource.gerrit.plugins.replication.ObservableQueue;
 import com.googlesource.gerrit.plugins.replication.pull.FetchResultProcessing.GitUpdateProcessing;
 import com.googlesource.gerrit.plugins.replication.pull.api.data.RevisionData;
@@ -65,6 +66,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+@Singleton
 public class ReplicationQueue
     implements ObservableQueue,
         EventListener,
@@ -79,6 +81,7 @@
   private static final String REF_UDPATED_EVENT_TYPE = new RefUpdatedEvent().type;
   private static final String ZEROS_OBJECTID = ObjectId.zeroId().getName();
   private final ReplicationStateListener stateLog;
+  private final ShutdownState shutdownState;
 
   private final WorkQueue workQueue;
   private final DynamicItem<EventDispatcher> dispatcher;
@@ -92,6 +95,7 @@
   private Provider<RevisionReader> revReaderProvider;
   private final ApplyObjectMetrics applyObjectMetrics;
   private final FetchReplicationMetrics fetchMetrics;
+  private final ReplicationQueueMetrics queueMetrics;
   private final String instanceId;
   private ApplyObjectsRefsFilter applyObjectsRefsFilter;
 
@@ -106,18 +110,22 @@
       Provider<RevisionReader> revReaderProvider,
       ApplyObjectMetrics applyObjectMetrics,
       FetchReplicationMetrics fetchMetrics,
+      ReplicationQueueMetrics queueMetrics,
       @GerritInstanceId String instanceId,
-      ApplyObjectsRefsFilter applyObjectsRefsFilter) {
+      ApplyObjectsRefsFilter applyObjectsRefsFilter,
+      ShutdownState shutdownState) {
     workQueue = wq;
     dispatcher = dis;
     sources = rd;
     stateLog = sl;
+    this.shutdownState = shutdownState;
     beforeStartupEventsQueue = Queues.newConcurrentLinkedQueue();
     this.fetchClientFactory = fetchClientFactory;
     this.refsFilter = refsFilter;
     this.revReaderProvider = revReaderProvider;
     this.applyObjectMetrics = applyObjectMetrics;
     this.fetchMetrics = fetchMetrics;
+    this.queueMetrics = queueMetrics;
     this.instanceId = instanceId;
     this.applyObjectsRefsFilter = applyObjectsRefsFilter;
   }
@@ -126,6 +134,7 @@
   public void start() {
     if (!running) {
       sources.get().startup(workQueue);
+      queueMetrics.start(this);
       fetchCallsTimeout =
           2
               * sources.get().getAll().stream()
@@ -141,10 +150,12 @@
   @Override
   public void stop() {
     running = false;
+    shutdownState.setIsShuttingDown(true);
     int discarded = sources.get().shutdown();
     if (discarded > 0) {
       repLog.warn("Canceled {} replication events during shutdown", discarded);
     }
+    queueMetrics.stop();
   }
 
   @Override
@@ -211,6 +222,8 @@
           "Replication plugin did not finish startup before event, event replication is postponed",
           state);
       beforeStartupEventsQueue.add(event);
+
+      queueMetrics.incrementQueuedBeforStartup();
       return;
     }
     ForkJoinPool fetchCallsPool = null;
@@ -538,6 +551,10 @@
                     .forEach(apiUrl -> s.scheduleUpdateHead(apiUrl, p, event.getNewHeadName())));
   }
 
+  SourcesCollection sourcesCollection() {
+    return sources.get();
+  }
+
   @AutoValue
   abstract static class ReferenceUpdatedEvent {
 
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/ReplicationQueueMetrics.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/ReplicationQueueMetrics.java
new file mode 100644
index 0000000..741927f
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/ReplicationQueueMetrics.java
@@ -0,0 +1,276 @@
+// Copyright (C) 2023 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.pull;
+
+import com.google.gerrit.extensions.annotations.PluginName;
+import com.google.gerrit.extensions.registration.RegistrationHandle;
+import com.google.gerrit.metrics.CallbackMetric1;
+import com.google.gerrit.metrics.Counter0;
+import com.google.gerrit.metrics.Counter1;
+import com.google.gerrit.metrics.Description;
+import com.google.gerrit.metrics.Field;
+import com.google.gerrit.metrics.MetricMaker;
+import com.google.gerrit.server.logging.PluginMetadata;
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+import com.google.inject.name.Named;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.function.Function;
+
+@Singleton
+public class ReplicationQueueMetrics {
+  private static final String EVENTS = "events";
+  private static final String TASKS = "tasks";
+  public static final String REPLICATION_QUEUE_METRICS = "ReplicationQueueMetrics";
+
+  private final Counter1<String> tasksScheduled;
+  private final Counter1<String> tasksCancelled;
+  private final Counter1<String> tasksNotScheduled;
+  private final Counter1<String> tasksRescheduled;
+  private final Counter1<String> tasksCompleted;
+  private final Counter1<String> tasksMerged;
+  private final Counter1<String> tasksFailed;
+  private final Counter1<String> tasksRetrying;
+  private final Counter0 eventsQueuedBeforeStartup;
+  private final Counter1<String> tasksCancelledMaxRetries;
+  private final MetricMaker metricMaker;
+  private final Field<String> sourceField;
+  private final Counter1<String> tasksStarted;
+  private final Set<RegistrationHandle> metricsHandles;
+
+  public class RunnableWithMetrics implements Runnable {
+    private final Source source;
+    private final Runnable runnable;
+
+    public RunnableWithMetrics(Source source, Runnable runnable) {
+      this.source = source;
+      this.runnable = runnable;
+    }
+
+    @Override
+    public void run() {
+      incrementTaskStarted(source);
+      runnable.run();
+      if (runnable instanceof Completable) {
+        if (((Completable) runnable).hasSucceeded()) {
+          incrementTaskCompleted(source);
+        }
+      }
+    }
+  }
+
+  @Inject
+  public ReplicationQueueMetrics(
+      @PluginName String pluginName, @Named(REPLICATION_QUEUE_METRICS) MetricMaker metricMaker) {
+    metricsHandles = new HashSet<>();
+
+    sourceField =
+        Field.ofString(
+                "source",
+                (metadataBuilder, fieldValue) ->
+                    metadataBuilder
+                        .pluginName(pluginName)
+                        .addPluginMetadata(PluginMetadata.create("source", fieldValue)))
+            .build();
+
+    eventsQueuedBeforeStartup =
+        registerMetric(
+            metricMaker.newCounter(
+                "events/queued_before_startup",
+                new Description("Replication events queued before startup")
+                    .setCumulative()
+                    .setUnit(EVENTS)));
+
+    tasksScheduled =
+        registerMetric(
+            metricMaker.newCounter(
+                "tasks/scheduled",
+                new Description("Replication tasks scheduled").setCumulative().setUnit(TASKS),
+                sourceField));
+
+    tasksStarted =
+        registerMetric(
+            metricMaker.newCounter(
+                "tasks/started",
+                new Description("Replication tasks started").setCumulative().setUnit(TASKS),
+                sourceField));
+
+    tasksRescheduled =
+        registerMetric(
+            metricMaker.newCounter(
+                "tasks/rescheduled",
+                new Description("Replication tasks rescheduled").setCumulative().setUnit(TASKS),
+                sourceField));
+
+    tasksCancelled =
+        registerMetric(
+            metricMaker.newCounter(
+                "tasks/cancelled",
+                new Description("Replication tasks cancelled").setCumulative().setUnit(TASKS),
+                sourceField));
+
+    tasksFailed =
+        registerMetric(
+            metricMaker.newCounter(
+                "tasks/failed",
+                new Description("Replication tasks failed").setCumulative().setUnit(TASKS),
+                sourceField));
+
+    tasksRetrying =
+        registerMetric(
+            metricMaker.newCounter(
+                "tasks/retrying",
+                new Description("Replication tasks retrying").setCumulative().setUnit(TASKS),
+                sourceField));
+
+    tasksNotScheduled =
+        registerMetric(
+            metricMaker.newCounter(
+                "tasks/not_scheduled",
+                new Description("Replication tasks not scheduled").setCumulative().setUnit(TASKS),
+                sourceField));
+
+    tasksCompleted =
+        registerMetric(
+            metricMaker.newCounter(
+                "tasks/completed",
+                new Description("Replication tasks completed").setCumulative().setUnit(TASKS),
+                sourceField));
+
+    tasksMerged =
+        registerMetric(
+            metricMaker.newCounter(
+                "tasks/merged",
+                new Description("Replication tasks merged").setCumulative().setUnit(TASKS),
+                sourceField));
+
+    tasksCancelledMaxRetries =
+        registerMetric(
+            metricMaker.newCounter(
+                "tasks/failed_max_retries",
+                new Description("Replication tasks cancelled for maximum number of retries")
+                    .setCumulative()
+                    .setUnit(TASKS),
+                sourceField));
+
+    this.metricMaker = metricMaker;
+  }
+
+  private <T extends RegistrationHandle> T registerMetric(T metricHandle) {
+    metricsHandles.add(metricHandle);
+    return metricHandle;
+  }
+
+  void start(ReplicationQueue queue) {
+    initCallbackMetrics(
+        queue,
+        Source::inflightTasksCount,
+        "tasks/inflight",
+        "In-flight replication tasks per source");
+    initCallbackMetrics(
+        queue, Source::pendingTasksCount, "tasks/pending", "Pending replication tasks per source");
+  }
+
+  void stop() {
+    metricsHandles.forEach(RegistrationHandle::remove);
+  }
+
+  private void initCallbackMetrics(
+      ReplicationQueue queue,
+      Function<Source, Long> sourceMetricFunc,
+      String metricName,
+      String description) {
+    CallbackMetric1<String, Long> metric =
+        registerMetric(
+            metricMaker.newCallbackMetric(
+                metricName,
+                Long.class,
+                new Description(description).setGauge().setUnit(TASKS),
+                sourceField));
+    registerMetric(
+        metricMaker.newTrigger(
+            metric,
+            () -> {
+              List<Source> sources = queue.sourcesCollection().getAll();
+              if (sources.isEmpty()) {
+                metric.forceCreate("");
+              } else {
+                sources.forEach(
+                    (source) -> {
+                      metric.set(source.getRemoteConfigName(), sourceMetricFunc.apply(source));
+                      metric.prune();
+                    });
+              }
+            }));
+  }
+
+  public void incrementTaskScheduled(Source source) {
+    tasksScheduled.increment(source.getRemoteConfigName());
+  }
+
+  public void incrementTaskCancelled(Source source) {
+    tasksCancelled.increment(source.getRemoteConfigName());
+  }
+
+  public void incrementQueuedBeforStartup() {
+    eventsQueuedBeforeStartup.increment();
+  }
+
+  public void incrementTaskCompleted(Source source) {
+    tasksCompleted.increment(source.getRemoteConfigName());
+  }
+
+  public void failTask(Source source) {
+    tasksFailed.increment(source.getRemoteConfigName());
+  }
+
+  public void incrementTaskMerged(Source source) {
+    tasksMerged.increment(source.getRemoteConfigName());
+  }
+
+  public void incrementTaskNotScheduled(Source source) {
+    tasksNotScheduled.increment(source.getRemoteConfigName());
+  }
+
+  public void incrementTaskRescheduled(Source source) {
+    tasksRescheduled.increment(source.getRemoteConfigName());
+  }
+
+  public void incrementTaskFailed(Source source) {
+    tasksFailed.increment(source.getRemoteConfigName());
+  }
+
+  public void incrementTaskRetrying(Source source) {
+    tasksRetrying.increment(source.getRemoteConfigName());
+  }
+
+  public void incrementTaskCancelledMaxRetries(Source source) {
+    tasksCancelledMaxRetries.increment(source.getRemoteConfigName());
+  }
+
+  public void incrementTaskStarted(Source source) {
+    tasksStarted.increment(source.getRemoteConfigName());
+  }
+
+  public Runnable runWithMetrics(Source source, Runnable runnableTask) {
+    if (runnableTask instanceof RunnableWithMetrics) {
+      return runnableTask;
+    }
+
+    return new RunnableWithMetrics(source, runnableTask);
+  }
+}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/ShutdownState.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/ShutdownState.java
new file mode 100644
index 0000000..0f99a7f
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/ShutdownState.java
@@ -0,0 +1,41 @@
+// Copyright (C) 2023 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.pull;
+
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * Singleton object representing this plugin shutdown state. This class can be injected any consumer
+ * interested in knowing if a shutdown was initiated.
+ */
+@Singleton
+public class ShutdownState {
+  private final AtomicBoolean isShuttingDown;
+
+  @Inject
+  public ShutdownState() {
+    isShuttingDown = new AtomicBoolean(false);
+  }
+
+  public void setIsShuttingDown(boolean shutting) {
+    isShuttingDown.set(shutting);
+  }
+
+  public boolean isShuttingDown() {
+    return isShuttingDown.get();
+  }
+}
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 f75ea7b..5ab3859 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,10 @@
 
 import static com.googlesource.gerrit.plugins.replication.ReplicationFileBasedConfig.replaceName;
 import static com.googlesource.gerrit.plugins.replication.pull.ReplicationType.SYNC;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static java.util.concurrent.TimeUnit.SECONDS;
 
+import com.google.common.base.Stopwatch;
 import com.google.common.base.Throwables;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
@@ -72,6 +75,7 @@
 import java.io.UnsupportedEncodingException;
 import java.net.URISyntaxException;
 import java.net.URLEncoder;
+import java.time.Duration;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -84,6 +88,7 @@
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
+import java.util.stream.Collectors;
 import org.apache.commons.io.FilenameUtils;
 import org.apache.http.impl.client.CloseableHttpClient;
 import org.eclipse.jgit.errors.TransportException;
@@ -119,6 +124,9 @@
   private final DynamicItem<EventDispatcher> eventDispatcher;
   private CloseableHttpClient httpClient;
   private final DeleteProjectTask.Factory deleteProjectFactory;
+  private final ReplicationQueueMetrics queueMetrics;
+  private static final int DRAINED_CHECK_FREQUENCY_MS = 50;
+  private static final int DRAINED_LOGGING_FREQUENCY_SECS = 5;
 
   protected enum RetryReason {
     TRANSPORT_ERROR,
@@ -148,7 +156,8 @@
       GroupBackend groupBackend,
       ReplicationStateListeners stateLog,
       GroupIncludeCache groupIncludeCache,
-      DynamicItem<EventDispatcher> eventDispatcher) {
+      DynamicItem<EventDispatcher> eventDispatcher,
+      ReplicationQueueMetrics queueMetrics) {
     config = cfg;
     this.eventDispatcher = eventDispatcher;
     gitManager = gitRepositoryManager;
@@ -156,6 +165,7 @@
     this.userProvider = userProvider;
     this.projectCache = projectCache;
     this.stateLog = stateLog;
+    this.queueMetrics = queueMetrics;
 
     CurrentUser remoteUser;
     if (!cfg.getAuthGroupNames().isEmpty()) {
@@ -256,7 +266,15 @@
   public synchronized int shutdown() {
     int cnt = 0;
     if (pool != null) {
-      cnt = pool.shutdownNow().size();
+      try {
+        waitUntil(this::isDrained, Duration.ofSeconds(config.getShutDownDrainTimeout()));
+        cnt = pool.shutdownNow().size();
+      } catch (InterruptedException e) {
+        logger.atSevere().withCause(e).log("Interrupted during termination.");
+        List<Runnable> fetchTasks = pool.shutdownNow();
+        logInterruptedShutdownStatus(fetchTasks);
+        cnt = fetchTasks.size();
+      }
       pool = null;
     }
     if (httpClient != null) {
@@ -271,6 +289,46 @@
     return cnt;
   }
 
+  private void logInterruptedShutdownStatus(List<Runnable> fetchTasks) {
+    String neverExecutedTasks =
+        fetchTasks.stream().map(r -> r.toString()).collect(Collectors.joining(","));
+    String pendingTasks =
+        pending.values().stream().map(FetchOne::toString).collect(Collectors.joining(","));
+    String inFlightTasks =
+        inFlight.values().stream().map(FetchOne::toString).collect(Collectors.joining(","));
+
+    repLog.error("Never executed tasks: {}", neverExecutedTasks);
+    repLog.error("Pending tasks: {}", pendingTasks);
+    repLog.error("In-flight tasks: {}", inFlightTasks);
+  }
+
+  private boolean isDrained() {
+    int numberOfPending = pending.size();
+    int numberOfInFlight = inFlight.size();
+
+    boolean drained = numberOfPending == 0 && numberOfInFlight == 0;
+
+    if (!drained) {
+      logger.atWarning().atMostEvery(DRAINED_LOGGING_FREQUENCY_SECS, SECONDS).log(
+          "Queue not drained: %d pending|%d in-flight", numberOfPending, numberOfInFlight);
+    } else {
+      logger.atWarning().log("Queue drained");
+    }
+
+    return drained;
+  }
+
+  private static void waitUntil(Supplier<Boolean> waitCondition, Duration timeout)
+      throws InterruptedException {
+    Stopwatch stopwatch = Stopwatch.createStarted();
+    while (!waitCondition.get()) {
+      if (stopwatch.elapsed().compareTo(timeout) > 0) {
+        throw new InterruptedException();
+      }
+      MILLISECONDS.sleep(DRAINED_CHECK_FREQUENCY_MS);
+    }
+  }
+
   private boolean shouldReplicate(ProjectState state, CurrentUser user)
       throws PermissionBackendException {
     if (!config.replicateHiddenProjects()
@@ -402,6 +460,7 @@
 
     repLog.info("scheduling replication {}:{} => {}", uri, ref, project);
     if (!shouldReplicate(project, ref, state)) {
+      queueMetrics.incrementTaskNotScheduled(this);
       return CompletableFuture.completedFuture(null);
     }
 
@@ -417,14 +476,17 @@
             if (head != null
                 && head.isSymbolic()
                 && RefNames.REFS_CONFIG.equals(head.getLeaf().getName())) {
+              queueMetrics.incrementTaskNotScheduled(this);
               return CompletableFuture.completedFuture(null);
             }
           } catch (IOException err) {
             stateLog.error(String.format("cannot check type of project %s", project), err, state);
+            queueMetrics.incrementTaskNotScheduled(this);
             return CompletableFuture.completedFuture(null);
           }
         } catch (IOException err) {
           stateLog.error(String.format("source project %s not available", project), err, state);
+          queueMetrics.incrementTaskNotScheduled(this);
           return CompletableFuture.completedFuture(null);
         }
       }
@@ -438,10 +500,18 @@
         addRef(e, ref);
         e.addState(ref, state);
         pending.put(uri, e);
-        f = pool.schedule(e, isSyncCall(replicationType) ? 0 : config.getDelay(), TimeUnit.SECONDS);
+        f =
+            pool.schedule(
+                queueMetrics.runWithMetrics(this, e),
+                isSyncCall(replicationType) ? 0 : config.getDelay(),
+                TimeUnit.SECONDS);
+        queueMetrics.incrementTaskScheduled(this);
       } else if (!e.getRefs().contains(ref)) {
         addRef(e, ref);
         e.addState(ref, state);
+        queueMetrics.incrementTaskMerged(this);
+      } else {
+        queueMetrics.incrementTaskNotScheduled(this);
       }
       state.increaseFetchTaskCount(project.get(), ref);
       repLog.info("scheduled {}:{} => {} to run after {}s", e, ref, project, config.getDelay());
@@ -452,13 +522,18 @@
   void scheduleDeleteProject(String uri, Project.NameKey project) {
     @SuppressWarnings("unused")
     ScheduledFuture<?> ignored =
-        pool.schedule(deleteProjectFactory.create(this, uri, project), 0, TimeUnit.SECONDS);
+        pool.schedule(
+            queueMetrics.runWithMetrics(this, deleteProjectFactory.create(this, uri, project)),
+            0,
+            TimeUnit.SECONDS);
+    queueMetrics.incrementTaskScheduled(this);
   }
 
   void fetchWasCanceled(FetchOne fetchOp) {
     synchronized (stateLock) {
       URIish uri = fetchOp.getURI();
       pending.remove(uri);
+      queueMetrics.incrementTaskCancelled(this);
     }
   }
 
@@ -522,6 +597,8 @@
                   fetchOp.getTaskIdHex(), fetchOp.getURI(), pendingFetchOp.getTaskIdHex()),
               fetchOp.getStatesAsArray());
 
+          queueMetrics.incrementTaskRescheduled(this);
+
         } else {
           // The one pending is one that is NOT retrying, it was just
           // scheduled believing no problem would happen. The one pending
@@ -546,6 +623,8 @@
                   "[%s] Merging the pending fetch from [%s] with task [%s] and rescheduling",
                   pendingFetchOp.getTaskIdHex(), pendingFetchOp.getURI(), fetchOp.getTaskIdHex()),
               pendingFetchOp.getStatesAsArray());
+
+          queueMetrics.incrementTaskMerged(this);
         }
       }
 
@@ -553,7 +632,11 @@
         pending.put(uri, fetchOp);
         switch (reason) {
           case COLLISION:
-            pool.schedule(fetchOp, config.getRescheduleDelay(), TimeUnit.SECONDS);
+            queueMetrics.incrementTaskRescheduled(this);
+            pool.schedule(
+                queueMetrics.runWithMetrics(this, fetchOp),
+                config.getRescheduleDelay(),
+                TimeUnit.SECONDS);
             break;
           case TRANSPORT_ERROR:
           case REPOSITORY_MISSING:
@@ -563,15 +646,19 @@
                     ? RefUpdate.Result.NOT_ATTEMPTED
                     : RefUpdate.Result.REJECTED_OTHER_REASON;
             postReplicationFailedEvent(fetchOp, trackingRefUpdate);
+            queueMetrics.incrementTaskFailed(this);
+
             if (fetchOp.setToRetry()) {
               postReplicationScheduledEvent(fetchOp);
               pool.schedule(fetchOp, config.getRetryDelay(), TimeUnit.MINUTES);
+              queueMetrics.incrementTaskRetrying(this);
             } else {
               fetchOp.canceledByReplication();
               pending.remove(uri);
               stateLog.error(
                   "Fetch from " + fetchOp.getURI() + " cancelled after maximum number of retries",
                   fetchOp.getStatesAsArray());
+              queueMetrics.incrementTaskCancelledMaxRetries(this);
             }
             break;
         }
@@ -794,13 +881,25 @@
       @SuppressWarnings("unused")
       ScheduledFuture<?> ignored =
           pool.schedule(
-              updateHeadFactory.create(this, apiURI, project, newHead), 0, TimeUnit.SECONDS);
+              queueMetrics.runWithMetrics(
+                  this, updateHeadFactory.create(this, apiURI, project, newHead)),
+              0,
+              TimeUnit.SECONDS);
+      queueMetrics.incrementTaskScheduled(this);
     } catch (URISyntaxException e) {
       logger.atSevere().withCause(e).log(
           "Could not schedule HEAD pull-replication for project %s", project.get());
     }
   }
 
+  long inflightTasksCount() {
+    return inFlight.size();
+  }
+
+  long pendingTasksCount() {
+    return pending.size();
+  }
+
   private static boolean matches(URIish uri, String urlMatch) {
     if (urlMatch == null || urlMatch.equals("") || urlMatch.equals("*")) {
       return true;
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/SourceConfiguration.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/SourceConfiguration.java
index aff1266..0e840bd 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/SourceConfiguration.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/SourceConfiguration.java
@@ -29,6 +29,7 @@
   static final int DEFAULT_MAX_CONNECTION_INACTIVITY_MS = 10000;
   static final int DEFAULT_CONNECTION_TIMEOUT_MS = 5000;
   static final int DEFAULT_CONNECTIONS_PER_ROUTE = 100;
+  static final int DEFAULT_DRAIN_SHUTDOWN_TIMEOUT_SECS = 300;
 
   private final int delay;
   private final int rescheduleDelay;
@@ -51,6 +52,7 @@
   private final int maxConnectionsPerRoute;
   private final int maxConnections;
   private final int maxRetries;
+  private final int shutDownDrainTimeout;
   private int slowLatencyThreshold;
   private boolean useCGitClient;
   private int refsBatchSize;
@@ -98,6 +100,16 @@
                 "slowLatencyThreshold",
                 DEFAULT_SLOW_LATENCY_THRESHOLD_SECS,
                 TimeUnit.SECONDS);
+
+    shutDownDrainTimeout =
+        (int)
+            ConfigUtil.getTimeUnit(
+                cfg,
+                "replication",
+                null,
+                "shutDownDrainTimeout",
+                DEFAULT_DRAIN_SHUTDOWN_TIMEOUT_SECS,
+                TimeUnit.SECONDS);
   }
 
   @Override
@@ -216,4 +228,8 @@
   public int getPushBatchSize() {
     return 0;
   }
+
+  public int getShutDownDrainTimeout() {
+    return shutDownDrainTimeout;
+  }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/SourcesCollection.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/SourcesCollection.java
index 53adaaa..8bd1d95 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/SourcesCollection.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/SourcesCollection.java
@@ -40,7 +40,7 @@
 
   private final Source.Factory sourceFactory;
   private volatile Map<String, Source> sources;
-  private boolean shuttingDown;
+  private final ShutdownState shutdownState;
   private final Provider<ReplicationQueue> replicationQueue;
 
   @Inject
@@ -49,9 +49,11 @@
       ConfigParser configParser,
       Source.Factory sourceFactory,
       EventBus eventBus,
-      Provider<ReplicationQueue> replicationQueue)
+      Provider<ReplicationQueue> replicationQueue,
+      ShutdownState shutdownState)
       throws ConfigInvalidException {
     this.sourceFactory = sourceFactory;
+    this.shutdownState = shutdownState;
     this.sources =
         allSources(sourceFactory, configParser.parseRemotes(replicationConfig.getConfig()));
     this.replicationQueue = replicationQueue;
@@ -78,7 +80,7 @@
 
   @Override
   public void startup(WorkQueue workQueue) {
-    shuttingDown = false;
+    shutdownState.setIsShuttingDown(false);
     for (Source cfg : sources.values()) {
       cfg.start(workQueue);
     }
@@ -99,9 +101,7 @@
    */
   @Override
   public int shutdown() {
-    synchronized (this) {
-      shuttingDown = true;
-    }
+    shutdownState.setIsShuttingDown(true);
 
     int discarded = 0;
     for (Source cfg : sources.values()) {
@@ -117,7 +117,7 @@
 
   @Subscribe
   public synchronized void onReload(List<RemoteConfiguration> sourceConfigurations) {
-    if (shuttingDown) {
+    if (shutdownState.isShuttingDown()) {
       logger.atWarning().log("Shutting down: configuration reload ignored");
       return;
     }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/UpdateHeadTask.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/UpdateHeadTask.java
index fabe3cd..9dfe26e 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/UpdateHeadTask.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/UpdateHeadTask.java
@@ -27,7 +27,7 @@
 import java.io.IOException;
 import org.eclipse.jgit.transport.URIish;
 
-public class UpdateHeadTask implements Runnable {
+public class UpdateHeadTask implements Runnable, Completable {
   private static final FluentLogger logger = FluentLogger.forEnclosingClass();
   private final FetchApiClient.Factory fetchClientFactory;
   private final Source source;
@@ -35,6 +35,7 @@
   private final Project.NameKey project;
   private final String newHead;
   private final int id;
+  private boolean succeeded;
 
   interface Factory {
     UpdateHeadTask create(Source source, URIish apiURI, Project.NameKey project, String newHead);
@@ -64,6 +65,7 @@
       if (!httpResult.isSuccessful()) {
         throw new IOException(httpResult.getMessage().orElse("Unknown"));
       }
+      succeeded = true;
       logger.atFine().log(
           "Successfully updated HEAD of project %s on remote %s",
           project.get(), apiURI.toASCIIString());
@@ -83,4 +85,9 @@
         "[%s] update-head %s at %s to %s",
         HexFormat.fromInt(id), project.get(), apiURI.toASCIIString(), newHead);
   }
+
+  @Override
+  public boolean hasSucceeded() {
+    return succeeded;
+  }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/event/EventsBrokerMessageConsumer.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/event/EventsBrokerMessageConsumer.java
index 6fda9d3..d06681b 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/event/EventsBrokerMessageConsumer.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/event/EventsBrokerMessageConsumer.java
@@ -24,22 +24,26 @@
 import com.google.gerrit.server.permissions.PermissionBackendException;
 import com.google.inject.Inject;
 import com.google.inject.name.Named;
+import com.googlesource.gerrit.plugins.replication.pull.ShutdownState;
 import java.util.function.Consumer;
 
 public class EventsBrokerMessageConsumer implements Consumer<Event>, LifecycleListener {
 
   private final DynamicItem<BrokerApi> eventsBroker;
   private final StreamEventListener eventListener;
+  private final ShutdownState shutdownState;
   private final String eventsTopicName;
 
   @Inject
   public EventsBrokerMessageConsumer(
       DynamicItem<BrokerApi> eventsBroker,
       StreamEventListener eventListener,
+      ShutdownState shutdownState,
       @Named(STREAM_EVENTS_TOPIC_NAME) String eventsTopicName) {
 
     this.eventsBroker = eventsBroker;
     this.eventListener = eventListener;
+    this.shutdownState = shutdownState;
     this.eventsTopicName = eventsTopicName;
   }
 
@@ -47,6 +51,7 @@
   public void accept(Event event) {
     try {
       eventListener.fetchRefsForEvent(event);
+      if (shutdownState.isShuttingDown()) stop();
     } catch (AuthException | PermissionBackendException e) {
       throw new EventRejectedException(event, e);
     }
@@ -59,6 +64,7 @@
 
   @Override
   public void stop() {
+    shutdownState.setIsShuttingDown(true);
     eventsBroker.get().disconnect();
   }
 }
diff --git a/src/main/resources/Documentation/config.md b/src/main/resources/Documentation/config.md
index 65b03b8..6287773 100644
--- a/src/main/resources/Documentation/config.md
+++ b/src/main/resources/Documentation/config.md
@@ -341,6 +341,30 @@
 
 	Default: 10000
 
+replication.shutDownDrainTimeout
+:   Maximum duration to wait for pending, retrying and executing fetch tasks to
+	complete after a request of plugin shutdown. Values should use common unit
+	suffixes to express their setting:
+
+	s, sec, second, seconds
+
+	m, min, minute, minutes
+
+	h, hr, hour, hours
+
+	d, day, days
+
+	w, week, weeks (1 week is treated as 7 days)
+
+	mon, month, months (1 month is treated as 30 days)
+
+	y, year, years (1 year is treated as 365 days)
+
+	If a unit suffix is not specified, seconds is assumed. If 0 is supplied, the maximum age
+	is infinite and items are never purged except when the cache is full.
+
+	Default: 5 minutes
+
 remote.NAME.url
 :	Address of the remote server to fetch from. Single URL can be
 	specified within a single remote block. A remote node can request
diff --git a/src/main/resources/Documentation/metrics.md b/src/main/resources/Documentation/metrics.md
new file mode 100644
index 0000000..6fd6cf2
--- /dev/null
+++ b/src/main/resources/Documentation/metrics.md
@@ -0,0 +1,90 @@
+Metrics
+=======
+
+The @PLUGIN@ plugin reports the status of its internal components
+to Gerrit metric system.
+
+The metrics are all prefixed with @PLUGIN@ and then divided in groups,
+depending on the component involved.
+
+### plugins/@PLUGIN@/events
+
+This prefix represents the component involved in the processing of incoming
+ref-update events from Gerrit or other event-broker sources.
+
+- `queued_before_startup` Counter of the number of events that have been received
+  when the plugin was still in its starting phase and not ready yet to process events.
+
+### plugins/@PLUGIN@/tasks/<metric>/<source>
+
+This prefix represents the tasks scheduling and execution system, also
+known as _replication queue_. The replication task represents one operation
+performed against a Git repository from a remote source. The task may involve
+one or more refs.
+
+The `<metric>` field can have one of the values described here below,
+while the `<source>` represent the replication source endpoint.
+
+> Bear in mind that the _replication queue_ is a lot more than a simple
+> queueing system, as it contains the logic for selecting, merging, retrying
+> and cancelling incoming tasks.
+
+- `scheduled`: (counter) number of tasks triggered and scheduled for
+  execution.
+
+- `started`: (counter) number of tasks started.
+
+- `rescheduled`: (counter) number of tasks re-scheduled for execution.
+
+- `cancelled`: (counter) number of tasks that were previously scheduled
+  and then cancelled before being executed.
+
+- `failed`: (counter) number of tasks executed but failed altogether or
+  partially. A total failure is when the entire operation returned an
+  error and none of the operations took place; a partial failure is when
+  some of the operations in the tasks succeeded but other failed.
+
+- `retrying`: (counter) number of tasks being retired for execution.
+
+- `not_scheduled`: (counter) number of tasks which have been discarded before
+  being executed, because redundant (duplicate of existing scheduled tasks)
+  or not needed anymore (e.g. ref updated already by other means or other
+  checks)
+
+- `completed`: (counter) number of tasks completed successfully.
+
+- `merged`: (counter) number of tasks not being executed because they had
+  been consolidated with an existing scheduled task.
+
+- `failed_max_retries`: (counter) number of tasks that have reached their maximum
+  retry count but never succeeded.
+
+### plugins/@PLUGIN@
+
+- `apply_object_latency`: (timer) execution time statistics for the
+  synchronous replication using the _apply-object_ REST-API.
+
+- `apply_object_end_2_end_latency`: (timer) execution time statistics
+  for the end-2-end replication from when the event is triggered in
+  Gerrit until the successful execution of the synchronous replication
+  using the _apply-object_ REST-API.
+
+- `apply_object_max_api_payload_reached`: (counter) number of times that
+  the apply-object REST-API did fallback to the fetch REST-API because
+  it reached its maximum payload to transfer.
+
+- `replication_latency`: (timer) execution time statistics for the
+  synchronous replication using a _git fetch_.
+
+- `replication_end_2_end_latency`: (timer) execution time statistics
+  for the end-2-end replication from when the event is triggered in
+  Gerrit until the successful execution of the _git fetch_ for the
+  associated refs.
+
+- `replication_delay`: (timer) interval from when the ref-update event
+  is triggered to the start of the _git fetch_ command.
+
+- `replication_retries`: (counter) number of times that a replication task
+  has been retried.
+
+
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/FetchOneTest.java b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/FetchOneTest.java
index 0756c5c..d5d4546 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/FetchOneTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/FetchOneTest.java
@@ -112,15 +112,27 @@
 
   @Test
   public void shouldIncludeTheTaskIndexInItsStringRepresentation() {
-    String expected = "[" + objectUnderTest.getTaskIdHex() + "] fetch " + URI_PATTERN;
+    objectUnderTest.addRefs(Set.of("refs/heads/foo", "refs/heads/bar"));
+    String expected =
+        "["
+            + objectUnderTest.getTaskIdHex()
+            + "] fetch "
+            + URI_PATTERN
+            + " [refs/heads/bar,refs/heads/foo]";
 
     assertThat(objectUnderTest.toString()).isEqualTo(expected);
   }
 
   @Test
   public void shouldIncludeTheRetryCountInItsStringRepresentationWhenATaskIsRetried() {
+    objectUnderTest.addRefs(Set.of("refs/heads/bar", "refs/heads/foo"));
     objectUnderTest.setToRetry();
-    String expected = "(retry 1) [" + objectUnderTest.getTaskIdHex() + "] fetch " + URI_PATTERN;
+    String expected =
+        "(retry 1) ["
+            + objectUnderTest.getTaskIdHex()
+            + "] fetch "
+            + URI_PATTERN
+            + " [refs/heads/bar,refs/heads/foo]";
 
     assertThat(objectUnderTest.toString()).isEqualTo(expected);
   }
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/InMemoryMetricMaker.java b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/InMemoryMetricMaker.java
new file mode 100644
index 0000000..bd85f4f
--- /dev/null
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/InMemoryMetricMaker.java
@@ -0,0 +1,103 @@
+// Copyright (C) 2023 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.pull;
+
+import com.google.gerrit.metrics.Counter0;
+import com.google.gerrit.metrics.Counter1;
+import com.google.gerrit.metrics.Description;
+import com.google.gerrit.metrics.DisabledMetricMaker;
+import com.google.gerrit.metrics.Field;
+import com.google.inject.Singleton;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+
+/** Exports no metrics but counters kept in memory for testing purposes */
+@Singleton
+public class InMemoryMetricMaker extends DisabledMetricMaker {
+
+  public static class InMemoryCounter0 extends Counter0 {
+    private AtomicLong counter = new AtomicLong();
+
+    @Override
+    public void remove() {}
+
+    @Override
+    public void incrementBy(long delta) {
+      counter.addAndGet(delta);
+    }
+
+    public long getValue() {
+      return counter.get();
+    }
+
+    @Override
+    public String toString() {
+      return counter.toString();
+    }
+  }
+
+  public static class InMemoryCounter1<F> extends Counter1<F> {
+    private ConcurrentHashMap<F, Long> countersMap = new ConcurrentHashMap<>();
+
+    @Override
+    public void remove() {}
+
+    @Override
+    public void incrementBy(F field, long delta) {
+      countersMap.compute(field, (f, counter) -> counter == null ? delta : counter + delta);
+    }
+
+    public Optional<Long> getValue(Object fieldValue) {
+      return Optional.ofNullable(countersMap.get(fieldValue));
+    }
+
+    @Override
+    public String toString() {
+      return countersMap.toString();
+    }
+  }
+
+  private ConcurrentHashMap<String, InMemoryCounter0> counters0Map = new ConcurrentHashMap<>();
+  private ConcurrentHashMap<String, InMemoryCounter1<?>> counters1Map = new ConcurrentHashMap<>();
+
+  @Override
+  public Counter0 newCounter(String name, Description desc) {
+    InMemoryCounter0 counter = new InMemoryCounter0();
+    counters0Map.put(name, counter);
+    return counter;
+  }
+
+  @Override
+  public <F> Counter1<F> newCounter(String name, Description desc, Field<F> field) {
+    InMemoryCounter1<F> counter = new InMemoryCounter1<>();
+    counters1Map.put(name, counter);
+    return counter;
+  }
+
+  public Optional<Long> counterValue(String name) {
+    return Optional.ofNullable(counters0Map.get(name)).map(InMemoryCounter0::getValue);
+  }
+
+  public <F> Optional<Long> counterValue(String name, F fieldValue) {
+    return Optional.ofNullable(counters1Map.get(name))
+        .flatMap(counter -> counter.getValue(fieldValue));
+  }
+
+  @Override
+  public String toString() {
+    return counters0Map.toString() + "\n" + counters1Map.toString();
+  }
+}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationAsyncIT.java b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationAsyncIT.java
index be017d0..0ebd0ae 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationAsyncIT.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationAsyncIT.java
@@ -26,7 +26,8 @@
 @UseLocalDisk
 @TestPlugin(
     name = "pull-replication",
-    sysModule = "com.googlesource.gerrit.plugins.replication.pull.PullReplicationModule",
+    sysModule =
+        "com.googlesource.gerrit.plugins.replication.pull.PullReplicationITAbstract$PullReplicationTestModule",
     httpModule = "com.googlesource.gerrit.plugins.replication.pull.api.HttpModule")
 public class PullReplicationAsyncIT extends PullReplicationITAbstract {
   @Inject private SitePaths sitePaths;
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationIT.java b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationIT.java
index 4c00bae..29bf7e4 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationIT.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationIT.java
@@ -58,7 +58,8 @@
 @UseLocalDisk
 @TestPlugin(
     name = "pull-replication",
-    sysModule = "com.googlesource.gerrit.plugins.replication.pull.PullReplicationModule",
+    sysModule =
+        "com.googlesource.gerrit.plugins.replication.pull.PullReplicationITAbstract$PullReplicationTestModule",
     httpModule = "com.googlesource.gerrit.plugins.replication.pull.api.HttpModule")
 public class PullReplicationIT extends PullReplicationSetupBase {
 
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 ba812e2..58693bc 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
@@ -15,6 +15,7 @@
 package com.googlesource.gerrit.plugins.replication.pull;
 
 import static com.google.common.truth.Truth.assertThat;
+import static com.google.common.truth.Truth8.assertThat;
 import static com.google.gerrit.acceptance.GitUtil.fetch;
 import static com.google.gerrit.acceptance.GitUtil.pushOne;
 import static com.google.gerrit.acceptance.testsuite.project.TestProjectUpdate.allow;
@@ -31,6 +32,8 @@
 import com.google.gerrit.extensions.events.HeadUpdatedListener;
 import com.google.gerrit.extensions.events.ProjectDeletedListener;
 import com.google.gerrit.extensions.restapi.RestApiException;
+import com.google.gerrit.server.config.SitePaths;
+import com.google.inject.Inject;
 import com.googlesource.gerrit.plugins.replication.AutoReloadConfigDecorator;
 import java.io.IOException;
 import java.util.Collection;
@@ -52,6 +55,13 @@
 /** Base class to run regular and async acceptance tests */
 public abstract class PullReplicationITAbstract extends PullReplicationSetupBase {
 
+  public static class PullReplicationTestModule extends PullReplicationModule {
+    @Inject
+    public PullReplicationTestModule(SitePaths site, InMemoryMetricMaker memMetric) {
+      super(site, memMetric);
+    }
+  }
+
   @Override
   protected void setReplicationSource(
       String remoteName, List<String> replicaSuffixes, Optional<String> project)
@@ -91,6 +101,7 @@
             sourceCommit.getId().getName(),
             TEST_REPLICATION_REMOTE);
     pullReplicationQueue.onEvent(event);
+    waitUntilReplicationCompleted(1);
 
     try (Repository repo = repoManager.openRepository(project)) {
       waitUntil(() -> checkedGetRef(repo, sourceRef) != null);
@@ -101,6 +112,13 @@
     }
   }
 
+  private void assertTasksMetricScheduledAndCompleted(int numTasks) {
+    assertTasksMetric("scheduled", numTasks);
+    assertTasksMetric("started", numTasks);
+    assertTasksMetric("completed", numTasks);
+    assertEmptyTasksMetric("failed");
+  }
+
   @Test
   @GerritConfig(name = "gerrit.instanceId", value = TEST_REPLICATION_REMOTE)
   public void shouldReplicateNewBranch() throws Exception {
@@ -124,6 +142,7 @@
             branchRevision,
             TEST_REPLICATION_REMOTE);
     pullReplicationQueue.onEvent(event);
+    waitUntilReplicationCompleted(1);
 
     try (Repository repo = repoManager.openRepository(project);
         Repository sourceRepo = repoManager.openRepository(project)) {
@@ -133,6 +152,8 @@
       assertThat(targetBranchRef).isNotNull();
       assertThat(targetBranchRef.getObjectId().getName()).isEqualTo(branchRevision);
     }
+
+    assertTasksMetricScheduledAndCompleted(1);
   }
 
   @Test
@@ -167,6 +188,7 @@
             branchRevision,
             TEST_REPLICATION_REMOTE);
     pullReplicationQueue.onEvent(event);
+    waitUntilReplicationCompleted(1);
 
     try (Repository repo = repoManager.openRepository(project)) {
       waitUntil(() -> checkedGetRef(repo, newBranch) != null);
@@ -193,6 +215,7 @@
             amendedCommit.getId().getName(),
             TEST_REPLICATION_REMOTE);
     pullReplicationQueue.onEvent(forcedPushEvent);
+    waitUntilReplicationCompleted(2);
 
     try (Repository repo = repoManager.openRepository(project);
         Repository sourceRepo = repoManager.openRepository(project)) {
@@ -204,6 +227,8 @@
                       .getName()
                       .equals(amendedCommit.getId().getName()));
     }
+
+    assertTasksMetricScheduledAndCompleted(2);
   }
 
   @Test
@@ -232,6 +257,7 @@
             sourceCommit.getId().getName(),
             TEST_REPLICATION_REMOTE);
     pullReplicationQueue.onEvent(event);
+    waitUntilReplicationCompleted(1);
 
     try (Repository repo = repoManager.openRepository(project)) {
       waitUntil(() -> checkedGetRef(repo, sourceRef) != null);
@@ -240,6 +266,8 @@
       assertThat(targetBranchRef).isNotNull();
       assertThat(targetBranchRef.getObjectId()).isEqualTo(sourceCommit.getId());
     }
+
+    assertTasksMetricScheduledAndCompleted(1);
   }
 
   @Test
@@ -273,6 +301,7 @@
             branchRevision,
             TEST_REPLICATION_REMOTE);
     pullReplicationQueue.onEvent(event);
+    waitUntilReplicationCompleted(1);
 
     try (Repository repo = repoManager.openRepository(project);
         Repository sourceRepo = repoManager.openRepository(project)) {
@@ -282,6 +311,8 @@
       assertThat(targetBranchRef).isNotNull();
       assertThat(targetBranchRef.getObjectId().getName()).isEqualTo(branchRevision);
     }
+
+    assertTasksMetricScheduledAndCompleted(1);
   }
 
   @Test
@@ -309,8 +340,11 @@
     for (ProjectDeletedListener l : deletedListeners) {
       l.onProjectDeleted(event);
     }
+    waitUntilReplicationCompleted(1);
 
     waitUntil(() -> !repoManager.list().contains(project));
+
+    assertTasksMetricScheduledAndCompleted(1);
   }
 
   @Test
@@ -335,6 +369,7 @@
 
     HeadUpdatedListener.Event event = new FakeHeadUpdateEvent(master, newBranch, testProjectName);
     pullReplicationQueue.onHeadUpdated(event);
+    waitUntilReplicationCompleted(1);
 
     waitUntil(
         () -> {
@@ -344,6 +379,8 @@
             return false;
           }
         });
+
+    assertTasksMetricScheduledAndCompleted(1);
   }
 
   @Ignore
@@ -365,6 +402,7 @@
             sourceCommit.getId().getName(),
             TEST_REPLICATION_REMOTE);
     pullReplicationQueue.onEvent(event);
+    waitUntilReplicationCompleted(1);
 
     try (Repository repo = repoManager.openRepository(project)) {
       waitUntil(() -> checkedGetRef(repo, sourceRef) != null);
@@ -374,4 +412,27 @@
       assertThat(targetBranchRef.getObjectId()).isEqualTo(sourceCommit.getId());
     }
   }
+
+  private void waitUntilReplicationCompleted(int expected) throws InterruptedException {
+    waitUntil(
+        () ->
+            inMemoryMetrics()
+                .counterValue("tasks/completed", TEST_REPLICATION_REMOTE)
+                .filter(counter -> counter == expected)
+                .isPresent());
+  }
+
+  private InMemoryMetricMaker inMemoryMetrics() {
+    return getInstance(InMemoryMetricMaker.class);
+  }
+
+  private void assertTasksMetric(String taskMetric, long value) {
+    assertThat(inMemoryMetrics().counterValue("tasks/" + taskMetric, TEST_REPLICATION_REMOTE))
+        .hasValue(value);
+  }
+
+  private void assertEmptyTasksMetric(String taskMetric) {
+    assertThat(inMemoryMetrics().counterValue("tasks/" + taskMetric, TEST_REPLICATION_REMOTE))
+        .isEmpty();
+  }
 }
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/ReplicationQueueTest.java b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/ReplicationQueueTest.java
index 9b7e8c1..e40b90c 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/ReplicationQueueTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/ReplicationQueueTest.java
@@ -98,6 +98,8 @@
   @Mock ApplyObjectsRefsFilter applyObjectsRefsFilter;
   ApplyObjectMetrics applyObjectMetrics;
   FetchReplicationMetrics fetchMetrics;
+  ReplicationQueueMetrics queueMetrics;
+  ShutdownState shutdownState;
 
   @Captor ArgumentCaptor<String> stringCaptor;
   @Captor ArgumentCaptor<Project.NameKey> projectNameKeyCaptor;
@@ -157,6 +159,8 @@
 
     applyObjectMetrics = new ApplyObjectMetrics("pull-replication", new DisabledMetricMaker());
     fetchMetrics = new FetchReplicationMetrics("pull-replication", new DisabledMetricMaker());
+    queueMetrics = new ReplicationQueueMetrics("pull-replication", new DisabledMetricMaker());
+    shutdownState = new ShutdownState();
 
     objectUnderTest =
         new ReplicationQueue(
@@ -169,8 +173,10 @@
             () -> revReader,
             applyObjectMetrics,
             fetchMetrics,
+            queueMetrics,
             LOCAL_INSTANCE_ID,
-            applyObjectsRefsFilter);
+            applyObjectsRefsFilter,
+            shutdownState);
   }
 
   @Test
@@ -345,8 +351,10 @@
             () -> revReader,
             applyObjectMetrics,
             fetchMetrics,
+            queueMetrics,
             LOCAL_INSTANCE_ID,
-            applyObjectsRefsFilter);
+            applyObjectsRefsFilter,
+            shutdownState);
     Event event = new TestEvent("refs/multi-site/version");
     objectUnderTest.onEvent(event);
 
@@ -354,6 +362,12 @@
   }
 
   @Test
+  public void shouldSetShutdownStateWhenStopping() throws IOException {
+    objectUnderTest.stop();
+    assertThat(shutdownState.isShuttingDown()).isTrue();
+  }
+
+  @Test
   public void shouldSkipEventWhenStarredChangesRef() {
     Event event = new TestEvent("refs/starred-changes/41/2941/1000000");
     objectUnderTest.onEvent(event);
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/event/EventsBrokerMessageConsumerTest.java b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/event/EventsBrokerMessageConsumerTest.java
index 394a8a6..9640b82 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/event/EventsBrokerMessageConsumerTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/event/EventsBrokerMessageConsumerTest.java
@@ -18,12 +18,16 @@
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.doNothing;
 import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 
 import com.gerritforge.gerrit.eventbroker.BrokerApi;
 import com.google.gerrit.extensions.registration.DynamicItem;
 import com.google.gerrit.extensions.restapi.AuthException;
 import com.google.gerrit.server.events.RefUpdatedEvent;
 import com.google.gerrit.server.permissions.PermissionBackendException;
+import com.googlesource.gerrit.plugins.replication.pull.ShutdownState;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -34,13 +38,18 @@
 public class EventsBrokerMessageConsumerTest {
 
   @Mock private StreamEventListener eventListener;
-  @Mock DynamicItem<BrokerApi> eventsBroker;
+  @Mock DynamicItem<BrokerApi> eventsBrokerDynamicItem;
+  @Mock BrokerApi eventsBroker;
 
   EventsBrokerMessageConsumer objectUnderTest;
+  ShutdownState shutdownState;
 
   @Before
   public void setup() {
-    objectUnderTest = new EventsBrokerMessageConsumer(eventsBroker, eventListener, "topicName");
+    shutdownState = new ShutdownState();
+    objectUnderTest =
+        new EventsBrokerMessageConsumer(
+            eventsBrokerDynamicItem, eventListener, shutdownState, "topicName");
   }
 
   @Test
@@ -63,4 +72,16 @@
     doNothing().when(eventListener).fetchRefsForEvent(any());
     objectUnderTest.accept(new RefUpdatedEvent());
   }
+
+  @Test
+  public void shouldStillAcceptLastEventDuringShutdownAndThenDisconnect()
+      throws AuthException, PermissionBackendException {
+    doNothing().when(eventListener).fetchRefsForEvent(any());
+    when(eventsBrokerDynamicItem.get()).thenReturn(eventsBroker);
+
+    shutdownState.setIsShuttingDown(true);
+
+    objectUnderTest.accept(new RefUpdatedEvent());
+    verify(eventsBroker, times(1)).disconnect();
+  }
 }