diff --git a/BUILD b/BUILD
index 0a54c3e..50615d8 100644
--- a/BUILD
+++ b/BUILD
@@ -21,7 +21,10 @@
 
 junit_tests(
     name = "replication_tests",
-    srcs = glob(["src/test/java/**/*Test.java"]),
+    srcs = glob([
+        "src/test/java/**/*Test.java",
+        "src/test/java/**/*IT.java",
+    ]),
     tags = ["replication"],
     visibility = ["//visibility:public"],
     deps = PLUGIN_TEST_DEPS + PLUGIN_DEPS + [
@@ -32,7 +35,7 @@
 
 java_library(
     name = "replication_util",
-    testonly = 1,
+    testonly = True,
     srcs = glob(
         ["src/test/java/**/*.java"],
         exclude = ["src/test/java/**/*Test.java"],
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/AdminApi.java b/src/main/java/com/googlesource/gerrit/plugins/replication/AdminApi.java
new file mode 100644
index 0000000..acbf763
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/AdminApi.java
@@ -0,0 +1,25 @@
+// Copyright (C) 2018 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.reviewdb.client.Project;
+
+public interface AdminApi {
+  public boolean createProject(Project.NameKey project, String head);
+
+  public boolean deleteProject(Project.NameKey project);
+
+  public boolean updateHead(Project.NameKey project, String newHead);
+}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/AdminApiFactory.java b/src/main/java/com/googlesource/gerrit/plugins/replication/AdminApiFactory.java
new file mode 100644
index 0000000..de6e91e
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/AdminApiFactory.java
@@ -0,0 +1,73 @@
+// Copyright (C) 2018 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.inject.Inject;
+import com.google.inject.Singleton;
+import java.util.Optional;
+import org.eclipse.jgit.transport.URIish;
+
+/** Factory for creating an {@link AdminApi} instance for a remote URI. */
+public interface AdminApiFactory {
+  /**
+   * Create an {@link AdminApi} for the given remote URI.
+   *
+   * @param uri the remote URI.
+   * @return An API for the given remote URI, or {@code Optional.empty} if there is no appropriate
+   *     API for the URI.
+   */
+  Optional<AdminApi> create(URIish uri);
+
+  @Singleton
+  static class DefaultAdminApiFactory implements AdminApiFactory {
+    protected final SshHelper sshHelper;
+
+    @Inject
+    public DefaultAdminApiFactory(SshHelper sshHelper) {
+      this.sshHelper = sshHelper;
+    }
+
+    @Override
+    public Optional<AdminApi> create(URIish uri) {
+      if (isGerrit(uri)) {
+        return Optional.of(new GerritSshApi(sshHelper, uri));
+      } else if (!uri.isRemote()) {
+        return Optional.of(new LocalFS(uri));
+      } else if (isSSH(uri)) {
+        return Optional.of(new RemoteSsh(sshHelper, uri));
+      }
+      return Optional.empty();
+    }
+  }
+
+  static boolean isGerrit(URIish uri) {
+    String scheme = uri.getScheme();
+    return scheme != null && scheme.toLowerCase().equals("gerrit+ssh");
+  }
+
+  static boolean isSSH(URIish uri) {
+    if (!uri.isRemote()) {
+      return false;
+    }
+    String scheme = uri.getScheme();
+    if (scheme != null && scheme.toLowerCase().contains("ssh")) {
+      return true;
+    }
+    if (scheme == null && uri.getHost() != null && uri.getPath() != null) {
+      return true;
+    }
+    return false;
+  }
+}
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 e73e049..02daa6d 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/AutoReloadConfigDecorator.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/AutoReloadConfigDecorator.java
@@ -13,35 +13,42 @@
 // limitations under the License.
 package com.googlesource.gerrit.plugins.replication;
 
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.flogger.FluentLogger;
 import com.google.gerrit.common.FileUtil;
 import com.google.gerrit.extensions.annotations.PluginData;
 import com.google.gerrit.server.config.SitePaths;
 import com.google.gerrit.server.git.WorkQueue;
 import com.google.inject.Inject;
+import com.google.inject.Provider;
 import com.google.inject.Singleton;
 import java.io.IOException;
 import java.nio.file.Path;
 import java.util.List;
 import org.eclipse.jgit.errors.ConfigInvalidException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 @Singleton
 public class AutoReloadConfigDecorator implements ReplicationConfig {
-  private static final Logger log = LoggerFactory.getLogger(AutoReloadConfigDecorator.class);
-  private ReplicationFileBasedConfig currentConfig;
+  private static final FluentLogger logger = FluentLogger.forEnclosingClass();
+
+  private volatile ReplicationFileBasedConfig currentConfig;
   private long currentConfigTs;
+  private long lastFailedConfigTs;
 
   private final SitePaths site;
-  private final WorkQueue workQueue;
-  private final DestinationFactory destinationFactory;
+  private final Destination.Factory destinationFactory;
   private final Path pluginDataDir;
+  // Use Provider<> instead of injecting the ReplicationQueue because of circular dependency with
+  // ReplicationConfig
+  private final Provider<ReplicationQueue> replicationQueue;
+
+  private volatile boolean shuttingDown;
 
   @Inject
   public AutoReloadConfigDecorator(
       SitePaths site,
-      WorkQueue workQueue,
-      DestinationFactory destinationFactory,
+      Destination.Factory destinationFactory,
+      Provider<ReplicationQueue> replicationQueue,
       @PluginData Path pluginDataDir)
       throws ConfigInvalidException, IOException {
     this.site = site;
@@ -49,7 +56,7 @@
     this.pluginDataDir = pluginDataDir;
     this.currentConfig = loadConfig();
     this.currentConfigTs = getLastModified(currentConfig);
-    this.workQueue = workQueue;
+    this.replicationQueue = replicationQueue;
   }
 
   private static long getLastModified(ReplicationFileBasedConfig cfg) {
@@ -71,25 +78,42 @@
   }
 
   private void reloadIfNeeded() {
-    try {
-      if (isAutoReload()) {
-        long lastModified = getLastModified(currentConfig);
-        if (lastModified > currentConfigTs) {
-          ReplicationFileBasedConfig newConfig = loadConfig();
-          newConfig.startup(workQueue);
-          int discarded = currentConfig.shutdown();
+    reload(false);
+  }
 
-          this.currentConfig = newConfig;
-          this.currentConfigTs = lastModified;
-          log.info(
-              "Configuration reloaded: {} destinations, {} replication events discarded",
-              currentConfig.getDestinations(FilterType.ALL).size(),
-              discarded);
+  @VisibleForTesting
+  public void forceReload() {
+    reload(true);
+  }
+
+  private void reload(boolean force) {
+    if (force || isAutoReload()) {
+      ReplicationQueue queue = replicationQueue.get();
+
+      long lastModified = getLastModified(currentConfig);
+      try {
+        if (force
+            || (!shuttingDown
+                && lastModified > currentConfigTs
+                && lastModified > lastFailedConfigTs
+                && queue.isRunning()
+                && !queue.isReplaying())) {
+          queue.stop();
+          currentConfig = loadConfig();
+          currentConfigTs = lastModified;
+          lastFailedConfigTs = 0;
+          logger.atInfo().log(
+              "Configuration reloaded: %d destinations",
+              currentConfig.getDestinations(FilterType.ALL).size());
         }
+      } catch (Exception e) {
+        logger.atSevere().withCause(e).log(
+            "Cannot reload replication configuration: keeping existing settings");
+        lastFailedConfigTs = lastModified;
+        return;
+      } finally {
+        queue.start();
       }
-    } catch (Exception e) {
-      log.error("Cannot reload replication configuration: keeping existing settings", e);
-      return;
     }
   }
 
@@ -113,13 +137,28 @@
     return currentConfig.getEventsDirectory();
   }
 
+  /* shutdown() cannot be set as a synchronized method because
+   * it may need to wait for pending events to complete;
+   * e.g. when enabling the drain of replication events before
+   * shutdown.
+   *
+   * As a rule of thumb for synchronized methods, because they
+   * implicitly define a critical section and associated lock,
+   * they should never hold waiting for another resource, otherwise
+   * the risk of deadlock is very high.
+   *
+   * See more background about deadlocks, what they are and how to
+   * prevent them at: https://en.wikipedia.org/wiki/Deadlock
+   */
   @Override
-  public synchronized int shutdown() {
+  public int shutdown() {
+    this.shuttingDown = true;
     return currentConfig.shutdown();
   }
 
   @Override
   public synchronized void startup(WorkQueue workQueue) {
+    shuttingDown = false;
     currentConfig.startup(workQueue);
   }
 
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 f8737b6..29a7ee6 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/AutoReloadSecureCredentialsFactoryDecorator.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/AutoReloadSecureCredentialsFactoryDecorator.java
@@ -16,18 +16,16 @@
 
 import static com.google.gerrit.common.FileUtil.lastModified;
 
+import com.google.common.flogger.FluentLogger;
 import com.google.gerrit.server.config.SitePaths;
 import com.google.inject.Inject;
 import java.io.IOException;
 import java.nio.file.Files;
 import java.util.concurrent.atomic.AtomicReference;
 import org.eclipse.jgit.errors.ConfigInvalidException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 public class AutoReloadSecureCredentialsFactoryDecorator implements CredentialsFactory {
-  private static final Logger log =
-      LoggerFactory.getLogger(AutoReloadSecureCredentialsFactoryDecorator.class);
+  private static final FluentLogger logger = FluentLogger.forEnclosingClass();
 
   private final AtomicReference<SecureCredentialsFactory> secureCredentialsFactory;
   private volatile long secureCredentialsFactoryLoadTs;
@@ -58,13 +56,12 @@
         secureCredentialsFactory.compareAndSet(
             secureCredentialsFactory.get(), new SecureCredentialsFactory(site));
         secureCredentialsFactoryLoadTs = getSecureConfigLastEditTs();
-        log.info("secure.config reloaded as it was updated on the file system");
+        logger.atInfo().log("secure.config reloaded as it was updated on the file system");
       }
     } catch (Exception e) {
-      log.error(
+      logger.atSevere().withCause(e).log(
           "Unexpected error while trying to reload "
-              + "secure.config: keeping existing credentials",
-          e);
+              + "secure.config: keeping existing credentials");
     }
 
     return secureCredentialsFactory.get().create(remoteName);
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 0a06093..36960a1 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/Destination.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/Destination.java
@@ -24,9 +24,7 @@
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.ImmutableSet.Builder;
 import com.google.common.collect.Lists;
-import com.google.gerrit.common.EventDispatcher;
 import com.google.gerrit.common.data.GroupReference;
-import com.google.gerrit.extensions.client.ProjectState;
 import com.google.gerrit.extensions.config.FactoryModule;
 import com.google.gerrit.extensions.registration.DynamicItem;
 import com.google.gerrit.extensions.restapi.AuthException;
@@ -42,6 +40,7 @@
 import com.google.gerrit.server.account.GroupIncludeCache;
 import com.google.gerrit.server.account.ListGroupMembership;
 import com.google.gerrit.server.config.RequestScopedReviewDbProvider;
+import com.google.gerrit.server.events.EventDispatcher;
 import com.google.gerrit.server.git.GitRepositoryManager;
 import com.google.gerrit.server.git.PerThreadRequestScope;
 import com.google.gerrit.server.git.WorkQueue;
@@ -50,15 +49,18 @@
 import com.google.gerrit.server.permissions.ProjectPermission;
 import com.google.gerrit.server.permissions.RefPermission;
 import com.google.gerrit.server.project.NoSuchProjectException;
-import com.google.gerrit.server.project.PerRequestProjectControlCache;
-import com.google.gerrit.server.project.ProjectControl;
+import com.google.gerrit.server.project.ProjectCache;
+import com.google.gerrit.server.project.ProjectState;
 import com.google.gerrit.server.util.RequestContext;
+import com.google.inject.Inject;
 import com.google.inject.Injector;
 import com.google.inject.Provider;
 import com.google.inject.Provides;
+import com.google.inject.assistedinject.Assisted;
 import com.google.inject.assistedinject.FactoryModuleBuilder;
 import com.google.inject.servlet.RequestScoped;
 import com.googlesource.gerrit.plugins.replication.ReplicationState.RefPushResult;
+import com.googlesource.gerrit.plugins.replication.ReplicationTasksStorage.ReplicateRefUpdate;
 import java.io.IOException;
 import java.io.UnsupportedEncodingException;
 import java.net.URLEncoder;
@@ -83,18 +85,25 @@
 
 public class Destination {
   private static final Logger repLog = ReplicationQueue.repLog;
+
+  public interface Factory {
+    Destination create(DestinationConfiguration config);
+  }
+
   private final ReplicationStateListener stateLog;
   private final Object stateLock = new Object();
   private final Map<URIish, PushOne> pending = new HashMap<>();
   private final Map<URIish, PushOne> inFlight = new HashMap<>();
   private final PushOne.Factory opFactory;
-  private final ProjectControl.Factory projectControlFactory;
   private final GitRepositoryManager gitManager;
   private final PermissionBackend permissionBackend;
+  private final Provider<CurrentUser> userProvider;
+  private final ProjectCache projectCache;
   private volatile ScheduledExecutorService pool;
   private final PerThreadRequestScope.Scoper threadScoper;
   private final DestinationConfiguration config;
   private final DynamicItem<EventDispatcher> eventDispatcher;
+  private final Provider<ReplicationTasksStorage> replicationTasksStorage;
 
   protected enum RetryReason {
     TRANSPORT_ERROR,
@@ -112,23 +121,28 @@
     }
   }
 
+  @Inject
   protected Destination(
       Injector injector,
-      DestinationConfiguration cfg,
-      RemoteSiteUser.Factory replicationUserFactory,
       PluginUser pluginUser,
       GitRepositoryManager gitRepositoryManager,
       PermissionBackend permissionBackend,
+      Provider<CurrentUser> userProvider,
+      ProjectCache projectCache,
       GroupBackend groupBackend,
-      ReplicationStateListener stateLog,
+      ReplicationStateListeners stateLog,
       GroupIncludeCache groupIncludeCache,
-      DynamicItem<EventDispatcher> eventDispatcher) {
-    config = cfg;
+      DynamicItem<EventDispatcher> eventDispatcher,
+      Provider<ReplicationTasksStorage> rts,
+      @Assisted DestinationConfiguration cfg) {
     this.eventDispatcher = eventDispatcher;
     gitManager = gitRepositoryManager;
     this.permissionBackend = permissionBackend;
+    this.userProvider = userProvider;
+    this.projectCache = projectCache;
     this.stateLog = stateLog;
-
+    this.replicationTasksStorage = rts;
+    config = cfg;
     CurrentUser remoteUser;
     if (!cfg.getAuthGroupNames().isEmpty()) {
       ImmutableSet.Builder<AccountGroup.UUID> builder = ImmutableSet.builder();
@@ -141,7 +155,7 @@
           repLog.warn("Group \"{}\" not recognized, removing from authGroup", name);
         }
       }
-      remoteUser = replicationUserFactory.create(new ListGroupMembership(builder.build()));
+      remoteUser = new RemoteSiteUser(new ListGroupMembership(builder.build()));
     } else {
       remoteUser = pluginUser;
     }
@@ -153,7 +167,6 @@
               protected void configure() {
                 bindScope(RequestScoped.class, PerThreadRequestScope.REQUEST);
                 bind(PerThreadRequestScope.Propagator.class);
-                bind(PerRequestProjectControlCache.class).in(RequestScoped.class);
 
                 bind(Destination.class).toInstance(Destination.this);
                 bind(RemoteConfig.class).toInstance(config.getRemoteConfig());
@@ -185,7 +198,6 @@
               }
             });
 
-    projectControlFactory = child.getInstance(ProjectControl.Factory.class);
     opFactory = child.getInstance(PushOne.Factory.class);
     threadScoper = child.getInstance(PerThreadRequestScope.Scoper.class);
   }
@@ -245,15 +257,21 @@
     }
   }
 
-  private boolean shouldReplicate(ProjectControl ctl) throws PermissionBackendException {
-    if (!config.replicateHiddenProjects() && ctl.getProject().getState() == ProjectState.HIDDEN) {
+  private boolean shouldReplicate(ProjectState state, CurrentUser user)
+      throws PermissionBackendException {
+    if (!config.replicateHiddenProjects()
+        && state.getProject().getState()
+            == com.google.gerrit.extensions.client.ProjectState.HIDDEN) {
       return false;
     }
+
+    // Hidden projects(permitsRead = false) should only be accessible by the project owners.
+    // READ_CONFIG is checked here because it's only allowed to project owners(ACCESS may also
+    // be allowed for other users).
+    ProjectPermission permissionToCheck =
+        state.statePermitsRead() ? ProjectPermission.ACCESS : ProjectPermission.READ_CONFIG;
     try {
-      permissionBackend
-          .user(ctl.getUser())
-          .project(ctl.getProject().getNameKey())
-          .check(ProjectPermission.ACCESS);
+      permissionBackend.user(user).project(state.getNameKey()).check(permissionToCheck);
       return true;
     } catch (AuthException e) {
       return false;
@@ -268,8 +286,19 @@
               new Callable<Boolean>() {
                 @Override
                 public Boolean call() throws NoSuchProjectException, PermissionBackendException {
-                  ProjectControl projectControl = controlFor(project);
-                  if (!shouldReplicate(projectControl)) {
+                  ProjectState projectState;
+                  try {
+                    projectState = projectCache.checkedGet(project);
+                  } catch (IOException e) {
+                    return false;
+                  }
+                  if (projectState == null) {
+                    throw new NoSuchProjectException(project);
+                  }
+                  if (!projectState.statePermitsRead()) {
+                    return false;
+                  }
+                  if (!shouldReplicate(projectState, userProvider.get())) {
                     return false;
                   }
                   if (PushOne.ALL_REFS.equals(ref)) {
@@ -277,7 +306,7 @@
                   }
                   try {
                     permissionBackend
-                        .user(projectControl.getUser())
+                        .user(userProvider.get())
                         .project(project)
                         .ref(ref)
                         .check(RefPermission.READ);
@@ -304,7 +333,16 @@
               new Callable<Boolean>() {
                 @Override
                 public Boolean call() throws NoSuchProjectException, PermissionBackendException {
-                  return shouldReplicate(controlFor(project));
+                  ProjectState projectState;
+                  try {
+                    projectState = projectCache.checkedGet(project);
+                  } catch (IOException e) {
+                    return false;
+                  }
+                  if (projectState == null) {
+                    throw new NoSuchProjectException(project);
+                  }
+                  return shouldReplicate(projectState, userProvider.get());
                 }
               })
           .call();
@@ -354,21 +392,21 @@
     }
 
     synchronized (stateLock) {
-      PushOne e = getPendingPush(uri);
-      if (e == null) {
-        e = opFactory.create(project, uri);
-        addRef(e, ref);
-        e.addState(ref, state);
+      PushOne task = getPendingPush(uri);
+      if (task == null) {
+        task = opFactory.create(project, uri);
+        addRef(task, ref);
+        task.addState(ref, state);
         @SuppressWarnings("unused")
         ScheduledFuture<?> ignored =
-            pool.schedule(e, now ? 0 : config.getDelay(), TimeUnit.SECONDS);
-        pending.put(uri, e);
-      } else if (!e.getRefs().contains(ref)) {
-        addRef(e, ref);
-        e.addState(ref, state);
+            pool.schedule(task, now ? 0 : config.getDelay(), TimeUnit.SECONDS);
+        pending.put(uri, task);
+      } else if (!task.getRefs().contains(ref)) {
+        addRef(task, ref);
+        task.addState(ref, state);
       }
       state.increasePushTaskCount(project.get(), ref);
-      repLog.info("scheduled {}:{} => {} to run after {}s", project, ref, e, config.getDelay());
+      repLog.info("scheduled {}:{} => {} to run after {}s", project, ref, task, config.getDelay());
     }
   }
 
@@ -492,10 +530,6 @@
     }
   }
 
-  ProjectControl controlFor(Project.NameKey project) throws NoSuchProjectException {
-    return projectControlFactory.controlFor(project);
-  }
-
   RunwayStatus requestRunway(PushOne op) {
     synchronized (stateLock) {
       if (op.wasCanceled()) {
@@ -511,12 +545,31 @@
     return RunwayStatus.allowed();
   }
 
-  void notifyFinished(PushOne op) {
+  void notifyFinished(PushOne task) {
     synchronized (stateLock) {
-      inFlight.remove(op.getURI());
+      inFlight.remove(task.getURI());
+      if (!task.wasCanceled()) {
+        for (String ref : task.getRefs()) {
+          if (!refHasPendingPush(task.getURI(), ref)) {
+            replicationTasksStorage
+                .get()
+                .delete(
+                    new ReplicateRefUpdate(
+                        task.getProjectNameKey().get(), ref, task.getURI(), getRemoteConfigName()));
+          }
+        }
+      }
     }
   }
 
+  private boolean refHasPendingPush(URIish opUri, String ref) {
+    return pushContainsRef(pending.get(opUri), ref) || pushContainsRef(inFlight.get(opUri), ref);
+  }
+
+  private boolean pushContainsRef(PushOne op, String ref) {
+    return op != null && op.getRefs().contains(ref);
+  }
+
   boolean wouldPushProject(Project.NameKey project) {
     if (!shouldReplicate(project)) {
       return false;
@@ -649,6 +702,14 @@
     return config.getMaxRetries();
   }
 
+  public int getDrainQueueAttempts() {
+    return config.getDrainQueueAttempts();
+  }
+
+  public int getReplicationDelaySeconds() {
+    return config.getDelay() * 1000;
+  }
+
   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/DestinationConfiguration.java b/src/main/java/com/googlesource/gerrit/plugins/replication/DestinationConfiguration.java
index b2d0de2..f688cfc 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/DestinationConfiguration.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/DestinationConfiguration.java
@@ -22,10 +22,12 @@
 public class DestinationConfiguration {
   static final int DEFAULT_REPLICATION_DELAY = 15;
   static final int DEFAULT_RESCHEDULE_DELAY = 3;
+  static final int DEFAULT_DRAIN_QUEUE_ATTEMPTS = 0;
 
   private final int delay;
   private final int rescheduleDelay;
   private final int retryDelay;
+  private final int drainQueueAttempts;
   private final int lockErrorMaxRetries;
   private final ImmutableList<String> adminUrls;
   private final int poolThreads;
@@ -50,6 +52,8 @@
     projects = ImmutableList.copyOf(cfg.getStringList("remote", name, "projects"));
     adminUrls = ImmutableList.copyOf(cfg.getStringList("remote", name, "adminUrl"));
     retryDelay = Math.max(0, getInt(remoteConfig, cfg, "replicationretry", 1));
+    drainQueueAttempts =
+        Math.max(0, getInt(remoteConfig, cfg, "drainQueueAttempts", DEFAULT_DRAIN_QUEUE_ATTEMPTS));
     poolThreads = Math.max(0, getInt(remoteConfig, cfg, "threads", 1));
     authGroupNames = ImmutableList.copyOf(cfg.getStringList("remote", name, "authGroup"));
     lockErrorMaxRetries = cfg.getInt("replication", "lockErrorMaxRetries", 0);
@@ -77,6 +81,10 @@
     return retryDelay;
   }
 
+  public int getDrainQueueAttempts() {
+    return drainQueueAttempts;
+  }
+
   public int getPoolThreads() {
     return poolThreads;
   }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/DestinationFactory.java b/src/main/java/com/googlesource/gerrit/plugins/replication/DestinationFactory.java
deleted file mode 100644
index 83eab86..0000000
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/DestinationFactory.java
+++ /dev/null
@@ -1,75 +0,0 @@
-// Copyright (C) 2016 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.common.EventDispatcher;
-import com.google.gerrit.extensions.registration.DynamicItem;
-import com.google.gerrit.server.PluginUser;
-import com.google.gerrit.server.account.GroupBackend;
-import com.google.gerrit.server.account.GroupIncludeCache;
-import com.google.gerrit.server.git.GitRepositoryManager;
-import com.google.gerrit.server.permissions.PermissionBackend;
-import com.google.inject.Inject;
-import com.google.inject.Injector;
-import com.google.inject.Singleton;
-
-@Singleton
-public class DestinationFactory {
-  private final Injector injector;
-  private final RemoteSiteUser.Factory replicationUserFactory;
-  private final PluginUser pluginUser;
-  private final GitRepositoryManager gitRepositoryManager;
-  private final PermissionBackend permissionBackend;
-  private final GroupBackend groupBackend;
-  private final ReplicationStateListener stateLog;
-  private final GroupIncludeCache groupIncludeCache;
-  private final DynamicItem<EventDispatcher> eventDispatcher;
-
-  @Inject
-  public DestinationFactory(
-      Injector injector,
-      RemoteSiteUser.Factory replicationUserFactory,
-      PluginUser pluginUser,
-      GitRepositoryManager gitRepositoryManager,
-      PermissionBackend permissionBackend,
-      GroupBackend groupBackend,
-      ReplicationStateListener stateLog,
-      GroupIncludeCache groupIncludeCache,
-      DynamicItem<EventDispatcher> eventDispatcher) {
-    this.injector = injector;
-    this.replicationUserFactory = replicationUserFactory;
-    this.pluginUser = pluginUser;
-    this.gitRepositoryManager = gitRepositoryManager;
-    this.permissionBackend = permissionBackend;
-    this.groupBackend = groupBackend;
-    this.stateLog = stateLog;
-    this.groupIncludeCache = groupIncludeCache;
-    this.eventDispatcher = eventDispatcher;
-  }
-
-  Destination create(DestinationConfiguration config) {
-    return new Destination(
-        injector,
-        config,
-        replicationUserFactory,
-        pluginUser,
-        gitRepositoryManager,
-        permissionBackend,
-        groupBackend,
-        stateLog,
-        groupIncludeCache,
-        eventDispatcher);
-  }
-}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/EventsStorage.java b/src/main/java/com/googlesource/gerrit/plugins/replication/EventsStorage.java
deleted file mode 100644
index c567876..0000000
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/EventsStorage.java
+++ /dev/null
@@ -1,110 +0,0 @@
-// Copyright (C) 2018 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 java.nio.charset.StandardCharsets.UTF_8;
-
-import com.google.common.hash.Hashing;
-import com.google.gson.Gson;
-import com.google.inject.Inject;
-import com.google.inject.ProvisionException;
-import com.google.inject.Singleton;
-import java.io.IOException;
-import java.nio.file.DirectoryStream;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.util.ArrayList;
-import java.util.List;
-import org.eclipse.jgit.lib.ObjectId;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-@Singleton
-public class EventsStorage {
-  private static final Logger log = LoggerFactory.getLogger(EventsStorage.class);
-
-  public static class ReplicateRefUpdate {
-    public String project;
-    public String ref;
-  }
-
-  private static Gson GSON = new Gson();
-
-  private final Path refUpdates;
-
-  @Inject
-  EventsStorage(ReplicationConfig config) {
-    refUpdates = config.getEventsDirectory().resolve("ref-updates");
-  }
-
-  public String persist(String project, String ref) {
-    ReplicateRefUpdate r = new ReplicateRefUpdate();
-    r.project = project;
-    r.ref = ref;
-
-    String json = GSON.toJson(r) + "\n";
-    String eventKey = sha1(json).name();
-    Path file = refUpdates().resolve(eventKey);
-
-    if (Files.exists(file)) {
-      return eventKey;
-    }
-
-    try {
-      Files.write(file, json.getBytes(UTF_8));
-    } catch (IOException e) {
-      log.warn("Couldn't persist event {}", json);
-    }
-    return eventKey;
-  }
-
-  public void delete(String eventKey) {
-    if (eventKey != null) {
-      try {
-        Files.delete(refUpdates().resolve(eventKey));
-      } catch (IOException e) {
-        log.error("Error while deleting event {}", eventKey, e);
-      }
-    }
-  }
-
-  public List<ReplicateRefUpdate> list() {
-    ArrayList<ReplicateRefUpdate> result = new ArrayList<>();
-    try (DirectoryStream<Path> events = Files.newDirectoryStream(refUpdates())) {
-      for (Path e : events) {
-        if (Files.isRegularFile(e)) {
-          String json = new String(Files.readAllBytes(e), UTF_8);
-          result.add(GSON.fromJson(json, ReplicateRefUpdate.class));
-          Files.delete(e);
-        }
-      }
-    } catch (IOException e) {
-      log.error("Error when firing pending events", e);
-    }
-    return result;
-  }
-
-  private ObjectId sha1(String s) {
-    return ObjectId.fromRaw(Hashing.sha1().hashString(s, UTF_8).asBytes());
-  }
-
-  private Path refUpdates() {
-    try {
-      return Files.createDirectories(refUpdates);
-    } catch (IOException e) {
-      throw new ProvisionException(String.format("Couldn't create %s", refUpdates), e);
-    }
-  }
-}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/GerritSshApi.java b/src/main/java/com/googlesource/gerrit/plugins/replication/GerritSshApi.java
index b46a0d9..ccd8bf4 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/GerritSshApi.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/GerritSshApi.java
@@ -14,33 +14,34 @@
 
 package com.googlesource.gerrit.plugins.replication;
 
+import com.google.common.flogger.FluentLogger;
 import com.google.gerrit.reviewdb.client.Project;
 import com.google.gerrit.server.ssh.SshAddressesModule;
-import com.google.inject.Inject;
 import java.io.IOException;
 import java.io.OutputStream;
 import java.net.URISyntaxException;
 import java.util.HashSet;
 import java.util.Set;
 import org.eclipse.jgit.transport.URIish;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
-public class GerritSshApi {
-  static int SSH_COMMAND_FAILED = -1;
-  private static final Logger log = LoggerFactory.getLogger(GerritSshApi.class);
-  private static String GERRIT_ADMIN_PROTOCOL_PREFIX = "gerrit+";
+public class GerritSshApi implements AdminApi {
+  private static final FluentLogger logger = FluentLogger.forEnclosingClass();
 
-  private final SshHelper sshHelper;
+  static final int SSH_COMMAND_FAILED = -1;
+  private static final String GERRIT_ADMIN_PROTOCOL_PREFIX = "gerrit+";
+
+  protected final SshHelper sshHelper;
+  protected final URIish uri;
 
   private final Set<URIish> withoutDeleteProjectPlugin = new HashSet<>();
 
-  @Inject
-  protected GerritSshApi(SshHelper sshHelper) {
+  protected GerritSshApi(SshHelper sshHelper, URIish uri) {
     this.sshHelper = sshHelper;
+    this.uri = uri;
   }
 
-  protected boolean createProject(URIish uri, Project.NameKey projectName, String head) {
+  @Override
+  public boolean createProject(Project.NameKey projectName, String head) {
     OutputStream errStream = sshHelper.newErrorBufferStream();
     String cmd = "gerrit create-project --branch " + head + " " + projectName.get();
     try {
@@ -52,7 +53,8 @@
     return true;
   }
 
-  protected boolean deleteProject(URIish uri, Project.NameKey projectName) {
+  @Override
+  public boolean deleteProject(Project.NameKey projectName) {
     if (!withoutDeleteProjectPlugin.contains(uri)) {
       OutputStream errStream = sshHelper.newErrorBufferStream();
       String cmd = "deleteproject delete --yes-really-delete --force " + projectName.get();
@@ -64,30 +66,23 @@
         return false;
       }
       if (exitCode == 1) {
-        log.info(
-            "DeleteProject plugin is not installed on {}; will not try to forward this operation to that host");
+        logger.atInfo().log(
+            "DeleteProject plugin is not installed on %s;"
+                + " will not try to forward this operation to that host");
         withoutDeleteProjectPlugin.add(uri);
-        return true;
       }
     }
     return true;
   }
 
-  protected boolean updateHead(URIish uri, Project.NameKey projectName, String newHead) {
+  @Override
+  public boolean updateHead(Project.NameKey projectName, String newHead) {
     OutputStream errStream = sshHelper.newErrorBufferStream();
     String cmd = "gerrit set-head " + projectName.get() + " --new-head " + newHead;
     try {
       execute(uri, cmd, errStream);
     } catch (IOException e) {
-      log.error(
-          "Error updating HEAD of remote repository at {} to {}:\n"
-              + "  Exception: {}\n  Command: {}\n  Output: {}",
-          uri,
-          newHead,
-          e,
-          cmd,
-          errStream,
-          e);
+      logError("updating HEAD of", uri, errStream, cmd, e);
       return false;
     }
     return true;
@@ -114,19 +109,14 @@
       URIish sshUri = toSshUri(uri);
       return sshHelper.executeRemoteSsh(sshUri, cmd, errStream);
     } catch (URISyntaxException e) {
-      log.error("Cannot convert {} to SSH uri", uri, e);
+      logger.atSevere().withCause(e).log("Cannot convert %s to SSH uri", uri);
     }
     return SSH_COMMAND_FAILED;
   }
 
   public void logError(String msg, URIish uri, OutputStream errStream, String cmd, IOException e) {
-    log.error(
-        "Error {} remote repository at {}:\n  Exception: {}\n  Command: {}\n  Output: {}",
-        msg,
-        uri,
-        e,
-        cmd,
-        errStream,
-        e);
+    logger.atSevere().withCause(e).log(
+        "Error %s remote repository at %s:\n  Exception: %s\n  Command: %s\n  Output: %s",
+        msg, uri, e, cmd, errStream);
   }
 }
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 fa17dce..07978ab 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/ListCommand.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/ListCommand.java
@@ -72,7 +72,7 @@
   }
 
   private void addQueueDetails(JsonObject obj, String key, Collection<PushOne> values) {
-    if (values.size() > 0) {
+    if (!values.isEmpty()) {
       JsonArray list = new JsonArray();
       for (PushOne p : values) {
         list.add(new JsonPrimitive(p.toString()));
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/LocalFS.java b/src/main/java/com/googlesource/gerrit/plugins/replication/LocalFS.java
new file mode 100644
index 0000000..aa6e16c
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/LocalFS.java
@@ -0,0 +1,97 @@
+// Copyright (C) 2018 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.googlesource.gerrit.plugins.replication.ReplicationQueue.repLog;
+
+import com.google.gerrit.reviewdb.client.Project;
+import java.io.File;
+import java.io.IOException;
+import org.eclipse.jgit.internal.storage.file.FileRepository;
+import org.eclipse.jgit.lib.Constants;
+import org.eclipse.jgit.lib.RefUpdate;
+import org.eclipse.jgit.lib.Repository;
+import org.eclipse.jgit.transport.URIish;
+
+public class LocalFS implements AdminApi {
+
+  private final URIish uri;
+
+  public LocalFS(URIish uri) {
+    this.uri = uri;
+  }
+
+  @Override
+  public boolean createProject(Project.NameKey project, String head) {
+    try (Repository repo = new FileRepository(uri.getPath())) {
+      repo.create(true /* bare */);
+
+      if (head != null && head.startsWith(Constants.R_REFS)) {
+        RefUpdate u = repo.updateRef(Constants.HEAD);
+        u.disableRefLog();
+        u.link(head);
+      }
+      repLog.info("Created local repository: {}", uri);
+    } catch (IOException e) {
+      repLog.error("Error creating local repository {}", uri.getPath(), e);
+      return false;
+    }
+    return true;
+  }
+
+  @Override
+  public boolean deleteProject(Project.NameKey project) {
+    try {
+      recursivelyDelete(new File(uri.getPath()));
+      repLog.info("Deleted local repository: {}", uri);
+    } catch (IOException e) {
+      repLog.error("Error deleting local repository {}:\n", uri.getPath(), e);
+      return false;
+    }
+    return true;
+  }
+
+  @Override
+  public boolean updateHead(Project.NameKey project, String newHead) {
+    try (Repository repo = new FileRepository(uri.getPath())) {
+      if (newHead != null) {
+        RefUpdate u = repo.updateRef(Constants.HEAD);
+        u.link(newHead);
+      }
+    } catch (IOException e) {
+      repLog.error("Failed to update HEAD of repository {} to {}", uri.getPath(), newHead, e);
+      return false;
+    }
+    return true;
+  }
+
+  private static void recursivelyDelete(File dir) throws IOException {
+    File[] contents = dir.listFiles();
+    if (contents != null) {
+      for (File d : contents) {
+        if (d.isDirectory()) {
+          recursivelyDelete(d);
+        } else {
+          if (!d.delete()) {
+            throw new IOException("Failed to delete: " + d.getAbsolutePath());
+          }
+        }
+      }
+    }
+    if (!dir.delete()) {
+      throw new IOException("Failed to delete: " + dir.getAbsolutePath());
+    }
+  }
+}
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 2d60d86..8b0aa3d 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/OnStartStop.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/OnStartStop.java
@@ -15,13 +15,12 @@
 package com.googlesource.gerrit.plugins.replication;
 
 import com.google.common.util.concurrent.Atomics;
-import com.google.gerrit.common.EventDispatcher;
 import com.google.gerrit.extensions.events.LifecycleListener;
 import com.google.gerrit.extensions.registration.DynamicItem;
 import com.google.gerrit.extensions.systemstatus.ServerInformation;
+import com.google.gerrit.server.events.EventDispatcher;
 import com.google.inject.Inject;
 import com.googlesource.gerrit.plugins.replication.PushResultProcessing.GitUpdateProcessing;
-import com.googlesource.gerrit.plugins.replication.ReplicationState.Factory;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
@@ -30,36 +29,27 @@
   private final AtomicReference<Future<?>> pushAllFuture;
   private final ServerInformation srvInfo;
   private final PushAll.Factory pushAll;
-  private final ReplicationQueue queue;
   private final ReplicationConfig config;
   private final DynamicItem<EventDispatcher> eventDispatcher;
-  private final Factory replicationStateFactory;
 
   @Inject
   protected OnStartStop(
       ServerInformation srvInfo,
       PushAll.Factory pushAll,
-      ReplicationQueue queue,
       ReplicationConfig config,
-      DynamicItem<EventDispatcher> eventDispatcher,
-      ReplicationState.Factory replicationStateFactory) {
+      DynamicItem<EventDispatcher> eventDispatcher) {
     this.srvInfo = srvInfo;
     this.pushAll = pushAll;
-    this.queue = queue;
     this.config = config;
     this.eventDispatcher = eventDispatcher;
-    this.replicationStateFactory = replicationStateFactory;
     this.pushAllFuture = Atomics.newReference();
   }
 
   @Override
   public void start() {
-    queue.start();
-
     if (srvInfo.getState() == ServerInformation.State.STARTUP
         && config.isReplicateAllOnPluginStart()) {
-      ReplicationState state =
-          replicationStateFactory.create(new GitUpdateProcessing(eventDispatcher.get()));
+      ReplicationState state = new ReplicationState(new GitUpdateProcessing(eventDispatcher.get()));
       pushAllFuture.set(
           pushAll
               .create(null, ReplicationFilter.all(), state, false)
@@ -73,6 +63,5 @@
     if (f != null) {
       f.cancel(true);
     }
-    queue.stop();
   }
 }
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 db067e2..833b02b 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/PushAll.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/PushAll.java
@@ -43,7 +43,7 @@
       WorkQueue wq,
       ProjectCache projectCache,
       ReplicationQueue rq,
-      ReplicationStateListener stateLog,
+      ReplicationStateListeners stateLog,
       @Assisted @Nullable String urlMatch,
       @Assisted ReplicationFilter filter,
       @Assisted ReplicationState state,
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 5f0c066..5794f6e 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/PushOne.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/PushOne.java
@@ -22,20 +22,23 @@
 import com.google.common.collect.ListMultimap;
 import com.google.common.collect.Sets;
 import com.google.gerrit.extensions.events.GitReferenceUpdatedListener;
+import com.google.gerrit.extensions.registration.DynamicItem;
 import com.google.gerrit.extensions.restapi.AuthException;
+import com.google.gerrit.extensions.restapi.ResourceConflictException;
 import com.google.gerrit.metrics.Timer1;
 import com.google.gerrit.reviewdb.client.Project;
 import com.google.gerrit.reviewdb.client.RefNames;
 import com.google.gerrit.server.git.GitRepositoryManager;
 import com.google.gerrit.server.git.PerThreadRequestScope;
 import com.google.gerrit.server.git.ProjectRunnable;
-import com.google.gerrit.server.git.VisibleRefFilter;
 import com.google.gerrit.server.git.WorkQueue.CanceledWhileRunning;
+import com.google.gerrit.server.ioutil.HexFormat;
 import com.google.gerrit.server.permissions.PermissionBackend;
+import com.google.gerrit.server.permissions.PermissionBackend.RefFilterOptions;
 import com.google.gerrit.server.permissions.PermissionBackendException;
 import com.google.gerrit.server.permissions.ProjectPermission;
-import com.google.gerrit.server.project.NoSuchProjectException;
-import com.google.gerrit.server.project.ProjectControl;
+import com.google.gerrit.server.project.ProjectCache;
+import com.google.gerrit.server.project.ProjectState;
 import com.google.gerrit.server.util.IdGenerator;
 import com.google.inject.Inject;
 import com.google.inject.assistedinject.Assisted;
@@ -92,7 +95,6 @@
   private final RemoteConfig config;
   private final CredentialsProvider credentialsProvider;
   private final PerThreadRequestScope.Scoper threadScoper;
-  private final VisibleRefFilter.Factory refFilterFactory;
   private final ReplicationQueue replicationQueue;
 
   private final Project.NameKey projectName;
@@ -110,7 +112,10 @@
   private final int id;
   private final long createdAt;
   private final ReplicationMetrics metrics;
+  private final ProjectCache projectCache;
   private final AtomicBoolean canceledWhileRunning;
+  private final TransportFactory transportFactory;
+  private DynamicItem<ReplicationPushFilter> replicationPushFilter;
 
   @Inject
   PushOne(
@@ -118,20 +123,20 @@
       PermissionBackend permissionBackend,
       Destination p,
       RemoteConfig c,
-      VisibleRefFilter.Factory rff,
       CredentialsFactory cpFactory,
       PerThreadRequestScope.Scoper ts,
       ReplicationQueue rq,
       IdGenerator ig,
-      ReplicationStateListener sl,
+      ReplicationStateListeners sl,
       ReplicationMetrics m,
+      ProjectCache pc,
+      TransportFactory tf,
       @Assisted Project.NameKey d,
       @Assisted URIish u) {
     gitManager = grm;
     this.permissionBackend = permissionBackend;
     pool = p;
     config = c;
-    refFilterFactory = rff;
     credentialsProvider = cpFactory.create(c.getName());
     threadScoper = ts;
     replicationQueue = rq;
@@ -143,13 +148,20 @@
     stateLog = sl;
     createdAt = System.nanoTime();
     metrics = m;
+    projectCache = pc;
     canceledWhileRunning = new AtomicBoolean(false);
     maxRetries = p.getMaxRetries();
+    transportFactory = tf;
+  }
+
+  @Inject(optional = true)
+  public void setReplicationPushFilter(DynamicItem<ReplicationPushFilter> replicationPushFilter) {
+    this.replicationPushFilter = replicationPushFilter;
   }
 
   @Override
   public void cancel() {
-    repLog.info("Replication [{}] to {} was canceled", IdGenerator.format(id), getURI());
+    repLog.info("Replication [{}] to {} was canceled", HexFormat.fromInt(id), getURI());
     canceledByReplication();
     pool.pushWasCanceled(this);
   }
@@ -158,7 +170,7 @@
   public void setCanceledWhileRunning() {
     repLog.info(
         "Replication [{}] to {} was canceled while being executed",
-        IdGenerator.format(id),
+        HexFormat.fromInt(id),
         getURI());
     canceledWhileRunning.set(true);
   }
@@ -180,7 +192,7 @@
 
   @Override
   public String toString() {
-    String print = "[" + IdGenerator.format(id) + "] push " + uri;
+    String print = "[" + HexFormat.fromInt(id) + "] push " + uri;
 
     if (retryCount > 0) {
       print = "(retry " + retryCount + ") " + print;
@@ -301,7 +313,7 @@
     // we start replication (instead a new instance, with the same URI, is
     // created and scheduled for a future point in time.)
     //
-    MDC.put(ID_MDC_KEY, IdGenerator.format(id));
+    MDC.put(ID_MDC_KEY, HexFormat.fromInt(id));
     RunwayStatus status = pool.requestRunway(this);
     if (!status.isAllowed()) {
       if (status.isCanceled()) {
@@ -310,7 +322,7 @@
         repLog.info(
             "Rescheduling replication to {} to avoid collision with the in-flight push [{}].",
             uri,
-            IdGenerator.format(status.getInFlightPushId()));
+            HexFormat.fromInt(status.getInFlightPushId()));
         pool.reschedule(this, Destination.RetryReason.COLLISION);
       }
       return;
@@ -407,15 +419,12 @@
     if (pool.isCreateMissingRepos()) {
       try {
         Ref head = git.exactRef(Constants.HEAD);
-        if (replicationQueue.createProject(projectName, head != null ? getName(head) : null)) {
+        if (replicationQueue.createProject(
+            config.getName(), projectName, head != null ? getName(head) : null)) {
           repLog.warn("Missing repository created; retry replication to {}", uri);
           pool.reschedule(this, Destination.RetryReason.REPOSITORY_MISSING);
         } else {
-          repLog.warn(
-              "Missing repository could not be created when replicating {}. "
-                  + "You can only create missing repositories locally, over SSH or when "
-                  + "using adminUrl in replication.config. See documentation for more information.",
-              uri);
+          repLog.warn("Missing repository could not be created when replicating {}", uri);
         }
       } catch (IOException ioe) {
         stateLog.error(
@@ -438,7 +447,7 @@
 
   private void runImpl() throws IOException, PermissionBackendException {
     PushResult res;
-    try (Transport tn = Transport.open(git, uri)) {
+    try (Transport tn = transportFactory.open(git, uri)) {
       res = pushVia(tn);
     }
     updateStates(res.getRemoteUpdates());
@@ -465,19 +474,19 @@
 
   private List<RemoteRefUpdate> generateUpdates(Transport tn)
       throws IOException, PermissionBackendException {
-    ProjectControl pc;
-    try {
-      pc = pool.controlFor(projectName);
-    } catch (NoSuchProjectException e) {
+    ProjectState projectState = projectCache.checkedGet(projectName);
+    if (projectState == null) {
       return Collections.emptyList();
     }
 
     Map<String, Ref> local = git.getAllRefs();
     boolean filter;
+    PermissionBackend.ForProject forProject = permissionBackend.currentUser().project(projectName);
     try {
-      permissionBackend.user(pc.getUser()).project(projectName).check(ProjectPermission.READ);
+      projectState.checkStatePermitsRead();
+      forProject.check(ProjectPermission.READ);
       filter = false;
-    } catch (AuthException e) {
+    } catch (AuthException | ResourceConflictException e) {
       filter = true;
     }
     if (filter) {
@@ -494,10 +503,15 @@
         }
         local = n;
       }
-      local = refFilterFactory.create(pc.getProjectState(), git).filter(local, true);
+      local = forProject.filter(local, git, RefFilterOptions.builder().setFilterMeta(true).build());
     }
 
-    return pushAllRefs ? doPushAll(tn, local) : doPushDelta(local);
+    List<RemoteRefUpdate> remoteUpdatesList =
+        pushAllRefs ? doPushAll(tn, local) : doPushDelta(local);
+
+    return replicationPushFilter == null || replicationPushFilter.get() == null
+        ? remoteUpdatesList
+        : replicationPushFilter.get().filter(projectName.get(), remoteUpdatesList);
   }
 
   private List<RemoteRefUpdate> doPushAll(Transport tn, Map<String, Ref> local)
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/PushResultProcessing.java b/src/main/java/com/googlesource/gerrit/plugins/replication/PushResultProcessing.java
index c71a792..ae0662d 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/PushResultProcessing.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/PushResultProcessing.java
@@ -14,7 +14,8 @@
 
 package com.googlesource.gerrit.plugins.replication;
 
-import com.google.gerrit.common.EventDispatcher;
+import com.google.common.flogger.FluentLogger;
+import com.google.gerrit.server.events.EventDispatcher;
 import com.google.gerrit.server.events.RefEvent;
 import com.google.gerrit.server.permissions.PermissionBackendException;
 import com.google.gwtorm.server.OrmException;
@@ -23,8 +24,6 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import org.eclipse.jgit.transport.RemoteRefUpdate;
 import org.eclipse.jgit.transport.URIish;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 public abstract class PushResultProcessing {
 
@@ -159,7 +158,7 @@
   }
 
   public static class GitUpdateProcessing extends PushResultProcessing {
-    private static final Logger log = LoggerFactory.getLogger(GitUpdateProcessing.class);
+    private static final FluentLogger logger = FluentLogger.forEnclosingClass();
 
     private final EventDispatcher dispatcher;
 
@@ -189,7 +188,7 @@
       try {
         dispatcher.postEvent(event);
       } catch (OrmException | PermissionBackendException e) {
-        log.error("Cannot post event", e);
+        logger.atSevere().withCause(e).log("Cannot post event");
       }
     }
   }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/RemoteSiteUser.java b/src/main/java/com/googlesource/gerrit/plugins/replication/RemoteSiteUser.java
index 91fce7f..c3556af 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/RemoteSiteUser.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/RemoteSiteUser.java
@@ -16,18 +16,11 @@
 
 import com.google.gerrit.server.CurrentUser;
 import com.google.gerrit.server.account.GroupMembership;
-import com.google.inject.Inject;
-import com.google.inject.assistedinject.Assisted;
 
 public class RemoteSiteUser extends CurrentUser {
-  public interface Factory {
-    RemoteSiteUser create(@Assisted GroupMembership authGroups);
-  }
-
   private final GroupMembership effectiveGroups;
 
-  @Inject
-  RemoteSiteUser(@Assisted GroupMembership authGroups) {
+  public RemoteSiteUser(GroupMembership authGroups) {
     effectiveGroups = authGroups;
   }
 
@@ -35,4 +28,10 @@
   public GroupMembership getEffectiveGroups() {
     return effectiveGroups;
   }
+
+  @Override
+  public Object getCacheKey() {
+    // Never cache a remote user
+    return new Object();
+  }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/RemoteSsh.java b/src/main/java/com/googlesource/gerrit/plugins/replication/RemoteSsh.java
new file mode 100644
index 0000000..ee9d4c0
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/RemoteSsh.java
@@ -0,0 +1,110 @@
+// Copyright (C) 2018 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.googlesource.gerrit.plugins.replication.ReplicationQueue.repLog;
+
+import com.google.gerrit.reviewdb.client.Project;
+import java.io.IOException;
+import java.io.OutputStream;
+import org.eclipse.jgit.transport.URIish;
+import org.eclipse.jgit.util.QuotedString;
+
+public class RemoteSsh implements AdminApi {
+
+  private final SshHelper sshHelper;
+  private URIish uri;
+
+  RemoteSsh(SshHelper sshHelper, URIish uri) {
+    this.sshHelper = sshHelper;
+    this.uri = uri;
+  }
+
+  @Override
+  public boolean createProject(Project.NameKey project, String head) {
+    String quotedPath = QuotedString.BOURNE.quote(uri.getPath());
+    String cmd = "mkdir -p " + quotedPath + " && cd " + quotedPath + " && git init --bare";
+    if (head != null) {
+      cmd = cmd + " && git symbolic-ref HEAD " + QuotedString.BOURNE.quote(head);
+    }
+    OutputStream errStream = sshHelper.newErrorBufferStream();
+    try {
+      sshHelper.executeRemoteSsh(uri, cmd, errStream);
+      repLog.info("Created remote repository: {}", uri);
+    } catch (IOException e) {
+      repLog.error(
+          "Error creating remote repository at {}:\n"
+              + "  Exception: {}\n"
+              + "  Command: {}\n"
+              + "  Output: {}",
+          uri,
+          e,
+          cmd,
+          errStream,
+          e);
+      return false;
+    }
+    return true;
+  }
+
+  @Override
+  public boolean deleteProject(Project.NameKey project) {
+    String quotedPath = QuotedString.BOURNE.quote(uri.getPath());
+    String cmd = "rm -rf " + quotedPath;
+    OutputStream errStream = sshHelper.newErrorBufferStream();
+    try {
+      sshHelper.executeRemoteSsh(uri, cmd, errStream);
+      repLog.info("Deleted remote repository: {}", uri);
+    } catch (IOException e) {
+      repLog.error(
+          "Error deleting remote repository at {}:\n"
+              + "  Exception: {}\n"
+              + "  Command: {}\n"
+              + "  Output: {}",
+          uri,
+          e,
+          cmd,
+          errStream,
+          e);
+      return false;
+    }
+    return true;
+  }
+
+  @Override
+  public boolean updateHead(Project.NameKey project, String newHead) {
+    String quotedPath = QuotedString.BOURNE.quote(uri.getPath());
+    String cmd =
+        "cd " + quotedPath + " && git symbolic-ref HEAD " + QuotedString.BOURNE.quote(newHead);
+    OutputStream errStream = sshHelper.newErrorBufferStream();
+    try {
+      sshHelper.executeRemoteSsh(uri, cmd, errStream);
+    } catch (IOException e) {
+      repLog.error(
+          "Error updating HEAD of remote repository at {} to {}:\n"
+              + "  Exception: {}\n"
+              + "  Command: {}\n"
+              + "  Output: {}",
+          uri,
+          newHead,
+          e,
+          cmd,
+          errStream,
+          e);
+      return false;
+    }
+    return true;
+  }
+}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationExtensionPointModule.java b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationExtensionPointModule.java
new file mode 100644
index 0000000..b92a54a
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationExtensionPointModule.java
@@ -0,0 +1,32 @@
+// Copyright (C) 2019 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.extensions.registration.DynamicItem;
+import com.google.inject.AbstractModule;
+
+/**
+ * Gerrit libModule for applying a ref-filter for outgoing replications.
+ *
+ * <p>It should be used only when an actual filter is defined, otherwise the default replication
+ * plugin behaviour will be pushing all refs without any filtering.
+ */
+public class ReplicationExtensionPointModule extends AbstractModule {
+
+  @Override
+  protected void configure() {
+    DynamicItem.itemOf(binder(), ReplicationPushFilter.class);
+  }
+}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationFileBasedConfig.java b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationFileBasedConfig.java
index d68505c..683fb9b 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationFileBasedConfig.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationFileBasedConfig.java
@@ -20,6 +20,7 @@
 import com.google.common.base.Strings;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Lists;
+import com.google.common.flogger.FluentLogger;
 import com.google.gerrit.extensions.annotations.PluginData;
 import com.google.gerrit.server.config.ConfigUtil;
 import com.google.gerrit.server.config.SitePaths;
@@ -40,12 +41,10 @@
 import org.eclipse.jgit.transport.RemoteConfig;
 import org.eclipse.jgit.transport.URIish;
 import org.eclipse.jgit.util.FS;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 @Singleton
 public class ReplicationFileBasedConfig implements ReplicationConfig {
-  private static final Logger log = LoggerFactory.getLogger(ReplicationFileBasedConfig.class);
+  private static final FluentLogger logger = FluentLogger.forEnclosingClass();
   private static final int DEFAULT_SSH_CONNECTION_TIMEOUT_MS = 2 * 60 * 1000; // 2 minutes
 
   private List<Destination> destinations;
@@ -60,7 +59,7 @@
 
   @Inject
   public ReplicationFileBasedConfig(
-      SitePaths site, DestinationFactory destinationFactory, @PluginData Path pluginDataDir)
+      SitePaths site, Destination.Factory destinationFactory, @PluginData Path pluginDataDir)
       throws ConfigInvalidException, IOException {
     this.site = site;
     this.cfgPath = site.etc_dir.resolve("replication.config");
@@ -93,14 +92,14 @@
     return destinations.stream().filter(Objects::nonNull).filter(filter).collect(toList());
   }
 
-  private List<Destination> allDestinations(DestinationFactory destinationFactory)
+  private List<Destination> allDestinations(Destination.Factory destinationFactory)
       throws ConfigInvalidException, IOException {
     if (!config.getFile().exists()) {
-      log.warn("Config file {} does not exist; not replicating", config.getFile());
+      logger.atWarning().log("Config file %s does not exist; not replicating", config.getFile());
       return Collections.emptyList();
     }
     if (config.getFile().length() == 0) {
-      log.info("Config file {} is empty; not replicating", config.getFile());
+      logger.atInfo().log("Config file %s is empty; not replicating", config.getFile());
       return Collections.emptyList();
     }
 
@@ -192,7 +191,7 @@
         result.add(new RemoteConfig(cfg, name));
       } catch (URISyntaxException e) {
         throw new ConfigInvalidException(
-            String.format("remote %s has invalid URL in %s", name, cfg.getFile()));
+            String.format("remote %s has invalid URL in %s", name, cfg.getFile()), e);
       }
     }
     return result;
@@ -223,11 +222,54 @@
   public int shutdown() {
     int discarded = 0;
     for (Destination cfg : destinations) {
-      discarded += cfg.shutdown();
+      try {
+        drainReplicationEvents(cfg);
+      } catch (EventQueueNotEmptyException e) {
+        logger.atWarning().log("Event queue not empty: %s", e.getMessage());
+      } finally {
+        discarded += cfg.shutdown();
+      }
     }
     return discarded;
   }
 
+  void drainReplicationEvents(Destination destination) throws EventQueueNotEmptyException {
+    int drainQueueAttempts = destination.getDrainQueueAttempts();
+    if (drainQueueAttempts == 0) {
+      return;
+    }
+    int pending = destination.getQueueInfo().pending.size();
+    int inFlight = destination.getQueueInfo().inFlight.size();
+
+    while ((inFlight > 0 || pending > 0) && drainQueueAttempts > 0) {
+      try {
+        logger.atInfo().log(
+            "Draining replication events, postpone shutdown. Events left: inFlight %d, pending %d",
+            inFlight, pending);
+        Thread.sleep(destination.getReplicationDelaySeconds());
+      } catch (InterruptedException ie) {
+        logger.atWarning().withCause(ie).log(
+            "Wait for replication events to drain has been interrupted");
+      }
+      pending = destination.getQueueInfo().pending.size();
+      inFlight = destination.getQueueInfo().inFlight.size();
+      drainQueueAttempts--;
+    }
+
+    if (pending > 0 || inFlight > 0) {
+      throw new EventQueueNotEmptyException(
+          String.format("Pending: %d - InFlight: %d", pending, inFlight));
+    }
+  }
+
+  public static class EventQueueNotEmptyException extends Exception {
+    private static final long serialVersionUID = 1L;
+
+    public EventQueueNotEmptyException(String errorMessage) {
+      super(errorMessage);
+    }
+  }
+
   FileBasedConfig getConfig() {
     return config;
   }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationFilter.java b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationFilter.java
index 7b3486b..05bbb03 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationFilter.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationFilter.java
@@ -15,7 +15,7 @@
 package com.googlesource.gerrit.plugins.replication;
 
 import com.google.gerrit.common.data.AccessSection;
-import com.google.gerrit.reviewdb.client.Project.NameKey;
+import com.google.gerrit.reviewdb.client.Project;
 import java.util.Collections;
 import java.util.List;
 
@@ -46,7 +46,7 @@
     projectPatterns = patterns;
   }
 
-  public boolean matches(NameKey name) {
+  public boolean matches(Project.NameKey name) {
     if (projectPatterns.isEmpty()) {
       return true;
     }
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 5e7c978..5fdb375 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationModule.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationModule.java
@@ -21,8 +21,8 @@
 import com.google.gerrit.extensions.events.GitReferenceUpdatedListener;
 import com.google.gerrit.extensions.events.HeadUpdatedListener;
 import com.google.gerrit.extensions.events.LifecycleListener;
-import com.google.gerrit.extensions.events.NewProjectCreatedListener;
 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;
@@ -34,11 +34,13 @@
 class ReplicationModule extends AbstractModule {
   @Override
   protected void configure() {
-    bind(DestinationFactory.class).in(Scopes.SINGLETON);
+    install(new FactoryModuleBuilder().build(Destination.Factory.class));
     bind(ReplicationQueue.class).in(Scopes.SINGLETON);
+    bind(LifecycleListener.class)
+        .annotatedWith(UniqueAnnotations.create())
+        .to(ReplicationQueue.class);
 
     DynamicSet.bind(binder(), GitReferenceUpdatedListener.class).to(ReplicationQueue.class);
-    DynamicSet.bind(binder(), NewProjectCreatedListener.class).to(ReplicationQueue.class);
     DynamicSet.bind(binder(), ProjectDeletedListener.class).to(ReplicationQueue.class);
     DynamicSet.bind(binder(), HeadUpdatedListener.class).to(ReplicationQueue.class);
 
@@ -55,15 +57,20 @@
         .to(StartReplicationCapability.class);
 
     install(new FactoryModuleBuilder().build(PushAll.Factory.class));
-    install(new FactoryModuleBuilder().build(RemoteSiteUser.Factory.class));
-    install(new FactoryModuleBuilder().build(ReplicationState.Factory.class));
 
     bind(ReplicationConfig.class).to(AutoReloadConfigDecorator.class);
-    bind(ReplicationStateListener.class).to(ReplicationStateLogger.class);
+    DynamicSet.setOf(binder(), ReplicationStateListener.class);
+    DynamicSet.bind(binder(), ReplicationStateListener.class).to(ReplicationStateLogger.class);
 
     EventTypes.register(RefReplicatedEvent.TYPE, RefReplicatedEvent.class);
     EventTypes.register(RefReplicationDoneEvent.TYPE, RefReplicationDoneEvent.class);
     EventTypes.register(ReplicationScheduledEvent.TYPE, ReplicationScheduledEvent.class);
     bind(SshSessionFactory.class).toProvider(ReplicationSshSessionFactoryProvider.class);
+
+    DynamicItem.itemOf(binder(), AdminApiFactory.class);
+    DynamicItem.bind(binder(), AdminApiFactory.class)
+        .to(AdminApiFactory.DefaultAdminApiFactory.class);
+
+    bind(TransportFactory.class).to(TransportFactoryImpl.class).in(Scopes.SINGLETON);
   }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationPushFilter.java b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationPushFilter.java
new file mode 100644
index 0000000..eb6ba90
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationPushFilter.java
@@ -0,0 +1,30 @@
+// Copyright (C) 2019 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.extensions.annotations.ExtensionPoint;
+import java.util.List;
+import org.eclipse.jgit.transport.RemoteRefUpdate;
+
+/**
+ * Filter that is invoked before list of remote ref updates is pushed to remote instance.
+ *
+ * <p>It can be used to filter out unwanted updates.
+ */
+@ExtensionPoint
+public interface ReplicationPushFilter {
+
+  public List<RemoteRefUpdate> filter(String projectName, List<RemoteRefUpdate> remoteUpdatesList);
+}
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 da5a4d2..d73ab7b 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationQueue.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationQueue.java
@@ -14,34 +14,33 @@
 
 package com.googlesource.gerrit.plugins.replication;
 
+import static com.googlesource.gerrit.plugins.replication.AdminApiFactory.isGerrit;
+import static com.googlesource.gerrit.plugins.replication.AdminApiFactory.isSSH;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Objects;
 import com.google.common.base.Strings;
 import com.google.common.collect.Queues;
-import com.google.gerrit.common.EventDispatcher;
+import com.google.gerrit.common.Nullable;
 import com.google.gerrit.extensions.events.GitReferenceUpdatedListener;
 import com.google.gerrit.extensions.events.HeadUpdatedListener;
 import com.google.gerrit.extensions.events.LifecycleListener;
-import com.google.gerrit.extensions.events.NewProjectCreatedListener;
 import com.google.gerrit.extensions.events.ProjectDeletedListener;
 import com.google.gerrit.extensions.registration.DynamicItem;
 import com.google.gerrit.reviewdb.client.Project;
+import com.google.gerrit.server.events.EventDispatcher;
 import com.google.gerrit.server.git.WorkQueue;
 import com.google.inject.Inject;
 import com.googlesource.gerrit.plugins.replication.PushResultProcessing.GitUpdateProcessing;
 import com.googlesource.gerrit.plugins.replication.ReplicationConfig.FilterType;
-import java.io.File;
-import java.io.IOException;
-import java.io.OutputStream;
+import com.googlesource.gerrit.plugins.replication.ReplicationTasksStorage.ReplicateRefUpdate;
 import java.net.URISyntaxException;
 import java.util.Collections;
 import java.util.HashSet;
+import java.util.Optional;
 import java.util.Queue;
 import java.util.Set;
-import org.eclipse.jgit.internal.storage.file.FileRepository;
-import org.eclipse.jgit.lib.Constants;
-import org.eclipse.jgit.lib.RefUpdate;
-import org.eclipse.jgit.lib.Repository;
 import org.eclipse.jgit.transport.URIish;
-import org.eclipse.jgit.util.QuotedString;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -49,7 +48,6 @@
 public class ReplicationQueue
     implements LifecycleListener,
         GitReferenceUpdatedListener,
-        NewProjectCreatedListener,
         ProjectDeletedListener,
         HeadUpdatedListener {
   static final String REPLICATION_LOG_NAME = "replication_log";
@@ -70,42 +68,39 @@
   }
 
   private final WorkQueue workQueue;
-  private final SshHelper sshHelper;
   private final DynamicItem<EventDispatcher> dispatcher;
   private final ReplicationConfig config;
-  private final GerritSshApi gerritAdmin;
-  private final ReplicationState.Factory replicationStateFactory;
-  private final EventsStorage eventsStorage;
+  private final DynamicItem<AdminApiFactory> adminApiFactory;
+  private final ReplicationTasksStorage replicationTasksStorage;
   private volatile boolean running;
-  private final Queue<GitReferenceUpdatedListener.Event> beforeStartupEventsQueue;
+  private volatile boolean replaying;
+  private final Queue<ReferenceUpdatedEvent> beforeStartupEventsQueue;
 
   @Inject
   ReplicationQueue(
       WorkQueue wq,
-      SshHelper sh,
-      GerritSshApi ga,
+      DynamicItem<AdminApiFactory> aaf,
       ReplicationConfig rc,
       DynamicItem<EventDispatcher> dis,
-      ReplicationStateListener sl,
-      ReplicationState.Factory rsf,
-      EventsStorage es) {
+      ReplicationStateListeners sl,
+      ReplicationTasksStorage rts) {
     workQueue = wq;
-    sshHelper = sh;
     dispatcher = dis;
     config = rc;
     stateLog = sl;
-    gerritAdmin = ga;
-    replicationStateFactory = rsf;
-    eventsStorage = es;
+    adminApiFactory = aaf;
+    replicationTasksStorage = rts;
     beforeStartupEventsQueue = Queues.newConcurrentLinkedQueue();
   }
 
   @Override
   public void start() {
-    config.startup(workQueue);
-    running = true;
-    firePendingEvents();
-    fireBeforeStartupEvents();
+    if (!running) {
+      config.startup(workQueue);
+      running = true;
+      firePendingEvents();
+      fireBeforeStartupEvents();
+    }
   }
 
   @Override
@@ -117,11 +112,20 @@
     }
   }
 
+  public boolean isRunning() {
+    return running;
+  }
+
+  public boolean isReplaying() {
+    return replaying;
+  }
+
   void scheduleFullSync(Project.NameKey project, String urlMatch, ReplicationState state) {
     scheduleFullSync(project, urlMatch, state, false);
   }
 
-  void scheduleFullSync(
+  @VisibleForTesting
+  public void scheduleFullSync(
       Project.NameKey project, String urlMatch, ReplicationState state, boolean now) {
     if (!running) {
       stateLog.warn("Replication plugin did not finish startup before event", state);
@@ -132,6 +136,9 @@
       if (cfg.wouldPushProject(project)) {
         for (URIish uri : cfg.getURIs(project, urlMatch)) {
           cfg.schedule(project, PushOne.ALL_REFS, uri, state, now);
+          replicationTasksStorage.persist(
+              new ReplicateRefUpdate(
+                  project.get(), PushOne.ALL_REFS, uri, cfg.getRemoteConfigName()));
         }
       }
     }
@@ -139,28 +146,25 @@
 
   @Override
   public void onGitReferenceUpdated(GitReferenceUpdatedListener.Event event) {
-    ReplicationState state =
-        replicationStateFactory.create(new GitUpdateProcessing(dispatcher.get()));
-    if (!running) {
-      stateLog.warn(
-          "Replication plugin did not finish startup before event, event replication is postponed",
-          state);
-      beforeStartupEventsQueue.add(event);
-      return;
-    }
     onGitReferenceUpdated(event.getProjectName(), event.getRefName());
   }
 
   private void onGitReferenceUpdated(String projectName, String refName) {
-    ReplicationState state =
-        replicationStateFactory.create(new GitUpdateProcessing(dispatcher.get()));
+    ReplicationState state = new ReplicationState(new GitUpdateProcessing(dispatcher.get()));
+    if (!running) {
+      stateLog.warn(
+          "Replication plugin did not finish startup before event, event replication is postponed",
+          state);
+      beforeStartupEventsQueue.add(new ReferenceUpdatedEvent(projectName, refName));
+      return;
+    }
 
     Project.NameKey project = new Project.NameKey(projectName);
     for (Destination cfg : config.getDestinations(FilterType.ALL)) {
       if (cfg.wouldPushProject(project) && cfg.wouldPushRef(refName)) {
-        String eventKey = eventsStorage.persist(projectName, refName);
-        state.setEventKey(eventKey);
         for (URIish uri : cfg.getURIs(project, null)) {
+          replicationTasksStorage.persist(
+              new ReplicateRefUpdate(projectName, refName, uri, cfg.getRemoteConfigName()));
           cfg.schedule(project, refName, uri, state);
         }
       }
@@ -169,28 +173,26 @@
   }
 
   private void firePendingEvents() {
-    for (EventsStorage.ReplicateRefUpdate e : eventsStorage.list()) {
-      if (e == null) {
-        repLog.warn("Encountered null replication event in EventsStorage");
-      } else {
-        repLog.info("Firing pending event {}", e);
-        onGitReferenceUpdated(e.project, e.ref);
+    try {
+      Set<String> eventsReplayed = new HashSet<>();
+      replaying = true;
+      for (ReplicationTasksStorage.ReplicateRefUpdate t : replicationTasksStorage.list()) {
+        String eventKey = String.format("%s:%s", t.project, t.ref);
+        if (!eventsReplayed.contains(eventKey)) {
+          repLog.info("Firing pending task {}", eventKey);
+          onGitReferenceUpdated(t.project, t.ref);
+          eventsReplayed.add(eventKey);
+        }
       }
-    }
-  }
-
-  @Override
-  public void onNewProjectCreated(NewProjectCreatedListener.Event event) {
-    Project.NameKey projectName = new Project.NameKey(event.getProjectName());
-    for (URIish uri : getURIs(projectName, FilterType.PROJECT_CREATION)) {
-      createProject(uri, projectName, event.getHeadName());
+    } finally {
+      replaying = false;
     }
   }
 
   @Override
   public void onProjectDeleted(ProjectDeletedListener.Event event) {
     Project.NameKey projectName = new Project.NameKey(event.getProjectName());
-    for (URIish uri : getURIs(projectName, FilterType.PROJECT_DELETION)) {
+    for (URIish uri : getURIs(null, projectName, FilterType.PROJECT_DELETION)) {
       deleteProject(uri, projectName);
     }
   }
@@ -198,24 +200,25 @@
   @Override
   public void onHeadUpdated(HeadUpdatedListener.Event event) {
     Project.NameKey project = new Project.NameKey(event.getProjectName());
-    for (URIish uri : getURIs(project, FilterType.ALL)) {
+    for (URIish uri : getURIs(null, project, FilterType.ALL)) {
       updateHead(uri, project, event.getNewHeadName());
     }
   }
 
   private void fireBeforeStartupEvents() {
     Set<String> eventsReplayed = new HashSet<>();
-    for (GitReferenceUpdatedListener.Event event : beforeStartupEventsQueue) {
+    for (ReferenceUpdatedEvent event : beforeStartupEventsQueue) {
       String eventKey = String.format("%s:%s", event.getProjectName(), event.getRefName());
       if (!eventsReplayed.contains(eventKey)) {
         repLog.info("Firing pending task {}", event);
-        onGitReferenceUpdated(event);
+        onGitReferenceUpdated(event.getProjectName(), event.getRefName());
         eventsReplayed.add(eventKey);
       }
     }
   }
 
-  private Set<URIish> getURIs(Project.NameKey projectName, FilterType filterType) {
+  private Set<URIish> getURIs(
+      @Nullable String remoteName, Project.NameKey projectName, FilterType filterType) {
     if (config.getDestinations(filterType).isEmpty()) {
       return Collections.emptySet();
     }
@@ -230,6 +233,10 @@
         continue;
       }
 
+      if (remoteName != null && !config.getRemoteConfigName().equals(remoteName)) {
+        continue;
+      }
+
       boolean adminURLUsed = false;
 
       for (String url : config.getAdminUrls()) {
@@ -272,201 +279,75 @@
     return uris;
   }
 
-  public boolean createProject(Project.NameKey project, String head) {
+  public boolean createProject(String remoteName, Project.NameKey project, String head) {
     boolean success = true;
-    for (URIish uri : getURIs(project, FilterType.PROJECT_CREATION)) {
+    for (URIish uri : getURIs(remoteName, project, FilterType.PROJECT_CREATION)) {
       success &= createProject(uri, project, head);
     }
     return success;
   }
 
   private boolean createProject(URIish replicateURI, Project.NameKey projectName, String head) {
-    if (isGerrit(replicateURI)) {
-      gerritAdmin.createProject(replicateURI, projectName, head);
-    } else if (!replicateURI.isRemote()) {
-      createLocally(replicateURI, head);
-    } else if (isSSH(replicateURI)) {
-      createRemoteSsh(replicateURI, head);
-    } else {
-      repLog.warn(
-          "Cannot create new project on remote site {}."
-              + " Only local paths and SSH URLs are supported"
-              + " for remote repository creation",
-          replicateURI);
-      return false;
-    }
-    return true;
-  }
-
-  private static void createLocally(URIish uri, String head) {
-    try (Repository repo = new FileRepository(uri.getPath())) {
-      repo.create(true /* bare */);
-
-      if (head != null && head.startsWith(Constants.R_REFS)) {
-        RefUpdate u = repo.updateRef(Constants.HEAD);
-        u.disableRefLog();
-        u.link(head);
-      }
-      repLog.info("Created local repository: {}", uri);
-    } catch (IOException e) {
-      repLog.error("Error creating local repository {}:\n", uri.getPath(), e);
-    }
-  }
-
-  private void createRemoteSsh(URIish uri, String head) {
-    String quotedPath = QuotedString.BOURNE.quote(uri.getPath());
-    String cmd = "mkdir -p " + quotedPath + " && cd " + quotedPath + " && git init --bare";
-    if (head != null) {
-      cmd = cmd + " && git symbolic-ref HEAD " + QuotedString.BOURNE.quote(head);
-    }
-    OutputStream errStream = sshHelper.newErrorBufferStream();
-    try {
-      sshHelper.executeRemoteSsh(uri, cmd, errStream);
-      repLog.info("Created remote repository: {}", uri);
-    } catch (IOException e) {
-      repLog.error(
-          "Error creating remote repository at {}:\n"
-              + "  Exception: {}\n"
-              + "  Command: {}\n"
-              + "  Output: {}",
-          uri,
-          e,
-          cmd,
-          errStream,
-          e);
-    }
-  }
-
-  private void deleteProject(URIish replicateURI, Project.NameKey projectName) {
-    if (isGerrit(replicateURI)) {
-      gerritAdmin.deleteProject(replicateURI, projectName);
-      repLog.info("Deleted remote repository: " + replicateURI);
-    } else if (!replicateURI.isRemote()) {
-      deleteLocally(replicateURI);
-    } else if (isSSH(replicateURI)) {
-      deleteRemoteSsh(replicateURI);
-    } else {
-      repLog.warn(
-          "Cannot delete project on remote site {}. "
-              + "Only local paths and SSH URLs are supported"
-              + " for remote repository deletion",
-          replicateURI);
-    }
-  }
-
-  private static void deleteLocally(URIish uri) {
-    try {
-      recursivelyDelete(new File(uri.getPath()));
-      repLog.info("Deleted local repository: {}", uri);
-    } catch (IOException e) {
-      repLog.error("Error deleting local repository {}:\n", uri.getPath(), e);
-    }
-  }
-
-  private static void recursivelyDelete(File dir) throws IOException {
-    File[] contents = dir.listFiles();
-    if (contents != null) {
-      for (File d : contents) {
-        if (d.isDirectory()) {
-          recursivelyDelete(d);
-        } else {
-          if (!d.delete()) {
-            throw new IOException("Failed to delete: " + d.getAbsolutePath());
-          }
-        }
-      }
-    }
-    if (!dir.delete()) {
-      throw new IOException("Failed to delete: " + dir.getAbsolutePath());
-    }
-  }
-
-  private void deleteRemoteSsh(URIish uri) {
-    String quotedPath = QuotedString.BOURNE.quote(uri.getPath());
-    String cmd = "rm -rf " + quotedPath;
-    OutputStream errStream = sshHelper.newErrorBufferStream();
-    try {
-      sshHelper.executeRemoteSsh(uri, cmd, errStream);
-      repLog.info("Deleted remote repository: {}", uri);
-    } catch (IOException e) {
-      repLog.error(
-          "Error deleting remote repository at {}:\n"
-              + "  Exception: {}\n"
-              + "  Command: {}\n"
-              + "  Output: {}",
-          uri,
-          e,
-          cmd,
-          errStream,
-          e);
-    }
-  }
-
-  private void updateHead(URIish replicateURI, Project.NameKey projectName, String newHead) {
-    if (isGerrit(replicateURI)) {
-      gerritAdmin.updateHead(replicateURI, projectName, newHead);
-    } else if (!replicateURI.isRemote()) {
-      updateHeadLocally(replicateURI, newHead);
-    } else if (isSSH(replicateURI)) {
-      updateHeadRemoteSsh(replicateURI, newHead);
-    } else {
-      repLog.warn(
-          "Cannot update HEAD of project on remote site {}."
-              + " Only local paths and SSH URLs are supported"
-              + " for remote HEAD update.",
-          replicateURI);
-    }
-  }
-
-  private void updateHeadRemoteSsh(URIish uri, String newHead) {
-    String quotedPath = QuotedString.BOURNE.quote(uri.getPath());
-    String cmd =
-        "cd " + quotedPath + " && git symbolic-ref HEAD " + QuotedString.BOURNE.quote(newHead);
-    OutputStream errStream = sshHelper.newErrorBufferStream();
-    try {
-      sshHelper.executeRemoteSsh(uri, cmd, errStream);
-    } catch (IOException e) {
-      repLog.error(
-          "Error updating HEAD of remote repository at {} to {}:\n"
-              + "  Exception: {}\n"
-              + "  Command: {}\n"
-              + "  Output: {}",
-          uri,
-          newHead,
-          e,
-          cmd,
-          errStream,
-          e);
-    }
-  }
-
-  private static void updateHeadLocally(URIish uri, String newHead) {
-    try (Repository repo = new FileRepository(uri.getPath())) {
-      if (newHead != null) {
-        RefUpdate u = repo.updateRef(Constants.HEAD);
-        u.link(newHead);
-      }
-    } catch (IOException e) {
-      repLog.error("Failed to update HEAD of repository {} to {}", uri.getPath(), newHead, e);
-    }
-  }
-
-  private static boolean isSSH(URIish uri) {
-    String scheme = uri.getScheme();
-    if (!uri.isRemote()) {
-      return false;
-    }
-    if (scheme != null && scheme.toLowerCase().contains("ssh")) {
+    Optional<AdminApi> adminApi = adminApiFactory.get().create(replicateURI);
+    if (adminApi.isPresent() && adminApi.get().createProject(projectName, head)) {
       return true;
     }
-    if (scheme == null && uri.getHost() != null && uri.getPath() != null) {
-      return true;
-    }
+
+    warnCannotPerform("create new project", replicateURI);
     return false;
   }
 
-  private static boolean isGerrit(URIish uri) {
-    String scheme = uri.getScheme();
-    return scheme != null && scheme.toLowerCase().equals("gerrit+ssh");
+  private void deleteProject(URIish replicateURI, Project.NameKey projectName) {
+    Optional<AdminApi> adminApi = adminApiFactory.get().create(replicateURI);
+    if (adminApi.isPresent()) {
+      adminApi.get().deleteProject(projectName);
+      return;
+    }
+
+    warnCannotPerform("delete project", replicateURI);
+  }
+
+  private void updateHead(URIish replicateURI, Project.NameKey projectName, String newHead) {
+    Optional<AdminApi> adminApi = adminApiFactory.get().create(replicateURI);
+    if (adminApi.isPresent()) {
+      adminApi.get().updateHead(projectName, newHead);
+      return;
+    }
+
+    warnCannotPerform("update HEAD of project", replicateURI);
+  }
+
+  private void warnCannotPerform(String op, URIish uri) {
+    repLog.warn("Cannot {} on remote site {}.", op, uri);
+  }
+
+  private static class ReferenceUpdatedEvent {
+    private String projectName;
+    private String refName;
+
+    public ReferenceUpdatedEvent(String projectName, String refName) {
+      this.projectName = projectName;
+      this.refName = refName;
+    }
+
+    public String getProjectName() {
+      return projectName;
+    }
+
+    public String getRefName() {
+      return refName;
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hashCode(projectName, refName);
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+      return (obj instanceof ReferenceUpdatedEvent)
+          && Objects.equal(projectName, ((ReferenceUpdatedEvent) obj).projectName)
+          && Objects.equal(refName, ((ReferenceUpdatedEvent) obj).refName);
+    }
   }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationScheduledEvent.java b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationScheduledEvent.java
index cd7a3cf..aa965fe 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationScheduledEvent.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationScheduledEvent.java
@@ -15,7 +15,6 @@
 package com.googlesource.gerrit.plugins.replication;
 
 import com.google.gerrit.reviewdb.client.Project;
-import com.google.gerrit.reviewdb.client.Project.NameKey;
 import com.google.gerrit.server.events.RefEvent;
 
 public class ReplicationScheduledEvent extends RefEvent {
@@ -38,7 +37,7 @@
   }
 
   @Override
-  public NameKey getProjectNameKey() {
+  public Project.NameKey getProjectNameKey() {
     return new Project.NameKey(project);
   }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationState.java b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationState.java
index 6f0803a..df8f3f4 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationState.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationState.java
@@ -16,8 +16,6 @@
 
 import com.google.common.collect.HashBasedTable;
 import com.google.common.collect.Table;
-import com.google.inject.assistedinject.Assisted;
-import com.google.inject.assistedinject.AssistedInject;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
@@ -26,12 +24,7 @@
 
 public class ReplicationState {
 
-  public interface Factory {
-    ReplicationState create(PushResultProcessing processing);
-  }
-
   private boolean allScheduled;
-  private final EventsStorage eventsStorage;
   private final PushResultProcessing pushResultProcessing;
 
   private final Lock countingLock = new ReentrantLock();
@@ -57,11 +50,7 @@
   private int totalPushTasksCount;
   private int finishedPushTasksCount;
 
-  private String eventKey;
-
-  @AssistedInject
-  ReplicationState(EventsStorage storage, @Assisted PushResultProcessing processing) {
-    eventsStorage = storage;
+  ReplicationState(PushResultProcessing processing) {
     pushResultProcessing = processing;
     statusByProjectRef = HashBasedTable.create();
   }
@@ -86,7 +75,6 @@
       URIish uri,
       RefPushResult status,
       RemoteRefUpdate.Status refUpdateStatus) {
-    deleteEvent();
     pushResultProcessing.onRefReplicatedToOneNode(project, ref, uri, status, refUpdateStatus);
 
     RefReplicationStatus completedRefStatus = null;
@@ -116,12 +104,6 @@
     }
   }
 
-  private void deleteEvent() {
-    if (eventKey != null) {
-      eventsStorage.delete(eventKey);
-    }
-  }
-
   public void markAllPushTasksScheduled() {
     countingLock.lock();
     try {
@@ -192,8 +174,4 @@
       return name().toLowerCase().replace("_", "-");
     }
   }
-
-  public void setEventKey(String eventKey) {
-    this.eventKey = eventKey;
-  }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationStateListeners.java b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationStateListeners.java
new file mode 100644
index 0000000..d7bf227
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationStateListeners.java
@@ -0,0 +1,48 @@
+// Copyright (C) 2019 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.extensions.registration.DynamicSet;
+import com.google.inject.Inject;
+
+public class ReplicationStateListeners implements ReplicationStateListener {
+  private final DynamicSet<ReplicationStateListener> listeners;
+
+  @Inject
+  ReplicationStateListeners(DynamicSet<ReplicationStateListener> stateListeners) {
+    this.listeners = stateListeners;
+  }
+
+  @Override
+  public void warn(String msg, ReplicationState... states) {
+    for (ReplicationStateListener listener : listeners) {
+      listener.warn(msg, states);
+    }
+  }
+
+  @Override
+  public void error(String msg, ReplicationState... states) {
+    for (ReplicationStateListener listener : listeners) {
+      listener.error(msg, states);
+    }
+  }
+
+  @Override
+  public void error(String msg, Throwable t, ReplicationState... states) {
+    for (ReplicationStateListener listener : listeners) {
+      listener.error(msg, t, states);
+    }
+  }
+}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorage.java b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorage.java
new file mode 100644
index 0000000..39158f3
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/ReplicationTasksStorage.java
@@ -0,0 +1,137 @@
+// Copyright (C) 2018 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 java.nio.charset.StandardCharsets.UTF_8;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.flogger.FluentLogger;
+import com.google.common.hash.Hashing;
+import com.google.gson.Gson;
+import com.google.inject.Inject;
+import com.google.inject.ProvisionException;
+import com.google.inject.Singleton;
+import java.io.IOException;
+import java.nio.file.DirectoryStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import org.eclipse.jgit.lib.ObjectId;
+import org.eclipse.jgit.transport.URIish;
+
+@Singleton
+public class ReplicationTasksStorage {
+  private static final FluentLogger logger = FluentLogger.forEnclosingClass();
+
+  private boolean disableDeleteForTesting;
+
+  public static class ReplicateRefUpdate {
+    public final String project;
+    public final String ref;
+    public final String uri;
+    public final String remote;
+
+    public ReplicateRefUpdate(String project, String ref, URIish uri, String remote) {
+      this.project = project;
+      this.ref = ref;
+      this.uri = uri.toASCIIString();
+      this.remote = remote;
+    }
+
+    @Override
+    public String toString() {
+      return "ref-update " + project + ":" + ref + " uri:" + uri + " remote:" + remote;
+    }
+  }
+
+  private static final Gson GSON = new Gson();
+
+  private final Path refUpdates;
+
+  @Inject
+  ReplicationTasksStorage(ReplicationConfig config) {
+    refUpdates = config.getEventsDirectory().resolve("ref-updates");
+  }
+
+  public String persist(ReplicateRefUpdate r) {
+    String json = GSON.toJson(r) + "\n";
+    String eventKey = sha1(json).name();
+    Path file = refUpdates().resolve(eventKey);
+
+    if (Files.exists(file)) {
+      return eventKey;
+    }
+
+    try {
+      logger.atFine().log("CREATE %s (%s:%s => %s)", file, r.project, r.ref, r.uri);
+      Files.write(file, json.getBytes(UTF_8));
+    } catch (IOException e) {
+      logger.atWarning().withCause(e).log("Couldn't persist event %s", json);
+    }
+    return eventKey;
+  }
+
+  @VisibleForTesting
+  public void disableDeleteForTesting(boolean deleteDisabled) {
+    this.disableDeleteForTesting = deleteDisabled;
+  }
+
+  public void delete(ReplicateRefUpdate r) {
+    String taskJson = GSON.toJson(r) + "\n";
+    String taskKey = sha1(taskJson).name();
+    Path file = refUpdates().resolve(taskKey);
+
+    if (disableDeleteForTesting) {
+      logger.atFine().log("DELETE %s (%s:%s => %s) DISABLED", file, r.project, r.ref, r.uri);
+      return;
+    }
+
+    try {
+      logger.atFine().log("DELETE %s (%s:%s => %s)", file, r.project, r.ref, r.uri);
+      Files.delete(file);
+    } catch (IOException e) {
+      logger.atSevere().withCause(e).log("Error while deleting event %s", taskKey);
+    }
+  }
+
+  public List<ReplicateRefUpdate> list() {
+    ArrayList<ReplicateRefUpdate> result = new ArrayList<>();
+    try (DirectoryStream<Path> events = Files.newDirectoryStream(refUpdates())) {
+      for (Path e : events) {
+        if (Files.isRegularFile(e)) {
+          String json = new String(Files.readAllBytes(e), UTF_8);
+          result.add(GSON.fromJson(json, ReplicateRefUpdate.class));
+        }
+      }
+    } catch (IOException e) {
+      logger.atSevere().withCause(e).log("Error when firing pending events");
+    }
+    return result;
+  }
+
+  @SuppressWarnings("deprecation")
+  private ObjectId sha1(String s) {
+    return ObjectId.fromRaw(Hashing.sha1().hashString(s, UTF_8).asBytes());
+  }
+
+  private Path refUpdates() {
+    try {
+      return Files.createDirectories(refUpdates);
+    } catch (IOException e) {
+      throw new ProvisionException(String.format("Couldn't create %s", refUpdates), e);
+    }
+  }
+}
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 2b0c16b..c518091 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/SecureCredentialsFactory.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/SecureCredentialsFactory.java
@@ -17,6 +17,7 @@
 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;
@@ -49,8 +50,8 @@
 
   @Override
   public SecureCredentialsProvider create(String remoteName) {
-    String user = config.getString("remote", remoteName, "username");
-    String pass = config.getString("remote", remoteName, "password");
+    String user = Objects.toString(config.getString("remote", remoteName, "username"), "");
+    String pass = Objects.toString(config.getString("remote", remoteName, "password"), "");
     return new SecureCredentialsProvider(user, pass);
   }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/SecureCredentialsProvider.java b/src/main/java/com/googlesource/gerrit/plugins/replication/SecureCredentialsProvider.java
index c4294a9..62b4036 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/SecureCredentialsProvider.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/SecureCredentialsProvider.java
@@ -20,7 +20,7 @@
 import org.eclipse.jgit.transport.URIish;
 
 /** Looks up a remote's password in secure.config. */
-class SecureCredentialsProvider extends CredentialsProvider {
+public class SecureCredentialsProvider extends CredentialsProvider {
   private final String cfgUser;
   private final String cfgPass;
 
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 2dbc7b7..fa8b44c 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/StartCommand.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/StartCommand.java
@@ -52,15 +52,14 @@
   @Inject private PushAll.Factory pushFactory;
 
   private final Object lock = new Object();
-  @Inject private ReplicationState.Factory replicationStateFactory;
 
   @Override
   protected void run() throws Failure {
-    if (all && projectPatterns.size() > 0) {
+    if (all && !projectPatterns.isEmpty()) {
       throw new UnloggedFailure(1, "error: cannot combine --all and PROJECT");
     }
 
-    ReplicationState state = replicationStateFactory.create(new CommandProcessing(this));
+    ReplicationState state = new ReplicationState(new CommandProcessing(this));
     Future<?> future = null;
 
     ReplicationFilter projectFilter;
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/TransportFactory.java b/src/main/java/com/googlesource/gerrit/plugins/replication/TransportFactory.java
new file mode 100644
index 0000000..ba14299
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/TransportFactory.java
@@ -0,0 +1,26 @@
+// Copyright (C) 2019 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 org.eclipse.jgit.errors.NotSupportedException;
+import org.eclipse.jgit.errors.TransportException;
+import org.eclipse.jgit.lib.Repository;
+import org.eclipse.jgit.transport.Transport;
+import org.eclipse.jgit.transport.URIish;
+
+public interface TransportFactory {
+
+  Transport open(Repository local, URIish uri) throws NotSupportedException, TransportException;
+}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/TransportFactoryImpl.java b/src/main/java/com/googlesource/gerrit/plugins/replication/TransportFactoryImpl.java
new file mode 100644
index 0000000..58c1214
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/TransportFactoryImpl.java
@@ -0,0 +1,30 @@
+// Copyright (C) 2019 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 org.eclipse.jgit.errors.NotSupportedException;
+import org.eclipse.jgit.errors.TransportException;
+import org.eclipse.jgit.lib.Repository;
+import org.eclipse.jgit.transport.Transport;
+import org.eclipse.jgit.transport.URIish;
+
+public class TransportFactoryImpl implements TransportFactory {
+
+  @Override
+  public Transport open(Repository git, URIish uri)
+      throws NotSupportedException, TransportException {
+    return Transport.open(git, uri);
+  }
+}
diff --git a/src/main/resources/Documentation/about.md b/src/main/resources/Documentation/about.md
index 69a371b..adb8d4c 100644
--- a/src/main/resources/Documentation/about.md
+++ b/src/main/resources/Documentation/about.md
@@ -24,7 +24,7 @@
 
 * `refs/users/*` (user branches)
 * `refs/meta/external-ids` (external IDs)
-* `refs/starred-changes/*` (star labels)
+* `refs/starred-changes/*` (star labels, not needed for Gerrit slaves)
 * `refs/sequences/accounts` (account sequence numbers, not needed for Gerrit
   slaves)
 
diff --git a/src/main/resources/Documentation/config.md b/src/main/resources/Documentation/config.md
index c479ab2..c9356b0 100644
--- a/src/main/resources/Documentation/config.md
+++ b/src/main/resources/Documentation/config.md
@@ -13,6 +13,11 @@
   sudo su -c 'ssh mirror1.us.some.org echo' gerrit2
 ```
 
+*NOTE:* make sure the local user's ssh keys format is PEM, here how to generate them:
+```
+  ssh-keygen -m PEM -t rsa -C "your_email@example.com"
+```
+
 <a name="example_file">
 Next, create `$site_path/etc/replication.config` as a Git-style config
 file, for example to replicate in parallel to four different hosts:</a>
@@ -283,6 +288,20 @@
 
 	By default, use replication.maxRetries.
 
+remote.NAME.drainQueueAttempts
+:	Maximum number of attempts to drain the replication event queue before
+	stopping the plugin.
+
+	When stopping the plugin, the shutdown will be delayed trying to drain
+	the event queue.
+
+	The maximum delay is "drainQueueAttempts" * "replicationDelay" seconds.
+
+	When not set or set to 0, the queue is not drained and the pending
+	replication events are cancelled.
+
+	By default, do not drain replication events.
+
 remote.NAME.threads
 :	Number of worker threads to dedicate to pushing to the
 	repositories described by this remote.  Each thread can push
@@ -309,7 +328,7 @@
 	If the remote site was not available at the moment when a new
 	project was created, it will be created if during the replication
 	of a ref it is found to be missing.
-	
+
 	If false, repositories are never created automatically on this
 	remote.
 
@@ -409,7 +428,7 @@
 File `~/.ssh/config`
 --------------------
 
-If present, Gerrit reads and caches `~/.ssh/config` at startup, and
+Gerrit reads and caches the `~/.ssh/config` at startup, and
 supports most SSH configuration options.  For example:
 
 ```
@@ -423,6 +442,15 @@
     PreferredAuthentications publickey
 ```
 
+*IdentityFile* and *PreferredAuthentications* must be defined for all the hosts.
+Here an example of the minimum `~/.ssh/config` needed:
+
+```
+  Host *
+    IdentityFile ~/.ssh/id_rsa
+    PreferredAuthentications publickey
+```
+
 Supported options:
 
   * Host
diff --git a/src/main/resources/Documentation/extension-point.md b/src/main/resources/Documentation/extension-point.md
new file mode 100644
index 0000000..076aded
--- /dev/null
+++ b/src/main/resources/Documentation/extension-point.md
@@ -0,0 +1,53 @@
+@PLUGIN@ extension points
+==============
+
+The replication plugin exposes an extension point to allow influencing its behaviour from another plugin or a script.
+Extension points can be defined from the replication plugin only when it is loaded as [libModule](/config-gerrit.html#gerrit.installModule) and
+implemented by another plugin by declaring a `provided` dependency from the replication plugin.
+
+### Install extension libModule
+
+The replication plugin's extension points are defined in the `c.g.g.p.r.ReplicationExtensionPointModule`
+that needs to be configured as libModule.
+
+Create a symbolic link from `$GERRIT_SITE/plugins/replication.jar` into `$GERRIT_SITE/lib`
+and then add the replication extension module to the `gerrit.config`.
+
+Example:
+
+```
+[gerrit]
+  installModule = com.googlesource.gerrit.plugins.replication.ReplicationExtensionPointModule
+```
+
+> **NOTE**: Use and configuration of the replication plugin as library module requires a Gerrit server restart and does not support hot plugin install or upgrade.
+
+
+### Extension points
+
+* `com.googlesource.gerrit.plugins.replication.ReplicationPushFilter`
+
+  Filter out the ref updates pushed to a remote instance.
+  Only one filter at a time is supported. Filter implementation needs to bind a `DynamicItem`.
+
+  Default: no filtering
+
+  Example:
+
+  ```
+  DynamicItem.bind(binder(), ReplicationPushFilter.class).to(ReplicationPushFilterImpl.class);
+  ```
+
+* `com.googlesource.gerrit.plugins.replication.AdminApiFactory`
+
+  Create an instance of `AdminApi` for a given remote URL. The default implementation
+  provides API instances for local FS, remote SSH, and remote Gerrit.
+
+  Only one factory at a time is supported. The implementation needs to be bound as a
+  `DynamicItem`.
+
+  Example:
+
+  ```
+  DynamicItem.bind(binder(), AdminApiFactory.class).to(AdminApiFactoryImpl.class);
+  ```
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/GitUpdateProcessingTest.java b/src/test/java/com/googlesource/gerrit/plugins/replication/GitUpdateProcessingTest.java
index 337bd1d..5fa7b98 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/GitUpdateProcessingTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/GitUpdateProcessingTest.java
@@ -22,8 +22,8 @@
 import static org.easymock.EasyMock.reset;
 import static org.easymock.EasyMock.verify;
 
-import com.google.gerrit.common.EventDispatcher;
 import com.google.gerrit.reviewdb.server.ReviewDb;
+import com.google.gerrit.server.events.EventDispatcher;
 import com.google.gerrit.server.permissions.PermissionBackendException;
 import com.google.gwtorm.client.KeyUtil;
 import com.google.gwtorm.server.OrmException;
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/PushOneTest.java b/src/test/java/com/googlesource/gerrit/plugins/replication/PushOneTest.java
new file mode 100644
index 0000000..af065b3
--- /dev/null
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/PushOneTest.java
@@ -0,0 +1,375 @@
+// Copyright (C) 2019 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.googlesource.gerrit.plugins.replication.RemoteRefUpdateCollectionMatcher.eqRemoteRef;
+import static org.easymock.EasyMock.anyObject;
+import static org.easymock.EasyMock.createNiceMock;
+import static org.easymock.EasyMock.expect;
+import static org.easymock.EasyMock.getCurrentArguments;
+import static org.easymock.EasyMock.replay;
+import static org.easymock.EasyMock.verify;
+import static org.eclipse.jgit.lib.Ref.Storage.NEW;
+
+import com.google.common.collect.ImmutableList;
+import com.google.gerrit.extensions.registration.DynamicItem;
+import com.google.gerrit.metrics.Timer1;
+import com.google.gerrit.reviewdb.client.Project;
+import com.google.gerrit.server.git.GitRepositoryManager;
+import com.google.gerrit.server.git.PerThreadRequestScope;
+import com.google.gerrit.server.permissions.PermissionBackend;
+import com.google.gerrit.server.permissions.PermissionBackend.ForProject;
+import com.google.gerrit.server.permissions.PermissionBackend.WithUser;
+import com.google.gerrit.server.project.ProjectCache;
+import com.google.gerrit.server.project.ProjectState;
+import com.google.gerrit.server.util.IdGenerator;
+import java.io.File;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import junit.framework.AssertionFailedError;
+import org.easymock.IAnswer;
+import org.eclipse.jgit.errors.NotSupportedException;
+import org.eclipse.jgit.errors.TransportException;
+import org.eclipse.jgit.lib.Config;
+import org.eclipse.jgit.lib.ObjectId;
+import org.eclipse.jgit.lib.ObjectIdRef;
+import org.eclipse.jgit.lib.Ref;
+import org.eclipse.jgit.lib.RefUpdate;
+import org.eclipse.jgit.lib.Repository;
+import org.eclipse.jgit.storage.file.FileBasedConfig;
+import org.eclipse.jgit.transport.FetchConnection;
+import org.eclipse.jgit.transport.PushConnection;
+import org.eclipse.jgit.transport.PushResult;
+import org.eclipse.jgit.transport.RefSpec;
+import org.eclipse.jgit.transport.RemoteConfig;
+import org.eclipse.jgit.transport.RemoteRefUpdate;
+import org.eclipse.jgit.transport.Transport;
+import org.eclipse.jgit.transport.URIish;
+import org.eclipse.jgit.util.FS;
+import org.junit.Before;
+import org.junit.Test;
+
+public class PushOneTest {
+  private static final int TEST_PUSH_TIMEOUT_SECS = 10;
+
+  private GitRepositoryManager gitRepositoryManagerMock;
+  private Repository repositoryMock;
+  private PermissionBackend permissionBackendMock;
+  private PermissionBackend.WithUser withUserMock;
+  private PermissionBackend.ForProject forProjectMock;
+
+  private Destination destinationMock;
+  private RemoteConfig remoteConfigMock;
+  private RefSpec refSpecMock;
+  private CredentialsFactory credentialsFactory;
+  private PerThreadRequestScope.Scoper threadRequestScoperMock;
+  private ReplicationQueue replicationQueueMock;
+  private IdGenerator idGeneratorMock;
+  private ReplicationStateListeners replicationStateListenersMock;
+  private ReplicationMetrics replicationMetricsMock;
+  private Timer1.Context timerContextMock;
+  private ProjectCache projectCacheMock;
+  private TransportFactory transportFactoryMock;
+  private Transport transportMock;
+  private FetchConnection fetchConnection;
+  private PushConnection pushConnection;
+  private ProjectState projectStateMock;
+  private RefUpdate refUpdateMock;
+
+  private Project.NameKey projectNameKey;
+  private URIish urish;
+  private Map<String, Ref> localRefs;
+
+  private Map<String, Ref> remoteRefs;
+  private CountDownLatch isCallFinished;
+  private Ref newLocalRef;
+
+  @Before
+  public void setup() throws Exception {
+    projectNameKey = new Project.NameKey("fooProject");
+    urish = new URIish("http://foo.com/fooProject.git");
+
+    newLocalRef =
+        new ObjectIdRef.Unpeeled(
+            NEW, "foo", ObjectId.fromString("0000000000000000000000000000000000000001"));
+
+    localRefs = new HashMap<>();
+    localRefs.put("fooProject", newLocalRef);
+
+    Ref remoteRef = new ObjectIdRef.Unpeeled(NEW, "foo", ObjectId.zeroId());
+    remoteRefs = new HashMap<>();
+    remoteRefs.put("fooProject", remoteRef);
+
+    isCallFinished = new CountDownLatch(1);
+
+    setupMocks();
+  }
+
+  private void setupMocks() throws Exception {
+    FileBasedConfig config = new FileBasedConfig(new Config(), new File("/foo"), FS.DETECTED);
+    config.setString("remote", "Replication", "push", "foo");
+
+    setupRefUpdateMock();
+    setupRepositoryMock(config);
+    setupGitRepoManagerMock();
+
+    projectStateMock = createNiceMock(ProjectState.class);
+    forProjectMock = createNiceMock(ForProject.class);
+    setupWithUserMock();
+    setupPermissionBackedMock();
+
+    setupDestinationMock();
+
+    setupRefSpecMock();
+    setupRemoteConfigMock();
+
+    credentialsFactory = createNiceMock(CredentialsFactory.class);
+
+    setupFetchConnectionMock();
+    setupPushConnectionMock();
+    setupRequestScopeMock();
+    replicationQueueMock = createNiceMock(ReplicationQueue.class);
+    idGeneratorMock = createNiceMock(IdGenerator.class);
+    replicationStateListenersMock = createNiceMock(ReplicationStateListeners.class);
+
+    timerContextMock = createNiceMock(Timer1.Context.class);
+    setupReplicationMetricsMock();
+
+    setupTransportMock();
+
+    setupProjectCacheMock();
+
+    replay(
+        gitRepositoryManagerMock,
+        refUpdateMock,
+        repositoryMock,
+        permissionBackendMock,
+        destinationMock,
+        remoteConfigMock,
+        credentialsFactory,
+        threadRequestScoperMock,
+        replicationQueueMock,
+        idGeneratorMock,
+        replicationStateListenersMock,
+        replicationMetricsMock,
+        projectCacheMock,
+        timerContextMock,
+        transportFactoryMock,
+        projectStateMock,
+        withUserMock,
+        forProjectMock,
+        fetchConnection,
+        pushConnection,
+        refSpecMock);
+  }
+
+  @Test
+  public void shouldPushAllRefsWhenNoFilters() throws InterruptedException, IOException {
+    shouldPushAllRefsWithDynamicItemFilter(DynamicItem.itemOf(ReplicationPushFilter.class, null));
+  }
+
+  @Test
+  public void shouldPushAllRefsWhenNoFiltersSetup() throws InterruptedException, IOException {
+    shouldPushAllRefsWithDynamicItemFilter(null);
+  }
+
+  private void shouldPushAllRefsWithDynamicItemFilter(
+      DynamicItem<ReplicationPushFilter> replicationPushFilter)
+      throws IOException, NotSupportedException, TransportException, InterruptedException {
+    List<RemoteRefUpdate> expectedUpdates =
+        Arrays.asList(
+            new RemoteRefUpdate(
+                repositoryMock,
+                newLocalRef.getName(),
+                newLocalRef.getObjectId(),
+                "fooProject",
+                false,
+                "fooProject",
+                null));
+
+    PushResult pushResult = new PushResult();
+
+    expect(transportMock.push(anyObject(), eqRemoteRef(expectedUpdates)))
+        .andReturn(pushResult)
+        .once();
+    replay(transportMock);
+
+    PushOne pushOne = createPushOne(replicationPushFilter);
+
+    pushOne.addRef(PushOne.ALL_REFS);
+    pushOne.run();
+
+    isCallFinished.await(TEST_PUSH_TIMEOUT_SECS, TimeUnit.SECONDS);
+
+    verify(transportMock);
+  }
+
+  @Test
+  public void shouldBlockReplicationUsingPushFilter() throws InterruptedException, IOException {
+    DynamicItem<ReplicationPushFilter> replicationPushFilter =
+        DynamicItem.itemOf(
+            ReplicationPushFilter.class,
+            new ReplicationPushFilter() {
+
+              @Override
+              public List<RemoteRefUpdate> filter(
+                  String projectName, List<RemoteRefUpdate> remoteUpdatesList) {
+                return Collections.emptyList();
+              }
+            });
+
+    // easymock way to check if method was never called
+    expect(transportMock.push(anyObject(), anyObject()))
+        .andThrow(new AssertionFailedError())
+        .anyTimes();
+    replay(transportMock);
+
+    PushOne pushOne = createPushOne(replicationPushFilter);
+
+    pushOne.addRef(PushOne.ALL_REFS);
+    pushOne.run();
+
+    isCallFinished.await(10, TimeUnit.SECONDS);
+
+    verify(transportMock);
+  }
+
+  private PushOne createPushOne(DynamicItem<ReplicationPushFilter> replicationPushFilter) {
+    PushOne push =
+        new PushOne(
+            gitRepositoryManagerMock,
+            permissionBackendMock,
+            destinationMock,
+            remoteConfigMock,
+            credentialsFactory,
+            threadRequestScoperMock,
+            replicationQueueMock,
+            idGeneratorMock,
+            replicationStateListenersMock,
+            replicationMetricsMock,
+            projectCacheMock,
+            transportFactoryMock,
+            projectNameKey,
+            urish);
+
+    push.setReplicationPushFilter(replicationPushFilter);
+    return push;
+  }
+
+  private void setupProjectCacheMock() throws IOException {
+    projectCacheMock = createNiceMock(ProjectCache.class);
+    expect(projectCacheMock.checkedGet(projectNameKey)).andReturn(projectStateMock);
+  }
+
+  private void setupTransportMock() throws NotSupportedException, TransportException {
+    transportMock = createNiceMock(Transport.class);
+    expect(transportMock.openFetch()).andReturn(fetchConnection);
+    transportFactoryMock = createNiceMock(TransportFactory.class);
+    expect(transportFactoryMock.open(repositoryMock, urish)).andReturn(transportMock).anyTimes();
+  }
+
+  private void setupReplicationMetricsMock() {
+    replicationMetricsMock = createNiceMock(ReplicationMetrics.class);
+    expect(replicationMetricsMock.start(anyObject())).andReturn(timerContextMock);
+  }
+
+  private void setupRequestScopeMock() {
+    threadRequestScoperMock = createNiceMock(PerThreadRequestScope.Scoper.class);
+    expect(threadRequestScoperMock.scope(anyObject()))
+        .andAnswer(
+            new IAnswer<Callable<Object>>() {
+              @SuppressWarnings("unchecked")
+              @Override
+              public Callable<Object> answer() throws Throwable {
+                Callable<Object> originalCall = (Callable<Object>) getCurrentArguments()[0];
+                return new Callable<Object>() {
+
+                  @Override
+                  public Object call() throws Exception {
+                    Object result = originalCall.call();
+                    isCallFinished.countDown();
+                    return result;
+                  }
+                };
+              }
+            })
+        .anyTimes();
+  }
+
+  private void setupPushConnectionMock() {
+    pushConnection = createNiceMock(PushConnection.class);
+    expect(pushConnection.getRefsMap()).andReturn(remoteRefs);
+  }
+
+  private void setupFetchConnectionMock() {
+    fetchConnection = createNiceMock(FetchConnection.class);
+    expect(fetchConnection.getRefsMap()).andReturn(remoteRefs);
+  }
+
+  private void setupRemoteConfigMock() {
+    remoteConfigMock = createNiceMock(RemoteConfig.class);
+    expect(remoteConfigMock.getPushRefSpecs()).andReturn(ImmutableList.of(refSpecMock));
+  }
+
+  private void setupRefSpecMock() {
+    refSpecMock = createNiceMock(RefSpec.class);
+    expect(refSpecMock.matchSource(anyObject(String.class))).andReturn(true);
+    expect(refSpecMock.expandFromSource(anyObject(String.class))).andReturn(refSpecMock);
+    expect(refSpecMock.getDestination()).andReturn("fooProject").anyTimes();
+    expect(refSpecMock.isForceUpdate()).andReturn(false).anyTimes();
+  }
+
+  private void setupDestinationMock() {
+    destinationMock = createNiceMock(Destination.class);
+    expect(destinationMock.requestRunway(anyObject())).andReturn(RunwayStatus.allowed());
+  }
+
+  private void setupPermissionBackedMock() {
+    permissionBackendMock = createNiceMock(PermissionBackend.class);
+    expect(permissionBackendMock.currentUser()).andReturn(withUserMock);
+  }
+
+  private void setupWithUserMock() {
+    withUserMock = createNiceMock(WithUser.class);
+    expect(withUserMock.project(projectNameKey)).andReturn(forProjectMock);
+  }
+
+  private void setupGitRepoManagerMock() throws IOException {
+    gitRepositoryManagerMock = createNiceMock(GitRepositoryManager.class);
+    expect(gitRepositoryManagerMock.openRepository(projectNameKey)).andReturn(repositoryMock);
+  }
+
+  @SuppressWarnings("deprecation")
+  private void setupRepositoryMock(FileBasedConfig config) throws IOException {
+    repositoryMock = createNiceMock(Repository.class);
+    expect(repositoryMock.getConfig()).andReturn(config).anyTimes();
+    expect(repositoryMock.getAllRefs()).andReturn(localRefs);
+    expect(repositoryMock.updateRef("fooProject")).andReturn(refUpdateMock);
+  }
+
+  private void setupRefUpdateMock() {
+    refUpdateMock = createNiceMock(RefUpdate.class);
+    expect(refUpdateMock.getOldObjectId())
+        .andReturn(ObjectId.fromString("0000000000000000000000000000000000000001"))
+        .anyTimes();
+  }
+}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/RemoteRefUpdateCollectionMatcher.java b/src/test/java/com/googlesource/gerrit/plugins/replication/RemoteRefUpdateCollectionMatcher.java
new file mode 100644
index 0000000..111a792
--- /dev/null
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/RemoteRefUpdateCollectionMatcher.java
@@ -0,0 +1,64 @@
+// Copyright (C) 2019 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.Collection;
+import java.util.Objects;
+import org.easymock.EasyMock;
+import org.easymock.IArgumentMatcher;
+import org.eclipse.jgit.transport.RemoteRefUpdate;
+
+public class RemoteRefUpdateCollectionMatcher implements IArgumentMatcher {
+  Collection<RemoteRefUpdate> expectedRemoteRefs;
+
+  public static Collection<RemoteRefUpdate> eqRemoteRef(
+      Collection<RemoteRefUpdate> expectedRemoteRefs) {
+    EasyMock.reportMatcher(new RemoteRefUpdateCollectionMatcher(expectedRemoteRefs));
+    return null;
+  }
+
+  public RemoteRefUpdateCollectionMatcher(Collection<RemoteRefUpdate> expectedRemoteRefs) {
+    this.expectedRemoteRefs = expectedRemoteRefs;
+  }
+
+  @Override
+  public boolean matches(Object argument) {
+    if (!(argument instanceof Collection)) return false;
+
+    @SuppressWarnings("unchecked")
+    Collection<RemoteRefUpdate> refs = (Collection<RemoteRefUpdate>) argument;
+
+    if (expectedRemoteRefs.size() != refs.size()) return false;
+    return refs.stream()
+        .allMatch(
+            ref -> expectedRemoteRefs.stream().anyMatch(expectedRef -> compare(ref, expectedRef)));
+  }
+
+  @Override
+  public void appendTo(StringBuffer buffer) {
+    buffer.append("expected:" + expectedRemoteRefs.toString());
+  }
+
+  private boolean compare(RemoteRefUpdate ref, RemoteRefUpdate expectedRef) {
+    return Objects.equals(ref.getRemoteName(), expectedRef.getRemoteName())
+        && Objects.equals(ref.getStatus(), expectedRef.getStatus())
+        && Objects.equals(ref.getExpectedOldObjectId(), expectedRef.getExpectedOldObjectId())
+        && Objects.equals(ref.getNewObjectId(), expectedRef.getNewObjectId())
+        && Objects.equals(ref.isFastForward(), expectedRef.isFastForward())
+        && Objects.equals(ref.getSrcRef(), expectedRef.getSrcRef())
+        && Objects.equals(ref.isForceUpdate(), expectedRef.isForceUpdate())
+        && Objects.equals(ref.getMessage(), expectedRef.getMessage());
+  }
+}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationIT.java b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationIT.java
new file mode 100644
index 0000000..61a53f3
--- /dev/null
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationIT.java
@@ -0,0 +1,377 @@
+// Copyright (C) 2019 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 com.google.gerrit.testing.GerritJUnit.assertThrows;
+import static java.util.stream.Collectors.toList;
+
+import com.google.common.flogger.FluentLogger;
+import com.google.gerrit.acceptance.LightweightPluginDaemonTest;
+import com.google.gerrit.acceptance.PushOneCommit.Result;
+import com.google.gerrit.acceptance.TestPlugin;
+import com.google.gerrit.acceptance.UseLocalDisk;
+import com.google.gerrit.extensions.annotations.PluginData;
+import com.google.gerrit.extensions.api.projects.BranchInput;
+import com.google.gerrit.extensions.common.ProjectInfo;
+import com.google.gerrit.reviewdb.client.Project;
+import com.google.gerrit.server.config.SitePaths;
+import com.google.inject.Inject;
+import com.google.inject.Key;
+import com.googlesource.gerrit.plugins.replication.ReplicationTasksStorage.ReplicateRefUpdate;
+import java.io.IOException;
+import java.nio.file.DirectoryStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Optional;
+import java.util.function.Supplier;
+import java.util.regex.Pattern;
+import org.eclipse.jgit.lib.Constants;
+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.RemoteRefUpdate;
+import org.eclipse.jgit.transport.URIish;
+import org.eclipse.jgit.util.FS;
+import org.junit.Test;
+
+@UseLocalDisk
+@TestPlugin(
+    name = "replication",
+    sysModule = "com.googlesource.gerrit.plugins.replication.ReplicationModule")
+public class ReplicationIT extends LightweightPluginDaemonTest {
+  private static final Optional<String> ALL_PROJECTS = Optional.empty();
+  private static final FluentLogger logger = FluentLogger.forEnclosingClass();
+  private static final int TEST_REPLICATION_DELAY = 1;
+  private static final Duration TEST_TIMEOUT = Duration.ofSeconds(TEST_REPLICATION_DELAY * 2);
+
+  @Inject private SitePaths sitePaths;
+  private Path pluginDataDir;
+  private Path gitPath;
+  private Path storagePath;
+  private FileBasedConfig config;
+  private ReplicationTasksStorage tasksStorage;
+
+  @Override
+  public void setUpTestPlugin() throws Exception {
+    gitPath = sitePaths.site_path.resolve("git");
+
+    config =
+        new FileBasedConfig(sitePaths.etc_dir.resolve("replication.config").toFile(), FS.DETECTED);
+    setReplicationDestination(
+        "remote1",
+        "suffix1",
+        Optional.of("not-used-project")); // Simulates a full replication.config initialization
+    config.save();
+
+    super.setUpTestPlugin();
+
+    pluginDataDir = plugin.getSysInjector().getInstance(Key.get(Path.class, PluginData.class));
+    storagePath = pluginDataDir.resolve("ref-updates");
+    tasksStorage = plugin.getSysInjector().getInstance(ReplicationTasksStorage.class);
+    cleanupReplicationTasks();
+    tasksStorage.disableDeleteForTesting(true);
+  }
+
+  @Test
+  public void shouldReplicateNewProject() throws Exception {
+    setReplicationDestination("foo", "replica", ALL_PROJECTS);
+    reloadConfig();
+
+    Project.NameKey sourceProject = createTestProject("foo");
+
+    assertThat(listReplicationTasks("refs/meta/config")).hasSize(1);
+
+    waitUntil(() -> projectExists(new Project.NameKey(sourceProject + "replica.git")));
+
+    ProjectInfo replicaProject = gApi.projects().name(sourceProject + "replica").get();
+    assertThat(replicaProject).isNotNull();
+  }
+
+  @Test
+  public void shouldReplicateNewChangeRef() throws Exception {
+    Project.NameKey targetProject = createTestProject("projectreplica");
+
+    setReplicationDestination("foo", "replica", ALL_PROJECTS);
+    reloadConfig();
+
+    Result pushResult = createChange();
+    RevCommit sourceCommit = pushResult.getCommit();
+    String sourceRef = pushResult.getPatchSet().getRefName();
+
+    assertThat(listReplicationTasks("refs/changes/\\d*/\\d*/\\d*")).hasSize(1);
+
+    try (Repository repo = repoManager.openRepository(targetProject)) {
+      waitUntil(() -> checkedGetRef(repo, sourceRef) != null);
+
+      Ref targetBranchRef = getRef(repo, sourceRef);
+      assertThat(targetBranchRef).isNotNull();
+      assertThat(targetBranchRef.getObjectId()).isEqualTo(sourceCommit.getId());
+    }
+  }
+
+  @Test
+  public void shouldReplicateNewBranch() throws Exception {
+    setReplicationDestination("foo", "replica", ALL_PROJECTS);
+    reloadConfig();
+
+    Project.NameKey targetProject = createTestProject("projectreplica");
+    String newBranch = "refs/heads/mybranch";
+    String master = "refs/heads/master";
+    BranchInput input = new BranchInput();
+    input.revision = master;
+    gApi.projects().name(project.get()).branch(newBranch).create(input);
+
+    assertThat(listReplicationTasks("refs/heads/(mybranch|master)")).hasSize(2);
+
+    try (Repository repo = repoManager.openRepository(targetProject);
+        Repository sourceRepo = repoManager.openRepository(project)) {
+      waitUntil(() -> checkedGetRef(repo, newBranch) != null);
+
+      Ref masterRef = getRef(sourceRepo, master);
+      Ref targetBranchRef = getRef(repo, newBranch);
+      assertThat(targetBranchRef).isNotNull();
+      assertThat(targetBranchRef.getObjectId()).isEqualTo(masterRef.getObjectId());
+    }
+  }
+
+  @Test
+  public void shouldReplicateNewBranchToTwoRemotes() throws Exception {
+    Project.NameKey targetProject1 = createTestProject("projectreplica1");
+    Project.NameKey targetProject2 = createTestProject("projectreplica2");
+
+    setReplicationDestination("foo1", "replica1", ALL_PROJECTS);
+    setReplicationDestination("foo2", "replica2", ALL_PROJECTS);
+    reloadConfig();
+
+    Result pushResult = createChange();
+    RevCommit sourceCommit = pushResult.getCommit();
+    String sourceRef = pushResult.getPatchSet().getRefName();
+
+    assertThat(listReplicationTasks("refs/changes/\\d*/\\d*/\\d*")).hasSize(2);
+
+    try (Repository repo1 = repoManager.openRepository(targetProject1);
+        Repository repo2 = repoManager.openRepository(targetProject2)) {
+      waitUntil(
+          () ->
+              (checkedGetRef(repo1, sourceRef) != null && checkedGetRef(repo2, sourceRef) != null));
+
+      Ref targetBranchRef1 = getRef(repo1, sourceRef);
+      assertThat(targetBranchRef1).isNotNull();
+      assertThat(targetBranchRef1.getObjectId()).isEqualTo(sourceCommit.getId());
+
+      Ref targetBranchRef2 = getRef(repo2, sourceRef);
+      assertThat(targetBranchRef2).isNotNull();
+      assertThat(targetBranchRef2.getObjectId()).isEqualTo(sourceCommit.getId());
+    }
+  }
+
+  @Test
+  public void shouldCreateIndividualReplicationTasksForEveryRemoteUrlPair() throws Exception {
+    List<String> replicaSuffixes = Arrays.asList("replica1", "replica2");
+    createTestProject("projectreplica1");
+    createTestProject("projectreplica2");
+
+    setReplicationDestination("foo1", replicaSuffixes, ALL_PROJECTS);
+    setReplicationDestination("foo2", replicaSuffixes, ALL_PROJECTS);
+    config.setInt("remote", "foo1", "replicationDelay", TEST_REPLICATION_DELAY * 100);
+    config.setInt("remote", "foo2", "replicationDelay", TEST_REPLICATION_DELAY * 100);
+    reloadConfig();
+
+    createChange();
+
+    assertThat(listReplicationTasks("refs/changes/\\d*/\\d*/\\d*")).hasSize(4);
+
+    setReplicationDestination("foo1", replicaSuffixes, ALL_PROJECTS);
+    setReplicationDestination("foo2", replicaSuffixes, ALL_PROJECTS);
+  }
+
+  @Test
+  public void shouldCreateOneReplicationTaskWhenSchedulingRepoFullSync() throws Exception {
+    PushResultProcessing pushResultProcessing =
+        new PushResultProcessing() {
+
+          @Override
+          void onRefReplicatedToOneNode(
+              String project,
+              String ref,
+              URIish uri,
+              ReplicationState.RefPushResult status,
+              RemoteRefUpdate.Status refStatus) {}
+
+          @Override
+          void onRefReplicatedToAllNodes(String project, String ref, int nodesCount) {}
+
+          @Override
+          void onAllRefsReplicatedToAllNodes(int totalPushTasksCount) {}
+        };
+
+    createTestProject("projectreplica");
+
+    setReplicationDestination("foo", "replica", ALL_PROJECTS);
+    reloadConfig();
+
+    plugin
+        .getSysInjector()
+        .getInstance(ReplicationQueue.class)
+        .scheduleFullSync(project, null, new ReplicationState(pushResultProcessing), true);
+
+    assertThat(listReplicationTasks(".*all.*")).hasSize(1);
+  }
+
+  @Test
+  public void shouldReplicateHeadUpdate() throws Exception {
+    setReplicationDestination("foo", "replica", ALL_PROJECTS);
+    reloadConfig();
+
+    Project.NameKey targetProject = createTestProject("projectreplica");
+    String newHead = "refs/heads/newhead";
+    String master = "refs/heads/master";
+    BranchInput input = new BranchInput();
+    input.revision = master;
+    gApi.projects().name(project.get()).branch(newHead).create(input);
+    gApi.projects().name(project.get()).head(newHead);
+
+    try (Repository repo = repoManager.openRepository(targetProject)) {
+      waitUntil(() -> checkedGetRef(repo, newHead) != null);
+
+      Ref targetProjectHead = getRef(repo, Constants.HEAD);
+      assertThat(targetProjectHead).isNotNull();
+      assertThat(targetProjectHead.getTarget().getName()).isEqualTo(newHead);
+    }
+  }
+
+  @Test
+  public void shouldNotDrainTheQueueWhenReloading() throws Exception {
+    // Setup repo to replicate
+    Project.NameKey targetProject = createTestProject("projectreplica");
+    String remoteName = "doNotDrainQueue";
+    setReplicationDestination(remoteName, "replica", ALL_PROJECTS);
+
+    Result pushResult = createChange();
+    shutdownConfig();
+
+    pushResult.getCommit();
+    String sourceRef = pushResult.getPatchSet().getRefName();
+
+    assertThrows(
+        InterruptedException.class,
+        () -> {
+          try (Repository repo = repoManager.openRepository(targetProject)) {
+            waitUntil(() -> checkedGetRef(repo, sourceRef) != null);
+          }
+        });
+  }
+
+  @Test
+  public void shouldDrainTheQueueWhenReloading() throws Exception {
+    // Setup repo to replicate
+    Project.NameKey targetProject = createTestProject("projectreplica");
+    String remoteName = "drainQueue";
+    setReplicationDestination(remoteName, "replica", ALL_PROJECTS);
+
+    config.setInt("remote", remoteName, "drainQueueAttempts", 2);
+    config.save();
+    reloadConfig();
+
+    Result pushResult = createChange();
+    shutdownConfig();
+
+    RevCommit sourceCommit = pushResult.getCommit();
+    String sourceRef = pushResult.getPatchSet().getRefName();
+
+    try (Repository repo = repoManager.openRepository(targetProject)) {
+      waitUntil(() -> checkedGetRef(repo, sourceRef) != null);
+      Ref targetBranchRef = getRef(repo, sourceRef);
+      assertThat(targetBranchRef).isNotNull();
+      assertThat(targetBranchRef.getObjectId()).isEqualTo(sourceCommit.getId());
+    }
+  }
+
+  private Project.NameKey createTestProject(String name) throws Exception {
+    return createProject(name);
+  }
+
+  private Ref getRef(Repository repo, String branchName) throws IOException {
+    return repo.getRefDatabase().exactRef(branchName);
+  }
+
+  private Ref checkedGetRef(Repository repo, String branchName) {
+    try {
+      return repo.getRefDatabase().exactRef(branchName);
+    } catch (Exception e) {
+      logger.atSevere().withCause(e).log("failed to get ref %s in repo %s", branchName, repo);
+      return null;
+    }
+  }
+
+  private void setReplicationDestination(
+      String remoteName, String replicaSuffix, Optional<String> project) throws IOException {
+    setReplicationDestination(remoteName, Arrays.asList(replicaSuffix), project);
+  }
+
+  private void setReplicationDestination(
+      String remoteName, List<String> replicaSuffixes, Optional<String> project)
+      throws IOException {
+
+    List<String> replicaUrls =
+        replicaSuffixes.stream()
+            .map(suffix -> gitPath.resolve("${name}" + suffix + ".git").toString())
+            .collect(toList());
+    config.setStringList("remote", remoteName, "url", replicaUrls);
+    config.setInt("remote", remoteName, "replicationDelay", TEST_REPLICATION_DELAY);
+    project.ifPresent(prj -> config.setString("remote", remoteName, "projects", prj));
+    config.save();
+  }
+
+  private void waitUntil(Supplier<Boolean> waitCondition) throws InterruptedException {
+    WaitUtil.waitUntil(waitCondition, TEST_TIMEOUT);
+  }
+
+  private void reloadConfig() {
+    plugin.getSysInjector().getInstance(AutoReloadConfigDecorator.class).forceReload();
+  }
+
+  private void shutdownConfig() {
+    plugin.getSysInjector().getInstance(AutoReloadConfigDecorator.class).shutdown();
+  }
+
+  private List<ReplicateRefUpdate> listReplicationTasks(String refRegex) {
+    Pattern refmaskPattern = Pattern.compile(refRegex);
+    return tasksStorage.list().stream()
+        .filter(task -> refmaskPattern.matcher(task.ref).matches())
+        .collect(toList());
+  }
+
+  private void cleanupReplicationTasks() throws IOException {
+    try (DirectoryStream<Path> files = Files.newDirectoryStream(storagePath)) {
+      for (Path path : files) {
+        path.toFile().delete();
+      }
+    }
+  }
+
+  private boolean projectExists(Project.NameKey name) {
+    try (Repository r = repoManager.openRepository(name)) {
+      return true;
+    } catch (Exception e) {
+      return false;
+    }
+  }
+}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationQueueIT.java b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationQueueIT.java
index 881a282..2a395ca 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationQueueIT.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationQueueIT.java
@@ -56,13 +56,13 @@
   private FileBasedConfig config;
 
   @Override
-  public void setUp() throws Exception {
+  public void setUpTestPlugin() throws Exception {
     gitPath = sitePaths.site_path.resolve("git");
     config =
         new FileBasedConfig(sitePaths.etc_dir.resolve("replication.config").toFile(), FS.DETECTED);
 
     setReplicationDestination("foo", "replica");
-    super.setUp();
+    super.setUpTestPlugin();
   }
 
   @Test
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationStateTest.java b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationStateTest.java
index cf6715e..193af1e 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationStateTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/ReplicationStateTest.java
@@ -32,15 +32,12 @@
 
   private ReplicationState replicationState;
   private PushResultProcessing pushResultProcessingMock;
-  private EventsStorage eventsStorage;
 
   @Before
   public void setUp() throws Exception {
     pushResultProcessingMock = createNiceMock(PushResultProcessing.class);
     replay(pushResultProcessingMock);
-    eventsStorage = createNiceMock(EventsStorage.class);
-    replay(eventsStorage);
-    replicationState = new ReplicationState(eventsStorage, pushResultProcessingMock);
+    replicationState = new ReplicationState(pushResultProcessingMock);
   }
 
   @Test
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/WaitUtil.java b/src/test/java/com/googlesource/gerrit/plugins/replication/WaitUtil.java
new file mode 100644
index 0000000..586b56c
--- /dev/null
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/WaitUtil.java
@@ -0,0 +1,34 @@
+// Copyright (C) 2019 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 java.util.concurrent.TimeUnit.MILLISECONDS;
+
+import com.google.common.base.Stopwatch;
+import java.time.Duration;
+import java.util.function.Supplier;
+
+public class WaitUtil {
+  public 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(50);
+    }
+  }
+}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/WaitUtilTest.java b/src/test/java/com/googlesource/gerrit/plugins/replication/WaitUtilTest.java
new file mode 100644
index 0000000..0ccb0af
--- /dev/null
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/WaitUtilTest.java
@@ -0,0 +1,40 @@
+// Copyright (C) 2019 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.gerrit.testing.GerritJUnit.assertThrows;
+import static com.googlesource.gerrit.plugins.replication.WaitUtil.waitUntil;
+
+import java.time.Duration;
+import org.junit.Test;
+
+public class WaitUtilTest {
+
+  @Test
+  public void shouldFailWhenConditionNotMetWithinTimeout() throws Exception {
+    assertThrows(
+        InterruptedException.class,
+        () -> waitUntil(() -> returnTrue() == false, Duration.ofSeconds(1)));
+  }
+
+  @Test
+  public void shouldNotFailWhenConditionIsMetWithinTimeout() throws Exception {
+    waitUntil(() -> returnTrue() == true, Duration.ofSeconds(1));
+  }
+
+  private static boolean returnTrue() {
+    return true;
+  }
+}