diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/FetchOne.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/FetchOne.java
index cdf573f..4fad8f9 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/FetchOne.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/FetchOne.java
@@ -66,7 +66,7 @@
  */
 public class FetchOne implements ProjectRunnable, CanceledWhileRunning {
   private final ReplicationStateListener stateLog;
-  static final String ALL_REFS = "..all..";
+  public static final String ALL_REFS = "..all..";
   static final String ID_KEY = "fetchOneId";
 
   interface Factory {
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationModule.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationModule.java
index c17d5df..3e51e03 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationModule.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationModule.java
@@ -42,18 +42,21 @@
 import com.googlesource.gerrit.plugins.replication.ReplicationConfig;
 import com.googlesource.gerrit.plugins.replication.ReplicationFileBasedConfig;
 import com.googlesource.gerrit.plugins.replication.StartReplicationCapability;
+import com.googlesource.gerrit.plugins.replication.pull.api.FetchJob;
 import com.googlesource.gerrit.plugins.replication.pull.api.PullReplicationApiModule;
 import com.googlesource.gerrit.plugins.replication.pull.client.FetchApiClient;
 import com.googlesource.gerrit.plugins.replication.pull.client.FetchRestApiClient;
 import com.googlesource.gerrit.plugins.replication.pull.client.HttpClient;
 import com.googlesource.gerrit.plugins.replication.pull.client.SourceHttpClient;
 import com.googlesource.gerrit.plugins.replication.pull.event.FetchRefReplicatedEventModule;
+import com.googlesource.gerrit.plugins.replication.pull.event.StreamEventModule;
 import com.googlesource.gerrit.plugins.replication.pull.fetch.ApplyObject;
 import java.io.File;
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import org.eclipse.jgit.errors.ConfigInvalidException;
+import org.eclipse.jgit.lib.Config;
 import org.eclipse.jgit.storage.file.FileBasedConfig;
 import org.eclipse.jgit.util.FS;
 
@@ -72,7 +75,7 @@
 
     bind(RevisionReader.class).in(Scopes.SINGLETON);
     bind(ApplyObject.class);
-
+    install(new FactoryModuleBuilder().build(FetchJob.Factory.class));
     install(new PullReplicationApiModule());
 
     install(new FetchRefReplicatedEventModule());
@@ -120,7 +123,8 @@
 
     bind(ConfigParser.class).to(SourceConfigParser.class).in(Scopes.SINGLETON);
 
-    if (getReplicationConfig().getBoolean("gerrit", "autoReload", false)) {
+    Config replicationConfig = getReplicationConfig();
+    if (replicationConfig.getBoolean("gerrit", "autoReload", false)) {
       bind(ReplicationConfig.class)
           .annotatedWith(MainReplicationConfig.class)
           .to(getReplicationConfigClass());
@@ -132,6 +136,10 @@
       bind(ReplicationConfig.class).to(getReplicationConfigClass()).in(Scopes.SINGLETON);
     }
 
+    if (replicationConfig.getBoolean("replication", "consumeStreamEvents", false)) {
+      install(new StreamEventModule());
+    }
+
     DynamicSet.setOf(binder(), ReplicationStateListener.class);
     DynamicSet.bind(binder(), ReplicationStateListener.class).to(PullReplicationStateLogger.class);
     EventTypes.register(FetchRefReplicatedEvent.TYPE, FetchRefReplicatedEvent.class);
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchAction.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchAction.java
index 48964db..fdb4f8f 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchAction.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchAction.java
@@ -17,7 +17,6 @@
 import static com.google.common.base.Preconditions.checkState;
 
 import com.google.common.base.Strings;
-import com.google.common.flogger.FluentLogger;
 import com.google.gerrit.entities.Project;
 import com.google.gerrit.extensions.registration.DynamicItem;
 import com.google.gerrit.extensions.restapi.AuthException;
@@ -32,6 +31,7 @@
 import com.google.gerrit.server.project.ProjectResource;
 import com.google.inject.Inject;
 import com.googlesource.gerrit.plugins.replication.pull.api.FetchAction.Input;
+import com.googlesource.gerrit.plugins.replication.pull.api.FetchJob.Factory;
 import com.googlesource.gerrit.plugins.replication.pull.api.exception.RemoteConfigurationMissingException;
 import java.util.Optional;
 import java.util.concurrent.ExecutionException;
@@ -42,17 +42,20 @@
   private final WorkQueue workQueue;
   private final DynamicItem<UrlFormatter> urlFormatter;
   private final FetchPreconditions preConditions;
+  private final Factory fetchJobFactory;
 
   @Inject
   public FetchAction(
       FetchCommand command,
       WorkQueue workQueue,
       DynamicItem<UrlFormatter> urlFormatter,
-      FetchPreconditions preConditions) {
+      FetchPreconditions preConditions,
+      FetchJob.Factory fetchJobFactory) {
     this.command = command;
     this.workQueue = workQueue;
     this.urlFormatter = urlFormatter;
     this.preConditions = preConditions;
+    this.fetchJobFactory = fetchJobFactory;
   }
 
   public static class Input {
@@ -104,7 +107,7 @@
             workQueue
                 .getDefaultQueue()
                 .submit(
-                    new FetchJob(command, project, input, PullReplicationApiRequestMetrics.get()));
+                    fetchJobFactory.create(project, input, PullReplicationApiRequestMetrics.get()));
     Optional<String> url =
         urlFormatter
             .get()
@@ -113,38 +116,4 @@
     checkState(url.isPresent());
     return Response.accepted(url.get());
   }
-
-  private static class FetchJob implements Runnable {
-    private static final FluentLogger log = FluentLogger.forEnclosingClass();
-
-    private FetchCommand command;
-    private Project.NameKey project;
-    private FetchAction.Input input;
-    private final PullReplicationApiRequestMetrics apiRequestMetrics;
-
-    public FetchJob(
-        FetchCommand command,
-        Project.NameKey project,
-        FetchAction.Input input,
-        PullReplicationApiRequestMetrics apiRequestMetrics) {
-      this.command = command;
-      this.project = project;
-      this.input = input;
-      this.apiRequestMetrics = apiRequestMetrics;
-    }
-
-    @Override
-    public void run() {
-      try {
-        command.fetchAsync(project, input.label, input.refName, apiRequestMetrics);
-      } catch (InterruptedException
-          | ExecutionException
-          | RemoteConfigurationMissingException
-          | TimeoutException e) {
-        log.atSevere().withCause(e).log(
-            "Exception during the async fetch call for project %s, label %s and ref name %s",
-            project.get(), input.label, input.refName);
-      }
-    }
-  }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchJob.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchJob.java
new file mode 100644
index 0000000..e15dd68
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchJob.java
@@ -0,0 +1,63 @@
+// Copyright (C) 2022 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.googlesource.gerrit.plugins.replication.pull.api;
+
+import com.google.common.flogger.FluentLogger;
+import com.google.gerrit.entities.Project;
+import com.google.inject.Inject;
+import com.google.inject.assistedinject.Assisted;
+import com.googlesource.gerrit.plugins.replication.pull.api.exception.RemoteConfigurationMissingException;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeoutException;
+
+public class FetchJob implements Runnable {
+  private static final FluentLogger log = FluentLogger.forEnclosingClass();
+
+  public interface Factory {
+    FetchJob create(
+        Project.NameKey project, FetchAction.Input input, PullReplicationApiRequestMetrics metrics);
+  }
+
+  private FetchCommand command;
+  private Project.NameKey project;
+  private FetchAction.Input input;
+  private final PullReplicationApiRequestMetrics metrics;
+
+  @Inject
+  public FetchJob(
+      FetchCommand command,
+      @Assisted Project.NameKey project,
+      @Assisted FetchAction.Input input,
+      @Assisted PullReplicationApiRequestMetrics metrics) {
+    this.command = command;
+    this.project = project;
+    this.input = input;
+    this.metrics = metrics;
+  }
+
+  @Override
+  public void run() {
+    try {
+      command.fetchAsync(project, input.label, input.refName, metrics);
+    } catch (InterruptedException
+        | ExecutionException
+        | RemoteConfigurationMissingException
+        | TimeoutException e) {
+      log.atSevere().withCause(e).log(
+          "Exception during the async fetch call for project %s, label %s and ref name %s",
+          project.get(), input.label, input.refName);
+    }
+  }
+}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchPreconditions.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchPreconditions.java
index ca1557a..161bcf4 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchPreconditions.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchPreconditions.java
@@ -40,8 +40,10 @@
   }
 
   public Boolean canCallFetchApi() {
-    PermissionBackend.WithUser userPermission = permissionBackend.user(userProvider.get());
-    return userPermission.testOrFalse(GlobalPermission.ADMINISTRATE_SERVER)
+    CurrentUser currentUser = userProvider.get();
+    PermissionBackend.WithUser userPermission = permissionBackend.user(currentUser);
+    return currentUser.isInternalUser()
+        || userPermission.testOrFalse(GlobalPermission.ADMINISTRATE_SERVER)
         || userPermission.testOrFalse(new PluginPermission(pluginName, CALL_FETCH_ACTION));
   }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/HttpModule.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/HttpModule.java
index b140cb4..83e8487 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/HttpModule.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/HttpModule.java
@@ -31,6 +31,10 @@
 
   @Override
   protected void configureServlets() {
+    DynamicSet.bind(binder(), AllRequestFilter.class)
+        .to(PullReplicationApiMetricsFilter.class)
+        .in(Scopes.SINGLETON);
+
     if (isReplica) {
       DynamicSet.bind(binder(), AllRequestFilter.class)
           .to(PullReplicationFilter.class)
@@ -38,9 +42,5 @@
     } else {
       serveRegex("/init-project/.*$").with(ProjectInitializationAction.class);
     }
-
-    DynamicSet.bind(binder(), AllRequestFilter.class)
-        .to(PullReplicationApiMetricsFilter.class)
-        .in(Scopes.SINGLETON);
   }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/ProjectDeletionAction.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/ProjectDeletionAction.java
index 8915e78..2e1c5d4 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/ProjectDeletionAction.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/ProjectDeletionAction.java
@@ -22,9 +22,11 @@
 import com.google.gerrit.extensions.restapi.Response;
 import com.google.gerrit.extensions.restapi.RestModifyView;
 import com.google.gerrit.extensions.restapi.UnprocessableEntityException;
+import com.google.gerrit.server.CurrentUser;
 import com.google.gerrit.server.permissions.PermissionBackend;
 import com.google.gerrit.server.project.ProjectResource;
 import com.google.inject.Inject;
+import com.google.inject.Provider;
 import com.googlesource.gerrit.plugins.replication.LocalFS;
 import com.googlesource.gerrit.plugins.replication.pull.GerritConfigOps;
 import java.util.Optional;
@@ -37,20 +39,29 @@
 
   static class DeleteInput {}
 
+  private final Provider<CurrentUser> userProvider;
   private final GerritConfigOps gerritConfigOps;
   private final PermissionBackend permissionBackend;
 
   @Inject
-  ProjectDeletionAction(GerritConfigOps gerritConfigOps, PermissionBackend permissionBackend) {
+  ProjectDeletionAction(
+      GerritConfigOps gerritConfigOps,
+      PermissionBackend permissionBackend,
+      Provider<CurrentUser> userProvider) {
     this.gerritConfigOps = gerritConfigOps;
     this.permissionBackend = permissionBackend;
+    this.userProvider = userProvider;
   }
 
   @Override
   public Response<?> apply(ProjectResource projectResource, DeleteInput input)
       throws AuthException, BadRequestException, ResourceConflictException, Exception {
 
-    permissionBackend.user(projectResource.getUser()).check(DELETE_PROJECT);
+    // When triggered internally(for example by consuming stream events) user is not provided
+    // and internal user is returned. Project deletion should be always allowed for internal user.
+    if (!userProvider.get().isInternalUser()) {
+      permissionBackend.user(projectResource.getUser()).check(DELETE_PROJECT);
+    }
 
     Optional<URIish> maybeRepoURI =
         gerritConfigOps.getGitRepositoryURI(String.format("%s.git", projectResource.getName()));
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/ProjectInitializationAction.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/ProjectInitializationAction.java
index 3426c03..c1174c9 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/ProjectInitializationAction.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/ProjectInitializationAction.java
@@ -14,7 +14,6 @@
 
 package com.googlesource.gerrit.plugins.replication.pull.api;
 
-import static com.googlesource.gerrit.plugins.replication.pull.api.FetchApiCapability.CALL_FETCH_ACTION;
 import static com.googlesource.gerrit.plugins.replication.pull.api.HttpServletOps.checkAcceptHeader;
 import static com.googlesource.gerrit.plugins.replication.pull.api.HttpServletOps.setResponse;
 
@@ -70,17 +69,8 @@
       return;
     }
 
-    if (!userProvider.get().isIdentifiedUser()) {
-      setResponse(
-          httpServletResponse,
-          HttpServletResponse.SC_UNAUTHORIZED,
-          "Unauthorized user. '" + CALL_FETCH_ACTION + "' capability needed.");
-      return;
-    }
-
     String path = httpServletRequest.getRequestURI();
     String projectName = Url.decode(path.substring(path.lastIndexOf('/') + 1));
-
     try {
       if (initProject(projectName)) {
         setResponse(
@@ -103,10 +93,12 @@
         "Cannot initialize project " + projectName);
   }
 
-  protected boolean initProject(String projectName)
-      throws AuthException, PermissionBackendException {
-    permissionBackend.user(userProvider.get()).check(GlobalPermission.CREATE_PROJECT);
-
+  public boolean initProject(String projectName) throws AuthException, PermissionBackendException {
+    // When triggered internally(for example by consuming stream events) user is not provided
+    // and internal user is returned. Project creation should be always allowed for internal user.
+    if (!userProvider.get().isInternalUser()) {
+      permissionBackend.user(userProvider.get()).check(GlobalPermission.CREATE_PROJECT);
+    }
     Optional<URIish> maybeUri = gerritConfigOps.getGitRepositoryURI(projectName);
     if (!maybeUri.isPresent()) {
       logger.atSevere().log("Cannot initialize project '%s'", projectName);
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/PullReplicationApiRequestMetrics.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/PullReplicationApiRequestMetrics.java
index 597b66f..8e4e43f 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/PullReplicationApiRequestMetrics.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/PullReplicationApiRequestMetrics.java
@@ -16,6 +16,7 @@
 
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 
+import com.google.gerrit.server.events.Event;
 import com.google.inject.Inject;
 import com.googlesource.gerrit.plugins.replication.pull.FetchReplicationMetrics;
 import java.util.Optional;
@@ -28,7 +29,7 @@
 
   public static final String HTTP_HEADER_X_START_TIME_NANOS = "X-StartTimeNanos";
 
-  private Optional<Long> startTimeNanos;
+  private Optional<Long> startTimeNanos = Optional.empty();
   private final AtomicBoolean initialised = new AtomicBoolean();
   private final FetchReplicationMetrics metrics;
 
@@ -59,6 +60,13 @@
             .map(nanoTime -> Math.min(currentTimeNanos(), nanoTime));
   }
 
+  public void start(Event event) {
+    if (!initialised.compareAndSet(false, true)) {
+      throw new IllegalStateException("PullReplicationApiRequestMetrics already initialised");
+    }
+    startTimeNanos = Optional.of(event.eventCreatedOn * 1000 * 1000 * 1000);
+  }
+
   public Optional<Long> stop(String replicationSourceName) {
     return startTimeNanos.map(
         start -> {
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/PullReplicationFilter.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/PullReplicationFilter.java
index 0a9b266..8c47801 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/PullReplicationFilter.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/api/PullReplicationFilter.java
@@ -23,9 +23,7 @@
 import static javax.servlet.http.HttpServletResponse.SC_FORBIDDEN;
 import static javax.servlet.http.HttpServletResponse.SC_INTERNAL_SERVER_ERROR;
 import static javax.servlet.http.HttpServletResponse.SC_OK;
-import static javax.servlet.http.HttpServletResponse.SC_UNAUTHORIZED;
 
-import com.google.common.base.Splitter;
 import com.google.common.flogger.FluentLogger;
 import com.google.gerrit.extensions.annotations.PluginName;
 import com.google.gerrit.extensions.api.projects.HeadInput;
@@ -38,11 +36,9 @@
 import com.google.gerrit.extensions.restapi.RestApiException;
 import com.google.gerrit.extensions.restapi.TopLevelResource;
 import com.google.gerrit.extensions.restapi.UnprocessableEntityException;
-import com.google.gerrit.extensions.restapi.Url;
 import com.google.gerrit.httpd.AllRequestFilter;
 import com.google.gerrit.httpd.restapi.RestApiServlet;
 import com.google.gerrit.json.OutputFormat;
-import com.google.gerrit.server.CurrentUser;
 import com.google.gerrit.server.permissions.PermissionBackendException;
 import com.google.gerrit.server.project.ProjectResource;
 import com.google.gerrit.server.restapi.project.ProjectsCollection;
@@ -51,7 +47,6 @@
 import com.google.gson.stream.JsonReader;
 import com.google.gson.stream.MalformedJsonException;
 import com.google.inject.Inject;
-import com.google.inject.Provider;
 import com.google.inject.TypeLiteral;
 import com.googlesource.gerrit.plugins.replication.pull.api.FetchAction.Input;
 import com.googlesource.gerrit.plugins.replication.pull.api.data.RevisionInput;
@@ -61,9 +56,11 @@
 import java.io.EOFException;
 import java.io.IOException;
 import java.io.PrintWriter;
-import java.util.ArrayList;
-import java.util.List;
 import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Optional;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 import javax.servlet.FilterChain;
 import javax.servlet.ServletException;
 import javax.servlet.ServletRequest;
@@ -74,6 +71,10 @@
 public class PullReplicationFilter extends AllRequestFilter {
   private static final FluentLogger logger = FluentLogger.forEnclosingClass();
 
+  private static final Pattern projectNameInGerritUrl = Pattern.compile(".*/projects/([^/]+)/.*");
+  private static final Pattern projectNameInitProjectUrl =
+      Pattern.compile(".*/init-project/([^/]+.git)");
+
   private FetchAction fetchAction;
   private ApplyObjectAction applyObjectAction;
   private ApplyObjectsAction applyObjectsAction;
@@ -82,7 +83,6 @@
   private ProjectDeletionAction projectDeletionAction;
   private ProjectsCollection projectsCollection;
   private Gson gson;
-  private Provider<CurrentUser> userProvider;
   private String pluginName;
 
   @Inject
@@ -94,7 +94,6 @@
       UpdateHeadAction updateHEADAction,
       ProjectDeletionAction projectDeletionAction,
       ProjectsCollection projectsCollection,
-      Provider<CurrentUser> userProvider,
       @PluginName String pluginName) {
     this.fetchAction = fetchAction;
     this.applyObjectAction = applyObjectAction;
@@ -103,7 +102,6 @@
     this.updateHEADAction = updateHEADAction;
     this.projectDeletionAction = projectDeletionAction;
     this.projectsCollection = projectsCollection;
-    this.userProvider = userProvider;
     this.pluginName = pluginName;
     this.gson = OutputFormat.JSON.newGsonBuilder().create();
   }
@@ -120,44 +118,20 @@
     HttpServletRequest httpRequest = (HttpServletRequest) request;
     try {
       if (isFetchAction(httpRequest)) {
-        if (userProvider.get().isIdentifiedUser()) {
-          writeResponse(httpResponse, doFetch(httpRequest));
-        } else {
-          httpResponse.sendError(SC_UNAUTHORIZED);
-        }
+        writeResponse(httpResponse, doFetch(httpRequest));
       } else if (isApplyObjectAction(httpRequest)) {
-        if (userProvider.get().isIdentifiedUser()) {
-          writeResponse(httpResponse, doApplyObject(httpRequest));
-        } else {
-          httpResponse.sendError(SC_UNAUTHORIZED);
-        }
+        writeResponse(httpResponse, doApplyObject(httpRequest));
       } else if (isApplyObjectsAction(httpRequest)) {
-        if (userProvider.get().isIdentifiedUser()) {
-          writeResponse(httpResponse, doApplyObjects(httpRequest));
-        } else {
-          httpResponse.sendError(SC_UNAUTHORIZED);
-        }
+        writeResponse(httpResponse, doApplyObjects(httpRequest));
       } else if (isInitProjectAction(httpRequest)) {
-        if (userProvider.get().isIdentifiedUser()) {
-          if (!checkAcceptHeader(httpRequest, httpResponse)) {
-            return;
-          }
-          doInitProject(httpRequest, httpResponse);
-        } else {
-          httpResponse.sendError(SC_UNAUTHORIZED);
+        if (!checkAcceptHeader(httpRequest, httpResponse)) {
+          return;
         }
+        doInitProject(httpRequest, httpResponse);
       } else if (isUpdateHEADAction(httpRequest)) {
-        if (userProvider.get().isIdentifiedUser()) {
-          writeResponse(httpResponse, doUpdateHEAD(httpRequest));
-        } else {
-          httpResponse.sendError(SC_UNAUTHORIZED);
-        }
+        writeResponse(httpResponse, doUpdateHEAD(httpRequest));
       } else if (isDeleteProjectAction(httpRequest)) {
-        if (userProvider.get().isIdentifiedUser()) {
-          writeResponse(httpResponse, doDeleteProject(httpRequest));
-        } else {
-          httpResponse.sendError(SC_UNAUTHORIZED);
-        }
+        writeResponse(httpResponse, doDeleteProject(httpRequest));
       } else {
         chain.doFilter(request, response);
       }
@@ -180,6 +154,9 @@
     } catch (InitProjectException | ResourceNotFoundException e) {
       RestApiServlet.replyError(
           httpRequest, httpResponse, SC_INTERNAL_SERVER_ERROR, e.getMessage(), e.caching(), e);
+    } catch (NoSuchElementException e) {
+      RestApiServlet.replyError(
+          httpRequest, httpResponse, SC_BAD_REQUEST, "Project name not present in the url", e);
     } catch (Exception e) {
       throw new ServletException(e);
     }
@@ -188,8 +165,8 @@
   private void doInitProject(HttpServletRequest httpRequest, HttpServletResponse httpResponse)
       throws RestApiException, IOException, PermissionBackendException {
 
-    String path = httpRequest.getRequestURI();
-    String projectName = Url.decode(path.substring(path.lastIndexOf('/') + 1));
+    IdString id = getInitProjectName(httpRequest).get();
+    String projectName = id.get();
     if (projectInitializationAction.initProject(projectName)) {
       setResponse(
           httpResponse, HttpServletResponse.SC_CREATED, "Project " + projectName + " initialized");
@@ -202,7 +179,7 @@
   private Response<Map<String, Object>> doApplyObject(HttpServletRequest httpRequest)
       throws RestApiException, IOException, PermissionBackendException {
     RevisionInput input = readJson(httpRequest, TypeLiteral.get(RevisionInput.class));
-    IdString id = getProjectName(httpRequest);
+    IdString id = getProjectName(httpRequest).get();
     ProjectResource projectResource = projectsCollection.parse(TopLevelResource.INSTANCE, id);
 
     return (Response<Map<String, Object>>) applyObjectAction.apply(projectResource, input);
@@ -212,7 +189,7 @@
   private Response<Map<String, Object>> doApplyObjects(HttpServletRequest httpRequest)
       throws RestApiException, IOException, PermissionBackendException {
     RevisionsInput input = readJson(httpRequest, TypeLiteral.get(RevisionsInput.class));
-    IdString id = getProjectName(httpRequest);
+    IdString id = getProjectName(httpRequest).get();
     ProjectResource projectResource = projectsCollection.parse(TopLevelResource.INSTANCE, id);
 
     return (Response<Map<String, Object>>) applyObjectsAction.apply(projectResource, input);
@@ -221,16 +198,16 @@
   @SuppressWarnings("unchecked")
   private Response<String> doUpdateHEAD(HttpServletRequest httpRequest) throws Exception {
     HeadInput input = readJson(httpRequest, TypeLiteral.get(HeadInput.class));
-    ProjectResource projectResource =
-        projectsCollection.parse(TopLevelResource.INSTANCE, getProjectName(httpRequest));
+    IdString id = getProjectName(httpRequest).get();
+    ProjectResource projectResource = projectsCollection.parse(TopLevelResource.INSTANCE, id);
 
     return (Response<String>) updateHEADAction.apply(projectResource, input);
   }
 
   @SuppressWarnings("unchecked")
   private Response<String> doDeleteProject(HttpServletRequest httpRequest) throws Exception {
-    ProjectResource projectResource =
-        projectsCollection.parse(TopLevelResource.INSTANCE, getProjectName(httpRequest));
+    IdString id = getProjectName(httpRequest).get();
+    ProjectResource projectResource = projectsCollection.parse(TopLevelResource.INSTANCE, id);
     return (Response<String>)
         projectDeletionAction.apply(projectResource, new ProjectDeletionAction.DeleteInput());
   }
@@ -239,7 +216,7 @@
   private Response<Map<String, Object>> doFetch(HttpServletRequest httpRequest)
       throws IOException, RestApiException, PermissionBackendException {
     Input input = readJson(httpRequest, TypeLiteral.get(Input.class));
-    IdString id = getProjectName(httpRequest);
+    IdString id = getProjectName(httpRequest).get();
     ProjectResource projectResource = projectsCollection.parse(TopLevelResource.INSTANCE, id);
 
     return (Response<Map<String, Object>>) fetchAction.apply(projectResource, input);
@@ -296,42 +273,48 @@
    * @param req
    * @return project name
    */
-  private IdString getProjectName(HttpServletRequest req) {
-    String path = req.getRequestURI();
+  private Optional<IdString> getInitProjectName(HttpServletRequest req) {
+    return extractProjectName(req, projectNameInitProjectUrl);
+  }
 
-    List<IdString> out = new ArrayList<>();
-    for (String p : Splitter.on('/').split(path)) {
-      out.add(IdString.fromUrl(p));
+  private Optional<IdString> getProjectName(HttpServletRequest req) {
+    return extractProjectName(req, projectNameInGerritUrl);
+  }
+
+  private Optional<IdString> extractProjectName(HttpServletRequest req, Pattern urlPattern) {
+    String path = req.getRequestURI();
+    Matcher projectGroupMatcher = urlPattern.matcher(path);
+
+    if (projectGroupMatcher.find()) {
+      return Optional.of(IdString.fromUrl(projectGroupMatcher.group(1)));
     }
-    if (!out.isEmpty() && out.get(out.size() - 1).isEmpty()) {
-      out.remove(out.size() - 1);
-    }
-    return out.get(3);
+
+    return Optional.empty();
   }
 
   private boolean isApplyObjectAction(HttpServletRequest httpRequest) {
-    return httpRequest.getRequestURI().endsWith("pull-replication~apply-object");
+    return httpRequest.getRequestURI().endsWith(String.format("/%s~apply-object", pluginName));
   }
 
   private boolean isApplyObjectsAction(HttpServletRequest httpRequest) {
-    return httpRequest.getRequestURI().endsWith("pull-replication~apply-objects");
+    return httpRequest.getRequestURI().endsWith(String.format("/%s~apply-objects", pluginName));
   }
 
   private boolean isFetchAction(HttpServletRequest httpRequest) {
-    return httpRequest.getRequestURI().endsWith("pull-replication~fetch");
+    return httpRequest.getRequestURI().endsWith(String.format("/%s~fetch", pluginName));
   }
 
   private boolean isInitProjectAction(HttpServletRequest httpRequest) {
-    return httpRequest.getRequestURI().contains("pull-replication/init-project/");
+    return httpRequest.getRequestURI().contains(String.format("/%s/init-project/", pluginName));
   }
 
   private boolean isUpdateHEADAction(HttpServletRequest httpRequest) {
-    return httpRequest.getRequestURI().matches("(/a)?/projects/[^/]+/HEAD")
+    return httpRequest.getRequestURI().matches(".*/projects/[^/]+/HEAD")
         && "PUT".equals(httpRequest.getMethod());
   }
 
   private boolean isDeleteProjectAction(HttpServletRequest httpRequest) {
-    return httpRequest.getRequestURI().endsWith(String.format("%s~delete-project", pluginName))
+    return httpRequest.getRequestURI().endsWith(String.format("/%s~delete-project", pluginName))
         && "DELETE".equals(httpRequest.getMethod());
   }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/client/FetchRestApiClient.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/client/FetchRestApiClient.java
index ed919de..09139f0 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/client/FetchRestApiClient.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/client/FetchRestApiClient.java
@@ -26,6 +26,7 @@
 import com.google.gerrit.entities.Project.NameKey;
 import com.google.gerrit.extensions.annotations.PluginName;
 import com.google.gerrit.extensions.restapi.Url;
+import com.google.gerrit.server.config.GerritInstanceId;
 import com.google.gson.Gson;
 import com.google.gson.GsonBuilder;
 import com.google.inject.Inject;
@@ -44,17 +45,16 @@
 import java.util.Optional;
 import org.apache.http.HttpResponse;
 import org.apache.http.ParseException;
-import org.apache.http.auth.AuthScope;
+import org.apache.http.auth.AuthenticationException;
 import org.apache.http.auth.UsernamePasswordCredentials;
 import org.apache.http.client.ClientProtocolException;
-import org.apache.http.client.CredentialsProvider;
 import org.apache.http.client.ResponseHandler;
 import org.apache.http.client.methods.HttpDelete;
 import org.apache.http.client.methods.HttpPost;
 import org.apache.http.client.methods.HttpPut;
-import org.apache.http.client.protocol.HttpClientContext;
+import org.apache.http.client.methods.HttpRequestBase;
 import org.apache.http.entity.StringEntity;
-import org.apache.http.impl.client.BasicCredentialsProvider;
+import org.apache.http.impl.auth.BasicScheme;
 import org.apache.http.message.BasicHeader;
 import org.apache.http.util.EntityUtils;
 import org.eclipse.jgit.transport.CredentialItem;
@@ -70,7 +70,7 @@
   private final CredentialsFactory credentials;
   private final SourceHttpClient.Factory httpClientFactory;
   private final Source source;
-  private final String instanceLabel;
+  private final String instanceId;
   private final String pluginName;
   private final SyncRefsFilter syncRefsFilter;
 
@@ -81,18 +81,22 @@
       ReplicationConfig replicationConfig,
       SyncRefsFilter syncRefsFilter,
       @PluginName String pluginName,
+      @Nullable @GerritInstanceId String instanceId,
       @Assisted Source source) {
     this.credentials = credentials;
     this.httpClientFactory = httpClientFactory;
     this.source = source;
     this.pluginName = pluginName;
     this.syncRefsFilter = syncRefsFilter;
-    this.instanceLabel =
-        Strings.nullToEmpty(
+    this.instanceId =
+        Optional.ofNullable(
                 replicationConfig.getConfig().getString("replication", null, "instanceLabel"))
+            .orElse(instanceId)
             .trim();
+
     requireNonNull(
-        Strings.emptyToNull(instanceLabel), "replication.instanceLabel cannot be null or empty");
+        Strings.emptyToNull(this.instanceId),
+        "gerrit.instanceId or replication.instanceLabel must be set");
   }
 
   /* (non-Javadoc)
@@ -112,13 +116,13 @@
         new StringEntity(
             String.format(
                 "{\"label\":\"%s\", \"ref_name\": \"%s\", \"async\":%s}",
-                instanceLabel, refName, callAsync),
+                instanceId, refName, callAsync),
             StandardCharsets.UTF_8));
     post.addHeader(new BasicHeader("Content-Type", "application/json"));
     post.addHeader(
         PullReplicationApiRequestMetrics.HTTP_HEADER_X_START_TIME_NANOS,
         Long.toString(startTimeNanos));
-    return httpClientFactory.create(source).execute(post, this, getContext(targetUri));
+    return httpClientFactory.create(source).execute(withBasicAuthentication(targetUri, post), this);
   }
 
   /* (non-Javadoc)
@@ -132,7 +136,7 @@
     HttpPut put = new HttpPut(url);
     put.addHeader(new BasicHeader("Accept", MediaType.ANY_TEXT_TYPE.toString()));
     put.addHeader(new BasicHeader("Content-Type", MediaType.PLAIN_TEXT_UTF_8.toString()));
-    return httpClientFactory.create(source).execute(put, this, getContext(uri));
+    return httpClientFactory.create(source).execute(withBasicAuthentication(uri, put), this);
   }
 
   /* (non-Javadoc)
@@ -143,7 +147,7 @@
     String url =
         String.format("%s/%s", apiUri.toASCIIString(), getProjectDeletionUrl(project.get()));
     HttpDelete delete = new HttpDelete(url);
-    return httpClientFactory.create(source).execute(delete, this, getContext(apiUri));
+    return httpClientFactory.create(source).execute(withBasicAuthentication(apiUri, delete), this);
   }
 
   /* (non-Javadoc)
@@ -159,7 +163,7 @@
     req.setEntity(
         new StringEntity(String.format("{\"ref\": \"%s\"}", newHead), StandardCharsets.UTF_8));
     req.addHeader(new BasicHeader("Content-Type", MediaType.JSON_UTF_8.toString()));
-    return httpClientFactory.create(source).execute(req, this, getContext(apiUri));
+    return httpClientFactory.create(source).execute(withBasicAuthentication(apiUri, req), this);
   }
 
   /* (non-Javadoc)
@@ -180,14 +184,14 @@
     } else {
       requireNull(revisionData, "DELETE ref-updates cannot be associated with a RevisionData");
     }
-    RevisionInput input = new RevisionInput(instanceLabel, refName, revisionData);
+    RevisionInput input = new RevisionInput(instanceId, refName, revisionData);
 
     String url = formatUrl(project, targetUri, "apply-object");
 
     HttpPost post = new HttpPost(url);
     post.setEntity(new StringEntity(GSON.toJson(input)));
     post.addHeader(new BasicHeader("Content-Type", MediaType.JSON_UTF_8.toString()));
-    return httpClientFactory.create(source).execute(post, this, getContext(targetUri));
+    return httpClientFactory.create(source).execute(withBasicAuthentication(targetUri, post), this);
   }
 
   @Override
@@ -199,14 +203,13 @@
     }
 
     RevisionData[] inputData = new RevisionData[revisionData.size()];
-    RevisionsInput input =
-        new RevisionsInput(instanceLabel, refName, revisionData.toArray(inputData));
+    RevisionsInput input = new RevisionsInput(instanceId, refName, revisionData.toArray(inputData));
 
     String url = formatUrl(project, targetUri, "apply-objects");
     HttpPost post = new HttpPost(url);
     post.setEntity(new StringEntity(GSON.toJson(input)));
     post.addHeader(new BasicHeader("Content-Type", MediaType.JSON_UTF_8.toString()));
-    return httpClientFactory.create(source).execute(post, this, getContext(targetUri));
+    return httpClientFactory.create(source).execute(withBasicAuthentication(targetUri, post), this);
   }
 
   private String formatUrl(Project.NameKey project, URIish targetUri, String api) {
@@ -242,23 +245,21 @@
     return new HttpResult(response.getStatusLine().getStatusCode(), responseBody);
   }
 
-  private HttpClientContext getContext(URIish targetUri) {
-    HttpClientContext ctx = HttpClientContext.create();
-    ctx.setCredentialsProvider(adapt(credentials.create(source.getRemoteConfigName()), targetUri));
-    return ctx;
-  }
-
-  private CredentialsProvider adapt(org.eclipse.jgit.transport.CredentialsProvider cp, URIish uri) {
+  private HttpRequestBase withBasicAuthentication(URIish targetUri, HttpRequestBase req) {
+    org.eclipse.jgit.transport.CredentialsProvider cp =
+        credentials.create(source.getRemoteConfigName());
     CredentialItem.Username user = new CredentialItem.Username();
     CredentialItem.Password pass = new CredentialItem.Password();
-    if (cp.supports(user, pass) && cp.get(uri, user, pass)) {
-      CredentialsProvider adapted = new BasicCredentialsProvider();
-      adapted.setCredentials(
-          AuthScope.ANY,
-          new UsernamePasswordCredentials(user.getValue(), new String(pass.getValue())));
-      return adapted;
+    if (cp.supports(user, pass) && cp.get(targetUri, user, pass)) {
+      UsernamePasswordCredentials creds =
+          new UsernamePasswordCredentials(user.getValue(), new String(pass.getValue()));
+      try {
+        req.addHeader(new BasicScheme().authenticate(creds, req, null));
+      } catch (AuthenticationException e) {
+        logger.atFine().log(String.format("Anonymous Basic Authentication for uri: %s", targetUri));
+      }
     }
-    return null;
+    return req;
   }
 
   String getProjectDeletionUrl(String projectName) {
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/client/HttpClient.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/client/HttpClient.java
index 7bfc7d1..6254a42 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/client/HttpClient.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/client/HttpClient.java
@@ -18,14 +18,11 @@
 import org.apache.http.client.ClientProtocolException;
 import org.apache.http.client.ResponseHandler;
 import org.apache.http.client.methods.HttpUriRequest;
-import org.apache.http.protocol.HttpContext;
 
 /** HTTP client for executing URI requests to a remote site */
 public interface HttpClient {
 
   public <T> T execute(
-      final HttpUriRequest request,
-      final ResponseHandler<? extends T> responseHandler,
-      final HttpContext context)
+      final HttpUriRequest request, final ResponseHandler<? extends T> responseHandler)
       throws ClientProtocolException, IOException;
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/client/SourceHttpClient.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/client/SourceHttpClient.java
index ee0fe79..fa700e3 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/client/SourceHttpClient.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/client/SourceHttpClient.java
@@ -25,7 +25,6 @@
 import org.apache.http.conn.HttpClientConnectionManager;
 import org.apache.http.impl.client.HttpClients;
 import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
-import org.apache.http.protocol.HttpContext;
 
 /** Apache HTTP client implementation based on Source-specific parameters */
 public class SourceHttpClient implements HttpClient {
@@ -41,8 +40,7 @@
   }
 
   @Override
-  public <T> T execute(
-      HttpUriRequest request, ResponseHandler<? extends T> responseHandler, HttpContext context)
+  public <T> T execute(HttpUriRequest request, ResponseHandler<? extends T> responseHandler)
       throws ClientProtocolException, IOException {
     return source
         .memoize(
@@ -51,7 +49,7 @@
                     .setConnectionManager(customConnectionManager(source))
                     .setDefaultRequestConfig(customRequestConfig(source))
                     .build())
-        .execute(request, responseHandler, context);
+        .execute(request, responseHandler);
   }
 
   private static RequestConfig customRequestConfig(Source source) {
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/event/StreamEventListener.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/event/StreamEventListener.java
new file mode 100644
index 0000000..0f092e5
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/event/StreamEventListener.java
@@ -0,0 +1,120 @@
+// Copyright (C) 2022 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.googlesource.gerrit.plugins.replication.pull.event;
+
+import static java.util.Objects.requireNonNull;
+
+import com.google.common.base.Strings;
+import com.google.common.flogger.FluentLogger;
+import com.google.gerrit.common.Nullable;
+import com.google.gerrit.entities.Project.NameKey;
+import com.google.gerrit.entities.RefNames;
+import com.google.gerrit.extensions.restapi.AuthException;
+import com.google.gerrit.server.config.GerritInstanceId;
+import com.google.gerrit.server.events.Event;
+import com.google.gerrit.server.events.EventListener;
+import com.google.gerrit.server.events.ProjectCreatedEvent;
+import com.google.gerrit.server.events.RefUpdatedEvent;
+import com.google.gerrit.server.git.WorkQueue;
+import com.google.gerrit.server.permissions.PermissionBackendException;
+import com.google.inject.Inject;
+import com.google.inject.Provider;
+import com.googlesource.gerrit.plugins.replication.pull.FetchOne;
+import com.googlesource.gerrit.plugins.replication.pull.api.FetchAction;
+import com.googlesource.gerrit.plugins.replication.pull.api.FetchJob;
+import com.googlesource.gerrit.plugins.replication.pull.api.FetchJob.Factory;
+import com.googlesource.gerrit.plugins.replication.pull.api.ProjectInitializationAction;
+import com.googlesource.gerrit.plugins.replication.pull.api.PullReplicationApiRequestMetrics;
+import org.eclipse.jgit.lib.ObjectId;
+
+public class StreamEventListener implements EventListener {
+  private static final FluentLogger logger = FluentLogger.forEnclosingClass();
+
+  private String instanceId;
+  private WorkQueue workQueue;
+  private ProjectInitializationAction projectInitializationAction;
+
+  private Factory fetchJobFactory;
+  private final Provider<PullReplicationApiRequestMetrics> metricsProvider;
+
+  @Inject
+  public StreamEventListener(
+      @Nullable @GerritInstanceId String instanceId,
+      ProjectInitializationAction projectInitializationAction,
+      WorkQueue workQueue,
+      FetchJob.Factory fetchJobFactory,
+      Provider<PullReplicationApiRequestMetrics> metricsProvider) {
+    this.instanceId = instanceId;
+    this.projectInitializationAction = projectInitializationAction;
+    this.workQueue = workQueue;
+    this.fetchJobFactory = fetchJobFactory;
+    this.metricsProvider = metricsProvider;
+
+    requireNonNull(
+        Strings.emptyToNull(this.instanceId), "gerrit.instanceId cannot be null or empty");
+  }
+
+  @Override
+  public void onEvent(Event event) {
+    if (!instanceId.equals(event.instanceId)) {
+      PullReplicationApiRequestMetrics metrics = metricsProvider.get();
+      metrics.start(event);
+      if (event instanceof RefUpdatedEvent) {
+        RefUpdatedEvent refUpdatedEvent = (RefUpdatedEvent) event;
+        if (!isProjectDelete(refUpdatedEvent)) {
+          fetchRefsAsync(
+              refUpdatedEvent.getRefName(),
+              refUpdatedEvent.instanceId,
+              refUpdatedEvent.getProjectNameKey(),
+              metrics);
+        }
+      }
+      if (event instanceof ProjectCreatedEvent) {
+        ProjectCreatedEvent projectCreatedEvent = (ProjectCreatedEvent) event;
+        try {
+          projectInitializationAction.initProject(getProjectRepositoryName(projectCreatedEvent));
+          fetchRefsAsync(
+              FetchOne.ALL_REFS,
+              projectCreatedEvent.instanceId,
+              projectCreatedEvent.getProjectNameKey(),
+              metrics);
+        } catch (AuthException | PermissionBackendException e) {
+          logger.atSevere().withCause(e).log(
+              "Cannot initialise project:%s", projectCreatedEvent.projectName);
+        }
+      }
+    }
+  }
+
+  private boolean isProjectDelete(RefUpdatedEvent event) {
+    return RefNames.isConfigRef(event.getRefName())
+        && ObjectId.zeroId().equals(ObjectId.fromString(event.refUpdate.get().newRev));
+  }
+
+  protected void fetchRefsAsync(
+      String refName,
+      String sourceInstanceId,
+      NameKey projectNameKey,
+      PullReplicationApiRequestMetrics metrics) {
+    FetchAction.Input input = new FetchAction.Input();
+    input.refName = refName;
+    input.label = sourceInstanceId;
+    workQueue.getDefaultQueue().submit(fetchJobFactory.create(projectNameKey, input, metrics));
+  }
+
+  private String getProjectRepositoryName(ProjectCreatedEvent projectCreatedEvent) {
+    return String.format("%s.git", projectCreatedEvent.projectName);
+  }
+}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/event/StreamEventModule.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/event/StreamEventModule.java
new file mode 100644
index 0000000..2389678
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/event/StreamEventModule.java
@@ -0,0 +1,27 @@
+// Copyright (C) 2022 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.googlesource.gerrit.plugins.replication.pull.event;
+
+import com.google.gerrit.extensions.registration.DynamicSet;
+import com.google.gerrit.server.events.EventListener;
+import com.google.inject.AbstractModule;
+
+public class StreamEventModule extends AbstractModule {
+
+  @Override
+  protected void configure() {
+    DynamicSet.bind(binder(), EventListener.class).to(StreamEventListener.class);
+  }
+}
diff --git a/src/main/resources/Documentation/config.md b/src/main/resources/Documentation/config.md
index b2470a4..f29e572 100644
--- a/src/main/resources/Documentation/config.md
+++ b/src/main/resources/Documentation/config.md
@@ -31,8 +31,13 @@
     threads = 3
     authGroup = Public Mirror Group
     authGroup = Second Public Mirror Group
-  [replication]
-    instanceLabel = host-one
+```
+
+And make sure that instanceId is setup in `$site_path/etc/gerrit.config`:
+
+```
+[gerrit]
+    instanceId = host-one
 ```
 
 Then reload the replication plugin to pick up the new configuration:
@@ -121,6 +126,21 @@
 	provided in the remote configuration section which name is equal
 	to instanceLabel.
 
+	Deprecated: This property is kept for backward compatibility and
+	will be removed in the future release. Use [gerrit.instanceId](https://gerrit-review.googlesource.com/Documentation/config-gerrit.html#gerrit.instanceId)
+	instead.
+
+replication.consumeStreamEvents
+:	Use stream events to trigger pull-replication actions alongside the
+	REST approach. This mechanism is useful together with event-broker
+	and multi-site to provide backfill mechanism when a node has to
+	catch up with the events after being unreachable.
+
+	NOTE: When `consumeStreamEvents` is enabled gerrit.instanceId
+	instead of [replication.instanceLabel](https://gerrit.googlesource.com/plugins/pull-replication/+/refs/heads/stable-3.4/src/main/resources/Documentation/config.md#replication.instanceLabel) must be used.
+
+	Default: false
+
 replication.maxConnectionsPerRoute
 :	Maximum number of HTTP connections per one HTTP route.
 
@@ -441,7 +461,6 @@
     autoReload = true
     replicateOnStartup = false
 [replication]
-	instanceLabel = host-one
     lockErrorMaxRetries = 5
     maxRetries = 5
 ```
@@ -477,7 +496,6 @@
     autoReload = true
     replicateOnStartup = false
 [replication]
-    instanceLabel = host-one
     lockErrorMaxRetries = 5
     maxRetries = 5
 
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationFanoutConfigIT.java b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationFanoutConfigIT.java
index cb5af42..ee5876f 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationFanoutConfigIT.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationFanoutConfigIT.java
@@ -23,6 +23,7 @@
 import com.google.gerrit.acceptance.SkipProjectClone;
 import com.google.gerrit.acceptance.TestPlugin;
 import com.google.gerrit.acceptance.UseLocalDisk;
+import com.google.gerrit.acceptance.config.GerritConfig;
 import com.google.gerrit.acceptance.testsuite.project.ProjectOperations;
 import com.google.gerrit.entities.Project;
 import com.google.gerrit.extensions.api.projects.BranchInput;
@@ -95,6 +96,7 @@
   }
 
   @Test
+  @GerritConfig(name = "gerrit.instanceId", value = TEST_REPLICATION_REMOTE)
   public void shouldReplicateNewChangeRef() throws Exception {
     testRepo = cloneProject(createTestProject(project + TEST_REPLICATION_SUFFIX));
 
@@ -122,6 +124,7 @@
   }
 
   @Test
+  @GerritConfig(name = "gerrit.instanceId", value = TEST_REPLICATION_REMOTE)
   public void shouldReplicateNewChangeRefAfterConfigReloaded() throws Exception {
     testRepo = cloneProject(createTestProject(project + TEST_REPLICATION_SUFFIX));
 
@@ -157,6 +160,7 @@
   }
 
   @Test
+  @GerritConfig(name = "gerrit.instanceId", value = TEST_REPLICATION_REMOTE)
   public void shouldReplicateNewBranch() throws Exception {
     String testProjectName = project + TEST_REPLICATION_SUFFIX;
     createTestProject(testProjectName);
@@ -190,6 +194,7 @@
   }
 
   @Test
+  @GerritConfig(name = "gerrit.instanceId", value = TEST_REPLICATION_REMOTE)
   public void shouldAutoReloadConfiguration() throws Exception {
     SourcesCollection sources = getInstance(SourcesCollection.class);
     AutoReloadConfigDecorator autoReloadConfigDecorator =
@@ -202,6 +207,7 @@
   }
 
   @Test
+  @GerritConfig(name = "gerrit.instanceId", value = TEST_REPLICATION_REMOTE)
   public void shouldAutoReloadConfigurationWhenRemoteConfigAdded() throws Exception {
     FileBasedConfig newRemoteConfig =
         new FileBasedConfig(
@@ -224,6 +230,7 @@
   }
 
   @Test
+  @GerritConfig(name = "gerrit.instanceId", value = TEST_REPLICATION_REMOTE)
   public void shouldAutoReloadConfigurationWhenRemoteConfigDeleted() throws Exception {
     SourcesCollection sources = getInstance(SourcesCollection.class);
     AutoReloadConfigDecorator autoReloadConfigDecorator =
@@ -259,7 +266,6 @@
   }
 
   private void setReplicationSource(String remoteName) throws IOException {
-    config.setString("replication", null, "instanceLabel", remoteName);
     config.setBoolean("gerrit", null, "autoReload", true);
     config.save();
   }
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationIT.java b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationIT.java
index ec60793..187af3d 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationIT.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationIT.java
@@ -27,6 +27,7 @@
 import com.google.gerrit.acceptance.SkipProjectClone;
 import com.google.gerrit.acceptance.TestPlugin;
 import com.google.gerrit.acceptance.UseLocalDisk;
+import com.google.gerrit.acceptance.config.GerritConfig;
 import com.google.gerrit.acceptance.testsuite.project.ProjectOperations;
 import com.google.gerrit.entities.Permission;
 import com.google.gerrit.entities.Project;
@@ -124,6 +125,7 @@
   }
 
   @Test
+  @GerritConfig(name = "gerrit.instanceId", value = TEST_REPLICATION_REMOTE)
   public void shouldReplicateNewChangeRef() throws Exception {
     testRepo = cloneProject(createTestProject(project + TEST_REPLICATION_SUFFIX));
 
@@ -151,6 +153,7 @@
   }
 
   @Test
+  @GerritConfig(name = "gerrit.instanceId", value = TEST_REPLICATION_REMOTE)
   public void shouldReplicateNewBranch() throws Exception {
     String testProjectName = project + TEST_REPLICATION_SUFFIX;
     createTestProject(testProjectName);
@@ -185,6 +188,7 @@
 
   @Test
   @UseLocalDisk
+  @GerritConfig(name = "gerrit.instanceId", value = TEST_REPLICATION_REMOTE)
   public void shouldReplicateForceUpdatedBranch() throws Exception {
     boolean forcedPush = true;
     String testProjectName = project + TEST_REPLICATION_SUFFIX;
@@ -254,6 +258,7 @@
   }
 
   @Test
+  @GerritConfig(name = "gerrit.instanceId", value = TEST_REPLICATION_REMOTE)
   public void shouldReplicateNewChangeRefCGitClient() throws Exception {
     AutoReloadConfigDecorator autoReloadConfigDecorator =
         getInstance(AutoReloadConfigDecorator.class);
@@ -289,6 +294,7 @@
   }
 
   @Test
+  @GerritConfig(name = "gerrit.instanceId", value = TEST_REPLICATION_REMOTE)
   public void shouldReplicateNewBranchCGitClient() throws Exception {
     AutoReloadConfigDecorator autoReloadConfigDecorator =
         getInstance(AutoReloadConfigDecorator.class);
@@ -330,6 +336,7 @@
   }
 
   @Test
+  @GerritConfig(name = "gerrit.instanceId", value = TEST_REPLICATION_REMOTE)
   public void shouldReplicateProjectDeletion() throws Exception {
     String projectToDelete = project.get();
     setReplicationSource(TEST_REPLICATION_REMOTE, "", Optional.of(projectToDelete));
@@ -358,6 +365,7 @@
   }
 
   @Test
+  @GerritConfig(name = "gerrit.instanceId", value = TEST_REPLICATION_REMOTE)
   public void shouldReplicateHeadUpdate() throws Exception {
     String testProjectName = project.get();
     setReplicationSource(TEST_REPLICATION_REMOTE, "", Optional.of(testProjectName));
@@ -389,6 +397,35 @@
         });
   }
 
+  @Test
+  @GerritConfig(name = "gerrit.instanceId", value = TEST_REPLICATION_REMOTE)
+  @GerritConfig(name = "container.replica", value = "true")
+  public void shouldReplicateNewChangeRefToReplica() throws Exception {
+    testRepo = cloneProject(createTestProject(project + TEST_REPLICATION_SUFFIX));
+
+    Result pushResult = createChange();
+    RevCommit sourceCommit = pushResult.getCommit();
+    String sourceRef = pushResult.getPatchSet().refName();
+
+    ReplicationQueue pullReplicationQueue = getInstance(ReplicationQueue.class);
+    GitReferenceUpdatedListener.Event event =
+        new FakeGitReferenceUpdatedEvent(
+            project,
+            sourceRef,
+            ObjectId.zeroId().getName(),
+            sourceCommit.getId().getName(),
+            ReceiveCommand.Type.CREATE);
+    pullReplicationQueue.onGitReferenceUpdated(event);
+
+    try (Repository repo = repoManager.openRepository(project)) {
+      waitUntil(() -> checkedGetRef(repo, sourceRef) != null);
+
+      Ref targetBranchRef = getRef(repo, sourceRef);
+      assertThat(targetBranchRef).isNotNull();
+      assertThat(targetBranchRef.getObjectId()).isEqualTo(sourceCommit.getId());
+    }
+  }
+
   private Ref getRef(Repository repo, String branchName) throws IOException {
     return repo.getRefDatabase().exactRef(branchName);
   }
@@ -416,7 +453,6 @@
         replicaSuffixes.stream()
             .map(suffix -> gitPath.resolve("${name}" + suffix + ".git").toString())
             .collect(toList());
-    config.setString("replication", null, "instanceLabel", remoteName);
     config.setStringList("remote", remoteName, "url", replicaUrls);
     config.setString("remote", remoteName, "apiUrl", adminRestSession.url());
     config.setString("remote", remoteName, "fetch", "+refs/*:refs/*");
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/ActionITBase.java b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/ActionITBase.java
index 1019b86..55ad15c 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/ActionITBase.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/ActionITBase.java
@@ -43,17 +43,16 @@
 import java.util.List;
 import java.util.Optional;
 import org.apache.http.HttpResponse;
-import org.apache.http.auth.AuthScope;
+import org.apache.http.auth.AuthenticationException;
 import org.apache.http.auth.UsernamePasswordCredentials;
 import org.apache.http.client.ClientProtocolException;
-import org.apache.http.client.CredentialsProvider;
 import org.apache.http.client.ResponseHandler;
 import org.apache.http.client.methods.HttpDelete;
 import org.apache.http.client.methods.HttpPost;
 import org.apache.http.client.methods.HttpPut;
-import org.apache.http.client.protocol.HttpClientContext;
+import org.apache.http.client.methods.HttpRequestBase;
 import org.apache.http.entity.StringEntity;
-import org.apache.http.impl.client.BasicCredentialsProvider;
+import org.apache.http.impl.auth.BasicScheme;
 import org.apache.http.message.BasicHeader;
 import org.eclipse.jgit.lib.Repository;
 import org.eclipse.jgit.storage.file.FileBasedConfig;
@@ -170,17 +169,22 @@
     };
   }
 
-  protected HttpClientContext getContext() {
-    return getContextForAccount(admin);
+  protected HttpRequestBase withBasicAuthenticationAsAdmin(HttpRequestBase httpRequest)
+      throws AuthenticationException {
+    return withBasicAuthentication(httpRequest, admin);
   }
 
-  protected HttpClientContext getUserContext() {
-    return getContextForAccount(user);
+  protected HttpRequestBase withBasicAuthenticationAsUser(HttpRequestBase httpRequest)
+      throws AuthenticationException {
+    return withBasicAuthentication(httpRequest, user);
   }
 
-  protected HttpClientContext getAnonymousContext() {
-    HttpClientContext ctx = HttpClientContext.create();
-    return ctx;
+  private HttpRequestBase withBasicAuthentication(HttpRequestBase httpRequest, TestAccount account)
+      throws AuthenticationException {
+    UsernamePasswordCredentials creds =
+        new UsernamePasswordCredentials(account.username(), account.httpPassword());
+    httpRequest.addHeader(new BasicScheme().authenticate(creds, httpRequest, null));
+    return httpRequest;
   }
 
   private Project.NameKey createTestProject(String name) throws Exception {
@@ -200,7 +204,6 @@
         replicaSuffixes.stream()
             .map(suffix -> gitPath.resolve("${name}" + suffix + ".git").toString())
             .collect(toList());
-    config.setString("replication", null, "instanceLabel", remoteName);
     config.setStringList("remote", remoteName, "url", replicaUrls);
     config.setString("remote", remoteName, "apiUrl", adminRestSession.url());
     config.setString("remote", remoteName, "fetch", "+refs/tags/*:refs/tags/*");
@@ -217,13 +220,4 @@
     secureConfig.setString("remote", remoteName, "password", password);
     secureConfig.save();
   }
-
-  private HttpClientContext getContextForAccount(TestAccount account) {
-    HttpClientContext ctx = HttpClientContext.create();
-    CredentialsProvider adapted = new BasicCredentialsProvider();
-    adapted.setCredentials(
-        AuthScope.ANY, new UsernamePasswordCredentials(account.username(), account.httpPassword()));
-    ctx.setCredentialsProvider(adapted);
-    return ctx;
-  }
 }
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/ApplyObjectActionIT.java b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/ApplyObjectActionIT.java
index aad3903..a5fd63c 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/ApplyObjectActionIT.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/ApplyObjectActionIT.java
@@ -22,7 +22,6 @@
 import com.google.gerrit.extensions.restapi.Url;
 import com.googlesource.gerrit.plugins.replication.pull.api.data.RevisionData;
 import java.util.Optional;
-import org.apache.http.client.methods.HttpPost;
 import org.junit.Test;
 
 public class ApplyObjectActionIT extends ActionITBase {
@@ -41,8 +40,11 @@
     RevisionData revisionData = revisionDataOption.get();
     String sendObjectPayload = createPayload(payloadWithAsyncFieldTemplate, refName, revisionData);
 
-    HttpPost post = createRequest(sendObjectPayload);
-    httpClientFactory.create(source).execute(post, assertHttpResponseCode(201), getContext());
+    httpClientFactory
+        .create(source)
+        .execute(
+            withBasicAuthenticationAsAdmin(createRequest(sendObjectPayload)),
+            assertHttpResponseCode(201));
   }
 
   @Test
@@ -60,8 +62,11 @@
     String sendObjectPayload =
         createPayload(payloadWithoutAsyncFieldTemplate, refName, revisionData);
 
-    HttpPost post = createRequest(sendObjectPayload);
-    httpClientFactory.create(source).execute(post, assertHttpResponseCode(201), getContext());
+    httpClientFactory
+        .create(source)
+        .execute(
+            withBasicAuthenticationAsAdmin(createRequest(sendObjectPayload)),
+            assertHttpResponseCode(201));
   }
 
   @Test
@@ -80,8 +85,11 @@
     String sendObjectPayload =
         createPayload(payloadWithoutAsyncFieldTemplate, refName, revisionData);
 
-    HttpPost post = createRequest(sendObjectPayload);
-    httpClientFactory.create(source).execute(post, assertHttpResponseCode(201), getContext());
+    httpClientFactory
+        .create(source)
+        .execute(
+            withBasicAuthenticationAsAdmin(createRequest(sendObjectPayload)),
+            assertHttpResponseCode(201));
   }
 
   @Test
@@ -103,13 +111,16 @@
         String.format(
             "%s/a/projects/%s/pull-replication~apply-object",
             adminRestSession.url(), Url.encode(projectName.get()));
-    HttpPost post = createRequest(sendObjectPayload);
-    httpClientFactory.create(source).execute(post, assertHttpResponseCode(201), getContext());
+    httpClientFactory
+        .create(source)
+        .execute(
+            withBasicAuthenticationAsAdmin(createRequest(sendObjectPayload)),
+            assertHttpResponseCode(201));
   }
 
   @Test
   @GerritConfig(name = "container.replica", value = "true")
-  public void shouldReturnUnauthorizedWhenNodeIsAReplicaAndUSerIsAnonymous() throws Exception {
+  public void shouldReturnForbiddenWhenNodeIsAReplicaAndUSerIsAnonymous() throws Exception {
     String payloadWithoutAsyncFieldTemplate =
         "{\"label\":\""
             + TEST_REPLICATION_REMOTE
@@ -123,10 +134,9 @@
     String sendObjectPayload =
         createPayload(payloadWithoutAsyncFieldTemplate, refName, revisionData);
 
-    HttpPost post = createRequest(sendObjectPayload);
     httpClientFactory
         .create(source)
-        .execute(post, assertHttpResponseCode(401), getAnonymousContext());
+        .execute(createRequest(sendObjectPayload), assertHttpResponseCode(403));
   }
 
   @Test
@@ -142,8 +152,11 @@
     String sendObjectPayload =
         createPayload(payloadWithoutLabelFieldTemplate, refName, revisionData);
 
-    HttpPost post = createRequest(sendObjectPayload);
-    httpClientFactory.create(source).execute(post, assertHttpResponseCode(400), getContext());
+    httpClientFactory
+        .create(source)
+        .execute(
+            withBasicAuthenticationAsAdmin(createRequest(sendObjectPayload)),
+            assertHttpResponseCode(400));
   }
 
   @Test
@@ -160,8 +173,11 @@
     RevisionData revisionData = revisionDataOption.get();
     String sendObjectPayload = createPayload(wrongPayloadTemplate, refName, revisionData);
 
-    HttpPost post = createRequest(sendObjectPayload);
-    httpClientFactory.create(source).execute(post, assertHttpResponseCode(400), getContext());
+    httpClientFactory
+        .create(source)
+        .execute(
+            withBasicAuthenticationAsAdmin(createRequest(sendObjectPayload)),
+            assertHttpResponseCode(400));
   }
 
   private String createPayload(
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchActionIT.java b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchActionIT.java
index 4ad8ce6..5634b2a 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchActionIT.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchActionIT.java
@@ -35,7 +35,9 @@
 
     httpClientFactory
         .create(source)
-        .execute(createRequest(sendObjectPayload), assertHttpResponseCode(201), getContext());
+        .execute(
+            withBasicAuthenticationAsAdmin(createRequest(sendObjectPayload)),
+            assertHttpResponseCode(201));
   }
 
   @Test
@@ -55,12 +57,14 @@
             adminRestSession.url(), Url.encode(projectName.get()));
     httpClientFactory
         .create(source)
-        .execute(createRequest(sendObjectPayload), assertHttpResponseCode(201), getContext());
+        .execute(
+            withBasicAuthenticationAsAdmin(createRequest(sendObjectPayload)),
+            assertHttpResponseCode(201));
   }
 
   @Test
   @GerritConfig(name = "container.replica", value = "true")
-  public void shouldReturnUnauthorizedWhenNodeIsAReplicaAndUSerIsAnonymous() throws Exception {
+  public void shouldReturnForbiddenWhenNodeIsAReplicaAndUSerIsAnonymous() throws Exception {
     String refName = createRef();
     String sendObjectPayload =
         "{\"label\":\""
@@ -71,8 +75,7 @@
 
     httpClientFactory
         .create(source)
-        .execute(
-            createRequest(sendObjectPayload), assertHttpResponseCode(401), getAnonymousContext());
+        .execute(createRequest(sendObjectPayload), assertHttpResponseCode(403));
   }
 
   @Override
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchActionTest.java b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchActionTest.java
index fb7f3d1..ce0b9d3 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchActionTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/FetchActionTest.java
@@ -55,6 +55,8 @@
   int taskId = 1234;
 
   @Mock FetchCommand fetchCommand;
+  @Mock FetchJob fetchJob;
+  @Mock FetchJob.Factory fetchJobFactory;
   @Mock ProjectResource projectResource;
   @Mock WorkQueue workQueue;
   @Mock ScheduledExecutorService exceutorService;
@@ -65,6 +67,7 @@
 
   @Before
   public void setup() {
+    when(fetchJobFactory.create(any(), any(), any())).thenReturn(fetchJob);
     when(workQueue.getDefaultQueue()).thenReturn(exceutorService);
     when(urlFormatter.getRestUrl(anyString())).thenReturn(Optional.of(location));
     when(exceutorService.submit(any(Runnable.class)))
@@ -79,7 +82,9 @@
     when(task.getTaskId()).thenReturn(taskId);
     when(preConditions.canCallFetchApi()).thenReturn(true);
 
-    fetchAction = new FetchAction(fetchCommand, workQueue, urlFormatterDynamicItem, preConditions);
+    fetchAction =
+        new FetchAction(
+            fetchCommand, workQueue, urlFormatterDynamicItem, preConditions, fetchJobFactory);
   }
 
   @Test
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/ProjectDeletionActionIT.java b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/ProjectDeletionActionIT.java
index f5c8d4c..2f61cff 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/ProjectDeletionActionIT.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/ProjectDeletionActionIT.java
@@ -22,6 +22,7 @@
 import com.google.gerrit.server.group.SystemGroupBackend;
 import com.google.inject.Inject;
 import javax.servlet.http.HttpServletResponse;
+import org.apache.http.client.methods.HttpRequestBase;
 import org.junit.Test;
 
 public class ProjectDeletionActionIT extends ActionITBase {
@@ -35,9 +36,7 @@
     httpClientFactory
         .create(source)
         .execute(
-            createDeleteRequest(),
-            assertHttpResponseCode(HttpServletResponse.SC_UNAUTHORIZED),
-            getAnonymousContext());
+            createDeleteRequest(), assertHttpResponseCode(HttpServletResponse.SC_UNAUTHORIZED));
   }
 
   @Test
@@ -47,9 +46,8 @@
     httpClientFactory
         .create(source)
         .execute(
-            createDeleteRequest(),
-            assertHttpResponseCode(HttpServletResponse.SC_FORBIDDEN),
-            getUserContext());
+            withBasicAuthenticationAsUser(createDeleteRequest()),
+            assertHttpResponseCode(HttpServletResponse.SC_FORBIDDEN));
 
     projectOperations
         .project(allProjects)
@@ -60,9 +58,8 @@
     httpClientFactory
         .create(source)
         .execute(
-            createDeleteRequest(),
-            assertHttpResponseCode(HttpServletResponse.SC_OK),
-            getUserContext());
+            withBasicAuthenticationAsUser(createDeleteRequest()),
+            assertHttpResponseCode(HttpServletResponse.SC_OK));
   }
 
   @Test
@@ -73,7 +70,8 @@
     httpClientFactory
         .create(source)
         .execute(
-            createDeleteRequest(), assertHttpResponseCode(HttpServletResponse.SC_OK), getContext());
+            withBasicAuthenticationAsAdmin(createDeleteRequest()),
+            assertHttpResponseCode(HttpServletResponse.SC_OK));
   }
 
   @Test
@@ -83,20 +81,16 @@
     httpClientFactory
         .create(source)
         .execute(
-            createDeleteRequest(),
-            assertHttpResponseCode(HttpServletResponse.SC_INTERNAL_SERVER_ERROR),
-            getContext());
+            withBasicAuthenticationAsAdmin(createDeleteRequest()),
+            assertHttpResponseCode(HttpServletResponse.SC_INTERNAL_SERVER_ERROR));
   }
 
   @Test
   @GerritConfig(name = "container.replica", value = "true")
-  public void shouldReturnUnauthorizedForUserWithoutPermissionsOnReplica() throws Exception {
+  public void shouldReturnForbiddenForUserWithoutPermissionsOnReplica() throws Exception {
     httpClientFactory
         .create(source)
-        .execute(
-            createDeleteRequest(),
-            assertHttpResponseCode(HttpServletResponse.SC_UNAUTHORIZED),
-            getAnonymousContext());
+        .execute(createDeleteRequest(), assertHttpResponseCode(HttpServletResponse.SC_FORBIDDEN));
   }
 
   @Test
@@ -108,7 +102,8 @@
     httpClientFactory
         .create(source)
         .execute(
-            createDeleteRequest(), assertHttpResponseCode(HttpServletResponse.SC_OK), getContext());
+            withBasicAuthenticationAsAdmin(createDeleteRequest()),
+            assertHttpResponseCode(HttpServletResponse.SC_OK));
   }
 
   @Test
@@ -117,12 +112,11 @@
       throws Exception {
     String testProjectName = project.get();
     url = getURL(testProjectName);
+    HttpRequestBase deleteRequest = withBasicAuthenticationAsUser(createDeleteRequest());
+
     httpClientFactory
         .create(source)
-        .execute(
-            createDeleteRequest(),
-            assertHttpResponseCode(HttpServletResponse.SC_FORBIDDEN),
-            getUserContext());
+        .execute(deleteRequest, assertHttpResponseCode(HttpServletResponse.SC_FORBIDDEN));
 
     projectOperations
         .project(allProjects)
@@ -132,10 +126,7 @@
 
     httpClientFactory
         .create(source)
-        .execute(
-            createDeleteRequest(),
-            assertHttpResponseCode(HttpServletResponse.SC_OK),
-            getUserContext());
+        .execute(deleteRequest, assertHttpResponseCode(HttpServletResponse.SC_OK));
   }
 
   @Test
@@ -147,9 +138,8 @@
     httpClientFactory
         .create(source)
         .execute(
-            createDeleteRequest(),
-            assertHttpResponseCode(HttpServletResponse.SC_INTERNAL_SERVER_ERROR),
-            getContext());
+            withBasicAuthenticationAsAdmin(createDeleteRequest()),
+            assertHttpResponseCode(HttpServletResponse.SC_INTERNAL_SERVER_ERROR));
   }
 
   @Override
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/ProjectInitializationActionIT.java b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/ProjectInitializationActionIT.java
index ca4fa31..d1242a0 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/ProjectInitializationActionIT.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/ProjectInitializationActionIT.java
@@ -27,6 +27,7 @@
 import javax.servlet.http.HttpServletResponse;
 import org.apache.http.client.methods.HttpGet;
 import org.apache.http.client.methods.HttpPut;
+import org.apache.http.client.methods.HttpRequestBase;
 import org.apache.http.message.BasicHeader;
 import org.junit.Test;
 
@@ -40,8 +41,7 @@
         .create(source)
         .execute(
             createPutRequestWithHeaders(),
-            assertHttpResponseCode(HttpServletResponse.SC_UNAUTHORIZED),
-            getAnonymousContext());
+            assertHttpResponseCode(HttpServletResponse.SC_UNAUTHORIZED));
   }
 
   @Test
@@ -49,9 +49,8 @@
     httpClientFactory
         .create(source)
         .execute(
-            createPutRequestWithoutHeaders(),
-            assertHttpResponseCode(HttpServletResponse.SC_BAD_REQUEST),
-            getContext());
+            withBasicAuthenticationAsAdmin(createPutRequestWithoutHeaders()),
+            assertHttpResponseCode(HttpServletResponse.SC_BAD_REQUEST));
   }
 
   @Test
@@ -61,28 +60,26 @@
     httpClientFactory
         .create(source)
         .execute(
-            createPutRequestWithHeaders(),
-            assertHttpResponseCode(HttpServletResponse.SC_CREATED),
-            getContext());
+            withBasicAuthenticationAsAdmin(createPutRequestWithHeaders()),
+            assertHttpResponseCode(HttpServletResponse.SC_CREATED));
 
-    HttpGet getNewProjectRequest =
-        new HttpGet(userRestSession.url() + "/a/projects/" + Url.encode(newProjectName));
+    HttpRequestBase getNewProjectRequest =
+        withBasicAuthenticationAsAdmin(
+            new HttpGet(userRestSession.url() + "/a/projects/" + Url.encode(newProjectName)));
+
     httpClientFactory
         .create(source)
-        .execute(
-            getNewProjectRequest, assertHttpResponseCode(HttpServletResponse.SC_OK), getContext());
+        .execute(getNewProjectRequest, assertHttpResponseCode(HttpServletResponse.SC_OK));
   }
 
   @Test
   public void shouldCreateRepositoryWhenUserHasProjectCreationCapabilities() throws Exception {
     String newProjectName = "new/newProjectForUserWithCapabilities";
     url = getURL(newProjectName);
+    HttpRequestBase put = withBasicAuthenticationAsUser(createPutRequestWithHeaders());
     httpClientFactory
         .create(source)
-        .execute(
-            createPutRequestWithHeaders(),
-            assertHttpResponseCode(HttpServletResponse.SC_FORBIDDEN),
-            getUserContext());
+        .execute(put, assertHttpResponseCode(HttpServletResponse.SC_FORBIDDEN));
 
     projectOperations
         .project(allProjects)
@@ -94,10 +91,7 @@
 
     httpClientFactory
         .create(source)
-        .execute(
-            createPutRequestWithHeaders(),
-            assertHttpResponseCode(HttpServletResponse.SC_CREATED),
-            getUserContext());
+        .execute(put, assertHttpResponseCode(HttpServletResponse.SC_CREATED));
   }
 
   @Test
@@ -105,9 +99,8 @@
     httpClientFactory
         .create(source)
         .execute(
-            createPutRequestWithHeaders(),
-            assertHttpResponseCode(HttpServletResponse.SC_FORBIDDEN),
-            getUserContext());
+            withBasicAuthenticationAsUser(createPutRequestWithHeaders()),
+            assertHttpResponseCode(HttpServletResponse.SC_FORBIDDEN));
   }
 
   @Test
@@ -118,9 +111,8 @@
     httpClientFactory
         .create(source)
         .execute(
-            createPutRequestWithHeaders(),
-            assertHttpResponseCode(HttpServletResponse.SC_CREATED),
-            getContext());
+            withBasicAuthenticationAsAdmin(createPutRequestWithHeaders()),
+            assertHttpResponseCode(HttpServletResponse.SC_CREATED));
   }
 
   @Test
@@ -129,9 +121,8 @@
     httpClientFactory
         .create(source)
         .execute(
-            createPutRequestWithHeaders(),
-            assertHttpResponseCode(HttpServletResponse.SC_FORBIDDEN),
-            getUserContext());
+            withBasicAuthenticationAsUser(createPutRequestWithHeaders()),
+            assertHttpResponseCode(HttpServletResponse.SC_FORBIDDEN));
   }
 
   @Test
@@ -140,12 +131,10 @@
       throws Exception {
     String newProjectName = "new/newProjectForUserWithCapabilitiesReplica";
     url = getURL(newProjectName);
+    HttpRequestBase put = withBasicAuthenticationAsUser(createPutRequestWithHeaders());
     httpClientFactory
         .create(source)
-        .execute(
-            createPutRequestWithHeaders(),
-            assertHttpResponseCode(HttpServletResponse.SC_FORBIDDEN),
-            getUserContext());
+        .execute(put, assertHttpResponseCode(HttpServletResponse.SC_FORBIDDEN));
 
     projectOperations
         .project(allProjects)
@@ -157,10 +146,7 @@
 
     httpClientFactory
         .create(source)
-        .execute(
-            createPutRequestWithHeaders(),
-            assertHttpResponseCode(HttpServletResponse.SC_CREATED),
-            getUserContext());
+        .execute(put, assertHttpResponseCode(HttpServletResponse.SC_CREATED));
   }
 
   @Test
@@ -168,13 +154,11 @@
   public void shouldReturnInternalServerErrorIfProjectCannotBeCreatedWhenNodeIsAReplica()
       throws Exception {
     url = getURL(INVALID_TEST_PROJECT_NAME);
-
     httpClientFactory
         .create(source)
         .execute(
-            createPutRequestWithHeaders(),
-            assertHttpResponseCode(HttpServletResponse.SC_INTERNAL_SERVER_ERROR),
-            getContext());
+            withBasicAuthenticationAsAdmin(createPutRequestWithHeaders()),
+            assertHttpResponseCode(HttpServletResponse.SC_INTERNAL_SERVER_ERROR));
   }
 
   @Test
@@ -183,21 +167,18 @@
     httpClientFactory
         .create(source)
         .execute(
-            createPutRequestWithoutHeaders(),
-            assertHttpResponseCode(HttpServletResponse.SC_BAD_REQUEST),
-            getContext());
+            withBasicAuthenticationAsAdmin(createPutRequestWithoutHeaders()),
+            assertHttpResponseCode(HttpServletResponse.SC_BAD_REQUEST));
   }
 
   @Test
   @GerritConfig(name = "container.replica", value = "true")
-  public void shouldReturnUnauthorizedForUserWithoutPermissionsWhenNodeIsAReplica()
-      throws Exception {
+  public void shouldReturnForbiddenForUserWithoutPermissionsWhenNodeIsAReplica() throws Exception {
     httpClientFactory
         .create(source)
         .execute(
             createPutRequestWithHeaders(),
-            assertHttpResponseCode(HttpServletResponse.SC_UNAUTHORIZED),
-            getAnonymousContext());
+            assertHttpResponseCode(HttpServletResponse.SC_FORBIDDEN));
   }
 
   @Override
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/PullReplicationFilterTest.java b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/PullReplicationFilterTest.java
new file mode 100644
index 0000000..2ed1466
--- /dev/null
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/PullReplicationFilterTest.java
@@ -0,0 +1,331 @@
+package com.googlesource.gerrit.plugins.replication.pull.api;
+
+import static com.google.common.net.HttpHeaders.ACCEPT;
+import static com.google.gerrit.httpd.restapi.RestApiServlet.SC_UNPROCESSABLE_ENTITY;
+import static javax.servlet.http.HttpServletResponse.SC_CONFLICT;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.mockito.internal.verification.VerificationModeFactory.atLeastOnce;
+import static org.mockito.internal.verification.VerificationModeFactory.times;
+
+import com.google.common.net.MediaType;
+import com.google.gerrit.extensions.restapi.*;
+import com.google.gerrit.server.project.ProjectResource;
+import com.google.gerrit.server.restapi.project.ProjectsCollection;
+import java.io.*;
+import java.nio.charset.StandardCharsets;
+import javax.servlet.FilterChain;
+import javax.servlet.ServletOutputStream;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+@RunWith(MockitoJUnitRunner.class)
+public class PullReplicationFilterTest {
+
+  @Mock HttpServletRequest request;
+  @Mock HttpServletResponse response;
+  @Mock FilterChain filterChain;
+  @Mock private FetchAction fetchAction;
+  @Mock private ApplyObjectAction applyObjectAction;
+  @Mock private ApplyObjectsAction applyObjectsAction;
+  @Mock private ProjectInitializationAction projectInitializationAction;
+  @Mock private UpdateHeadAction updateHEADAction;
+  @Mock private ProjectDeletionAction projectDeletionAction;
+  @Mock private ProjectsCollection projectsCollection;
+  @Mock private ProjectResource projectResource;
+  @Mock private ServletOutputStream outputStream;
+  @Mock private PrintWriter printWriter;
+  private final String PLUGIN_NAME = "pull-replication";
+  private final String PROJECT_NAME = "some-project";
+  private final String PROJECT_NAME_GIT = "some-project.git";
+  private final String FETCH_URI =
+      String.format("any-prefix/projects/%s/%s~fetch", PROJECT_NAME, PLUGIN_NAME);
+  private final String APPLY_OBJECT_URI =
+      String.format("any-prefix/projects/%s/%s~apply-object", PROJECT_NAME, PLUGIN_NAME);
+  private final String APPLY_OBJECTS_URI =
+      String.format("any-prefix/projects/%s/%s~apply-objects", PROJECT_NAME, PLUGIN_NAME);
+  private final String HEAD_URI = String.format("any-prefix/projects/%s/HEAD", PROJECT_NAME);
+  private final String DELETE_PROJECT_URI =
+      String.format("any-prefix/projects/%s/%s~delete-project", PROJECT_NAME, PLUGIN_NAME);
+  private final String INIT_PROJECT_URI =
+      String.format("any-prefix/%s/init-project/%s", PLUGIN_NAME, PROJECT_NAME_GIT);
+
+  private final Response OK_RESPONSE = Response.ok();
+
+  private PullReplicationFilter createPullReplicationFilter() {
+    return new PullReplicationFilter(
+        fetchAction,
+        applyObjectAction,
+        applyObjectsAction,
+        projectInitializationAction,
+        updateHEADAction,
+        projectDeletionAction,
+        projectsCollection,
+        PLUGIN_NAME);
+  }
+
+  private void defineBehaviours(byte[] payload, String uri) throws Exception {
+    when(request.getRequestURI()).thenReturn(uri);
+    InputStream is = new ByteArrayInputStream(payload);
+    BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(is));
+    when(request.getReader()).thenReturn(bufferedReader);
+    when(projectsCollection.parse(TopLevelResource.INSTANCE, IdString.fromDecoded(PROJECT_NAME)))
+        .thenReturn(projectResource);
+    when(response.getWriter()).thenReturn(printWriter);
+  }
+
+  private void verifyBehaviours() throws Exception {
+    verify(request, atLeastOnce()).getRequestURI();
+    verify(request).getReader();
+    verify(projectsCollection).parse(TopLevelResource.INSTANCE, IdString.fromDecoded(PROJECT_NAME));
+    verify(response).getWriter();
+    verify(response).setContentType("application/json");
+    verify(response).setStatus(HttpServletResponse.SC_OK);
+  }
+
+  @Test
+  public void shouldFilterFetchAction() throws Exception {
+    byte[] payloadFetch =
+        ("{"
+                + "\"label\":\"Replication\", "
+                + "\"ref_name\": \"refs/heads/master\", "
+                + "\"async\":false"
+                + "}")
+            .getBytes(StandardCharsets.UTF_8);
+
+    defineBehaviours(payloadFetch, FETCH_URI);
+    when(fetchAction.apply(any(), any())).thenReturn(OK_RESPONSE);
+
+    PullReplicationFilter pullReplicationFilter = createPullReplicationFilter();
+    pullReplicationFilter.doFilter(request, response, filterChain);
+
+    verifyBehaviours();
+    verify(fetchAction).apply(eq(projectResource), any());
+  }
+
+  @Test
+  public void shouldFilterApplyObjectAction() throws Exception {
+
+    byte[] payloadApplyObject =
+        ("{\"label\":\"Replication\",\"ref_name\":\"refs/heads/master\","
+                + "\"revision_data\":{"
+                + "\"commit_object\":{\"type\":1,\"content\":\"some-content\"},"
+                + "\"tree_object\":{\"type\":2,\"content\":\"some-content\"},"
+                + "\"blobs\":[]}"
+                + "}")
+            .getBytes(StandardCharsets.UTF_8);
+
+    defineBehaviours(payloadApplyObject, APPLY_OBJECT_URI);
+
+    when(applyObjectAction.apply(any(), any())).thenReturn(OK_RESPONSE);
+
+    PullReplicationFilter pullReplicationFilter = createPullReplicationFilter();
+    pullReplicationFilter.doFilter(request, response, filterChain);
+
+    verifyBehaviours();
+    verify(applyObjectAction).apply(eq(projectResource), any());
+  }
+
+  @Test
+  public void shouldFilterApplyObjectsAction() throws Exception {
+
+    byte[] payloadApplyObjects =
+        ("{\"label\":\"Replication\",\"ref_name\":\"refs/heads/master\","
+                + "\"revisions_data\":[{"
+                + "\"commit_object\":{\"type\":1,\"content\":\"some-content\"},"
+                + "\"tree_object\":{\"type\":2,\"content\":\"some-content\"},"
+                + "\"blobs\":[]}]}")
+            .getBytes(StandardCharsets.UTF_8);
+
+    defineBehaviours(payloadApplyObjects, APPLY_OBJECTS_URI);
+
+    when(applyObjectsAction.apply(any(), any())).thenReturn(OK_RESPONSE);
+
+    PullReplicationFilter pullReplicationFilter = createPullReplicationFilter();
+    pullReplicationFilter.doFilter(request, response, filterChain);
+
+    verifyBehaviours();
+    verify(applyObjectsAction).apply(eq(projectResource), any());
+  }
+
+  @Test
+  public void shouldFilterProjectInitializationAction() throws Exception {
+
+    when(request.getRequestURI()).thenReturn(INIT_PROJECT_URI);
+    when(request.getHeader(ACCEPT)).thenReturn(MediaType.PLAIN_TEXT_UTF_8.toString());
+    when(projectInitializationAction.initProject(PROJECT_NAME_GIT)).thenReturn(true);
+    when(response.getWriter()).thenReturn(printWriter);
+
+    final PullReplicationFilter pullReplicationFilter = createPullReplicationFilter();
+    pullReplicationFilter.doFilter(request, response, filterChain);
+
+    verify(request, times(5)).getRequestURI();
+    verify(projectInitializationAction).initProject(eq(PROJECT_NAME_GIT));
+    verify(response).getWriter();
+  }
+
+  @Test
+  public void shouldFilterUpdateHEADAction() throws Exception {
+
+    byte[] payloadUpdateHead = "{\"ref\":\"some-ref\"}".getBytes(StandardCharsets.UTF_8);
+    defineBehaviours(payloadUpdateHead, HEAD_URI);
+    when(request.getMethod()).thenReturn("PUT");
+    when(updateHEADAction.apply(any(), any())).thenReturn(OK_RESPONSE);
+
+    final PullReplicationFilter pullReplicationFilter = createPullReplicationFilter();
+    pullReplicationFilter.doFilter(request, response, filterChain);
+
+    verifyBehaviours();
+    verify(updateHEADAction).apply(eq(projectResource), any());
+  }
+
+  @Test
+  public void shouldFilterProjectDeletionAction() throws Exception {
+    when(request.getRequestURI()).thenReturn(DELETE_PROJECT_URI);
+    when(request.getMethod()).thenReturn("DELETE");
+    when(projectsCollection.parse(TopLevelResource.INSTANCE, IdString.fromDecoded(PROJECT_NAME)))
+        .thenReturn(projectResource);
+    when(projectDeletionAction.apply(any(), any())).thenReturn(OK_RESPONSE);
+    when(response.getWriter()).thenReturn(printWriter);
+
+    final PullReplicationFilter pullReplicationFilter = createPullReplicationFilter();
+    pullReplicationFilter.doFilter(request, response, filterChain);
+
+    verify(request, times(7)).getRequestURI();
+    verify(projectsCollection).parse(TopLevelResource.INSTANCE, IdString.fromDecoded(PROJECT_NAME));
+    verify(projectDeletionAction).apply(eq(projectResource), any());
+    verify(response).getWriter();
+    verify(response).setContentType("application/json");
+    verify(response).setStatus(OK_RESPONSE.statusCode());
+  }
+
+  @Test
+  public void shouldGoNextInChainWhenUriDoesNotMatch() throws Exception {
+    when(request.getRequestURI()).thenReturn("any-url");
+    final PullReplicationFilter pullReplicationFilter = createPullReplicationFilter();
+    pullReplicationFilter.doFilter(request, response, filterChain);
+    verify(filterChain).doFilter(request, response);
+  }
+
+  @Test
+  public void shouldBe404WhenJsonIsMalformed() throws Exception {
+    byte[] payloadMalformedJson = "some-json-malformed".getBytes(StandardCharsets.UTF_8);
+    InputStream is = new ByteArrayInputStream(payloadMalformedJson);
+    BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(is));
+    when(request.getRequestURI()).thenReturn(FETCH_URI);
+    when(request.getReader()).thenReturn(bufferedReader);
+    when(response.getOutputStream()).thenReturn(outputStream);
+
+    PullReplicationFilter pullReplicationFilter = createPullReplicationFilter();
+    pullReplicationFilter.doFilter(request, response, filterChain);
+
+    verify(response).setStatus(HttpServletResponse.SC_BAD_REQUEST);
+  }
+
+  @Test
+  public void shouldBe500WhenProjectCannotBeInitiated() throws Exception {
+    when(request.getRequestURI()).thenReturn(INIT_PROJECT_URI);
+    when(request.getHeader(ACCEPT)).thenReturn(MediaType.PLAIN_TEXT_UTF_8.toString());
+    when(projectInitializationAction.initProject(PROJECT_NAME_GIT)).thenReturn(false);
+    when(response.getOutputStream()).thenReturn(outputStream);
+
+    final PullReplicationFilter pullReplicationFilter = createPullReplicationFilter();
+    pullReplicationFilter.doFilter(request, response, filterChain);
+
+    verify(response).setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);
+  }
+
+  @Test
+  public void shouldBe500WhenResourceNotFound() throws Exception {
+    when(request.getRequestURI()).thenReturn(DELETE_PROJECT_URI);
+    when(request.getMethod()).thenReturn("DELETE");
+    when(projectsCollection.parse(TopLevelResource.INSTANCE, IdString.fromDecoded(PROJECT_NAME)))
+        .thenReturn(projectResource);
+    when(projectDeletionAction.apply(any(), any()))
+        .thenThrow(new ResourceNotFoundException("resource not found"));
+    when(response.getOutputStream()).thenReturn(outputStream);
+
+    final PullReplicationFilter pullReplicationFilter = createPullReplicationFilter();
+    pullReplicationFilter.doFilter(request, response, filterChain);
+
+    verify(response).setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);
+  }
+
+  @Test
+  public void shouldBe403WhenUserIsNotAuthorised() throws Exception {
+    byte[] payloadFetchAction =
+        ("{"
+                + "\"label\":\"Replication\", "
+                + "\"ref_name\": \"refs/heads/master\", "
+                + "\"async\":false"
+                + "}")
+            .getBytes(StandardCharsets.UTF_8);
+
+    defineBehaviours(payloadFetchAction, FETCH_URI);
+    when(fetchAction.apply(any(), any()))
+        .thenThrow(new AuthException("The user is not authorised"));
+    when(response.getOutputStream()).thenReturn(outputStream);
+
+    PullReplicationFilter pullReplicationFilter = createPullReplicationFilter();
+    pullReplicationFilter.doFilter(request, response, filterChain);
+
+    verify(response).setStatus(HttpServletResponse.SC_FORBIDDEN);
+  }
+
+  @Test
+  public void shouldBe422WhenEntityCannotBeProcessed() throws Exception {
+    byte[] payloadFetchAction =
+        ("{"
+                + "\"label\":\"Replication\", "
+                + "\"ref_name\": \"refs/heads/master\", "
+                + "\"async\":false"
+                + "}")
+            .getBytes(StandardCharsets.UTF_8);
+
+    defineBehaviours(payloadFetchAction, FETCH_URI);
+    when(fetchAction.apply(any(), any()))
+        .thenThrow(new UnprocessableEntityException("Entity cannot be processed"));
+    when(response.getOutputStream()).thenReturn(outputStream);
+
+    PullReplicationFilter pullReplicationFilter = createPullReplicationFilter();
+    pullReplicationFilter.doFilter(request, response, filterChain);
+
+    verify(response).setStatus(SC_UNPROCESSABLE_ENTITY);
+  }
+
+  @Test
+  public void shouldBe409WhenThereIsResourceConflict() throws Exception {
+    when(request.getRequestURI()).thenReturn(DELETE_PROJECT_URI);
+    when(request.getMethod()).thenReturn("DELETE");
+    when(projectsCollection.parse(TopLevelResource.INSTANCE, IdString.fromDecoded(PROJECT_NAME)))
+        .thenReturn(projectResource);
+
+    when(projectDeletionAction.apply(any(), any()))
+        .thenThrow(new ResourceConflictException("Resource conflict"));
+    when(response.getOutputStream()).thenReturn(outputStream);
+
+    PullReplicationFilter pullReplicationFilter = createPullReplicationFilter();
+    pullReplicationFilter.doFilter(request, response, filterChain);
+
+    verify(response).setStatus(SC_CONFLICT);
+  }
+
+  @Test
+  public void shouldBe400WhenProjectNameIsNotPresentInURL() throws Exception {
+    when(request.getRequestURI())
+        .thenReturn(String.format("any-prefix/projects/%s~delete-project", PLUGIN_NAME));
+    when(request.getMethod()).thenReturn("DELETE");
+    when(response.getOutputStream()).thenReturn(outputStream);
+
+    PullReplicationFilter pullReplicationFilter = createPullReplicationFilter();
+    pullReplicationFilter.doFilter(request, response, filterChain);
+
+    verify(response).setStatus(HttpServletResponse.SC_BAD_REQUEST);
+  }
+}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/UpdateHeadActionIT.java b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/UpdateHeadActionIT.java
index aa07a7c..5355251 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/UpdateHeadActionIT.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/api/UpdateHeadActionIT.java
@@ -26,6 +26,7 @@
 import com.google.gson.Gson;
 import com.google.inject.Inject;
 import javax.servlet.http.HttpServletResponse;
+import org.apache.http.client.methods.HttpRequestBase;
 import org.junit.Test;
 
 public class UpdateHeadActionIT extends ActionITBase {
@@ -39,8 +40,7 @@
         .create(source)
         .execute(
             createPutRequest(headInput("some/branch")),
-            assertHttpResponseCode(HttpServletResponse.SC_UNAUTHORIZED),
-            getAnonymousContext());
+            assertHttpResponseCode(HttpServletResponse.SC_UNAUTHORIZED));
   }
 
   @Test
@@ -48,9 +48,8 @@
     httpClientFactory
         .create(source)
         .execute(
-            createPutRequest(headInput("")),
-            assertHttpResponseCode(HttpServletResponse.SC_BAD_REQUEST),
-            getContext());
+            withBasicAuthenticationAsAdmin(createPutRequest(headInput(""))),
+            assertHttpResponseCode(HttpServletResponse.SC_BAD_REQUEST));
   }
 
   @Test
@@ -61,13 +60,11 @@
     BranchInput input = new BranchInput();
     input.revision = master;
     gApi.projects().name(testProjectName).branch(newBranch).create(input);
-
     httpClientFactory
         .create(source)
         .execute(
-            createPutRequest(headInput(newBranch)),
-            assertHttpResponseCode(HttpServletResponse.SC_OK),
-            getContext());
+            withBasicAuthenticationAsAdmin(createPutRequest(headInput(newBranch))),
+            assertHttpResponseCode(HttpServletResponse.SC_OK));
 
     assertThat(gApi.projects().name(testProjectName).head()).isEqualTo(newBranch);
   }
@@ -78,9 +75,8 @@
     httpClientFactory
         .create(source)
         .execute(
-            createPutRequest(headInput("")),
-            assertHttpResponseCode(HttpServletResponse.SC_BAD_REQUEST),
-            getContext());
+            withBasicAuthenticationAsAdmin(createPutRequest(headInput(""))),
+            assertHttpResponseCode(HttpServletResponse.SC_BAD_REQUEST));
   }
 
   @Test
@@ -92,13 +88,11 @@
     BranchInput input = new BranchInput();
     input.revision = master;
     gApi.projects().name(testProjectName).branch(newBranch).create(input);
-
     httpClientFactory
         .create(source)
         .execute(
-            createPutRequest(headInput(newBranch)),
-            assertHttpResponseCode(HttpServletResponse.SC_OK),
-            getContext());
+            withBasicAuthenticationAsAdmin(createPutRequest(headInput(newBranch))),
+            assertHttpResponseCode(HttpServletResponse.SC_OK));
 
     assertThat(gApi.projects().name(testProjectName).head()).isEqualTo(newBranch);
   }
@@ -108,9 +102,8 @@
     httpClientFactory
         .create(source)
         .execute(
-            createPutRequest(headInput("some/new/head")),
-            assertHttpResponseCode(HttpServletResponse.SC_FORBIDDEN),
-            getUserContext());
+            withBasicAuthenticationAsUser(createPutRequest(headInput("some/new/head"))),
+            assertHttpResponseCode(HttpServletResponse.SC_FORBIDDEN));
   }
 
   @Test
@@ -121,13 +114,10 @@
     BranchInput input = new BranchInput();
     input.revision = master;
     gApi.projects().name(testProjectName).branch(newBranch).create(input);
-
+    HttpRequestBase put = withBasicAuthenticationAsUser(createPutRequest(headInput(newBranch)));
     httpClientFactory
         .create(source)
-        .execute(
-            createPutRequest(headInput(newBranch)),
-            assertHttpResponseCode(HttpServletResponse.SC_FORBIDDEN),
-            getUserContext());
+        .execute(put, assertHttpResponseCode(HttpServletResponse.SC_FORBIDDEN));
 
     projectOperations
         .project(project)
@@ -137,10 +127,7 @@
 
     httpClientFactory
         .create(source)
-        .execute(
-            createPutRequest(headInput(newBranch)),
-            assertHttpResponseCode(HttpServletResponse.SC_OK),
-            getUserContext());
+        .execute(put, assertHttpResponseCode(HttpServletResponse.SC_OK));
   }
 
   @Test
@@ -149,9 +136,8 @@
     httpClientFactory
         .create(source)
         .execute(
-            createPutRequest(headInput("some/new/head")),
-            assertHttpResponseCode(HttpServletResponse.SC_FORBIDDEN),
-            getUserContext());
+            withBasicAuthenticationAsUser(createPutRequest(headInput("some/new/head"))),
+            assertHttpResponseCode(HttpServletResponse.SC_FORBIDDEN));
   }
 
   private String headInput(String ref) {
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/client/FetchRestApiClientTest.java b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/client/FetchRestApiClientTest.java
index c45e7be..a6886f1 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/client/FetchRestApiClientTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/client/FetchRestApiClientTest.java
@@ -80,7 +80,7 @@
 
   String api = "http://gerrit-host";
   String pluginName = "pull-replication";
-  String label = "Replication";
+  String instanceId = "Replication";
   String refName = RefNames.REFS_HEADS + "master";
 
   String expectedPayload =
@@ -172,15 +172,20 @@
     when(replicationConfig.getConfig()).thenReturn(config);
     when(config.getStringList("replication", null, "syncRefs")).thenReturn(new String[0]);
     when(source.getRemoteConfigName()).thenReturn("Replication");
-    when(config.getString("replication", null, "instanceLabel")).thenReturn(label);
 
     HttpResult httpResult = new HttpResult(SC_CREATED, Optional.of("result message"));
-    when(httpClient.execute(any(HttpPost.class), any(), any())).thenReturn(httpResult);
+    when(httpClient.execute(any(HttpPost.class), any())).thenReturn(httpResult);
     when(httpClientFactory.create(any())).thenReturn(httpClient);
     syncRefsFilter = new SyncRefsFilter(replicationConfig);
     objectUnderTest =
         new FetchRestApiClient(
-            credentials, httpClientFactory, replicationConfig, syncRefsFilter, pluginName, source);
+            credentials,
+            httpClientFactory,
+            replicationConfig,
+            syncRefsFilter,
+            pluginName,
+            instanceId,
+            source);
   }
 
   @Test
@@ -189,7 +194,7 @@
 
     objectUnderTest.callFetch(Project.nameKey("test_repo"), refName, new URIish(api));
 
-    verify(httpClient, times(1)).execute(httpPostCaptor.capture(), any(), any());
+    verify(httpClient, times(1)).execute(httpPostCaptor.capture(), any());
 
     HttpPost httpPost = httpPostCaptor.getValue();
     assertThat(httpPost.getURI().getHost()).isEqualTo("gerrit-host");
@@ -204,11 +209,17 @@
     syncRefsFilter = new SyncRefsFilter(replicationConfig);
     objectUnderTest =
         new FetchRestApiClient(
-            credentials, httpClientFactory, replicationConfig, syncRefsFilter, pluginName, source);
+            credentials,
+            httpClientFactory,
+            replicationConfig,
+            syncRefsFilter,
+            pluginName,
+            instanceId,
+            source);
 
     objectUnderTest.callFetch(Project.nameKey("test_repo"), refName, new URIish(api));
 
-    verify(httpClient, times(1)).execute(httpPostCaptor.capture(), any(), any());
+    verify(httpClient, times(1)).execute(httpPostCaptor.capture(), any());
 
     HttpPost httpPost = httpPostCaptor.getValue();
     assertThat(readPayload(httpPost)).isEqualTo(expectedPayload);
@@ -223,11 +234,17 @@
     syncRefsFilter = new SyncRefsFilter(replicationConfig);
     objectUnderTest =
         new FetchRestApiClient(
-            credentials, httpClientFactory, replicationConfig, syncRefsFilter, pluginName, source);
+            credentials,
+            httpClientFactory,
+            replicationConfig,
+            syncRefsFilter,
+            pluginName,
+            instanceId,
+            source);
 
     objectUnderTest.callFetch(Project.nameKey("test_repo"), refName, new URIish(api));
 
-    verify(httpClient, times(1)).execute(httpPostCaptor.capture(), any(), any());
+    verify(httpClient, times(1)).execute(httpPostCaptor.capture(), any());
 
     HttpPost httpPost = httpPostCaptor.getValue();
     assertThat(readPayload(httpPost)).isEqualTo(expectedAsyncPayload);
@@ -245,15 +262,21 @@
     syncRefsFilter = new SyncRefsFilter(replicationConfig);
     objectUnderTest =
         new FetchRestApiClient(
-            credentials, httpClientFactory, replicationConfig, syncRefsFilter, pluginName, source);
+            credentials,
+            httpClientFactory,
+            replicationConfig,
+            syncRefsFilter,
+            pluginName,
+            instanceId,
+            source);
 
     objectUnderTest.callFetch(Project.nameKey("test_repo"), refName, new URIish(api));
-    verify(httpClient, times(1)).execute(httpPostCaptor.capture(), any(), any());
+    verify(httpClient, times(1)).execute(httpPostCaptor.capture(), any());
     HttpPost httpPost = httpPostCaptor.getValue();
     assertThat(readPayload(httpPost)).isEqualTo(expectedAsyncPayload);
 
     objectUnderTest.callFetch(Project.nameKey("test_repo"), metaRefName, new URIish(api));
-    verify(httpClient, times(2)).execute(httpPostCaptor.capture(), any(), any());
+    verify(httpClient, times(2)).execute(httpPostCaptor.capture(), any());
     httpPost = httpPostCaptor.getValue();
     assertThat(readPayload(httpPost)).isEqualTo(expectedMetaRefPayload);
   }
@@ -264,7 +287,7 @@
 
     objectUnderTest.callFetch(Project.nameKey("test_repo"), refName, new URIish(api));
 
-    verify(httpClient, times(1)).execute(httpPostCaptor.capture(), any(), any());
+    verify(httpClient, times(1)).execute(httpPostCaptor.capture(), any());
 
     HttpPost httpPost = httpPostCaptor.getValue();
     assertThat(readPayload(httpPost)).isEqualTo(expectedPayload);
@@ -276,7 +299,7 @@
 
     objectUnderTest.callFetch(Project.nameKey("test_repo"), refName, new URIish(api));
 
-    verify(httpClient, times(1)).execute(httpPostCaptor.capture(), any(), any());
+    verify(httpClient, times(1)).execute(httpPostCaptor.capture(), any());
 
     HttpPost httpPost = httpPostCaptor.getValue();
     assertThat(httpPost.getLastHeader("Content-Type").getValue())
@@ -294,7 +317,7 @@
         createSampleRevisionData(),
         new URIish(api));
 
-    verify(httpClient, times(1)).execute(httpPostCaptor.capture(), any(), any());
+    verify(httpClient, times(1)).execute(httpPostCaptor.capture(), any());
 
     HttpPost httpPost = httpPostCaptor.getValue();
     assertThat(httpPost.getURI().getHost()).isEqualTo("gerrit-host");
@@ -313,7 +336,7 @@
         createSampleRevisionData(),
         new URIish(api));
 
-    verify(httpClient, times(1)).execute(httpPostCaptor.capture(), any(), any());
+    verify(httpClient, times(1)).execute(httpPostCaptor.capture(), any());
 
     HttpPost httpPost = httpPostCaptor.getValue();
     assertThat(readPayload(httpPost)).isEqualTo(expectedSendObjectPayload);
@@ -325,7 +348,7 @@
 
     objectUnderTest.callFetch(Project.nameKey("test_repo"), refName, new URIish(api));
 
-    verify(httpClient, times(1)).execute(httpPostCaptor.capture(), any(), any());
+    verify(httpClient, times(1)).execute(httpPostCaptor.capture(), any());
 
     HttpPost httpPost = httpPostCaptor.getValue();
     assertThat(httpPost.getLastHeader("Content-Type").getValue())
@@ -334,7 +357,6 @@
 
   @Test
   public void shouldThrowExceptionWhenInstanceLabelIsNull() {
-    when(config.getString("replication", null, "instanceLabel")).thenReturn(null);
     assertThrows(
         NullPointerException.class,
         () ->
@@ -344,12 +366,12 @@
                 replicationConfig,
                 syncRefsFilter,
                 pluginName,
+                null,
                 source));
   }
 
   @Test
   public void shouldTrimInstanceLabel() {
-    when(config.getString("replication", null, "instanceLabel")).thenReturn(" ");
     assertThrows(
         NullPointerException.class,
         () ->
@@ -359,12 +381,12 @@
                 replicationConfig,
                 syncRefsFilter,
                 pluginName,
+                " ",
                 source));
   }
 
   @Test
   public void shouldThrowExceptionWhenInstanceLabelIsEmpty() {
-    when(config.getString("replication", null, "instanceLabel")).thenReturn("");
     assertThrows(
         NullPointerException.class,
         () ->
@@ -374,15 +396,37 @@
                 replicationConfig,
                 syncRefsFilter,
                 pluginName,
+                "",
                 source));
   }
 
   @Test
+  public void shouldUseReplicationLabelWhenProvided()
+      throws ClientProtocolException, IOException, URISyntaxException {
+    when(config.getString("replication", null, "instanceLabel")).thenReturn(instanceId);
+    FetchRestApiClient objectUnderTest =
+        new FetchRestApiClient(
+            credentials,
+            httpClientFactory,
+            replicationConfig,
+            syncRefsFilter,
+            pluginName,
+            "",
+            source);
+    objectUnderTest.callFetch(Project.nameKey("test_repo"), refName, new URIish(api));
+
+    verify(httpClient, times(1)).execute(httpPostCaptor.capture(), any());
+
+    HttpPost httpPost = httpPostCaptor.getValue();
+    assertThat(readPayload(httpPost)).isEqualTo(expectedPayload);
+  }
+
+  @Test
   public void shouldCallInitProjectEndpoint() throws IOException, URISyntaxException {
 
     objectUnderTest.initProject(Project.nameKey("test_repo"), new URIish(api));
 
-    verify(httpClient, times(1)).execute(httpPutCaptor.capture(), any(), any());
+    verify(httpClient, times(1)).execute(httpPutCaptor.capture(), any());
 
     HttpPut httpPut = httpPutCaptor.getValue();
     assertThat(httpPut.getURI().getHost()).isEqualTo("gerrit-host");
@@ -395,7 +439,7 @@
 
     objectUnderTest.deleteProject(Project.nameKey("test_repo"), new URIish(api));
 
-    verify(httpClient, times(1)).execute(httpDeleteCaptor.capture(), any(), any());
+    verify(httpClient, times(1)).execute(httpDeleteCaptor.capture(), any());
 
     HttpDelete httpDelete = httpDeleteCaptor.getValue();
     assertThat(httpDelete.getURI().getHost()).isEqualTo("gerrit-host");
@@ -409,7 +453,7 @@
     String projectName = "aProject";
     objectUnderTest.updateHead(Project.nameKey(projectName), newHead, new URIish(api));
 
-    verify(httpClient, times(1)).execute(httpPutCaptor.capture(), any(), any());
+    verify(httpClient, times(1)).execute(httpPutCaptor.capture(), any());
 
     HttpPut httpPut = httpPutCaptor.getValue();
     String payload =
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/event/StreamEventListenerTest.java b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/event/StreamEventListenerTest.java
new file mode 100644
index 0000000..c673011
--- /dev/null
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/event/StreamEventListenerTest.java
@@ -0,0 +1,150 @@
+// Copyright (C) 2022 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.googlesource.gerrit.plugins.replication.pull.event;
+
+import static com.google.common.truth.Truth.assertThat;
+import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.eq;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import com.google.gerrit.entities.Project;
+import com.google.gerrit.entities.RefNames;
+import com.google.gerrit.extensions.restapi.AuthException;
+import com.google.gerrit.server.data.RefUpdateAttribute;
+import com.google.gerrit.server.events.Event;
+import com.google.gerrit.server.events.ProjectCreatedEvent;
+import com.google.gerrit.server.events.RefUpdatedEvent;
+import com.google.gerrit.server.git.WorkQueue;
+import com.google.gerrit.server.permissions.PermissionBackendException;
+import com.googlesource.gerrit.plugins.replication.pull.FetchOne;
+import com.googlesource.gerrit.plugins.replication.pull.api.FetchAction.Input;
+import com.googlesource.gerrit.plugins.replication.pull.api.FetchJob;
+import com.googlesource.gerrit.plugins.replication.pull.api.ProjectInitializationAction;
+import com.googlesource.gerrit.plugins.replication.pull.api.PullReplicationApiRequestMetrics;
+import java.util.concurrent.ScheduledExecutorService;
+import org.eclipse.jgit.lib.ObjectId;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Captor;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+@RunWith(MockitoJUnitRunner.class)
+public class StreamEventListenerTest {
+
+  private static final String TEST_REF_NAME = "refs/changes/01/1/1";
+  private static final String TEST_PROJECT = "test-project";
+  private static final String INSTANCE_ID = "node_instance_id";
+  private static final String REMOTE_INSTANCE_ID = "remote_node_instance_id";
+
+  @Mock private ProjectInitializationAction projectInitializationAction;
+  @Mock private WorkQueue workQueue;
+  @Mock private ScheduledExecutorService executor;
+  @Mock private FetchJob fetchJob;
+  @Mock private FetchJob.Factory fetchJobFactory;
+  @Captor ArgumentCaptor<Input> inputCaptor;
+  @Mock private PullReplicationApiRequestMetrics metrics;
+
+  private StreamEventListener objectUnderTest;
+
+  @Before
+  public void setup() {
+    when(workQueue.getDefaultQueue()).thenReturn(executor);
+    when(fetchJobFactory.create(eq(Project.nameKey(TEST_PROJECT)), any(), any()))
+        .thenReturn(fetchJob);
+    objectUnderTest =
+        new StreamEventListener(
+            INSTANCE_ID, projectInitializationAction, workQueue, fetchJobFactory, () -> metrics);
+  }
+
+  @Test
+  public void shouldSkipEventsGeneratedByTheSameInstance() {
+    Event event = new RefUpdatedEvent();
+    event.instanceId = INSTANCE_ID;
+    objectUnderTest.onEvent(event);
+
+    verify(executor, never()).submit(any(Runnable.class));
+  }
+
+  @Test
+  public void shouldSkipFetchForProjectDeleteEvent() {
+    RefUpdatedEvent event = new RefUpdatedEvent();
+    RefUpdateAttribute refUpdate = new RefUpdateAttribute();
+    refUpdate.refName = RefNames.REFS_CONFIG;
+    refUpdate.newRev = ObjectId.zeroId().getName();
+    refUpdate.project = TEST_PROJECT;
+
+    event.instanceId = REMOTE_INSTANCE_ID;
+    event.refUpdate = () -> refUpdate;
+
+    objectUnderTest.onEvent(event);
+
+    verify(executor, never()).submit(any(Runnable.class));
+  }
+
+  @Test
+  public void shouldScheduleFetchJobForRefUpdateEvent() {
+    RefUpdatedEvent event = new RefUpdatedEvent();
+    RefUpdateAttribute refUpdate = new RefUpdateAttribute();
+    refUpdate.refName = TEST_REF_NAME;
+    refUpdate.project = TEST_PROJECT;
+
+    event.instanceId = REMOTE_INSTANCE_ID;
+    event.refUpdate = () -> refUpdate;
+
+    objectUnderTest.onEvent(event);
+
+    verify(fetchJobFactory).create(eq(Project.nameKey(TEST_PROJECT)), inputCaptor.capture(), any());
+
+    Input input = inputCaptor.getValue();
+    assertThat(input.label).isEqualTo(REMOTE_INSTANCE_ID);
+    assertThat(input.refName).isEqualTo(TEST_REF_NAME);
+
+    verify(executor).submit(any(FetchJob.class));
+  }
+
+  @Test
+  public void shouldCreateProjectForProjectCreatedEvent()
+      throws AuthException, PermissionBackendException {
+    ProjectCreatedEvent event = new ProjectCreatedEvent();
+    event.instanceId = REMOTE_INSTANCE_ID;
+    event.projectName = TEST_PROJECT;
+
+    objectUnderTest.onEvent(event);
+
+    verify(projectInitializationAction).initProject(String.format("%s.git", TEST_PROJECT));
+  }
+
+  @Test
+  public void shouldScheduleAllRefsFetchForProjectCreatedEvent() {
+    ProjectCreatedEvent event = new ProjectCreatedEvent();
+    event.instanceId = REMOTE_INSTANCE_ID;
+    event.projectName = TEST_PROJECT;
+
+    objectUnderTest.onEvent(event);
+
+    verify(fetchJobFactory).create(eq(Project.nameKey(TEST_PROJECT)), inputCaptor.capture(), any());
+
+    Input input = inputCaptor.getValue();
+    assertThat(input.label).isEqualTo(REMOTE_INSTANCE_ID);
+    assertThat(input.refName).isEqualTo(FetchOne.ALL_REFS);
+
+    verify(executor).submit(any(FetchJob.class));
+  }
+}