diff --git a/BUILD b/BUILD
index d7bd3d8..d95dc33 100644
--- a/BUILD
+++ b/BUILD
@@ -18,9 +18,10 @@
     ],
     resources = glob(["src/main/resources/**/*"]),
     deps = [
+        ":events-broker-neverlink",
+        ":global-refdb-neverlink",
+        ":pull-replication-neverlink",
         ":replication-neverlink",
-        "@events-broker//jar",
-        "@global-refdb//jar",
     ],
 )
 
@@ -30,6 +31,24 @@
     exports = ["//plugins/replication"],
 )
 
+java_library(
+    name = "pull-replication-neverlink",
+    neverlink = 1,
+    exports = ["//plugins/pull-replication"],
+)
+
+java_library(
+    name = "events-broker-neverlink",
+    neverlink = 1,
+    exports = ["//plugins/events-broker"],
+)
+
+java_library(
+    name = "global-refdb-neverlink",
+    neverlink = 1,
+    exports = ["//plugins/global-refdb"],
+)
+
 junit_tests(
     name = "multi_site_tests",
     srcs = glob(["src/test/java/**/*.java"]),
@@ -49,8 +68,9 @@
     visibility = ["//visibility:public"],
     exports = PLUGIN_DEPS + PLUGIN_TEST_DEPS + [
         ":multi-site__plugin",
-        "@global-refdb//jar",
-        "@events-broker//jar",
+        "//plugins/events-broker",
+        "//plugins/global-refdb",
+        "//plugins/pull-replication",
         "//plugins/replication",
     ],
 )
@@ -74,12 +94,6 @@
     srcs = [
         "e2e-tests/test.sh",
     ],
-    data = [
-        "//plugins/multi-site",
-        "//plugins/multi-site:e2e_multi_site_test_dir",
-        "//plugins/multi-site:e2e_multi_site_setup_local_env_dir",
-        "external_plugin_deps.bzl",
-    ] + glob(["setup_local_env/**/*"]) + glob(["e2e-tests/**/*"]),
     args = [
         "--multisite-lib-file $(location //plugins/multi-site)",
         "--healthcheck-interval 5s",
@@ -88,6 +102,11 @@
         "--location '$(location //plugins/multi-site:e2e_multi_site_test_dir)'",
         "--local-env '$(location //plugins/multi-site:e2e_multi_site_setup_local_env_dir)'",
     ],
+    data = [
+        "//plugins/multi-site",
+        "//plugins/multi-site:e2e_multi_site_test_dir",
+        "//plugins/multi-site:e2e_multi_site_setup_local_env_dir",
+    ] + glob(["setup_local_env/**/*"]) + glob(["e2e-tests/**/*"]),
     tags = [
         "e2e-multi-site",
     ],
diff --git a/Jenkinsfile b/Jenkinsfile
index f008749..6ef43af 100644
--- a/Jenkinsfile
+++ b/Jenkinsfile
@@ -1,2 +1,4 @@
 pluginPipeline(formatCheckId: 'gerritforge:multi-site-format-47168e90078b0b3f11401610930e82830e76bff7',
-               buildCheckId: 'gerritforge:multi-site-47168e90078b0b3f11401610930e82830e76bff7')
+               buildCheckId: 'gerritforge:multi-site-47168e90078b0b3f11401610930e82830e76bff7',
+               extraPlugins: [ 'pull-replication' ],
+               extraModules: [ 'events-broker', 'global-refdb' ])
diff --git a/README.md b/README.md
index 79094d3..f683f4e 100644
--- a/README.md
+++ b/README.md
@@ -40,8 +40,6 @@
 
 cd gerrit/plugins
 ln -s ../../multi-site .
-rm external_plugin_deps.bzl
-ln -s multi-site/external_plugin_deps.bzl .
 ```
 
 Example of building the multi-site plugin:
@@ -65,6 +63,21 @@
 daemon running (/var/run/docker.sock accessible) or a DOCKER_HOST pointing to
 a Docker server.
 
+## Pre-requisites
+
+Each Gerrit server of the cluster must be identified with a globally unique
+[instance-id](https://gerrit-documentation.storage.googleapis.com/Documentation/3.4.5/config-gerrit.html#gerrit.instanceId)
+defined in `$GERRIT_SITE/etc/gerrit.config`.
+When migrating from a multi-site configuration with Gerrit v3.3 or earlier,
+you must reuse the instance-id value stored under `$GERRIT_SITE/data/multi-site`.
+
+Example:
+
+```
+[gerrit]
+  instanceId = 758fe5b7-1869-46e6-942a-3ae0ae7e3bd2
+```
+
 ## How to configure
 
 Install the multi-site plugin into the `$GERRIT_SITE/lib` directory of all
diff --git a/e2e-tests/test.sh b/e2e-tests/test.sh
index 19c8b04..3ecb9fd 100755
--- a/e2e-tests/test.sh
+++ b/e2e-tests/test.sh
@@ -16,11 +16,11 @@
 
 LOCATION="$( cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null 2>&1 && pwd )"
 LOCAL_ENV="$( cd "${LOCATION}/../setup_local_env" >/dev/null 2>&1 && pwd )"
-GERRIT_BRANCH=stable-3.3
+GERRIT_BRANCH=stable-3.4
 GERRIT_CI=https://archive-ci.gerritforge.com/view/Plugins-$GERRIT_BRANCH/job
 LAST_BUILD=lastSuccessfulBuild/artifact/bazel-bin/plugins
 DEF_MULTISITE_LOCATION=${LOCATION}/../../../bazel-bin/plugins/multi-site/multi-site.jar
-DEF_GERRIT_IMAGE=3.3.6-centos8
+DEF_GERRIT_IMAGE=3.4.0-centos8
 DEF_GERRIT_HEALTHCHECK_START_PERIOD=60s
 DEF_GERRIT_HEALTHCHECK_INTERVAL=5s
 DEF_GERRIT_HEALTHCHECK_TIMEOUT=5s
@@ -64,7 +64,7 @@
 
   export BROKER_HOST=$2
   export BROKER_PORT=$3
-  export GROUP_ID=$4
+  export INSTANCE_ID=$4
   export SSH_ADVERTISED_PORT=$5
   export LOCATION_TEST_SITE=/var/gerrit
   export REMOTE_DEBUG_PORT=5005
@@ -196,8 +196,6 @@
 done
 
 # Defaults
-EVENTS_BROKER_VER=`grep 'com.gerritforge:events-broker' ${LOCATION}/../external_plugin_deps.bzl | cut -d '"' -f 2 | cut -d ':' -f 3`
-GLOBAL_REFDB_VER=`grep 'com.gerritforge:global-refdb' ${LOCATION}/../external_plugin_deps.bzl | cut -d '"' -f 2 | cut -d ':' -f 3`
 DEPLOYMENT_LOCATION=$(mktemp -d || $(echo >&2 "Could not create temp dir" && exit 1))
 MULTISITE_LIB_LOCATION=${MULTISITE_LIB_LOCATION:-${DEF_MULTISITE_LOCATION}}
 BROKER_TYPE=${BROKER_TYPE:-"kafka"}
@@ -261,13 +259,11 @@
 docker cp ${CONTAINER_NAME}:/var/gerrit/plugins/replication.jar $COMMON_LIBS/
 docker rm -fv ${CONTAINER_NAME}
 
-echo "Downloading global-refdb library $GERRIT_BRANCH"
-wget https://repo1.maven.org/maven2/com/gerritforge/global-refdb/$GLOBAL_REFDB_VER/global-refdb-$GLOBAL_REFDB_VER.jar \
-  -O $COMMON_LIBS/global-refdb.jar || { echo >&2 "Cannot download global-refdb library: Check internet connection. Aborting"; exit 1; }
+echo "Copying global-refdb library $GERRIT_BRANCH"
+cp bazel-bin/plugins/global-refdb/global-refdb.jar $COMMON_LIBS/global-refdb.jar
 
 echo "Downloading events-broker library $GERRIT_BRANCH"
-wget https://repo1.maven.org/maven2/com/gerritforge/events-broker/$EVENTS_BROKER_VER/events-broker-$EVENTS_BROKER_VER.jar \
-  -O $COMMON_LIBS/events-broker.jar || { echo >&2 "Cannot download events-broker library: Check internet connection. Aborting"; exit 1; }
+cp bazel-bin/plugins/events-broker/events-broker.jar $COMMON_LIBS/events-broker.jar
 
 echo "Setting up directories"
 mkdir -p ${GERRIT_1_ETC} ${GERRIT_1_PLUGINS} ${GERRIT_1_LIBS} ${GERRIT_2_ETC} ${GERRIT_2_PLUGINS} ${GERRIT_2_LIBS}
diff --git a/external_plugin_deps.bzl b/external_plugin_deps.bzl
deleted file mode 100644
index 223778e..0000000
--- a/external_plugin_deps.bzl
+++ /dev/null
@@ -1,14 +0,0 @@
-load("//tools/bzl:maven_jar.bzl", "maven_jar")
-
-def external_plugin_deps():
-    maven_jar(
-        name = "global-refdb",
-        artifact = "com.gerritforge:global-refdb:3.3.1",
-        sha1 = "5df9dddad2fc67c922406f41549186b210cd957e",
-    )
-
-    maven_jar(
-        name = "events-broker",
-        artifact = "com.gerritforge:events-broker:3.3.2",
-        sha1 = "d8bcb77047cc12dd7c623b5b4de70a25499d3d6c",
-    )
diff --git a/setup_local_env/README.md b/setup_local_env/README.md
index 88b40d5..e6ec9e7 100644
--- a/setup_local_env/README.md
+++ b/setup_local_env/README.md
@@ -74,8 +74,6 @@
 [--gerrit2-httpd-port]          Gerrit Instance 2 http port; default 18081
 [--gerrit2-sshd-port]           Gerrit Instance 2 sshd port; default 49418
 
-[--replication-type]            Options [file,ssh]; default file
-[--replication-ssh-user]        SSH user for the replication plugin; default $(whoami)
 [--replication-delay]           Replication delay across the two instances in seconds
 
 [--just-cleanup-env]            Cleans up previous deployment; default false
diff --git a/setup_local_env/configs/gerrit.config b/setup_local_env/configs/gerrit.config
index 2cd8c9d..8a176f2 100644
--- a/setup_local_env/configs/gerrit.config
+++ b/setup_local_env/configs/gerrit.config
@@ -5,6 +5,7 @@
     installModule = com.gerritforge.gerrit.eventbroker.BrokerApiModule # events-broker module to setup BrokerApi dynamic item
     installModule = com.googlesource.gerrit.plugins.multisite.Module # multi-site needs to be a gerrit lib
     installDbModule = com.googlesource.gerrit.plugins.multisite.GitModule
+    instanceId = $INSTANCE_ID
 [database]
     type = h2
     database = $LOCATION_TEST_SITE/db/ReviewDB
@@ -43,7 +44,7 @@
 [plugin "events-kafka"]
     sendAsync = true
     bootstrapServers = $BROKER_HOST:$BROKER_PORT
-    groupId = $GROUP_ID
+    groupId = $INSTANCE_ID
     numberOfSubscribers = 6
     securityProtocol = PLAINTEXT
     pollingIntervalMs = 1000
@@ -55,12 +56,12 @@
     pollingIntervalMs = 1000
     region = us-east-1
     endpoint = http://localhost:$BROKER_PORT
-    applicationName = $GROUP_ID
+    applicationName = $INSTANCE_ID
     initialPosition = trim_horizon
 [plugin "events-gcloud-pubsub"]
     numberOfSubscribers = 6
     gcloudProject="test-project"
-    subscriptionId=$GROUP_ID
+    subscriptionId=$INSTANCE_ID
     privateKeyLocation="not used in local mode"
 [plugin "metrics-reporter-prometheus"]
     prometheusBearerToken = token
diff --git a/setup_local_env/configs/replication.config b/setup_local_env/configs/replication.config
index ece1b3e..1c2aac4 100644
--- a/setup_local_env/configs/replication.config
+++ b/setup_local_env/configs/replication.config
@@ -1,6 +1,7 @@
-[remote "Replication"]
-    $REPLICATION_URL
-    push = +refs/*:refs/*
+[remote "$REPLICA_INSTANCE_ID"]
+    $PULL_REPLICATION_URL
+    $PULL_REPLICATION_API_URL
+    fetch = +refs/*:refs/*
     mirror = true
     timeout = 600
     rescheduleDelay = 15
@@ -13,4 +14,8 @@
     replicateOnStartup = false
 [replication]
     lockErrorMaxRetries = 5
-    maxRetries = 5
\ No newline at end of file
+    maxRetries = 5
+    useCGitClient = false
+    consumeStreamEvents = false
+    syncRefs="ALL REFS ASYNC"
+    maxApiPayloadSize=40000
diff --git a/setup_local_env/configs/secure.config b/setup_local_env/configs/secure.config
new file mode 100644
index 0000000..faeb659
--- /dev/null
+++ b/setup_local_env/configs/secure.config
@@ -0,0 +1,3 @@
+[auth]
+	registerEmailPrivateKey = A3D00Ny0bLW2z7xdInvIf35Q5mDCi5u9o/E=
+	bearertoken = some-bearer-token
diff --git a/setup_local_env/setup.sh b/setup_local_env/setup.sh
index 11ae04f..247a721 100755
--- a/setup_local_env/setup.sh
+++ b/setup_local_env/setup.sh
@@ -16,17 +16,15 @@
 
 
 SCRIPT_DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null 2>&1 && pwd )"
-GERRIT_BRANCH=stable-3.3
+GERRIT_BRANCH=stable-3.4
 GERRIT_CI=https://archive-ci.gerritforge.com/view/Plugins-$GERRIT_BRANCH/job
 LAST_BUILD=lastSuccessfulBuild/artifact/bazel-bin/plugins
-EVENTS_BROKER_VER=`grep 'com.gerritforge:events-broker' $(dirname $0)/../external_plugin_deps.bzl | cut -d '"' -f 2 | cut -d ':' -f 3`
-GLOBAL_REFDB_VER=`grep 'com.gerritforge:global-refdb' $(dirname $0)/../external_plugin_deps.bzl | cut -d '"' -f 2 | cut -d ':' -f 3`
 
 function check_application_requirements {
   type haproxy >/dev/null 2>&1 || { echo >&2 "Require haproxy but it's not installed. Aborting."; exit 1; }
   type java >/dev/null 2>&1 || { echo >&2 "Require java but it's not installed. Aborting."; exit 1; }
   type docker >/dev/null 2>&1 || { echo >&2 "Require docker but it's not installed. Aborting."; exit 1; }
-  type docker-compose >/dev/null 2>&1 || { echo >&2 "Require docker-compose but it's not installed. Aborting."; exit 1; }
+  [ $($SUDO docker --version | awk '{print $3}' | cut -d '.' -f 1) -ge 20 ] || { echo >&2 "Require docker v20 or later. Aborting."; exit 1; }
   type wget >/dev/null 2>&1 || { echo >&2 "Require wget but it's not installed. Aborting."; exit 1; }
   type envsubst >/dev/null 2>&1 || { echo >&2 "Require envsubst but it's not installed. Aborting."; exit 1; }
   type openssl >/dev/null 2>&1 || { echo >&2 "Require openssl but it's not installed. Aborting."; exit 1; }
@@ -35,16 +33,16 @@
   fi
 }
 
-function get_replication_url {
-  REPLICATION_LOCATION_TEST_SITE=$1
-  REPLICATION_HOSTNAME=$2
-  USER=$REPLICATION_SSH_USER
+function get_pull_replication_api_url {
+  REPLICATION_HOSTNAME=$1
 
-  if [ "$REPLICATION_TYPE" = "file" ];then
-    echo "url = file://$REPLICATION_LOCATION_TEST_SITE/git/#{name}#.git"
-  elif [ "$REPLICATION_TYPE" = "ssh" ];then
-    echo "url = ssh://$USER@$REPLICATION_HOSTNAME:$REPLICATION_LOCATION_TEST_SITE/git/#{name}#.git"
-  fi
+  echo "apiUrl = http://$REPLICATION_HOSTNAME:$REPLICATION_HTTPD_PORT"
+}
+
+function get_pull_replication_url {
+  REPLICATION_HOSTNAME=$1
+
+  echo "url = http://$REPLICATION_HOSTNAME:$REPLICATION_HTTPD_PORT/#{name}#.git"
 }
 
 function deploy_tls_certificates {
@@ -66,12 +64,13 @@
     export LOCATION_TEST_SITE=$3
     export GERRIT_SSHD_PORT=$4
     export REPLICATION_HTTPD_PORT=$5
-    export REPLICATION_LOCATION_TEST_SITE=$6
-    export GERRIT_HOSTNAME=$7
-    export REPLICATION_HOSTNAME=$8
-    export REMOTE_DEBUG_PORT=$9
-    export GROUP_ID=${10}
-    export REPLICATION_URL=$(get_replication_url $REPLICATION_LOCATION_TEST_SITE $REPLICATION_HOSTNAME)
+    export GERRIT_HOSTNAME=$6
+    export REPLICATION_HOSTNAME=$7
+    export REMOTE_DEBUG_PORT=$8
+    export INSTANCE_ID=${9}
+    export REPLICA_INSTANCE_ID=${10}
+    export PULL_REPLICATION_URL=$(get_pull_replication_url $REPLICATION_HOSTNAME)
+    export PULL_REPLICATION_API_URL=$(get_pull_replication_api_url $REPLICATION_HOSTNAME)
 
     echo "Replacing variables for file $file and copying to $CONFIG_TEST_SITE/$file_name"
 
@@ -126,25 +125,25 @@
   GERRIT_SITE1_SSHD_PORT=$3
   CONFIG_TEST_SITE_1=$LOCATION_TEST_SITE_1/etc
   GERRIT_SITE1_REMOTE_DEBUG_PORT="5005"
-  GERRIT_SITE1_GROUP_ID="instance-1"
+  GERRIT_SITE1_INSTANCE_ID="instance-1"
   # SITE 2
   GERRIT_SITE2_HOSTNAME=$4
   GERRIT_SITE2_HTTPD_PORT=$5
   GERRIT_SITE2_SSHD_PORT=$6
   CONFIG_TEST_SITE_2=$LOCATION_TEST_SITE_2/etc
   GERRIT_SITE2_REMOTE_DEBUG_PORT="5006"
-  GERRIT_SITE2_GROUP_ID="instance-2"
+  GERRIT_SITE2_INSTANCE_ID="instance-2"
 
   # Set config SITE1
-  copy_config_files $CONFIG_TEST_SITE_1 $GERRIT_SITE1_HTTPD_PORT $LOCATION_TEST_SITE_1 $GERRIT_SITE1_SSHD_PORT $GERRIT_SITE2_HTTPD_PORT $LOCATION_TEST_SITE_2 $GERRIT_SITE1_HOSTNAME $GERRIT_SITE2_HOSTNAME $GERRIT_SITE1_REMOTE_DEBUG_PORT $GERRIT_SITE1_GROUP_ID
+  copy_config_files $CONFIG_TEST_SITE_1 $GERRIT_SITE1_HTTPD_PORT $LOCATION_TEST_SITE_1 $GERRIT_SITE1_SSHD_PORT $GERRIT_SITE2_HTTPD_PORT $GERRIT_SITE1_HOSTNAME $GERRIT_SITE2_HOSTNAME $GERRIT_SITE1_REMOTE_DEBUG_PORT $GERRIT_SITE1_INSTANCE_ID $GERRIT_SITE2_INSTANCE_ID
 
 
   # Set config SITE2
-  copy_config_files $CONFIG_TEST_SITE_2 $GERRIT_SITE2_HTTPD_PORT $LOCATION_TEST_SITE_2 $GERRIT_SITE2_SSHD_PORT $GERRIT_SITE1_HTTPD_PORT $LOCATION_TEST_SITE_1 $GERRIT_SITE1_HOSTNAME $GERRIT_SITE2_HOSTNAME $GERRIT_SITE2_REMOTE_DEBUG_PORT $GERRIT_SITE2_GROUP_ID
+  copy_config_files $CONFIG_TEST_SITE_2 $GERRIT_SITE2_HTTPD_PORT $LOCATION_TEST_SITE_2 $GERRIT_SITE2_SSHD_PORT $GERRIT_SITE1_HTTPD_PORT $GERRIT_SITE1_HOSTNAME $GERRIT_SITE2_HOSTNAME $GERRIT_SITE2_REMOTE_DEBUG_PORT $GERRIT_SITE2_INSTANCE_ID $GERRIT_SITE1_INSTANCE_ID
 }
 
 function is_docker_desktop {
-  echo $(docker info | grep "Operating System: Docker Desktop" | wc -l)
+  echo $($SUDO docker info | grep "Operating System: Docker Desktop" | wc -l)
 }
 
 function docker_host_env {
@@ -183,9 +182,10 @@
   kill $(ps -ax | grep haproxy | grep "gerrit_setup/ha-proxy-config" | awk '{print $1}') 2> /dev/null
 
   echo "Stopping $BROKER_TYPE docker container"
-  docker-compose -f "${SCRIPT_DIR}/docker-compose-${BROKER_TYPE}.yaml" down 2> /dev/null
+  printenv > "${SCRIPT_DIR}/docker-compose-${BROKER_TYPE}.env"
+  $SUDO docker compose -f "${SCRIPT_DIR}/docker-compose-${BROKER_TYPE}.yaml" --env-file "${SCRIPT_DIR}/docker-compose-${BROKER_TYPE}.env" down 2> /dev/null
   echo "Stopping core docker containers"
-  docker-compose -f "${SCRIPT_DIR}/docker-compose-core.yaml" down 2> /dev/null
+  $SUDO docker compose -f "${SCRIPT_DIR}/docker-compose-core.yaml" --env-file "${SCRIPT_DIR}/docker-compose-${BROKER_TYPE}.env" down 2> /dev/null
 
   echo "Stopping GERRIT instances"
   $1/bin/gerrit.sh stop 2> /dev/null
@@ -197,7 +197,7 @@
 
 function check_if_container_is_running {
   local container=$1;
-  echo $(docker inspect "$container" 2> /dev/null | grep '"Running": true' | wc -l)
+  echo $($SUDO docker inspect "$container" 2> /dev/null | grep '"Running": true' | wc -l)
 }
 
 function ensure_docker_compose_is_up_and_running {
@@ -207,8 +207,9 @@
 
   local is_container_running=$(check_if_container_is_running "$container_name")
   if [ "$is_container_running" -lt 1 ];then
+    printenv > "${SCRIPT_DIR}/${docker_compose_file}.env"
     echo "[$log_label] Starting docker containers"
-    docker-compose -f "${SCRIPT_DIR}/${docker_compose_file}" up -d
+    $SUDO docker compose -f "${SCRIPT_DIR}/${docker_compose_file}" --env-file "${SCRIPT_DIR}/${docker_compose_file}.env" up -d
 
     echo "[$log_label] Waiting for docker containers to start..."
     while [[ $(check_if_container_is_running "$container_name") -lt 1 ]];do sleep 10s; done
@@ -231,6 +232,8 @@
     echo
     echo "[--release-war-file]            Location to release.war file"
     echo "[--multisite-lib-file]          Location to lib multi-site.jar file"
+    echo "[--eventsbroker-lib-file]       Location to lib events-broker.jar file"
+    echo "[--globalrefdb-lib-file]        Location to lib global-refdb.jar file"
     echo
     echo "[--new-deployment]              Cleans up previous gerrit deployment and re-installs it. default true"
     echo "[--get-websession-plugin]       Download websession-broker plugin from CI lastSuccessfulBuild; default true"
@@ -247,8 +250,6 @@
     echo "[--gerrit2-httpd-port]          Gerrit Instance 2 http port; default 18081"
     echo "[--gerrit2-sshd-port]           Gerrit Instance 2 sshd port; default 49418"
     echo
-    echo "[--replication-type]            Options [file,ssh]; default ssh"
-    echo "[--replication-ssh-user]        SSH user for the replication plugin; default $(whoami)"
     echo "[--replication-delay]           Replication delay across the two instances in seconds"
     echo
     echo "[--just-cleanup-env]            Cleans up previous deployment; default false"
@@ -257,6 +258,8 @@
     echo
     echo "[--broker-type]                 events broker type; 'kafka', 'kinesis' or 'gcloud-pubsub'. Default 'kafka'"
     echo
+    echo "[--sudo]                        run docker commands with sudo"
+    echo
     exit 0
   ;;
   "--new-deployment")
@@ -284,6 +287,16 @@
     shift
     shift
   ;;
+  "--eventsbroker-lib-file" )
+    EVENTS_BROKER_LIB_LOCATION=$2
+    shift
+    shift
+  ;;
+  "--globalrefdb-lib-file" )
+    GLOBAL_REFDB_LIB_LOCATION=$2
+    shift
+    shift
+  ;;
   "--gerrit-canonical-host" )
     export GERRIT_CANONICAL_HOSTNAME=$2
     shift
@@ -319,16 +332,6 @@
     shift
     shift
   ;;
-  "--replication-ssh-user" )
-    export REPLICATION_SSH_USER=$2
-    shift
-    shift
-  ;;
-  "--replication-type")
-    export REPLICATION_TYPE=$2
-    shift
-    shift
-  ;;
   "--replication-delay")
     export REPLICATION_DELAY_SEC=$2
     shift
@@ -353,6 +356,11 @@
       exit 1
     fi
   ;;
+  "--sudo" )
+    SUDO=sudo
+    shift
+    shift
+  ;;
   *     )
     echo "Unknown option argument: $1"
     shift
@@ -376,8 +384,6 @@
 GERRIT_2_HTTPD_PORT=${GERRIT_2_HTTPD_PORT:-"18081"}
 GERRIT_1_SSHD_PORT=${GERRIT_1_SSHD_PORT:-"39418"}
 GERRIT_2_SSHD_PORT=${GERRIT_2_SSHD_PORT:-"49418"}
-REPLICATION_TYPE=${REPLICATION_TYPE:-"file"}
-REPLICATION_SSH_USER=${REPLICATION_SSH_USER:-$(whoami)}
 export REPLICATION_DELAY_SEC=${REPLICATION_DELAY_SEC:-"5"}
 export SSH_ADVERTISED_PORT=${SSH_ADVERTISED_PORT:-"29418"}
 HTTPS_ENABLED=${HTTPS_ENABLED:-"false"}
@@ -392,7 +398,8 @@
 
 RELEASE_WAR_FILE_LOCATION=${RELEASE_WAR_FILE_LOCATION:-bazel-bin/release.war}
 MULTISITE_LIB_LOCATION=${MULTISITE_LIB_LOCATION:-bazel-bin/plugins/multi-site/multi-site.jar}
-
+EVENTS_BROKER_LIB_LOCATION=${EVENTS_BROKER_LIB_LOCATION:-bazel-bin/plugins/events-broker/events-broker.jar}
+GLOBAL_REFDB_LIB_LOCATION=${GLOBAL_REFDB_LIB_LOCATION:-bazel-bin/plugins/global-refdb/global-refdb.jar}
 
 export FAKE_NFS=$COMMON_LOCATION/fake_nfs
 
@@ -413,10 +420,20 @@
 else
   cp -f $MULTISITE_LIB_LOCATION $DEPLOYMENT_LOCATION/multi-site.jar  >/dev/null 2>&1 || { echo >&2 "$MULTISITE_LIB_LOCATION: Not able to copy the file. Aborting"; exit 1; }
 fi
+
+echo "Copying events-broker library"
+  cp -f $EVENTS_BROKER_LIB_LOCATION $DEPLOYMENT_LOCATION/events-broker.jar  >/dev/null 2>&1 || { echo >&2 "$EVENTS_BROKER_LIB_LOCATION: Not able to copy the file. Aborting"; exit 1; }
+
+echo "Copying global-refdb library"
+  cp -f $GLOBAL_REFDB_LIB_LOCATION $DEPLOYMENT_LOCATION/global-refdb.jar  >/dev/null 2>&1 || { echo >&2 "$GLOBAL_REFDB_LIB_LOCATION: Not able to copy the file. Aborting"; exit 1; }
+
 if [ $DOWNLOAD_WEBSESSION_PLUGIN = "true" ];then
   echo "Downloading websession-broker plugin $GERRIT_BRANCH"
   wget $GERRIT_CI/plugin-websession-broker-bazel-$GERRIT_BRANCH/$LAST_BUILD/websession-broker/websession-broker.jar \
-  -O $DEPLOYMENT_LOCATION/websession-broker.jar || { echo >&2 "Cannot download websession-broker plugin: Check internet connection. Abort\
+  -O $DEPLOYMENT_LOCATION/websession-broker.jar || \
+  wget $GERRIT_CI/plugin-websession-broker-bazel-master-$GERRIT_BRANCH/$LAST_BUILD/websession-broker/websession-broker.jar \
+  -O $DEPLOYMENT_LOCATION/websession-broker.jar || \
+  { echo >&2 "Cannot download websession-broker plugin: Check internet connection. Abort\
 ing"; exit 1; }
   wget $GERRIT_CI/plugin-healthcheck-bazel-$GERRIT_BRANCH/$LAST_BUILD/healthcheck/healthcheck.jar \
   -O $DEPLOYMENT_LOCATION/healthcheck.jar || { echo >&2 "Cannot download healthcheck plugin: Check internet connection. Abort\
@@ -427,30 +444,29 @@
 
 echo "Downloading zookeeper plugin $GERRIT_BRANCH"
   wget $GERRIT_CI/plugin-zookeeper-refdb-bazel-$GERRIT_BRANCH/$LAST_BUILD/zookeeper-refdb/zookeeper-refdb.jar \
-  -O $DEPLOYMENT_LOCATION/zookeeper-refdb.jar || { echo >&2 "Cannot download zookeeper plugin: Check internet connection. Abort\
-ing"; exit 1; }
-
-echo "Downloading global-refdb library $GERRIT_BRANCH"
-  wget https://repo1.maven.org/maven2/com/gerritforge/global-refdb/$GLOBAL_REFDB_VER/global-refdb-$GLOBAL_REFDB_VER.jar \
-  -O $DEPLOYMENT_LOCATION/global-refdb.jar || { echo >&2 "Cannot download global-refdb library: Check internet connection. Abort\
-ing"; exit 1; }
-
-echo "Downloading events-broker library $GERRIT_BRANCH"
-  wget https://repo1.maven.org/maven2/com/gerritforge/events-broker/$EVENTS_BROKER_VER/events-broker-$EVENTS_BROKER_VER.jar \
-  -O $DEPLOYMENT_LOCATION/events-broker.jar || { echo >&2 "Cannot download events-broker library: Check internet connection. Abort\
+  -O $DEPLOYMENT_LOCATION/zookeeper-refdb.jar || \
+  wget $GERRIT_CI/plugin-zookeeper-refdb-bazel-master-$GERRIT_BRANCH/$LAST_BUILD/zookeeper-refdb/zookeeper-refdb.jar \
+  -O $DEPLOYMENT_LOCATION/zookeeper-refdb.jar || \
+  { echo >&2 "Cannot download zookeeper plugin: Check internet connection. Abort\
 ing"; exit 1; }
 
 if [ "$BROKER_TYPE" = "kafka" ]; then
 echo "Downloading events-kafka plugin $GERRIT_BRANCH"
   wget $GERRIT_CI/plugin-events-kafka-bazel-$GERRIT_BRANCH/$LAST_BUILD/events-kafka/events-kafka.jar \
-  -O $DEPLOYMENT_LOCATION/events-kafka.jar || { echo >&2 "Cannot download events-kafka plugin: Check internet connection. Abort\
+  -O $DEPLOYMENT_LOCATION/events-kafka.jar || \
+  wget $GERRIT_CI/plugin-events-kafka-bazel-master-$GERRIT_BRANCH/$LAST_BUILD/events-kafka/events-kafka.jar \
+  -O $DEPLOYMENT_LOCATION/events-kafka.jar || \
+  { echo >&2 "Cannot download events-kafka plugin: Check internet connection. Abort\
 ing"; exit 1; }
 fi
 
 if [ "$BROKER_TYPE" = "kinesis" ]; then
 echo "Downloading events-aws-kinesis plugin $GERRIT_BRANCH"
   wget $GERRIT_CI/plugin-events-aws-kinesis-bazel-$GERRIT_BRANCH/$LAST_BUILD/events-aws-kinesis/events-aws-kinesis.jar \
-  -O $DEPLOYMENT_LOCATION/events-aws-kinesis.jar || { echo >&2 "Cannot download events-aws-kinesis plugin: Check internet connection. Abort\
+  -O $DEPLOYMENT_LOCATION/events-aws-kinesis.jar || \
+  wget $GERRIT_CI/plugin-events-aws-kinesis-bazel-master-$GERRIT_BRANCH/$LAST_BUILD/events-aws-kinesis/events-aws-kinesis.jar \
+  -O $DEPLOYMENT_LOCATION/events-aws-kinesis.jar || \
+  { echo >&2 "Cannot download events-aws-kinesis plugin: Check internet connection. Abort\
 ing"; exit 1; }
 fi
 
@@ -458,19 +474,25 @@
 if [ "$BROKER_TYPE" = "gcloud-pubsub" ]; then
 echo "Downloading events-gcloud-pubsub plugin $GERRIT_BRANCH"
   wget $GERRIT_CI/plugin-events-gcloud-pubsub-bazel-$GERRIT_BRANCH/$LAST_BUILD/events-gcloud-pubsub/events-gcloud-pubsub.jar \
-  -O $DEPLOYMENT_LOCATION/events-gcloud-pubsub.jar || { echo >&2 "Cannot download events-gcloud-pubsub plugin: Check internet connection. Abort\
+  -O $DEPLOYMENT_LOCATION/events-gcloud-pubsub.jar || \
+  wget $GERRIT_CI/plugin-events-gcloud-pubsub-bazel-master-$GERRIT_BRANCH/$LAST_BUILD/events-gcloud-pubsub/events-gcloud-pubsub.jar \
+  -O $DEPLOYMENT_LOCATION/events-gcloud-pubsub.jar || \
+  { echo >&2 "Cannot download events-gcloud-pubsub plugin: Check internet connection. Abort\
 ing"; exit 1; }
 fi
 
 echo "Downloading metrics-reporter-prometheus plugin $GERRIT_BRANCH"
+  wget $GERRIT_CI/plugin-metrics-reporter-prometheus-bazel-$GERRIT_BRANCH/$LAST_BUILD/metrics-reporter-prometheus/metrics-reporter-prometheus.jar \
+  -O $DEPLOYMENT_LOCATION/metrics-reporter-prometheus.jar || \
   wget $GERRIT_CI/plugin-metrics-reporter-prometheus-bazel-master-$GERRIT_BRANCH/$LAST_BUILD/metrics-reporter-prometheus/metrics-reporter-prometheus.jar \
-  -O $DEPLOYMENT_LOCATION/metrics-reporter-prometheus.jar || { echo >&2 "Cannot download metrics-reporter-prometheus plugin: Check internet connection. Abort\
+  -O $DEPLOYMENT_LOCATION/metrics-reporter-prometheus.jar || \
+  { echo >&2 "Cannot download metrics-reporter-prometheus plugin: Check internet connection. Abort\
 ing"; exit 1; }
 
-if [ "$REPLICATION_TYPE" = "ssh" ];then
-  echo "Using 'SSH' replication type"
-  echo "Make sure ~/.ssh/authorized_keys and ~/.ssh/known_hosts are configured correctly"
-fi
+echo "Downloading pull-replication plugin $GERRIT_BRANCH"
+  wget $GERRIT_CI/plugin-pull-replication-bazel-$GERRIT_BRANCH/$LAST_BUILD/pull-replication/pull-replication.jar \
+  -O $DEPLOYMENT_LOCATION/pull-replication.jar || { echo >&2 "Cannot download pull-replication plugin: Check internet connection. Abort\
+ing"; exit 1; }
 
 if [ "$HTTPS_ENABLED" = "true" ];then
   export HTTP_PROTOCOL="https"
@@ -530,13 +552,18 @@
   echo "Replicating environment"
   cp -fR $LOCATION_TEST_SITE_1/* $LOCATION_TEST_SITE_2
 
-  echo "Link replication plugin"
-  ln -s $LOCATION_TEST_SITE_1/plugins/replication.jar $LOCATION_TEST_SITE_1/lib/replication.jar
-  ln -s $LOCATION_TEST_SITE_2/plugins/replication.jar $LOCATION_TEST_SITE_2/lib/replication.jar
+  echo "Link pullreplication plugin"
+  ln -s $LOCATION_TEST_SITE_1/plugins/pull-replication.jar $LOCATION_TEST_SITE_1/lib/pull-replication.jar
+  ln -s $LOCATION_TEST_SITE_2/plugins/pull-replication.jar $LOCATION_TEST_SITE_2/lib/pull-replication.jar
 
   echo "Link multi-site library to plugin directory"
   ln -s $LOCATION_TEST_SITE_1/lib/multi-site.jar $LOCATION_TEST_SITE_1/plugins/multi-site.jar
   ln -s $LOCATION_TEST_SITE_2/lib/multi-site.jar $LOCATION_TEST_SITE_2/plugins/multi-site.jar
+
+  echo "Copy pull-replication plugin"
+  cp -f $DEPLOYMENT_LOCATION/pull-replication.jar $LOCATION_TEST_SITE_1/plugins/pull-replication.jar
+  cp -f $DEPLOYMENT_LOCATION/pull-replication.jar $LOCATION_TEST_SITE_2/plugins/pull-replication.jar
+
 fi
 
 DOCKER_HOST_ENV=$(docker_host_env)
@@ -558,8 +585,12 @@
 
 echo "Re-deploying configuration files"
 deploy_config_files $GERRIT_1_HOSTNAME $GERRIT_1_HTTPD_PORT $GERRIT_1_SSHD_PORT $GERRIT_2_HOSTNAME $GERRIT_2_HTTPD_PORT $GERRIT_2_SSHD_PORT
+echo "Remove replication plugin from gerrit site 1"
+rm $LOCATION_TEST_SITE_1/plugins/replication.jar
 echo "Starting gerrit site 1"
 $LOCATION_TEST_SITE_1/bin/gerrit.sh restart
+echo "Remove replication plugin from gerrit site 2"
+rm $LOCATION_TEST_SITE_2/plugins/replication.jar
 echo "Starting gerrit site 2"
 $LOCATION_TEST_SITE_2/bin/gerrit.sh restart
 
@@ -574,8 +605,6 @@
 echo "==============================="
 echo "The admin password is 'secret'"
 echo "deployment-location=$DEPLOYMENT_LOCATION"
-echo "replication-type=$REPLICATION_TYPE"
-echo "replication-ssh-user=$REPLICATION_SSH_USER"
 echo "replication-delay=$REPLICATION_DELAY_SEC"
 echo "enable-https=$HTTPS_ENABLED"
 echo
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/Configuration.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/Configuration.java
index 6b77713..2d67eab 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/Configuration.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/Configuration.java
@@ -23,15 +23,19 @@
 import com.google.common.base.Strings;
 import com.google.common.base.Supplier;
 import com.google.common.collect.ImmutableList;
+import com.google.gerrit.server.config.ConfigUtil;
 import com.google.gerrit.server.config.SitePaths;
 import com.google.inject.Inject;
 import com.google.inject.Singleton;
 import com.google.inject.spi.Message;
 import java.io.IOException;
+import java.time.Duration;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
+import java.util.Random;
+import java.util.concurrent.TimeUnit;
 import org.eclipse.jgit.errors.ConfigInvalidException;
 import org.eclipse.jgit.lib.Config;
 import org.eclipse.jgit.storage.file.FileBasedConfig;
@@ -50,6 +54,10 @@
   static final String THREAD_POOL_SIZE_KEY = "threadPoolSize";
   static final int DEFAULT_THREAD_POOL_SIZE = 4;
 
+  private static final String REF_DATABASE = "ref-database";
+  private static final String REPLICATION_LAG_REFRESH_INTERVAL = "replicationLagRefreshInterval";
+  private static final Duration REPLICATION_LAG_REFRESH_INTERVAL_DEFAULT = Duration.ofSeconds(60);
+
   private static final String REPLICATION_CONFIG = "replication.config";
   // common parameters to cache and index sections
   private static final int DEFAULT_INDEX_MAX_TRIES = 2;
@@ -64,7 +72,9 @@
   private final Supplier<SharedRefDbConfiguration> sharedRefDb;
   private final Supplier<Collection<Message>> replicationConfigValidation;
   private final Supplier<Broker> broker;
+  private final Supplier<ReplicationFilter> replicationFilter;
   private final Config multiSiteConfig;
+  private final Supplier<Duration> replicationLagRefreshInterval;
 
   @Inject
   Configuration(SitePaths sitePaths) {
@@ -86,6 +96,18 @@
                 new SharedRefDbConfiguration(
                     enableSharedRefDbByDefault(lazyMultiSiteCfg.get()), PLUGIN_NAME));
     broker = memoize(() -> new Broker(lazyMultiSiteCfg));
+    replicationFilter = memoize(() -> new ReplicationFilter(lazyMultiSiteCfg));
+    replicationLagRefreshInterval =
+        memoize(
+            () ->
+                Duration.ofMillis(
+                    ConfigUtil.getTimeUnit(
+                        lazyMultiSiteCfg.get(),
+                        REF_DATABASE,
+                        null,
+                        REPLICATION_LAG_REFRESH_INTERVAL,
+                        REPLICATION_LAG_REFRESH_INTERVAL_DEFAULT.toMillis(),
+                        TimeUnit.MILLISECONDS)));
   }
 
   public Config getMultiSiteConfig() {
@@ -116,6 +138,14 @@
     return projects.get();
   }
 
+  public ReplicationFilter replicationFilter() {
+    return replicationFilter.get();
+  }
+
+  public Duration replicationLagRefreshInterval() {
+    return replicationLagRefreshInterval.get();
+  }
+
   public Collection<Message> validate() {
     return replicationConfigValidation.get();
   }
@@ -311,6 +341,81 @@
     }
   }
 
+  public static class ReplicationFilter {
+    static final String REPLICATION_FILTER_SECTION = "replication";
+    static final String REPLICATION_PUSH_FILTER_SUBSECTION = "push-filter";
+    static final String REPLICATION_FETCH_FILTER_SUBSECTION = "fetch-filter";
+
+    private static final String MIN_WAIT_BEFORE_RELOAD_LOCAL_VERSION_MS =
+        "minWaitBeforeReloadLocalVersionMs";
+    private static final String RANDOM_WAIT_MAX_BOUND_BEFORE_RELOAD_LOCAL_VERSION_MS =
+        "maxRandomWaitBeforeReloadLocalVersionMs";
+
+    private static final int MIN_WAIT_BEFORE_RELOAD_LOCAL_VERSION_MS_DEFAULT = 1000;
+    private static final int RANDOM_WAIT_MAX_BOUND_BEFORE_RELOAD_LOCAL_VERSION_MS_DEFAULT = 1000;
+    private final Supplier<Integer> fetchMinWaitBeforeReloadLocalVersionMs;
+    private final Supplier<Integer> fetchWaitBeforeReloadLocalVersionMs;
+    private final Supplier<Integer> pushMinWaitBeforeReloadLocalVersionMs;
+    private final Supplier<Integer> pushWaitBeforeReloadLocalVersionMs;
+
+    public ReplicationFilter(Supplier<Config> cfg) {
+      fetchMinWaitBeforeReloadLocalVersionMs =
+          memoize(
+              () ->
+                  cfg.get()
+                      .getInt(
+                          REPLICATION_FILTER_SECTION,
+                          REPLICATION_FETCH_FILTER_SUBSECTION,
+                          MIN_WAIT_BEFORE_RELOAD_LOCAL_VERSION_MS,
+                          MIN_WAIT_BEFORE_RELOAD_LOCAL_VERSION_MS_DEFAULT));
+      fetchWaitBeforeReloadLocalVersionMs =
+          memoize(
+              () ->
+                  cfg.get()
+                      .getInt(
+                          REPLICATION_FILTER_SECTION,
+                          REPLICATION_FETCH_FILTER_SUBSECTION,
+                          RANDOM_WAIT_MAX_BOUND_BEFORE_RELOAD_LOCAL_VERSION_MS,
+                          RANDOM_WAIT_MAX_BOUND_BEFORE_RELOAD_LOCAL_VERSION_MS_DEFAULT));
+      pushMinWaitBeforeReloadLocalVersionMs =
+          memoize(
+              () ->
+                  cfg.get()
+                      .getInt(
+                          REPLICATION_FILTER_SECTION,
+                          REPLICATION_PUSH_FILTER_SUBSECTION,
+                          MIN_WAIT_BEFORE_RELOAD_LOCAL_VERSION_MS,
+                          MIN_WAIT_BEFORE_RELOAD_LOCAL_VERSION_MS_DEFAULT));
+      pushWaitBeforeReloadLocalVersionMs =
+          memoize(
+              () ->
+                  cfg.get()
+                      .getInt(
+                          REPLICATION_FILTER_SECTION,
+                          REPLICATION_PUSH_FILTER_SUBSECTION,
+                          RANDOM_WAIT_MAX_BOUND_BEFORE_RELOAD_LOCAL_VERSION_MS,
+                          RANDOM_WAIT_MAX_BOUND_BEFORE_RELOAD_LOCAL_VERSION_MS_DEFAULT));
+    }
+
+    public boolean isFetchFilterRandomSleepEnabled() {
+      return fetchWaitBeforeReloadLocalVersionMs.get() != 0;
+    }
+
+    public Integer fetchFilterRandomSleepTimeMs() {
+      return fetchMinWaitBeforeReloadLocalVersionMs.get()
+          + new Random().nextInt(fetchWaitBeforeReloadLocalVersionMs.get());
+    }
+
+    public boolean isPushFilterRandomSleepEnabled() {
+      return pushWaitBeforeReloadLocalVersionMs.get() != 0;
+    }
+
+    public Integer pushFilterRandomSleepTimeMs() {
+      return pushMinWaitBeforeReloadLocalVersionMs.get()
+          + new Random().nextInt(pushWaitBeforeReloadLocalVersionMs.get());
+    }
+  }
+
   static boolean getBoolean(
       Supplier<Config> cfg, String section, String subsection, String name, boolean defaultValue) {
     try {
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/Log4jMessageLogger.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/Log4jMessageLogger.java
index 7c88655..95c417f 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/Log4jMessageLogger.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/Log4jMessageLogger.java
@@ -14,9 +14,9 @@
 
 package com.googlesource.gerrit.plugins.multisite;
 
-import com.gerritforge.gerrit.eventbroker.EventGsonProvider;
-import com.gerritforge.gerrit.eventbroker.EventMessage;
 import com.google.gerrit.extensions.systemstatus.ServerInformation;
+import com.google.gerrit.server.events.Event;
+import com.google.gerrit.server.events.EventGsonProvider;
 import com.google.gerrit.server.util.PluginLogFile;
 import com.google.gerrit.server.util.SystemLog;
 import com.google.gson.Gson;
@@ -41,7 +41,7 @@
   }
 
   @Override
-  public void log(Direction direction, String topic, EventMessage event) {
+  public void log(Direction direction, String topic, Event event) {
     msgLog.info("{} {} {}", direction, topic, gson.toJson(event));
   }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/Log4jProjectVersionLogger.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/Log4jProjectVersionLogger.java
index c2c4b46..23c720a 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/Log4jProjectVersionLogger.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/Log4jProjectVersionLogger.java
@@ -45,4 +45,9 @@
       verLog.info("{ \"project\":\"{}\", \"version\":{} }", projectName, currentVersion);
     }
   }
+
+  @Override
+  public void logDeleted(Project.NameKey projectName) {
+    verLog.info("{ \"project\":\"{}\", \"status\":\"DELETED\" }", projectName);
+  }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/MessageLogger.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/MessageLogger.java
index b1f3e79..cc64b02 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/MessageLogger.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/MessageLogger.java
@@ -14,7 +14,7 @@
 
 package com.googlesource.gerrit.plugins.multisite;
 
-import com.gerritforge.gerrit.eventbroker.EventMessage;
+import com.google.gerrit.server.events.Event;
 
 public interface MessageLogger {
 
@@ -23,5 +23,5 @@
     CONSUME;
   }
 
-  public void log(Direction direction, String topic, EventMessage event);
+  public void log(Direction direction, String topic, Event event);
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/Module.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/Module.java
index 91a927b..9392077 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/Module.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/Module.java
@@ -16,30 +16,17 @@
 
 import com.gerritforge.gerrit.globalrefdb.validation.LibModule;
 import com.google.gerrit.lifecycle.LifecycleModule;
-import com.google.gerrit.server.config.SitePaths;
 import com.google.inject.CreationException;
 import com.google.inject.Inject;
-import com.google.inject.Provides;
-import com.google.inject.Singleton;
 import com.google.inject.spi.Message;
+import com.googlesource.gerrit.plugins.multisite.broker.BrokerModule;
 import com.googlesource.gerrit.plugins.multisite.cache.CacheModule;
-import com.googlesource.gerrit.plugins.multisite.event.EventModule;
 import com.googlesource.gerrit.plugins.multisite.forwarder.ForwarderModule;
 import com.googlesource.gerrit.plugins.multisite.forwarder.router.RouterModule;
 import com.googlesource.gerrit.plugins.multisite.index.IndexModule;
-import java.io.BufferedReader;
-import java.io.BufferedWriter;
-import java.io.IOException;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.nio.file.Paths;
 import java.util.Collection;
-import java.util.UUID;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 public class Module extends LifecycleModule {
-  private static final Logger log = LoggerFactory.getLogger(Module.class);
   private Configuration config;
 
   @Inject
@@ -49,6 +36,7 @@
 
   @Override
   protected void configure() {
+    boolean brokerRouterNeeded = false;
 
     Collection<Message> validationErrors = config.validate();
     if (!validationErrors.isEmpty()) {
@@ -64,65 +52,19 @@
 
     if (config.cache().synchronize()) {
       install(new CacheModule());
+      brokerRouterNeeded = true;
     }
     if (config.event().synchronize()) {
-      install(new EventModule(config));
+      brokerRouterNeeded = true;
     }
     if (config.index().synchronize()) {
       install(new IndexModule());
+      brokerRouterNeeded = true;
     }
 
-    install(new RouterModule());
-  }
-
-  @Provides
-  @Singleton
-  @InstanceId
-  UUID getInstanceId(SitePaths sitePaths) throws IOException {
-    UUID instanceId = null;
-    Path dataDir = sitePaths.data_dir.resolve(Configuration.PLUGIN_NAME);
-    if (!dataDir.toFile().exists()) {
-      dataDir.toFile().mkdirs();
+    if (brokerRouterNeeded) {
+      install(new BrokerModule());
+      install(new RouterModule());
     }
-    String serverIdFile =
-        dataDir.toAbsolutePath().toString() + "/" + Configuration.INSTANCE_ID_FILE;
-
-    instanceId = tryToLoadSavedInstanceId(serverIdFile);
-
-    if (instanceId == null) {
-      instanceId = UUID.randomUUID();
-      Files.createFile(Paths.get(serverIdFile));
-      try (BufferedWriter writer = Files.newBufferedWriter(Paths.get(serverIdFile))) {
-        writer.write(instanceId.toString());
-      } catch (IOException e) {
-        log.warn(
-            String.format(
-                "Cannot write instance ID, a new one will be generated at instance restart. (%s)",
-                e.getMessage()));
-      }
-    }
-    return instanceId;
-  }
-
-  private UUID tryToLoadSavedInstanceId(String serverIdFile) {
-    if (Files.exists(Paths.get(serverIdFile))) {
-      try (BufferedReader br = Files.newBufferedReader(Paths.get(serverIdFile))) {
-        return UUID.fromString(br.readLine());
-      } catch (IOException e) {
-        log.warn(
-            String.format(
-                "Cannot read instance ID from path '%s', deleting the old file and generating a new ID: (%s)",
-                serverIdFile, e.getMessage()));
-        try {
-          Files.delete(Paths.get(serverIdFile));
-        } catch (IOException e1) {
-          log.warn(
-              String.format(
-                  "Cannot delete old instance ID file at path '%s' with instance ID while generating a new one: (%s)",
-                  serverIdFile, e1.getMessage()));
-        }
-      }
-    }
-    return null;
   }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/PluginModule.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/PluginModule.java
index 20aaef6..db06c9f 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/PluginModule.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/PluginModule.java
@@ -15,38 +15,99 @@
 package com.googlesource.gerrit.plugins.multisite;
 
 import com.gerritforge.gerrit.globalrefdb.validation.ProjectDeletedSharedDbCleanup;
+import com.google.common.collect.ImmutableList;
+import com.google.common.flogger.FluentLogger;
 import com.google.gerrit.extensions.events.ProjectDeletedListener;
 import com.google.gerrit.extensions.registration.DynamicSet;
 import com.google.gerrit.lifecycle.LifecycleModule;
+import com.google.gerrit.server.git.WorkQueue;
+import com.google.inject.AbstractModule;
 import com.google.inject.Inject;
+import com.google.inject.Injector;
+import com.google.inject.ProvisionException;
 import com.google.inject.Scopes;
 import com.googlesource.gerrit.plugins.multisite.broker.BrokerApiWrapper;
 import com.googlesource.gerrit.plugins.multisite.consumer.MultiSiteConsumerRunner;
 import com.googlesource.gerrit.plugins.multisite.consumer.ReplicationStatusModule;
 import com.googlesource.gerrit.plugins.multisite.consumer.SubscriberModule;
+import com.googlesource.gerrit.plugins.multisite.event.EventModule;
 import com.googlesource.gerrit.plugins.multisite.forwarder.broker.BrokerForwarderModule;
 
 public class PluginModule extends LifecycleModule {
-  private Configuration config;
+  private static final FluentLogger log = FluentLogger.forEnclosingClass();
+  private static final String[] FILTER_MODULES_CLASS_NAMES =
+      new String[] {
+        /* Class names are defined as String for avoiding this class failing to load
+         * if either replication or pull-replication plugins are missing.
+         */
+        "com.googlesource.gerrit.plugins.multisite.validation.PullReplicationFilterModule",
+        "com.googlesource.gerrit.plugins.multisite.validation.PushReplicationFilterModule"
+      };
+
+  private final Configuration config;
+  private final WorkQueue workQueue;
+  private final Injector parentInjector;
 
   @Inject
-  public PluginModule(Configuration config) {
+  public PluginModule(Configuration config, WorkQueue workQueue, Injector parentInjector) {
     this.config = config;
+    this.workQueue = workQueue;
+    this.parentInjector = parentInjector;
   }
 
   @Override
   protected void configure() {
-    bind(BrokerApiWrapper.class).in(Scopes.SINGLETON);
-    install(new SubscriberModule());
+    if (config.index().synchronize()
+        || config.cache().synchronize()
+        || config.event().synchronize()) {
+      install(new EventModule(config));
+      bind(BrokerApiWrapper.class).in(Scopes.SINGLETON);
+      install(new SubscriberModule());
 
-    install(new BrokerForwarderModule());
-    listener().to(MultiSiteConsumerRunner.class);
+      install(new BrokerForwarderModule());
+      listener().to(MultiSiteConsumerRunner.class);
 
-    install(new ReplicationStatusModule());
+      install(new ReplicationStatusModule(workQueue));
+    }
+
     if (config.getSharedRefDbConfiguration().getSharedRefDb().isEnabled()) {
       listener().to(PluginStartup.class);
       DynamicSet.bind(binder(), ProjectDeletedListener.class)
           .to(ProjectDeletedSharedDbCleanup.class);
     }
+
+    detectFilterModules()
+        .forEach(
+            mod -> {
+              install(mod);
+              log.atInfo().log(
+                  "Replication filter module %s installed successfully",
+                  mod.getClass().getSimpleName());
+            });
+  }
+
+  private Iterable<AbstractModule> detectFilterModules() {
+    ImmutableList.Builder<AbstractModule> filterModulesBuilder = ImmutableList.builder();
+
+    for (String filterClassName : FILTER_MODULES_CLASS_NAMES) {
+      try {
+        @SuppressWarnings("unchecked")
+        Class<AbstractModule> filterClass = (Class<AbstractModule>) Class.forName(filterClassName);
+
+        AbstractModule filterModule = parentInjector.getInstance(filterClass);
+        // Check if the filterModule would be valid for creating a child Guice Injector
+        parentInjector.createChildInjector(filterModule);
+
+        filterModulesBuilder.add(filterModule);
+      } catch (NoClassDefFoundError | ClassNotFoundException e) {
+        log.atFine().withCause(e).log(
+            "Not loading %s because of missing the associated replication plugin", filterClassName);
+      } catch (Exception e) {
+        throw new ProvisionException(
+            "Unable to instantiate replication filter " + filterClassName, e);
+      }
+    }
+
+    return filterModulesBuilder.build();
   }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/ProjectVersionLogger.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/ProjectVersionLogger.java
index 6ababb6..2ee2c13 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/ProjectVersionLogger.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/ProjectVersionLogger.java
@@ -19,4 +19,6 @@
 public interface ProjectVersionLogger {
 
   public void log(Project.NameKey projectName, long currentVersion, long replicationLag);
+
+  public void logDeleted(Project.NameKey projectName);
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/broker/BrokerApiWrapper.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/broker/BrokerApiWrapper.java
index 71be5e6..f58efa7 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/broker/BrokerApiWrapper.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/broker/BrokerApiWrapper.java
@@ -15,62 +15,101 @@
 package com.googlesource.gerrit.plugins.multisite.broker;
 
 import com.gerritforge.gerrit.eventbroker.BrokerApi;
-import com.gerritforge.gerrit.eventbroker.EventMessage;
 import com.gerritforge.gerrit.eventbroker.TopicSubscriber;
+import com.google.common.base.Strings;
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.SettableFuture;
 import com.google.gerrit.extensions.registration.DynamicItem;
 import com.google.gerrit.server.events.Event;
 import com.google.inject.Inject;
-import com.googlesource.gerrit.plugins.multisite.InstanceId;
 import com.googlesource.gerrit.plugins.multisite.MessageLogger;
 import com.googlesource.gerrit.plugins.multisite.MessageLogger.Direction;
 import com.googlesource.gerrit.plugins.multisite.forwarder.Context;
 import java.util.Set;
-import java.util.UUID;
+import java.util.concurrent.Executor;
 import java.util.function.Consumer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class BrokerApiWrapper implements BrokerApi {
+  private static final Logger log = LoggerFactory.getLogger(BrokerApiWrapper.class);
+  private final Executor executor;
   private final DynamicItem<BrokerApi> apiDelegate;
   private final BrokerMetrics metrics;
   private final MessageLogger msgLog;
-  private final UUID instanceId;
 
   @Inject
   public BrokerApiWrapper(
+      @BrokerExecutor Executor executor,
       DynamicItem<BrokerApi> apiDelegate,
       BrokerMetrics metrics,
-      MessageLogger msgLog,
-      @InstanceId UUID instanceId) {
+      MessageLogger msgLog) {
     this.apiDelegate = apiDelegate;
+    this.executor = executor;
     this.metrics = metrics;
     this.msgLog = msgLog;
-    this.instanceId = instanceId;
   }
 
-  public boolean send(String topic, Event event) {
-    return send(topic, apiDelegate.get().newMessage(instanceId, event));
-  }
-
-  @Override
-  public boolean send(String topic, EventMessage message) {
-    if (Context.isForwardedEvent()) {
-      return true;
-    }
-    boolean succeeded = false;
+  public boolean sendSync(String topic, Event event) {
     try {
-      succeeded = apiDelegate.get().send(topic, message);
-    } finally {
-      if (succeeded) {
-        msgLog.log(Direction.PUBLISH, topic, message);
-        metrics.incrementBrokerPublishedMessage();
-      } else {
-        metrics.incrementBrokerFailedToPublishMessage();
-      }
+      return send(topic, event).get();
+    } catch (Throwable e) {
+      log.error(
+          "Failed to publish event '{}' to topic '{}' - error: {} - stack trace: {}",
+          event,
+          topic,
+          e.getMessage(),
+          e.getStackTrace());
+      metrics.incrementBrokerFailedToPublishMessage();
+      return false;
     }
-    return succeeded;
   }
 
   @Override
-  public void receiveAsync(String topic, Consumer<EventMessage> messageConsumer) {
+  public ListenableFuture<Boolean> send(String topic, Event message) {
+    SettableFuture<Boolean> resultFuture = SettableFuture.create();
+    if (Context.isForwardedEvent()) {
+      resultFuture.set(true);
+      return resultFuture;
+    }
+
+    if (Strings.isNullOrEmpty(message.instanceId)) {
+      log.warn(
+          "Dropping event '{}' because event instance id cannot be null or empty",
+          message.toString());
+      resultFuture.set(true);
+      return resultFuture;
+    }
+
+    ListenableFuture<Boolean> resfultF = apiDelegate.get().send(topic, message);
+    Futures.addCallback(
+        resfultF,
+        new FutureCallback<Boolean>() {
+          @Override
+          public void onSuccess(Boolean result) {
+            msgLog.log(Direction.PUBLISH, topic, message);
+            metrics.incrementBrokerPublishedMessage();
+          }
+
+          @Override
+          public void onFailure(Throwable throwable) {
+            log.error(
+                "Failed to publish message '{}' to topic '{}' - error: {}",
+                message.toString(),
+                topic,
+                throwable.getMessage());
+            metrics.incrementBrokerFailedToPublishMessage();
+          }
+        },
+        executor);
+
+    return resfultF;
+  }
+
+  @Override
+  public void receiveAsync(String topic, Consumer<Event> messageConsumer) {
     apiDelegate.get().receiveAsync(topic, messageConsumer);
   }
 
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/InstanceId.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/broker/BrokerExecutor.java
similarity index 83%
rename from src/main/java/com/googlesource/gerrit/plugins/multisite/InstanceId.java
rename to src/main/java/com/googlesource/gerrit/plugins/multisite/broker/BrokerExecutor.java
index 87306a2..aa24eb1 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/InstanceId.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/broker/BrokerExecutor.java
@@ -1,4 +1,4 @@
-// Copyright (C) 2019 The Android Open Source Project
+// Copyright (C) 2021 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.
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package com.googlesource.gerrit.plugins.multisite;
+package com.googlesource.gerrit.plugins.multisite.broker;
 
 import static java.lang.annotation.RetentionPolicy.RUNTIME;
 
@@ -21,4 +21,4 @@
 
 @Retention(RUNTIME)
 @BindingAnnotation
-public @interface InstanceId {}
+@interface BrokerExecutor {}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/broker/BrokerExecutorProvider.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/broker/BrokerExecutorProvider.java
new file mode 100644
index 0000000..e843263
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/broker/BrokerExecutorProvider.java
@@ -0,0 +1,29 @@
+// Copyright (C) 2021 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.multisite.broker;
+
+import com.google.gerrit.server.git.WorkQueue;
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+import com.googlesource.gerrit.plugins.multisite.ExecutorProvider;
+
+@Singleton
+class BrokerExecutorProvider extends ExecutorProvider {
+
+  @Inject
+  BrokerExecutorProvider(WorkQueue workQueue) {
+    super(workQueue, 1, "Multi-Site-Broker");
+  }
+}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/broker/BrokerModule.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/broker/BrokerModule.java
new file mode 100644
index 0000000..a5dac4a
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/broker/BrokerModule.java
@@ -0,0 +1,28 @@
+// Copyright (C) 2021 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.multisite.broker;
+
+import com.google.gerrit.lifecycle.LifecycleModule;
+import java.util.concurrent.Executor;
+
+public class BrokerModule extends LifecycleModule {
+
+  @Override
+  protected void configure() {
+    bind(Executor.class)
+        .annotatedWith(BrokerExecutor.class)
+        .toProvider(BrokerExecutorProvider.class);
+  }
+}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/cache/CacheEvictionHandler.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/cache/CacheEvictionHandler.java
index 2990264..e418da5 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/cache/CacheEvictionHandler.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/cache/CacheEvictionHandler.java
@@ -17,6 +17,7 @@
 import com.google.common.cache.RemovalNotification;
 import com.google.gerrit.extensions.registration.DynamicSet;
 import com.google.gerrit.server.cache.CacheRemovalListener;
+import com.google.gerrit.server.config.GerritInstanceId;
 import com.google.inject.Inject;
 import com.googlesource.gerrit.plugins.multisite.forwarder.CacheEvictionForwarder;
 import com.googlesource.gerrit.plugins.multisite.forwarder.Context;
@@ -28,21 +29,25 @@
   private final Executor executor;
   private final DynamicSet<CacheEvictionForwarder> forwarders;
   private final CachePatternMatcher matcher;
+  private final String instanceId;
 
   @Inject
   CacheEvictionHandler(
       DynamicSet<CacheEvictionForwarder> forwarders,
       @CacheExecutor Executor executor,
-      CachePatternMatcher matcher) {
+      CachePatternMatcher matcher,
+      @GerritInstanceId String instanceId) {
     this.forwarders = forwarders;
     this.executor = executor;
     this.matcher = matcher;
+    this.instanceId = instanceId;
   }
 
   @Override
   public void onRemoval(String plugin, String cache, RemovalNotification<K, V> notification) {
     if (!Context.isForwardedEvent() && !notification.wasEvicted() && matcher.matches(cache)) {
-      executor.execute(new CacheEvictionTask(new CacheEvictionEvent(cache, notification.getKey())));
+      executor.execute(
+          new CacheEvictionTask(new CacheEvictionEvent(cache, notification.getKey(), instanceId)));
     }
   }
 
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/cache/CachePatternMatcher.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/cache/CachePatternMatcher.java
index b8521a3..2dcb09a 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/cache/CachePatternMatcher.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/cache/CachePatternMatcher.java
@@ -25,7 +25,7 @@
 
 @Singleton
 class CachePatternMatcher {
-  private static final List<String> DEFAULT_PATTERNS =
+  private static final ImmutableList<String> DEFAULT_PATTERNS =
       ImmutableList.of("^groups.*", "ldap_groups", "ldap_usernames", "projects", "sshkeys");
 
   private final Pattern pattern;
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/cache/ProjectListUpdateHandler.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/cache/ProjectListUpdateHandler.java
index fdc6fc3..44f8417 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/cache/ProjectListUpdateHandler.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/cache/ProjectListUpdateHandler.java
@@ -19,6 +19,7 @@
 import com.google.gerrit.extensions.events.ProjectDeletedListener;
 import com.google.gerrit.extensions.events.ProjectEvent;
 import com.google.gerrit.extensions.registration.DynamicSet;
+import com.google.gerrit.server.config.GerritInstanceId;
 import com.google.inject.Inject;
 import com.google.inject.Singleton;
 import com.googlesource.gerrit.plugins.multisite.forwarder.Context;
@@ -33,15 +34,18 @@
   private final DynamicSet<ProjectListUpdateForwarder> forwarders;
   private final Executor executor;
   private final ProjectsFilter projectsFilter;
+  private final String instanceId;
 
   @Inject
   public ProjectListUpdateHandler(
       DynamicSet<ProjectListUpdateForwarder> forwarders,
       @CacheExecutor Executor executor,
-      ProjectsFilter filter) {
+      ProjectsFilter filter,
+      @GerritInstanceId String instanceId) {
     this.forwarders = forwarders;
     this.executor = executor;
     this.projectsFilter = filter;
+    this.instanceId = instanceId;
   }
 
   @Override
@@ -59,7 +63,8 @@
   private void process(ProjectEvent event, boolean delete) {
     if (!Context.isForwardedEvent() && projectsFilter.matches(event.getProjectName())) {
       executor.execute(
-          new ProjectListUpdateTask(new ProjectListUpdateEvent(event.getProjectName(), delete)));
+          new ProjectListUpdateTask(
+              new ProjectListUpdateEvent(event.getProjectName(), delete, instanceId)));
     }
   }
 
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/AbstractSubcriber.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/AbstractSubcriber.java
index b37f434..7107b87 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/AbstractSubcriber.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/AbstractSubcriber.java
@@ -14,20 +14,19 @@
 
 package com.googlesource.gerrit.plugins.multisite.consumer;
 
-import com.gerritforge.gerrit.eventbroker.EventMessage;
 import com.google.common.base.Strings;
 import com.google.common.flogger.FluentLogger;
 import com.google.gerrit.extensions.registration.DynamicSet;
+import com.google.gerrit.server.config.GerritInstanceId;
+import com.google.gerrit.server.events.Event;
 import com.google.gerrit.server.permissions.PermissionBackendException;
 import com.googlesource.gerrit.plugins.multisite.Configuration;
-import com.googlesource.gerrit.plugins.multisite.InstanceId;
 import com.googlesource.gerrit.plugins.multisite.MessageLogger;
 import com.googlesource.gerrit.plugins.multisite.MessageLogger.Direction;
 import com.googlesource.gerrit.plugins.multisite.forwarder.CacheNotFoundException;
 import com.googlesource.gerrit.plugins.multisite.forwarder.events.EventTopic;
 import com.googlesource.gerrit.plugins.multisite.forwarder.router.ForwardedEventRouter;
 import java.io.IOException;
-import java.util.UUID;
 import java.util.function.Consumer;
 
 public abstract class AbstractSubcriber {
@@ -44,13 +43,13 @@
   public AbstractSubcriber(
       ForwardedEventRouter eventRouter,
       DynamicSet<DroppedEventListener> droppedEventListeners,
-      @InstanceId UUID instanceId,
+      @GerritInstanceId String gerritInstanceId,
       MessageLogger msgLog,
       SubscriberMetrics subscriberMetrics,
       Configuration cfg) {
     this.eventRouter = eventRouter;
     this.droppedEventListeners = droppedEventListeners;
-    this.instanceId = instanceId.toString();
+    this.instanceId = gerritInstanceId;
     this.msgLog = msgLog;
     this.subscriberMetrics = subscriberMetrics;
     this.cfg = cfg;
@@ -59,19 +58,22 @@
 
   protected abstract EventTopic getTopic();
 
-  public Consumer<EventMessage> getConsumer() {
+  protected abstract Boolean shouldConsumeEvent(Event event);
+
+  public Consumer<Event> getConsumer() {
     return this::processRecord;
   }
 
-  private void processRecord(EventMessage event) {
-    String sourceInstanceId = event.getHeader().sourceInstanceId;
+  private void processRecord(Event event) {
+    String sourceInstanceId = event.instanceId;
 
-    if (Strings.isNullOrEmpty(sourceInstanceId) || instanceId.equals(sourceInstanceId)) {
+    if ((Strings.isNullOrEmpty(sourceInstanceId) || instanceId.equals(sourceInstanceId))
+        || !shouldConsumeEvent(event)) {
       if (Strings.isNullOrEmpty(sourceInstanceId)) {
         logger.atWarning().log(
             String.format(
                 "Dropping event %s because sourceInstanceId cannot be null", event.toString()));
-      } else {
+      } else if (instanceId.equals(sourceInstanceId)) {
         logger.atFiner().log(
             String.format(
                 "Dropping event %s produced by our instanceId %s", event.toString(), instanceId));
@@ -80,16 +82,16 @@
     } else {
       try {
         msgLog.log(Direction.CONSUME, topic, event);
-        eventRouter.route(event.getEvent());
+        eventRouter.route(event);
         subscriberMetrics.incrementSubscriberConsumedMessage();
-        subscriberMetrics.updateReplicationStatusMetrics(event);
       } catch (IOException e) {
-        logger.atSevere().withCause(e).log("Malformed event '%s'", event.getHeader());
+        logger.atSevere().withCause(e).log("Malformed event '%s'", event);
         subscriberMetrics.incrementSubscriberFailedToConsumeMessage();
       } catch (PermissionBackendException | CacheNotFoundException e) {
-        logger.atSevere().withCause(e).log("Cannot handle message '%s'", event.getHeader());
+        logger.atSevere().withCause(e).log("Cannot handle message '%s'", event);
         subscriberMetrics.incrementSubscriberFailedToConsumeMessage();
       }
     }
+    subscriberMetrics.updateReplicationStatusMetrics(event);
   }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/BatchIndexEventSubscriber.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/BatchIndexEventSubscriber.java
index 5bbaee0..cdbf220 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/BatchIndexEventSubscriber.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/BatchIndexEventSubscriber.java
@@ -14,31 +14,49 @@
 
 package com.googlesource.gerrit.plugins.multisite.consumer;
 
+import com.gerritforge.gerrit.globalrefdb.validation.ProjectsFilter;
 import com.google.gerrit.extensions.registration.DynamicSet;
+import com.google.gerrit.server.config.GerritInstanceId;
+import com.google.gerrit.server.events.Event;
 import com.google.inject.Inject;
 import com.google.inject.Singleton;
 import com.googlesource.gerrit.plugins.multisite.Configuration;
-import com.googlesource.gerrit.plugins.multisite.InstanceId;
 import com.googlesource.gerrit.plugins.multisite.MessageLogger;
+import com.googlesource.gerrit.plugins.multisite.forwarder.events.ChangeIndexEvent;
 import com.googlesource.gerrit.plugins.multisite.forwarder.events.EventTopic;
+import com.googlesource.gerrit.plugins.multisite.forwarder.events.ProjectIndexEvent;
 import com.googlesource.gerrit.plugins.multisite.forwarder.router.IndexEventRouter;
-import java.util.UUID;
 
 @Singleton
 public class BatchIndexEventSubscriber extends AbstractSubcriber {
+  private final ProjectsFilter projectsFilter;
+
   @Inject
   public BatchIndexEventSubscriber(
       IndexEventRouter eventRouter,
       DynamicSet<DroppedEventListener> droppedEventListeners,
-      @InstanceId UUID instanceId,
+      @GerritInstanceId String instanceId,
       MessageLogger msgLog,
       SubscriberMetrics subscriberMetrics,
-      Configuration cfg) {
+      Configuration cfg,
+      ProjectsFilter projectsFilter) {
     super(eventRouter, droppedEventListeners, instanceId, msgLog, subscriberMetrics, cfg);
+    this.projectsFilter = projectsFilter;
   }
 
   @Override
   protected EventTopic getTopic() {
     return EventTopic.BATCH_INDEX_TOPIC;
   }
+
+  @Override
+  protected Boolean shouldConsumeEvent(Event event) {
+    if (event instanceof ChangeIndexEvent) {
+      return projectsFilter.matches(((ChangeIndexEvent) event).projectName);
+    }
+    if (event instanceof ProjectIndexEvent) {
+      return projectsFilter.matches(((ProjectIndexEvent) event).projectName);
+    }
+    return true;
+  }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/CacheEvictionEventSubscriber.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/CacheEvictionEventSubscriber.java
index eae66b4..d8342b0 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/CacheEvictionEventSubscriber.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/CacheEvictionEventSubscriber.java
@@ -15,14 +15,14 @@
 package com.googlesource.gerrit.plugins.multisite.consumer;
 
 import com.google.gerrit.extensions.registration.DynamicSet;
+import com.google.gerrit.server.config.GerritInstanceId;
+import com.google.gerrit.server.events.Event;
 import com.google.inject.Inject;
 import com.google.inject.Singleton;
 import com.googlesource.gerrit.plugins.multisite.Configuration;
-import com.googlesource.gerrit.plugins.multisite.InstanceId;
 import com.googlesource.gerrit.plugins.multisite.MessageLogger;
 import com.googlesource.gerrit.plugins.multisite.forwarder.events.EventTopic;
 import com.googlesource.gerrit.plugins.multisite.forwarder.router.CacheEvictionEventRouter;
-import java.util.UUID;
 
 @Singleton
 public class CacheEvictionEventSubscriber extends AbstractSubcriber {
@@ -30,7 +30,7 @@
   public CacheEvictionEventSubscriber(
       CacheEvictionEventRouter eventRouter,
       DynamicSet<DroppedEventListener> droppedEventListeners,
-      @InstanceId UUID instanceId,
+      @GerritInstanceId String instanceId,
       MessageLogger msgLog,
       SubscriberMetrics subscriberMetrics,
       Configuration cfg) {
@@ -42,4 +42,9 @@
   protected EventTopic getTopic() {
     return EventTopic.CACHE_TOPIC;
   }
+
+  @Override
+  protected Boolean shouldConsumeEvent(Event event) {
+    return true;
+  }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/DroppedEventListener.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/DroppedEventListener.java
index 6f4680c..b34e02c 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/DroppedEventListener.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/DroppedEventListener.java
@@ -14,7 +14,7 @@
 
 package com.googlesource.gerrit.plugins.multisite.consumer;
 
-import com.gerritforge.gerrit.eventbroker.EventMessage;
+import com.google.gerrit.server.events.Event;
 
 public interface DroppedEventListener {
   /**
@@ -22,5 +22,5 @@
    *
    * @param event information about the event.
    */
-  void onEventDropped(EventMessage event);
+  void onEventDropped(Event event);
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/IndexEventSubscriber.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/IndexEventSubscriber.java
index 49d470a..480fc50 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/IndexEventSubscriber.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/IndexEventSubscriber.java
@@ -14,31 +14,68 @@
 
 package com.googlesource.gerrit.plugins.multisite.consumer;
 
+import com.gerritforge.gerrit.globalrefdb.validation.ProjectsFilter;
+import com.google.gerrit.entities.Change;
 import com.google.gerrit.extensions.registration.DynamicSet;
+import com.google.gerrit.server.change.ChangeFinder;
+import com.google.gerrit.server.config.GerritInstanceId;
+import com.google.gerrit.server.events.Event;
 import com.google.inject.Inject;
 import com.google.inject.Singleton;
 import com.googlesource.gerrit.plugins.multisite.Configuration;
-import com.googlesource.gerrit.plugins.multisite.InstanceId;
 import com.googlesource.gerrit.plugins.multisite.MessageLogger;
+import com.googlesource.gerrit.plugins.multisite.forwarder.events.ChangeIndexEvent;
 import com.googlesource.gerrit.plugins.multisite.forwarder.events.EventTopic;
+import com.googlesource.gerrit.plugins.multisite.forwarder.events.ProjectIndexEvent;
 import com.googlesource.gerrit.plugins.multisite.forwarder.router.IndexEventRouter;
-import java.util.UUID;
+import java.util.Optional;
 
 @Singleton
 public class IndexEventSubscriber extends AbstractSubcriber {
+  private final ProjectsFilter projectsFilter;
+  private final ChangeFinder changeFinder;
+
   @Inject
   public IndexEventSubscriber(
       IndexEventRouter eventRouter,
       DynamicSet<DroppedEventListener> droppedEventListeners,
-      @InstanceId UUID instanceId,
+      @GerritInstanceId String instanceId,
       MessageLogger msgLog,
       SubscriberMetrics subscriberMetrics,
-      Configuration cfg) {
+      Configuration cfg,
+      ProjectsFilter projectsFilter,
+      ChangeFinder changeFinder) {
     super(eventRouter, droppedEventListeners, instanceId, msgLog, subscriberMetrics, cfg);
+    this.projectsFilter = projectsFilter;
+    this.changeFinder = changeFinder;
   }
 
   @Override
   protected EventTopic getTopic() {
     return EventTopic.INDEX_TOPIC;
   }
+
+  @Override
+  protected Boolean shouldConsumeEvent(Event event) {
+    if (event instanceof ChangeIndexEvent) {
+      ChangeIndexEvent changeIndexEvent = (ChangeIndexEvent) event;
+      String projectName = changeIndexEvent.projectName;
+      if (isDeletedChangeWithEmptyProject(changeIndexEvent)) {
+        projectName = findProjectFromChangeId(changeIndexEvent.changeId).orElse(projectName);
+      }
+      return projectsFilter.matches(projectName);
+    }
+    if (event instanceof ProjectIndexEvent) {
+      return projectsFilter.matches(((ProjectIndexEvent) event).projectName);
+    }
+    return true;
+  }
+
+  private boolean isDeletedChangeWithEmptyProject(ChangeIndexEvent changeIndexEvent) {
+    return changeIndexEvent.deleted && changeIndexEvent.projectName.isEmpty();
+  }
+
+  private Optional<String> findProjectFromChangeId(int changeId) {
+    return changeFinder.findOne(Change.id(changeId)).map(c -> c.getChange().getProject().get());
+  }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/ProjectUpdateEventSubscriber.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/ProjectUpdateEventSubscriber.java
index 6ff0969..069a516 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/ProjectUpdateEventSubscriber.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/ProjectUpdateEventSubscriber.java
@@ -14,31 +14,42 @@
 
 package com.googlesource.gerrit.plugins.multisite.consumer;
 
+import com.gerritforge.gerrit.globalrefdb.validation.ProjectsFilter;
 import com.google.gerrit.extensions.registration.DynamicSet;
+import com.google.gerrit.server.config.GerritInstanceId;
+import com.google.gerrit.server.events.Event;
 import com.google.inject.Inject;
 import com.google.inject.Singleton;
 import com.googlesource.gerrit.plugins.multisite.Configuration;
-import com.googlesource.gerrit.plugins.multisite.InstanceId;
 import com.googlesource.gerrit.plugins.multisite.MessageLogger;
 import com.googlesource.gerrit.plugins.multisite.forwarder.events.EventTopic;
+import com.googlesource.gerrit.plugins.multisite.forwarder.events.ProjectListUpdateEvent;
 import com.googlesource.gerrit.plugins.multisite.forwarder.router.ProjectListUpdateRouter;
-import java.util.UUID;
 
 @Singleton
 public class ProjectUpdateEventSubscriber extends AbstractSubcriber {
+  private final ProjectsFilter projectsFilter;
+
   @Inject
   public ProjectUpdateEventSubscriber(
       ProjectListUpdateRouter eventRouter,
       DynamicSet<DroppedEventListener> droppedEventListeners,
-      @InstanceId UUID instanceId,
+      @GerritInstanceId String instanceId,
       MessageLogger msgLog,
       SubscriberMetrics subscriberMetrics,
-      Configuration cfg) {
+      Configuration cfg,
+      ProjectsFilter projectsFilter) {
     super(eventRouter, droppedEventListeners, instanceId, msgLog, subscriberMetrics, cfg);
+    this.projectsFilter = projectsFilter;
   }
 
   @Override
   protected EventTopic getTopic() {
     return EventTopic.PROJECT_LIST_TOPIC;
   }
+
+  @Override
+  protected Boolean shouldConsumeEvent(Event event) {
+    return projectsFilter.matches(((ProjectListUpdateEvent) event).projectName);
+  }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/ReplicationStatus.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/ReplicationStatus.java
index 7360b8f..0253588 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/ReplicationStatus.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/ReplicationStatus.java
@@ -19,14 +19,20 @@
 import com.google.common.flogger.FluentLogger;
 import com.google.gerrit.entities.Project;
 import com.google.gerrit.extensions.events.LifecycleListener;
+import com.google.gerrit.extensions.events.ProjectDeletedListener;
+import com.google.gerrit.metrics.CallbackMetric1;
+import com.google.gerrit.metrics.MetricMaker;
 import com.google.gerrit.server.cache.CacheModule;
 import com.google.gerrit.server.cache.serialize.JavaCacheSerializer;
 import com.google.gerrit.server.cache.serialize.StringCacheSerializer;
+import com.google.gerrit.server.git.WorkQueue;
 import com.google.gerrit.server.project.ProjectCache;
 import com.google.inject.Inject;
 import com.google.inject.Module;
 import com.google.inject.Singleton;
 import com.google.inject.name.Named;
+import com.google.inject.name.Names;
+import com.googlesource.gerrit.plugins.multisite.Configuration;
 import com.googlesource.gerrit.plugins.multisite.ProjectVersionLogger;
 import com.googlesource.gerrit.plugins.multisite.validation.ProjectVersionRefUpdate;
 import java.util.Collection;
@@ -36,23 +42,29 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
 @Singleton
-public class ReplicationStatus implements LifecycleListener {
+public class ReplicationStatus implements LifecycleListener, ProjectDeletedListener {
   private static final FluentLogger logger = FluentLogger.forEnclosingClass();
 
   private final Map<String, Long> replicationStatusPerProject = new HashMap<>();
-  static final String REPLICATION_STATUS_CACHE = "replication_status";
+  static final String REPLICATION_STATUS = "replication_status";
 
-  public static Module cacheModule() {
+  public static Module cacheModule(WorkQueue queue) {
     return new CacheModule() {
       @Override
       protected void configure() {
-        persist(REPLICATION_STATUS_CACHE, String.class, Long.class)
+        persist(REPLICATION_STATUS, String.class, Long.class)
             .version(1)
             .keySerializer(StringCacheSerializer.INSTANCE)
             .valueSerializer(new JavaCacheSerializer<>());
+
+        bind(ScheduledExecutorService.class)
+            .annotatedWith(Names.named(REPLICATION_STATUS))
+            .toInstance(queue.createQueue(0, REPLICATION_STATUS));
       }
     };
   }
@@ -62,20 +74,35 @@
   private final Optional<ProjectVersionRefUpdate> projectVersionRefUpdate;
   private final ProjectVersionLogger verLogger;
   private final ProjectCache projectCache;
+  private final ScheduledExecutorService statusScheduler;
+
+  private final Configuration config;
+
+  private final MetricMaker metricMaker;
 
   @Inject
   public ReplicationStatus(
-      @Named(REPLICATION_STATUS_CACHE) Cache<String, Long> cache,
+      @Named(REPLICATION_STATUS) Cache<String, Long> cache,
       Optional<ProjectVersionRefUpdate> projectVersionRefUpdate,
       ProjectVersionLogger verLogger,
-      ProjectCache projectCache) {
+      ProjectCache projectCache,
+      @Named(REPLICATION_STATUS) ScheduledExecutorService statusScheduler,
+      Configuration config,
+      MetricMaker metricMaker) {
     this.cache = cache;
     this.projectVersionRefUpdate = projectVersionRefUpdate;
     this.verLogger = verLogger;
     this.projectCache = projectCache;
+    this.statusScheduler = statusScheduler;
+    this.config = config;
+    this.metricMaker = metricMaker;
   }
 
   public Long getMaxLag() {
+    return getMaxLagMillis() / 1000;
+  }
+
+  public Long getMaxLagMillis() {
     Collection<Long> lags = replicationStatusPerProject.values();
     if (lags.isEmpty()) {
       return 0L;
@@ -121,15 +148,75 @@
     }
   }
 
+  void removeProjectFromReplicationLagMetrics(Project.NameKey projectName) {
+    Optional<Long> localVersion =
+        projectVersionRefUpdate.get().getProjectLocalVersion(projectName.get());
+
+    if (!localVersion.isPresent() && replicationStatusPerProject.containsKey(projectName.get())) {
+      cache.invalidate(projectName.get());
+      replicationStatusPerProject.remove(projectName.get());
+      localVersionPerProject.remove(projectName.get());
+      verLogger.logDeleted(projectName);
+      logger.atFine().log("Removed project '%s' from replication lag metrics", projectName);
+    }
+  }
+
+  @VisibleForTesting
+  Runnable replicationLagMetricPerProject(CallbackMetric1<String, Long> metricCallback) {
+    return () -> {
+      if (replicationStatusPerProject.isEmpty()) {
+        metricCallback.forceCreate("");
+      } else {
+        replicationStatusPerProject.entrySet().stream()
+            .filter(e -> e.getValue() > 0)
+            .forEach(
+                e ->
+                    metricCallback.set(
+                        SubscriberMetrics.sanitizeProjectName(e.getKey()), e.getValue()));
+        metricCallback.prune();
+      }
+    };
+  }
+
   @VisibleForTesting
   public void doUpdateLag(Project.NameKey projectName, Long lag) {
     cache.put(projectName.get(), lag);
     replicationStatusPerProject.put(projectName.get(), lag);
   }
 
+  @VisibleForTesting
+  Long getReplicationStatus(String projectName) {
+    return replicationStatusPerProject.get(projectName);
+  }
+
+  @VisibleForTesting
+  Long getLocalVersion(String projectName) {
+    return localVersionPerProject.get(projectName);
+  }
+
   @Override
   public void start() {
     loadAllFromCache();
+
+    long replicationLagPollingInterval = config.replicationLagRefreshInterval().toMillis();
+
+    if (replicationLagPollingInterval > 0) {
+      statusScheduler.scheduleAtFixedRate(
+          this::refreshProjectsWithLag,
+          replicationLagPollingInterval,
+          replicationLagPollingInterval,
+          TimeUnit.MILLISECONDS);
+    }
+  }
+
+  @VisibleForTesting
+  public void refreshProjectsWithLag() {
+    logger.atFine().log("Refreshing projects version lags triggered ...");
+    replicationStatusPerProject.entrySet().stream()
+        .filter(entry -> entry.getValue() > 0)
+        .map(Map.Entry::getKey)
+        .map(Project::nameKey)
+        .forEach(this::updateReplicationLag);
   }
 
   @Override
@@ -140,4 +227,9 @@
         projectCache.all().stream().map(Project.NameKey::get).collect(Collectors.toSet());
     replicationStatusPerProject.putAll(cache.getAllPresent(cachedProjects));
   }
+
+  @Override
+  public void onProjectDeleted(Event event) {
+    removeProjectFromReplicationLagMetrics(Project.nameKey(event.getProjectName()));
+  }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/ReplicationStatusModule.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/ReplicationStatusModule.java
index e954bc4..605e119 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/ReplicationStatusModule.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/ReplicationStatusModule.java
@@ -14,14 +14,27 @@
 
 package com.googlesource.gerrit.plugins.multisite.consumer;
 
+import com.google.gerrit.extensions.events.ProjectDeletedListener;
+import com.google.gerrit.extensions.registration.DynamicSet;
 import com.google.gerrit.lifecycle.LifecycleModule;
+import com.google.gerrit.server.git.WorkQueue;
+import com.google.inject.Inject;
 import com.google.inject.Scopes;
 
 public class ReplicationStatusModule extends LifecycleModule {
+
+  private final WorkQueue workQueue;
+
+  @Inject
+  public ReplicationStatusModule(WorkQueue workQueue) {
+    this.workQueue = workQueue;
+  }
+
   @Override
   protected void configure() {
     bind(ReplicationStatus.class).in(Scopes.SINGLETON);
-    install(ReplicationStatus.cacheModule());
+    install(ReplicationStatus.cacheModule(workQueue));
     listener().to(ReplicationStatus.class);
+    DynamicSet.bind(binder(), ProjectDeletedListener.class).to(ReplicationStatus.class);
   }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/StreamEventSubscriber.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/StreamEventSubscriber.java
index 20c355e..ab64651 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/StreamEventSubscriber.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/StreamEventSubscriber.java
@@ -14,31 +14,45 @@
 
 package com.googlesource.gerrit.plugins.multisite.consumer;
 
+import com.gerritforge.gerrit.globalrefdb.validation.ProjectsFilter;
 import com.google.gerrit.extensions.registration.DynamicSet;
+import com.google.gerrit.server.config.GerritInstanceId;
+import com.google.gerrit.server.events.Event;
+import com.google.gerrit.server.events.ProjectEvent;
 import com.google.inject.Inject;
 import com.google.inject.Singleton;
 import com.googlesource.gerrit.plugins.multisite.Configuration;
-import com.googlesource.gerrit.plugins.multisite.InstanceId;
 import com.googlesource.gerrit.plugins.multisite.MessageLogger;
 import com.googlesource.gerrit.plugins.multisite.forwarder.events.EventTopic;
 import com.googlesource.gerrit.plugins.multisite.forwarder.router.StreamEventRouter;
-import java.util.UUID;
 
 @Singleton
 public class StreamEventSubscriber extends AbstractSubcriber {
+  private final ProjectsFilter projectsFilter;
+
   @Inject
   public StreamEventSubscriber(
       StreamEventRouter eventRouter,
       DynamicSet<DroppedEventListener> droppedEventListeners,
-      @InstanceId UUID instanceId,
+      @GerritInstanceId String instanceId,
       MessageLogger msgLog,
       SubscriberMetrics subscriberMetrics,
-      Configuration cfg) {
+      Configuration cfg,
+      ProjectsFilter projectsFilter) {
     super(eventRouter, droppedEventListeners, instanceId, msgLog, subscriberMetrics, cfg);
+    this.projectsFilter = projectsFilter;
   }
 
   @Override
   protected EventTopic getTopic() {
     return EventTopic.STREAM_EVENT_TOPIC;
   }
+
+  @Override
+  protected Boolean shouldConsumeEvent(Event event) {
+    if (event instanceof ProjectEvent) {
+      return projectsFilter.matches(((ProjectEvent) event).getProjectNameKey().get());
+    }
+    return true;
+  }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/SubscriberMetrics.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/SubscriberMetrics.java
index 899233e..bb971ff 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/SubscriberMetrics.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/SubscriberMetrics.java
@@ -14,32 +14,43 @@
 
 package com.googlesource.gerrit.plugins.multisite.consumer;
 
-import com.gerritforge.gerrit.eventbroker.EventMessage;
-import com.google.common.flogger.FluentLogger;
+import com.google.gerrit.metrics.CallbackMetric1;
 import com.google.gerrit.metrics.Counter1;
 import com.google.gerrit.metrics.Description;
+import com.google.gerrit.metrics.Field;
 import com.google.gerrit.metrics.MetricMaker;
 import com.google.gerrit.server.events.Event;
+import com.google.gerrit.server.events.ProjectEvent;
 import com.google.gerrit.server.events.RefUpdatedEvent;
+import com.google.gerrit.server.logging.Metadata;
 import com.google.inject.Inject;
 import com.google.inject.Singleton;
 import com.googlesource.gerrit.plugins.multisite.MultiSiteMetrics;
-import com.googlesource.gerrit.plugins.replication.RefReplicatedEvent;
-import com.googlesource.gerrit.plugins.replication.RefReplicationDoneEvent;
-import com.googlesource.gerrit.plugins.replication.ReplicationScheduledEvent;
+import com.googlesource.gerrit.plugins.replication.events.ProjectDeletionReplicationSucceededEvent;
+import com.googlesource.gerrit.plugins.replication.events.RefReplicatedEvent;
+import com.googlesource.gerrit.plugins.replication.events.RefReplicationDoneEvent;
+import com.googlesource.gerrit.plugins.replication.events.ReplicationScheduledEvent;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 @Singleton
 public class SubscriberMetrics extends MultiSiteMetrics {
-  private static final FluentLogger logger = FluentLogger.forEnclosingClass();
   private static final String SUBSCRIBER_SUCCESS_COUNTER = "subscriber_msg_consumer_counter";
   private static final String SUBSCRIBER_FAILURE_COUNTER =
       "subscriber_msg_consumer_failure_counter";
-  private static final String REPLICATION_LAG_SEC =
+  public static final String REPLICATION_LAG_SEC =
       "multi_site/subscriber/subscriber_replication_status/sec_behind";
+  private static final String REPLICATION_LAG_MSEC =
+      "multi_site/subscriber/subscriber_replication_status/msec_behind";
+  private static final String REPLICATION_LAG_MSEC_PROJECT =
+      "multi_site/subscriber/subscriber_replication_status/msec_behind/per_project";
 
   private final Counter1<String> subscriberSuccessCounter;
   private final Counter1<String> subscriberFailureCounter;
   private final ReplicationStatus replicationStatus;
+  private static final Pattern isValidMetricNamePattern = Pattern.compile("[a-zA-Z0-9_-]");
+  private static final Field<String> PROJECT_NAME =
+      Field.ofString("project_name", Metadata.Builder::cacheName).build();
 
   @Inject
   public SubscriberMetrics(MetricMaker metricMaker, ReplicationStatus replicationStatus) {
@@ -64,6 +75,51 @@
         Long.class,
         new Description("Replication lag (sec)").setGauge().setUnit(Description.Units.SECONDS),
         replicationStatus::getMaxLag);
+    metricMaker.newCallbackMetric(
+        REPLICATION_LAG_MSEC,
+        Long.class,
+        new Description("Replication lag (msec)")
+            .setGauge()
+            .setUnit(Description.Units.MILLISECONDS),
+        replicationStatus::getMaxLagMillis);
+
+    CallbackMetric1<String, Long> metrics =
+        metricMaker.newCallbackMetric(
+            SubscriberMetrics.REPLICATION_LAG_MSEC_PROJECT,
+            Long.class,
+            new Description("Per-project replication lag (msec)")
+                .setGauge()
+                .setUnit(Description.Units.MILLISECONDS),
+            PROJECT_NAME);
+    metricMaker.newTrigger(metrics, replicationStatus.replicationLagMetricPerProject(metrics));
+  }
+
+  /**
+   * Ensures that the generated metric is compatible with prometheus metric names, as the set of
+   * values that represent a valid metric name are different from the ones that represent a valid
+   * project name. Main differences: - All _ are replaced with __ - All characters that aren't a
+   * letter(uppercase or lowercase) or a hyphen(-) are replaced with `_<hex_code>` This is to avoid
+   * all chances of name clashes when modifying a project name.
+   *
+   * @param name name of the metric to sanitize
+   * @return sanitized metric name
+   */
+  public static String sanitizeProjectName(String name) {
+    StringBuilder sanitizedName = new StringBuilder();
+    for (int i = 0; i < name.length(); i++) {
+      Character c = name.charAt(i);
+      Matcher matcher = isValidMetricNamePattern.matcher(String.valueOf(c));
+      if (matcher.find()) {
+        if (c == '_') {
+          sanitizedName.append("__");
+        } else {
+          sanitizedName.append(c);
+        }
+      } else {
+        sanitizedName.append("_").append(Integer.toHexString((int) c));
+      }
+    }
+    return sanitizedName.toString();
   }
 
   public void incrementSubscriberConsumedMessage() {
@@ -74,20 +130,18 @@
     subscriberFailureCounter.increment(SUBSCRIBER_FAILURE_COUNTER);
   }
 
-  public void updateReplicationStatusMetrics(EventMessage eventMessage) {
-    Event event = eventMessage.getEvent();
-    if (event instanceof RefReplicationDoneEvent) {
-      RefReplicationDoneEvent replicationDone = (RefReplicationDoneEvent) event;
-      replicationStatus.updateReplicationLag(replicationDone.getProjectNameKey());
-    } else if (event instanceof RefReplicatedEvent) {
-      RefReplicatedEvent replicated = (RefReplicatedEvent) event;
-      replicationStatus.updateReplicationLag(replicated.getProjectNameKey());
-    } else if (event instanceof ReplicationScheduledEvent) {
-      ReplicationScheduledEvent updated = (ReplicationScheduledEvent) event;
-      replicationStatus.updateReplicationLag(updated.getProjectNameKey());
-    } else if (event instanceof RefUpdatedEvent) {
-      RefUpdatedEvent updated = (RefUpdatedEvent) event;
-      replicationStatus.updateReplicationLag(updated.getProjectNameKey());
+  public void updateReplicationStatusMetrics(Event event) {
+
+    if (event instanceof RefReplicationDoneEvent
+        || event instanceof RefReplicatedEvent
+        || event instanceof ReplicationScheduledEvent
+        || event instanceof RefUpdatedEvent) {
+      ProjectEvent projectEvent = (ProjectEvent) event;
+      replicationStatus.updateReplicationLag(projectEvent.getProjectNameKey());
+    } else if (event instanceof ProjectDeletionReplicationSucceededEvent) {
+      ProjectDeletionReplicationSucceededEvent projectDeletion =
+          (ProjectDeletionReplicationSucceededEvent) event;
+      replicationStatus.removeProjectFromReplicationLagMetrics(projectDeletion.getProjectNameKey());
     }
   }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/Context.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/Context.java
index 1c29d75..269388c 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/Context.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/Context.java
@@ -14,14 +14,22 @@
 
 package com.googlesource.gerrit.plugins.multisite.forwarder;
 
+import static com.googlesource.gerrit.plugins.replication.pull.api.PullReplicationEndpoints.APPLY_OBJECTS_API_ENDPOINT;
+import static com.googlesource.gerrit.plugins.replication.pull.api.PullReplicationEndpoints.APPLY_OBJECT_API_ENDPOINT;
+
 /** Allows to tag a forwarded event to avoid infinitely looping events. */
 public class Context {
+  public static final String PULL_REPLICATION_PLUGIN_NAME = "pull-replication";
   private static final ThreadLocal<Boolean> forwardedEvent = ThreadLocal.withInitial(() -> false);
 
   private Context() {}
 
   public static Boolean isForwardedEvent() {
-    return forwardedEvent.get();
+    return forwardedEvent.get()
+        ||
+        // When the event is a result of pull-replication event, is considered as
+        // "forwarded" action because did not happen on this node.
+        isPullReplicationApplyObjectIndexing();
   }
 
   public static void setForwardedEvent(Boolean b) {
@@ -31,4 +39,10 @@
   public static void unsetForwardedEvent() {
     forwardedEvent.remove();
   }
+
+  public static boolean isPullReplicationApplyObjectIndexing() {
+    String threadName = Thread.currentThread().getName();
+    return threadName.contains(PULL_REPLICATION_PLUGIN_NAME + "~" + APPLY_OBJECT_API_ENDPOINT)
+        || threadName.contains(PULL_REPLICATION_PLUGIN_NAME + "~" + APPLY_OBJECTS_API_ENDPOINT);
+  }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedIndexChangeHandler.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedIndexChangeHandler.java
index 8d41500..522a78f 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedIndexChangeHandler.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedIndexChangeHandler.java
@@ -63,17 +63,27 @@
   protected void attemptToIndex(String id, Optional<ChangeIndexEvent> indexEvent, int retryCount) {
     ChangeChecker checker = changeCheckerFactory.create(id);
     Optional<ChangeNotes> changeNotes = checker.getChangeNotes();
-    if (changeNotes.isPresent()) {
+    boolean changeIsPresent = changeNotes.isPresent();
+    boolean changeIsConsistent = checker.isChangeConsistent();
+    if (changeIsPresent && changeIsConsistent) {
       reindexAndCheckIsUpToDate(id, indexEvent, checker, retryCount);
     } else {
       log.warn(
-          "Change {} not present yet in local Git repository (event={}) after {} attempt(s)",
+          "Change {} {} in local Git repository (event={}) after {} attempt(s)",
           id,
+          !changeIsPresent
+              ? "not present yet"
+              : (changeIsConsistent ? "is" : "is not") + " consistent",
           indexEvent,
           retryCount);
       if (!rescheduleIndex(id, indexEvent, retryCount + 1)) {
         log.error(
-            "Change {} could not be found in the local Git repository (event={})", id, indexEvent);
+            "Change {} {} in the local Git repository (event={})",
+            id,
+            !changeIsPresent
+                ? "could not be found"
+                : (changeIsConsistent ? "was" : "was not") + " consistent",
+            indexEvent);
       }
     }
   }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/broker/BrokerForwarder.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/broker/BrokerForwarder.java
index 975916a..19936ef 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/broker/BrokerForwarder.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/broker/BrokerForwarder.java
@@ -49,6 +49,6 @@
       return true;
     }
 
-    return broker.send(eventTopic.topic(cfg), event);
+    return broker.sendSync(eventTopic.topic(cfg), event);
   }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/broker/BrokerStreamEventForwarder.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/broker/BrokerStreamEventForwarder.java
index 9ff4688..8a020fc 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/broker/BrokerStreamEventForwarder.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/broker/BrokerStreamEventForwarder.java
@@ -35,6 +35,6 @@
 
   @Override
   public boolean send(Event event) {
-    return broker.send(EventTopic.STREAM_EVENT_TOPIC.topic(cfg), event);
+    return broker.sendSync(EventTopic.STREAM_EVENT_TOPIC.topic(cfg), event);
   }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/AccountIndexEvent.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/AccountIndexEvent.java
index 317eb76..065ef7e 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/AccountIndexEvent.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/AccountIndexEvent.java
@@ -21,8 +21,8 @@
 
   public int accountId;
 
-  public AccountIndexEvent(int accountId) {
-    super(TYPE);
+  public AccountIndexEvent(int accountId, String instanceId) {
+    super(TYPE, instanceId);
     this.accountId = accountId;
   }
 
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/CacheEvictionEvent.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/CacheEvictionEvent.java
index 1c2185f..4444756 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/CacheEvictionEvent.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/CacheEvictionEvent.java
@@ -22,8 +22,8 @@
   public String cacheName;
   public Object key;
 
-  public CacheEvictionEvent(String cacheName, Object key) {
-    super(TYPE);
+  public CacheEvictionEvent(String cacheName, Object key, String instanceId) {
+    super(TYPE, instanceId);
     this.cacheName = cacheName;
     this.key = key;
   }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/ChangeIndexEvent.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/ChangeIndexEvent.java
index ab4ddf4..64fbdfb 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/ChangeIndexEvent.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/ChangeIndexEvent.java
@@ -28,8 +28,8 @@
   public String targetSha;
   public boolean deleted;
 
-  public ChangeIndexEvent(String projectName, int changeId, boolean deleted) {
-    super(TYPE);
+  public ChangeIndexEvent(String projectName, int changeId, boolean deleted, String instanceId) {
+    super(TYPE, instanceId);
     this.projectName = projectName;
     this.changeId = changeId;
     this.deleted = deleted;
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/GroupIndexEvent.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/GroupIndexEvent.java
index 4981b2f..05ac198 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/GroupIndexEvent.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/GroupIndexEvent.java
@@ -24,8 +24,8 @@
   public final String groupUUID;
   public final ObjectId sha1;
 
-  public GroupIndexEvent(String groupUUID, @Nullable ObjectId sha1) {
-    super(TYPE);
+  public GroupIndexEvent(String groupUUID, @Nullable ObjectId sha1, String instanceId) {
+    super(TYPE, instanceId);
     this.groupUUID = groupUUID;
     this.sha1 = sha1;
   }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/IndexEvent.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/IndexEvent.java
index ea2c3fb..2fdda72 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/IndexEvent.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/IndexEvent.java
@@ -15,7 +15,7 @@
 package com.googlesource.gerrit.plugins.multisite.forwarder.events;
 
 public abstract class IndexEvent extends MultiSiteEvent {
-  protected IndexEvent(String type) {
-    super(type);
+  protected IndexEvent(String type, String instanceId) {
+    super(type, instanceId);
   }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/MultiSiteEvent.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/MultiSiteEvent.java
index 404d168..f29204b 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/MultiSiteEvent.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/MultiSiteEvent.java
@@ -29,7 +29,8 @@
     register(ProjectListUpdateEvent.TYPE, ProjectListUpdateEvent.class);
   }
 
-  protected MultiSiteEvent(String type) {
+  protected MultiSiteEvent(String type, String instanceId) {
     super(type);
+    this.instanceId = instanceId;
   }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/ProjectIndexEvent.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/ProjectIndexEvent.java
index 8bdb7b5..954befb 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/ProjectIndexEvent.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/ProjectIndexEvent.java
@@ -21,8 +21,8 @@
 
   public String projectName;
 
-  public ProjectIndexEvent(String projectName) {
-    super(TYPE);
+  public ProjectIndexEvent(String projectName, String instanceId) {
+    super(TYPE, instanceId);
     this.projectName = projectName;
   }
 
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/ProjectListUpdateEvent.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/ProjectListUpdateEvent.java
index d030e5b..0e18b27 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/ProjectListUpdateEvent.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/ProjectListUpdateEvent.java
@@ -22,8 +22,8 @@
   public String projectName;
   public boolean remove;
 
-  public ProjectListUpdateEvent(String projectName, boolean remove) {
-    super(TYPE);
+  public ProjectListUpdateEvent(String projectName, boolean remove, String instanceId) {
+    super(TYPE, instanceId);
     this.projectName = projectName;
     this.remove = remove;
   }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/router/IndexEventRouter.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/router/IndexEventRouter.java
index 202fb42..a647bce 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/router/IndexEventRouter.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/router/IndexEventRouter.java
@@ -21,6 +21,10 @@
 import com.google.gerrit.entities.Account;
 import com.google.gerrit.extensions.events.LifecycleListener;
 import com.google.gerrit.server.config.AllUsersName;
+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.RefEvent;
 import com.google.inject.Inject;
 import com.googlesource.gerrit.plugins.multisite.forwarder.ForwardedIndexAccountHandler;
 import com.googlesource.gerrit.plugins.multisite.forwarder.ForwardedIndexChangeHandler;
@@ -32,12 +36,12 @@
 import com.googlesource.gerrit.plugins.multisite.forwarder.events.GroupIndexEvent;
 import com.googlesource.gerrit.plugins.multisite.forwarder.events.IndexEvent;
 import com.googlesource.gerrit.plugins.multisite.forwarder.events.ProjectIndexEvent;
-import com.googlesource.gerrit.plugins.replication.RefReplicationDoneEvent;
 import java.io.IOException;
 import java.util.Optional;
 import java.util.Set;
 
-public class IndexEventRouter implements ForwardedEventRouter<IndexEvent>, LifecycleListener {
+public class IndexEventRouter
+    implements ForwardedEventRouter<IndexEvent>, EventListener, LifecycleListener {
   private static final FluentLogger logger = FluentLogger.forEnclosingClass();
 
   private final ForwardedIndexAccountHandler indexAccountHandler;
@@ -45,6 +49,7 @@
   private final ForwardedIndexGroupHandler indexGroupHandler;
   private final ForwardedIndexProjectHandler indexProjectHandler;
   private final AllUsersName allUsersName;
+  private final String gerritInstanceId;
 
   @Inject
   public IndexEventRouter(
@@ -52,12 +57,14 @@
       ForwardedIndexChangeHandler indexChangeHandler,
       ForwardedIndexGroupHandler indexGroupHandler,
       ForwardedIndexProjectHandler indexProjectHandler,
-      AllUsersName allUsersName) {
+      AllUsersName allUsersName,
+      @GerritInstanceId String gerritInstanceId) {
     this.indexAccountHandler = indexAccountHandler;
     this.indexChangeHandler = indexChangeHandler;
     this.indexGroupHandler = indexGroupHandler;
     this.indexProjectHandler = indexProjectHandler;
     this.allUsersName = allUsersName;
+    this.gerritInstanceId = gerritInstanceId;
   }
 
   @Override
@@ -85,7 +92,7 @@
     }
   }
 
-  public void onRefReplicated(RefReplicationDoneEvent replicationEvent) throws IOException {
+  public void onRefReplicated(RefEvent replicationEvent) throws IOException {
     if (replicationEvent.getProjectNameKey().equals(allUsersName)) {
       Account.Id accountId = Account.Id.fromRef(replicationEvent.getRefName());
       if (accountId != null) {
@@ -97,6 +104,20 @@
   }
 
   @Override
+  public void onEvent(Event event) {
+    if (event instanceof RefEvent
+        && (event.getType().contains("fetch-ref-replicated")
+            || event.getType().contains("fetch-ref-replication-done"))
+        && gerritInstanceId.equals(event.instanceId)) {
+      try {
+        onRefReplicated((RefEvent) event);
+      } catch (IOException e) {
+        logger.atSevere().withCause(e).log("Error while processing event %s", event);
+      }
+    }
+  }
+
+  @Override
   public void start() {}
 
   @Override
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/router/RouterModule.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/router/RouterModule.java
index bac907e..2193034 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/router/RouterModule.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/router/RouterModule.java
@@ -14,7 +14,9 @@
 
 package com.googlesource.gerrit.plugins.multisite.forwarder.router;
 
+import com.google.gerrit.extensions.registration.DynamicSet;
 import com.google.gerrit.lifecycle.LifecycleModule;
+import com.google.gerrit.server.events.EventListener;
 import com.google.inject.Scopes;
 
 public class RouterModule extends LifecycleModule {
@@ -22,6 +24,8 @@
   protected void configure() {
     bind(IndexEventRouter.class).in(Scopes.SINGLETON);
     listener().to(IndexEventRouter.class).in(Scopes.SINGLETON);
+    DynamicSet.bind(binder(), EventListener.class).to(IndexEventRouter.class);
+
     bind(CacheEvictionEventRouter.class).in(Scopes.SINGLETON);
     bind(ProjectListUpdateRouter.class).in(Scopes.SINGLETON);
     bind(StreamEventRouter.class).in(Scopes.SINGLETON);
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/router/StreamEventRouter.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/router/StreamEventRouter.java
index 4ef3426..95a3e66 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/router/StreamEventRouter.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/router/StreamEventRouter.java
@@ -18,7 +18,7 @@
 import com.google.gerrit.server.permissions.PermissionBackendException;
 import com.google.inject.Inject;
 import com.googlesource.gerrit.plugins.multisite.forwarder.ForwardedEventHandler;
-import com.googlesource.gerrit.plugins.replication.RefReplicationDoneEvent;
+import com.googlesource.gerrit.plugins.replication.events.RefReplicationDoneEvent;
 import java.io.IOException;
 
 public class StreamEventRouter implements ForwardedEventRouter<Event> {
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/http/HttpModule.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/http/HttpModule.java
index 2ae2d9b..26bfb53 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/http/HttpModule.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/http/HttpModule.java
@@ -14,14 +14,25 @@
 
 package com.googlesource.gerrit.plugins.multisite.http;
 
+import com.google.inject.Inject;
 import com.google.inject.servlet.ServletModule;
+import com.googlesource.gerrit.plugins.multisite.Configuration;
 
 public class HttpModule extends ServletModule {
 
   public static final String LAG_ENDPOINT_SEGMENT = "replication-lag";
 
+  private final Configuration config;
+
+  @Inject
+  public HttpModule(Configuration config) {
+    this.config = config;
+  }
+
   @Override
   protected void configureServlets() {
-    serve(String.format("/%s", LAG_ENDPOINT_SEGMENT)).with(ReplicationStatusServlet.class);
+    if (config.event().synchronize()) {
+      serve(String.format("/%s", LAG_ENDPOINT_SEGMENT)).with(ReplicationStatusServlet.class);
+    }
   }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/index/ChangeChecker.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/index/ChangeChecker.java
index 9ee59eb..3277150 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/index/ChangeChecker.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/index/ChangeChecker.java
@@ -59,4 +59,11 @@
    * @throws IOException if an I/O error occurred while reading the local Change
    */
   public Optional<Long> getComputedChangeTs() throws IOException;
+
+  /**
+   * Check if the local Change contains current patchset refs
+   *
+   * @return true if local change contains meta and current patchset refs
+   */
+  public boolean isChangeConsistent();
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/index/ChangeCheckerImpl.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/index/ChangeCheckerImpl.java
index 08b26f7..4ef5b7f 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/index/ChangeCheckerImpl.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/index/ChangeCheckerImpl.java
@@ -19,6 +19,7 @@
 import com.google.gerrit.exceptions.StorageException;
 import com.google.gerrit.server.CommentsUtil;
 import com.google.gerrit.server.change.ChangeFinder;
+import com.google.gerrit.server.config.GerritInstanceId;
 import com.google.gerrit.server.git.GitRepositoryManager;
 import com.google.gerrit.server.notedb.ChangeNotes;
 import com.google.gerrit.server.util.ManualRequestContext;
@@ -30,8 +31,11 @@
 import java.sql.Timestamp;
 import java.util.Objects;
 import java.util.Optional;
+import org.eclipse.jgit.errors.MissingObjectException;
+import org.eclipse.jgit.lib.ObjectId;
 import org.eclipse.jgit.lib.Ref;
 import org.eclipse.jgit.lib.Repository;
+import org.eclipse.jgit.revwalk.RevWalk;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -42,6 +46,7 @@
   private final OneOffRequestContext oneOffReqCtx;
   private final String changeId;
   private final ChangeFinder changeFinder;
+  private final String instanceId;
   private Optional<Long> computedChangeTs = Optional.empty();
   private Optional<ChangeNotes> changeNotes = Optional.empty();
 
@@ -55,12 +60,14 @@
       CommentsUtil commentsUtil,
       ChangeFinder changeFinder,
       OneOffRequestContext oneOffReqCtx,
+      @GerritInstanceId String instanceId,
       @Assisted String changeId) {
     this.changeFinder = changeFinder;
     this.gitRepoMgr = gitRepoMgr;
     this.commentsUtil = commentsUtil;
     this.oneOffReqCtx = oneOffReqCtx;
     this.changeId = changeId;
+    this.instanceId = instanceId;
   }
 
   @Override
@@ -69,7 +76,8 @@
     return getComputedChangeTs()
         .map(
             ts -> {
-              ChangeIndexEvent event = new ChangeIndexEvent(projectName, changeId, deleted);
+              ChangeIndexEvent event =
+                  new ChangeIndexEvent(projectName, changeId, deleted, instanceId);
               event.eventCreatedOn = ts;
               event.targetSha = getBranchTargetSha();
               return event;
@@ -140,6 +148,34 @@
     }
   }
 
+  @Override
+  public boolean isChangeConsistent() {
+    Optional<ChangeNotes> notes = getChangeNotes();
+    if (!notes.isPresent()) {
+      log.warn("Unable to compute change notes for change {}", changeId);
+      return true;
+    }
+    ObjectId currentPatchSetCommitId = notes.get().getCurrentPatchSet().commitId();
+    try (Repository repo = gitRepoMgr.openRepository(changeNotes.get().getProjectName());
+        RevWalk walk = new RevWalk(repo)) {
+      walk.parseCommit(currentPatchSetCommitId);
+    } catch (StorageException | MissingObjectException e) {
+      log.warn(
+          String.format(
+              "Consistency check failed for change %s, missing current patchset commit %s",
+              changeId, currentPatchSetCommitId.getName()),
+          e);
+      return false;
+    } catch (IOException e) {
+      log.warn(
+          String.format(
+              "Cannot check consistency for change %s, current patchset commit %s. Assuming change is consistent",
+              changeId, currentPatchSetCommitId.getName()),
+          e);
+    }
+    return true;
+  }
+
   private Optional<Long> computeLastChangeTs() {
     return getChangeNotes().map(notes -> getTsFromChangeAndDraftComments(notes));
   }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/index/IndexEventHandler.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/index/IndexEventHandler.java
index eef3e4b..02f1b1c 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/index/IndexEventHandler.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/index/IndexEventHandler.java
@@ -21,6 +21,7 @@
 import com.google.gerrit.extensions.events.GroupIndexedListener;
 import com.google.gerrit.extensions.events.ProjectIndexedListener;
 import com.google.gerrit.extensions.registration.DynamicSet;
+import com.google.gerrit.server.config.GerritInstanceId;
 import com.google.inject.Inject;
 import com.googlesource.gerrit.plugins.multisite.forwarder.Context;
 import com.googlesource.gerrit.plugins.multisite.forwarder.ForwarderTask;
@@ -48,6 +49,7 @@
   private final ChangeCheckerImpl.Factory changeChecker;
   private final ProjectsFilter projectsFilter;
   private final GroupChecker groupChecker;
+  private final String instanceId;
 
   @Inject
   IndexEventHandler(
@@ -55,18 +57,20 @@
       DynamicSet<IndexEventForwarder> forwarders,
       ChangeCheckerImpl.Factory changeChecker,
       ProjectsFilter projectsFilter,
-      GroupChecker groupChecker) {
+      GroupChecker groupChecker,
+      @GerritInstanceId String instanceId) {
     this.forwarders = forwarders;
     this.executor = executor;
     this.changeChecker = changeChecker;
     this.projectsFilter = projectsFilter;
     this.groupChecker = groupChecker;
+    this.instanceId = instanceId;
   }
 
   @Override
   public void onAccountIndexed(int id) {
     if (!Context.isForwardedEvent()) {
-      IndexAccountTask task = new IndexAccountTask(new AccountIndexEvent(id));
+      IndexAccountTask task = new IndexAccountTask(new AccountIndexEvent(id, instanceId));
       if (queuedTasks.add(task)) {
         executor.execute(task);
       }
@@ -87,7 +91,8 @@
   public void onGroupIndexed(String groupUUID) {
     if (!Context.isForwardedEvent()) {
       IndexGroupTask task =
-          new IndexGroupTask(new GroupIndexEvent(groupUUID, groupChecker.getGroupHead(groupUUID)));
+          new IndexGroupTask(
+              new GroupIndexEvent(groupUUID, groupChecker.getGroupHead(groupUUID), instanceId));
       if (queuedTasks.add(task)) {
         executor.execute(task);
       }
@@ -97,7 +102,7 @@
   @Override
   public void onProjectIndexed(String projectName) {
     if (!Context.isForwardedEvent() && projectsFilter.matches(projectName)) {
-      IndexProjectTask task = new IndexProjectTask(new ProjectIndexEvent(projectName));
+      IndexProjectTask task = new IndexProjectTask(new ProjectIndexEvent(projectName, instanceId));
       if (queuedTasks.add(task)) {
         executor.execute(task);
       }
@@ -132,7 +137,7 @@
 
   private void executeDeleteChangeTask(int id) {
     if (!Context.isForwardedEvent()) {
-      IndexChangeTask task = new IndexChangeTask(new ChangeIndexEvent("", id, true));
+      IndexChangeTask task = new IndexChangeTask(new ChangeIndexEvent("", id, true, instanceId));
       if (queuedTasks.add(task)) {
         executor.execute(task);
       }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/MultisiteReplicationFetchFilter.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/MultisiteReplicationFetchFilter.java
new file mode 100644
index 0000000..d851420
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/MultisiteReplicationFetchFilter.java
@@ -0,0 +1,200 @@
+// Copyright (C) 2023 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.googlesource.gerrit.plugins.multisite.validation;
+
+import static com.googlesource.gerrit.plugins.multisite.validation.ProjectVersionRefUpdate.MULTI_SITE_VERSIONING_REF;
+import static com.googlesource.gerrit.plugins.replication.pull.PullReplicationLogger.repLog;
+
+import com.gerritforge.gerrit.globalrefdb.GlobalRefDbLockException;
+import com.gerritforge.gerrit.globalrefdb.validation.SharedRefDatabaseWrapper;
+import com.google.common.flogger.FluentLogger;
+import com.google.gerrit.entities.Project;
+import com.google.gerrit.entities.Project.NameKey;
+import com.google.gerrit.server.git.GitRepositoryManager;
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+import com.googlesource.gerrit.plugins.multisite.Configuration;
+import com.googlesource.gerrit.plugins.replication.pull.ReplicationFetchFilter;
+import java.io.IOException;
+import java.util.Collections;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.eclipse.jgit.lib.ObjectId;
+import org.eclipse.jgit.lib.Ref;
+import org.eclipse.jgit.lib.RefDatabase;
+import org.eclipse.jgit.lib.Repository;
+
+@Singleton
+public class MultisiteReplicationFetchFilter implements ReplicationFetchFilter {
+  private static final String ZERO_ID_NAME = ObjectId.zeroId().name();
+  private static final FluentLogger logger = FluentLogger.forEnclosingClass();
+
+  private final SharedRefDatabaseWrapper sharedRefDb;
+  private final GitRepositoryManager gitRepositoryManager;
+  private Configuration config;
+
+  @Inject
+  public MultisiteReplicationFetchFilter(
+      SharedRefDatabaseWrapper sharedRefDb,
+      GitRepositoryManager gitRepositoryManager,
+      Configuration config) {
+    this.sharedRefDb = sharedRefDb;
+    this.gitRepositoryManager = gitRepositoryManager;
+    this.config = config;
+  }
+
+  @Override
+  public Set<String> filter(String projectName, Set<String> refs) {
+    try (Repository repository =
+        gitRepositoryManager.openRepository(Project.nameKey(projectName))) {
+      RefDatabase refDb = repository.getRefDatabase();
+      return refs.stream()
+          .filter(ref -> !hasBeenRemovedFromGlobalRefDb(projectName, ref))
+          .filter(
+              ref -> {
+                if (shouldNotBeTrackedAnymoreOnGlobalRefDb(ref)) {
+                  return true;
+                }
+                Optional<ObjectId> localRefOid =
+                    getLocalSha1IfEqualsToExistingGlobalRefDb(
+                        repository, projectName, refDb, ref, true);
+                localRefOid.ifPresent(
+                    oid ->
+                        repLog.info(
+                            "{}:{}={} is already up-to-date with the shared-refdb and thus will NOT BE"
+                                + " fetched",
+                            projectName,
+                            ref,
+                            oid.getName()));
+
+                return !localRefOid.isPresent();
+              })
+          .collect(Collectors.toSet());
+    } catch (IOException ioe) {
+      String message = String.format("Error while opening project: '%s'", projectName);
+      repLog.error(message);
+      logger.atSevere().withCause(ioe).log(message);
+      return Collections.emptySet();
+    }
+  }
+
+  /*
+   * Since ac43a5f94c773c9db7a73d44035961d69d13fa53 the 'refs/multi-site/version' is
+   * not updated anymore on the global-refdb; however, the values stored already
+   * on the global-refdb could get in the way and prevent replication from happening
+   * as expected.
+   *
+   * Exclude the 'refs/multi-site/version' from local vs. global refdb checking
+   * pretending that the global-refdb for that ref did not exist.
+   */
+  private boolean shouldNotBeTrackedAnymoreOnGlobalRefDb(String ref) {
+    return MULTI_SITE_VERSIONING_REF.equals(ref);
+  }
+
+  /* If the ref to fetch has been set to all zeros on the global-refdb, it means
+   * that whatever is the situation locally, we do not need to fetch it:
+   * - If the remote still has it, fetching it will be useless because the global
+   *   state is that the ref should be removed.
+   * - If the remote doesn't have it anymore, trying to fetch the ref won't do
+   *   anything because you can't just remove local refs by fetching.
+   */
+  private boolean hasBeenRemovedFromGlobalRefDb(String projectName, String ref) {
+    if (foundAsZeroInSharedRefDb(Project.nameKey(projectName), ref)) {
+      repLog.info(
+          "{}:{} is found as zeros (removed) in shared-refdb thus will NOT BE fetched",
+          projectName,
+          ref);
+      return true;
+    }
+    return false;
+  }
+
+  private boolean foundAsZeroInSharedRefDb(NameKey projectName, String ref) {
+    return sharedRefDb
+        .get(projectName, ref, String.class)
+        .map(r -> ZERO_ID_NAME.equals(r))
+        .orElse(false);
+  }
+
+  private Optional<ObjectId> getLocalSha1IfEqualsToExistingGlobalRefDb(
+      Repository repository,
+      String projectName,
+      RefDatabase refDb,
+      String ref,
+      boolean retryWithRandomSleep) {
+    try {
+      Optional<ObjectId> localRefObjectId =
+          Optional.ofNullable(refDb.exactRef(ref))
+              .filter(
+                  r ->
+                      sharedRefDb
+                          .get(Project.nameKey(projectName), r.getName(), String.class)
+                          .map(sharedRefObjId -> r.getObjectId().getName().equals(sharedRefObjId))
+                          .orElse(false))
+              .map(Ref::getObjectId);
+
+      if (!localRefObjectId.isPresent() && retryWithRandomSleep) {
+        randomSleepForMitigatingConditionWhereLocalRefHaveJustBeenChanged(
+            projectName, localRefObjectId, ref);
+        localRefObjectId =
+            getLocalSha1IfEqualsToExistingGlobalRefDb(repository, projectName, refDb, ref, false);
+      }
+
+      return localRefObjectId;
+    } catch (GlobalRefDbLockException gle) {
+      String message = String.format("%s is locked on shared-refdb", ref);
+      repLog.error(message);
+      logger.atSevere().withCause(gle).log(message);
+      return Optional.empty();
+    } catch (IOException ioe) {
+      String message =
+          String.format("Error while extracting ref '%s' for project '%s'", ref, projectName);
+      repLog.error(message);
+      logger.atSevere().withCause(ioe).log(message);
+      return Optional.empty();
+    }
+  }
+
+  private void randomSleepForMitigatingConditionWhereLocalRefHaveJustBeenChanged(
+      String projectName, Optional<ObjectId> refObjectId, String ref) {
+    if (!config.replicationFilter().isFetchFilterRandomSleepEnabled()) {
+      repLog.debug(
+          "'{}' is not up-to-date for project '{}' [local='{}']. Random sleep is disabled,"
+              + " reload local ref without delay and re-check",
+          ref,
+          projectName,
+          refObjectId);
+      return;
+    }
+
+    int randomSleepTimeMsec = config.replicationFilter().fetchFilterRandomSleepTimeMs();
+    repLog.debug(
+        "'{}' is not up-to-date for project '{}' [local='{}']. Reload local ref in '{} ms' and"
+            + " re-check",
+        ref,
+        projectName,
+        refObjectId,
+        randomSleepTimeMsec);
+    try {
+      Thread.sleep(randomSleepTimeMsec);
+    } catch (InterruptedException ie) {
+      String message =
+          String.format("Error while waiting for next check for '%s', ref '%s'", projectName, ref);
+      repLog.error(message);
+      logger.atWarning().withCause(ie).log(message);
+    }
+  }
+}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/MultisiteReplicationPushFilter.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/MultisiteReplicationPushFilter.java
index 98f4897..6f9c2e7 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/MultisiteReplicationPushFilter.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/MultisiteReplicationPushFilter.java
@@ -22,12 +22,13 @@
 import com.google.gerrit.server.git.GitRepositoryManager;
 import com.google.inject.Inject;
 import com.google.inject.Singleton;
+import com.googlesource.gerrit.plugins.multisite.Configuration;
 import com.googlesource.gerrit.plugins.replication.ReplicationPushFilter;
 import java.io.IOException;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
-import java.util.Random;
+import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
 import org.eclipse.jgit.lib.ObjectId;
@@ -42,20 +43,22 @@
 public class MultisiteReplicationPushFilter implements ReplicationPushFilter {
   private static final FluentLogger logger = FluentLogger.forEnclosingClass();
   private static final String REF_META_SUFFIX = "/meta";
-  public static final int MIN_WAIT_BEFORE_RELOAD_LOCAL_VERSION_MS = 1000;
-  public static final int RANDOM_WAIT_BEFORE_RELOAD_LOCAL_VERSION_MS = 1000;
 
   static final String REPLICATION_LOG_NAME = "replication_log";
   static final Logger repLog = LoggerFactory.getLogger(REPLICATION_LOG_NAME);
 
   private final SharedRefDatabaseWrapper sharedRefDb;
   private final GitRepositoryManager gitRepositoryManager;
+  private Configuration config;
 
   @Inject
   public MultisiteReplicationPushFilter(
-      SharedRefDatabaseWrapper sharedRefDb, GitRepositoryManager gitRepositoryManager) {
+      SharedRefDatabaseWrapper sharedRefDb,
+      GitRepositoryManager gitRepositoryManager,
+      Configuration config) {
     this.sharedRefDb = sharedRefDb;
     this.gitRepositoryManager = gitRepositoryManager;
+    this.config = config;
   }
 
   @Override
@@ -66,19 +69,23 @@
         gitRepositoryManager.openRepository(Project.nameKey(projectName))) {
       List<RemoteRefUpdate> filteredRefUpdates =
           remoteUpdatesList.stream()
-              .filter(
+              .map(
                   refUpdate -> {
-                    boolean refUpToDate = isUpToDateWithRetry(projectName, repository, refUpdate);
-                    if (!refUpToDate) {
+                    Optional<RemoteRefUpdate> updatedRefUpdate =
+                        isUpToDateWithRetry(projectName, repository, refUpdate);
+                    if (!updatedRefUpdate.isPresent()) {
                       repLog.warn(
-                          "{} is not up-to-date with the shared-refdb and thus will NOT BE replicated",
+                          "{} is not up-to-date with the shared-refdb and thus will NOT BE"
+                              + " replicated",
                           refUpdate);
                       if (refUpdate.getSrcRef().endsWith(REF_META_SUFFIX)) {
                         outdatedChanges.add(getRootChangeRefPrefix(refUpdate.getSrcRef()));
                       }
                     }
-                    return refUpToDate;
+                    return updatedRefUpdate;
                   })
+              .filter(Optional::isPresent)
+              .map(Optional::get)
               .collect(Collectors.toList());
 
       return filteredRefUpdates.stream()
@@ -102,45 +109,67 @@
     }
   }
 
-  private boolean isUpToDateWithRetry(
+  private Optional<RemoteRefUpdate> isUpToDateWithRetry(
       String projectName, Repository repository, RemoteRefUpdate refUpdate) {
     String ref = refUpdate.getSrcRef();
     try {
       if (sharedRefDb.isUpToDate(
           Project.nameKey(projectName),
           new ObjectIdRef.Unpeeled(Ref.Storage.NETWORK, ref, refUpdate.getNewObjectId()))) {
-        return true;
+        return Optional.of(refUpdate);
       }
 
       randomSleepForMitigatingConditionWhereLocalRefHaveJustBeenChanged(
           projectName, refUpdate, ref);
 
+      ObjectId reloadedNewObjectId = getNotNullExactRef(repository, ref);
+      RemoteRefUpdate refUpdateReloaded =
+          newRemoteRefUpdateWithObjectId(repository, refUpdate, reloadedNewObjectId);
       return sharedRefDb.isUpToDate(
-          Project.nameKey(projectName),
-          new ObjectIdRef.Unpeeled(Ref.Storage.NETWORK, ref, getNotNullExactRef(repository, ref)));
+              Project.nameKey(projectName),
+              new ObjectIdRef.Unpeeled(
+                  Ref.Storage.NETWORK, ref, refUpdateReloaded.getNewObjectId()))
+          ? Optional.of(refUpdateReloaded)
+          : Optional.empty();
     } catch (GlobalRefDbLockException gle) {
       String message =
           String.format("%s is locked on shared-refdb and thus will NOT BE replicated", ref);
       repLog.error(message);
       logger.atSevere().withCause(gle).log(message);
-      return false;
+      return Optional.empty();
     } catch (IOException ioe) {
       String message =
           String.format("Error while extracting ref '%s' for project '%s'", ref, projectName);
       repLog.error(message);
       logger.atSevere().withCause(ioe).log(message);
-      return false;
+      return Optional.empty();
     }
   }
 
+  private RemoteRefUpdate newRemoteRefUpdateWithObjectId(
+      Repository localDb, RemoteRefUpdate refUpdate, ObjectId reloadedNewObjectId)
+      throws IOException {
+    return new RemoteRefUpdate(
+        localDb,
+        refUpdate.getSrcRef(),
+        reloadedNewObjectId,
+        refUpdate.getRemoteName(),
+        refUpdate.isForceUpdate(),
+        null,
+        refUpdate.getExpectedOldObjectId());
+  }
+
   private void randomSleepForMitigatingConditionWhereLocalRefHaveJustBeenChanged(
       String projectName, RemoteRefUpdate refUpdate, String ref) {
-    int randomSleepTimeMsec =
-        MIN_WAIT_BEFORE_RELOAD_LOCAL_VERSION_MS
-            + new Random().nextInt(RANDOM_WAIT_BEFORE_RELOAD_LOCAL_VERSION_MS);
+    if (!config.replicationFilter().isPushFilterRandomSleepEnabled()) {
+      return;
+    }
+
+    int randomSleepTimeMsec = config.replicationFilter().pushFilterRandomSleepTimeMs();
     repLog.debug(
         String.format(
-            "'%s' is not up-to-date for project '%s' [local='%s']. Reload local ref in '%d ms' and re-check",
+            "'%s' is not up-to-date for project '%s' [local='%s']. Reload local ref in '%d ms' and"
+                + " re-check",
             ref, projectName, refUpdate.getNewObjectId(), randomSleepTimeMsec));
     try {
       Thread.sleep(randomSleepTimeMsec);
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/ProjectVersionRefUpdateImpl.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/ProjectVersionRefUpdateImpl.java
index 469122b..e07fef0 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/ProjectVersionRefUpdateImpl.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/ProjectVersionRefUpdateImpl.java
@@ -20,27 +20,27 @@
 import com.gerritforge.gerrit.globalrefdb.GlobalRefDbSystemError;
 import com.gerritforge.gerrit.globalrefdb.validation.ProjectsFilter;
 import com.gerritforge.gerrit.globalrefdb.validation.SharedRefDatabaseWrapper;
+import com.google.common.base.Predicates;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.flogger.FluentLogger;
 import com.google.gerrit.entities.Project;
+import com.google.gerrit.entities.Project.NameKey;
 import com.google.gerrit.entities.RefNames;
 import com.google.gerrit.server.events.Event;
 import com.google.gerrit.server.events.EventListener;
 import com.google.gerrit.server.events.RefUpdatedEvent;
 import com.google.gerrit.server.extensions.events.GitReferenceUpdated;
 import com.google.gerrit.server.git.GitRepositoryManager;
-import com.google.gerrit.server.notedb.IntBlob;
 import com.google.inject.Inject;
 import com.googlesource.gerrit.plugins.multisite.ProjectVersionLogger;
 import com.googlesource.gerrit.plugins.multisite.forwarder.Context;
 import java.io.IOException;
+import java.nio.charset.StandardCharsets;
 import java.util.Optional;
 import java.util.Set;
 import org.eclipse.jgit.errors.RepositoryNotFoundException;
 import org.eclipse.jgit.lib.ObjectId;
-import org.eclipse.jgit.lib.ObjectIdRef;
 import org.eclipse.jgit.lib.ObjectInserter;
-import org.eclipse.jgit.lib.Ref;
 import org.eclipse.jgit.lib.RefUpdate;
 import org.eclipse.jgit.lib.Repository;
 
@@ -106,8 +106,7 @@
       if (newProjectVersionRefUpdate.isPresent()) {
         verLogger.log(projectNameKey, newVersion, 0L);
 
-        if (updateSharedProjectVersion(
-            projectNameKey, newProjectVersionRefUpdate.get().getNewObjectId(), newVersion)) {
+        if (updateSharedProjectVersion(projectNameKey, newVersion)) {
           gitReferenceUpdated.fire(projectNameKey, newProjectVersionRefUpdate.get(), null);
         }
       } else {
@@ -137,20 +136,9 @@
     return newId;
   }
 
-  private boolean updateSharedProjectVersion(
-      Project.NameKey projectNameKey, ObjectId newObjectId, Long newVersion)
+  private boolean updateSharedProjectVersion(Project.NameKey projectNameKey, Long newVersion)
       throws SharedProjectVersionUpdateException {
 
-    Ref sharedRef =
-        sharedRefDb
-            .get(projectNameKey, MULTI_SITE_VERSIONING_REF, String.class)
-            .map(
-                (String objectId) ->
-                    new ObjectIdRef.Unpeeled(
-                        Ref.Storage.NEW, MULTI_SITE_VERSIONING_REF, ObjectId.fromString(objectId)))
-            .orElse(
-                new ObjectIdRef.Unpeeled(
-                    Ref.Storage.NEW, MULTI_SITE_VERSIONING_REF, ObjectId.zeroId()));
     Optional<Long> sharedVersion =
         sharedRefDb
             .get(projectNameKey, MULTI_SITE_VERSIONING_VALUE_REF, String.class)
@@ -160,59 +148,50 @@
       if (sharedVersion.isPresent() && sharedVersion.get() >= newVersion) {
         logger.atWarning().log(
             String.format(
-                "NOT Updating project %s version %s (value=%d) in shared ref-db because is more recent than the local one %s (value=%d) ",
-                projectNameKey.get(),
-                newObjectId,
-                newVersion,
-                sharedRef.getObjectId().getName(),
-                sharedVersion.get()));
+                "NOT Updating project %s value=%d in shared ref-db because is more recent than the local value=%d",
+                projectNameKey.get(), newVersion, sharedVersion.get()));
         return false;
       }
 
       logger.atFine().log(
           String.format(
-              "Updating shared project %s version to %s (value=%d)",
-              projectNameKey.get(), newObjectId, newVersion));
+              "Updating shared project %s value to %d", projectNameKey.get(), newVersion));
 
-      boolean success = sharedRefDb.compareAndPut(projectNameKey, sharedRef, newObjectId);
-      if (!success) {
-        String message =
-            String.format(
-                "Project version blob update failed for %s. Current value %s, new value: %s",
-                projectNameKey.get(), safeGetObjectId(sharedRef), newObjectId);
-        logger.atSevere().log(message);
-        throw new SharedProjectVersionUpdateException(message);
-      }
-
-      success =
-          sharedRefDb.compareAndPut(
-              projectNameKey,
-              MULTI_SITE_VERSIONING_VALUE_REF,
-              sharedVersion.map(Object::toString).orElse(null),
-              newVersion.toString());
-      if (!success) {
-        String message =
-            String.format(
-                "Project version update failed for %s. Current value %s, new value: %s",
-                projectNameKey.get(), safeGetObjectId(sharedRef), newObjectId);
-        logger.atSevere().log(message);
-        throw new SharedProjectVersionUpdateException(message);
-      }
-
+      updateProjectVersionValue(projectNameKey, newVersion, sharedVersion);
       return true;
     } catch (GlobalRefDbSystemError refDbSystemError) {
       String message =
           String.format(
-              "Error while updating shared project version for %s. Current value %s, new value: %s. Error: %s",
+              "Error while updating shared project value for %s. Current value %s, new value: %s. Error: %s",
               projectNameKey.get(),
-              sharedRef.getObjectId(),
-              newObjectId,
+              sharedVersion.map(Object::toString).orElse(null),
+              newVersion,
               refDbSystemError.getMessage());
       logger.atSevere().withCause(refDbSystemError).log(message);
       throw new SharedProjectVersionUpdateException(message);
     }
   }
 
+  private void updateProjectVersionValue(
+      NameKey projectNameKey, Long newVersion, Optional<Long> sharedVersion) {
+    try {
+      if (sharedRefDb.isSetOperationSupported()) {
+        sharedRefDb.put(projectNameKey, MULTI_SITE_VERSIONING_VALUE_REF, newVersion.toString());
+        return;
+      }
+    } catch (NoSuchMethodError e) {
+      logger.atSevere().log(
+          "Global-refdb library is outdated and is not supporting "
+              + "'put' method, update global-refdb to the newest version. Falling back to 'compareAndPut'");
+    }
+
+    sharedRefDb.compareAndPut(
+        projectNameKey,
+        MULTI_SITE_VERSIONING_VALUE_REF,
+        sharedVersion.map(Object::toString).orElse(null),
+        newVersion.toString());
+  }
+
   /* (non-Javadoc)
    * @see com.googlesource.gerrit.plugins.multisite.validation.ProjectVersionRefUpdate#getProjectLocalVersion(java.lang.String)
    */
@@ -220,9 +199,9 @@
   public Optional<Long> getProjectLocalVersion(String projectName) {
     try (Repository repository =
         gitRepositoryManager.openRepository(Project.NameKey.parse(projectName))) {
-      Optional<IntBlob> blob = IntBlob.parse(repository, MULTI_SITE_VERSIONING_REF);
+      Optional<Long> blob = longBlobParse(repository, MULTI_SITE_VERSIONING_REF);
       if (blob.isPresent()) {
-        Long repoVersion = Integer.toUnsignedLong(blob.get().value());
+        Long repoVersion = blob.get();
         logger.atFine().log("Local project '%s' has version %d", projectName, repoVersion);
         return Optional.of(repoVersion);
       }
@@ -234,6 +213,22 @@
     return Optional.empty();
   }
 
+  private Optional<Long> longBlobParse(Repository repo, String refName) throws IOException {
+    return Optional.ofNullable(repo.exactRef(refName))
+        .map(
+            (ref) -> {
+              try {
+                return Long.parseLong(
+                    new String(repo.open(ref.getObjectId()).getBytes(), StandardCharsets.UTF_8));
+              } catch (IOException e) {
+                logger.atSevere().withCause(e).log(
+                    "Unable to extract long BLOB from %s:%s", repo.getDirectory(), ref);
+                return null;
+              }
+            })
+        .filter(Predicates.notNull());
+  }
+
   /* (non-Javadoc)
    * @see com.googlesource.gerrit.plugins.multisite.validation.ProjectVersionRefUpdate#getProjectRemoteVersion(java.lang.String)
    */
@@ -245,10 +240,6 @@
     return globalVersion.flatMap(longString -> getLongValueOf(longString));
   }
 
-  private Object safeGetObjectId(Ref currentRef) {
-    return currentRef == null ? "null" : currentRef.getObjectId();
-  }
-
   private Optional<Long> getLongValueOf(String longString) {
     try {
       return Optional.ofNullable(Long.parseLong(longString));
@@ -286,7 +277,7 @@
   }
 
   private long getCurrentGlobalVersionNumber() {
-    return System.currentTimeMillis() / 1000;
+    return System.currentTimeMillis();
   }
 
   private Boolean isSuccessful(RefUpdate.Result result) {
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/PullReplicationFilterModule.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/PullReplicationFilterModule.java
new file mode 100644
index 0000000..b8ad99d
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/PullReplicationFilterModule.java
@@ -0,0 +1,28 @@
+// Copyright (C) 2023 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.googlesource.gerrit.plugins.multisite.validation;
+
+import com.google.gerrit.extensions.registration.DynamicItem;
+import com.google.inject.AbstractModule;
+import com.googlesource.gerrit.plugins.replication.pull.ReplicationFetchFilter;
+
+public class PullReplicationFilterModule extends AbstractModule {
+
+  @Override
+  protected void configure() {
+    DynamicItem.bind(binder(), ReplicationFetchFilter.class)
+        .to(MultisiteReplicationFetchFilter.class);
+  }
+}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/PushReplicationFilterModule.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/PushReplicationFilterModule.java
new file mode 100644
index 0000000..8a8e826
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/PushReplicationFilterModule.java
@@ -0,0 +1,28 @@
+// Copyright (C) 2023 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.googlesource.gerrit.plugins.multisite.validation;
+
+import com.google.gerrit.extensions.registration.DynamicItem;
+import com.google.inject.AbstractModule;
+import com.googlesource.gerrit.plugins.replication.ReplicationPushFilter;
+
+public class PushReplicationFilterModule extends AbstractModule {
+
+  @Override
+  protected void configure() {
+    DynamicItem.bind(binder(), ReplicationPushFilter.class)
+        .to(MultisiteReplicationPushFilter.class);
+  }
+}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/ValidationModule.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/ValidationModule.java
index 8601cf2..fc4d505 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/ValidationModule.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/ValidationModule.java
@@ -30,14 +30,11 @@
 import com.gerritforge.gerrit.globalrefdb.validation.dfsrefdb.SharedRefEnforcement;
 import com.google.common.collect.ImmutableSet;
 import com.google.gerrit.extensions.config.FactoryModule;
-import com.google.gerrit.extensions.registration.DynamicItem;
 import com.google.gerrit.server.config.RepositoryConfig;
 import com.google.inject.Scopes;
 import com.google.inject.TypeLiteral;
 import com.google.inject.name.Names;
 import com.googlesource.gerrit.plugins.multisite.Configuration;
-import com.googlesource.gerrit.plugins.replication.ReplicationExtensionPointModule;
-import com.googlesource.gerrit.plugins.replication.ReplicationPushFilter;
 
 public class ValidationModule extends FactoryModule {
   private final Configuration cfg;
@@ -50,8 +47,6 @@
 
   @Override
   protected void configure() {
-    install(new ReplicationExtensionPointModule());
-
     bind(SharedRefDatabaseWrapper.class).in(Scopes.SINGLETON);
     bind(SharedRefLogger.class).to(Log4jSharedRefLogger.class);
     factory(LockWrapper.Factory.class);
@@ -70,8 +65,6 @@
                 ProjectVersionRefUpdate.MULTI_SITE_VERSIONING_REF,
                 ProjectVersionRefUpdate.MULTI_SITE_VERSIONING_VALUE_REF));
     install(new RepositoryManagerModule(repoConfig));
-    DynamicItem.bind(binder(), ReplicationPushFilter.class)
-        .to(MultisiteReplicationPushFilter.class);
 
     if (cfg.getSharedRefDbConfiguration().getSharedRefDb().getEnforcementRules().isEmpty()) {
       bind(SharedRefEnforcement.class).to(DefaultSharedRefEnforcement.class).in(Scopes.SINGLETON);
diff --git a/src/main/resources/Documentation/about.md b/src/main/resources/Documentation/about.md
index 623871b..d7c5648 100644
--- a/src/main/resources/Documentation/about.md
+++ b/src/main/resources/Documentation/about.md
@@ -7,7 +7,8 @@
 message broker for aligning with the other masters over different sites.
 
 The masters must be:
-
+* Gerrit instance id is mandatory for @PLUGIN@ plugin. All the master 
+  must have [gerrit.instanceId](https://gerrit-review.googlesource.com/Documentation/config-gerrit.html#gerrit) populated.
 * events-broker library must be installed as a library module in the
   `$GERRIT_SITE/lib` directory of all the masters
 * global-refdb library must be installed as a library module in the
@@ -98,3 +99,7 @@
 * Subscriber replication lag (sec behind the producer)
 
 `metric=site/multi_site/subscriber/subscriber_replication_status/sec_behind, type=com.google.gerrit.metrics.dropwizard.CallbackMetricImpl`
+
+* Subscriber replication lag (millisec behind the producer)
+
+`metric=site/multi_site/subscriber/subscriber_replication_status/msec_behind, type=com.google.gerrit.metrics.dropwizard.CallbackMetricImpl`
diff --git a/src/main/resources/Documentation/build.md b/src/main/resources/Documentation/build.md
index bb38288..146341e 100644
--- a/src/main/resources/Documentation/build.md
+++ b/src/main/resources/Documentation/build.md
@@ -3,15 +3,29 @@
 This plugin can be built with Bazel in the Gerrit tree.
 
 Clone or link this plugin to the plugins directory of Gerrit's
-source tree. Put the external dependency Bazel build file into
-the Gerrit /plugins directory, replacing the existing empty one.
+source tree.
+
+Clone the [pull-replication](https://gerrit.googlesource.com/plugins/pull-replication) on
+the same branch of the @PLUGIN@ plugin and link it to the `gerrit/plugins` directory.
 
 ```
+  export BRANCH=$(git --git-dir=@PLUGIN@ branch)
+  git clone https://gerrit.googlesource.com/plugins/pull-replication
   cd gerrit/plugins
   rm external_plugin_deps.bzl
   ln -s @PLUGIN@/external_plugin_deps.bzl .
 ```
 
+Clone the [global-refdb](git clone "https://gerrit.googlesource.com/modules/global-refdb") on
+the same branch of the @PLUGIN@ plugin and link it to the `gerrit/plugins` directory.
+
+```
+  export BRANCH=$(git --git-dir=@PLUGIN@ branch)
+  git clone "https://gerrit.googlesource.com/modules/global-refdb"
+  cd gerrit/plugins
+  ln -s ../../global-refdb .
+```
+
 From the Gerrit source tree issue the command:
 
 ```
diff --git a/src/main/resources/Documentation/config.md b/src/main/resources/Documentation/config.md
index 02806a1..e451cb3 100644
--- a/src/main/resources/Documentation/config.md
+++ b/src/main/resources/Documentation/config.md
@@ -3,15 +3,19 @@
 =========================
 
 The @PLUGIN@ plugin must be installed as a library module in the
-`$GERRIT_SITE/lib` folder of all the instances. Configuration should
-be specified in the `$site_path/etc/@PLUGIN@.config` file.
+`$GERRIT_SITE/lib` folder of all the instances and linked to
+`$GERRIT_SITE/plugins`, which enable to use it as both libModule
+and plugin.
+Configuration should be specified in the `$site_path/etc/@PLUGIN@.config` file.
 
 ## Configuration parameters
 
 ```cache.synchronize```
-:   Whether to run cache evictions synchronously. It requires disabling the
-    background cache evictions notifications in `gerrit.config` by setting
-    `cache.threads = 0`.
+:   Whether to synchronize cache evictions. Set to false when relying on
+    low cache TTLs and therefore cache eviction is not strictly needed.
+    It requires disabling the background cache evictions notifications in
+    `gerrit.config` by setting `cache.threads = 0`.
+
     Defaults to true.
 
 ```cache.threadPoolSize```
@@ -29,7 +33,8 @@
     forwarded.
 
 ```event.synchronize```
-:   Whether to synchronize stream events.
+:   Whether to synchronize stream events. Set to false when not using the SSH
+    stream events.
     Defaults to true.
 
 ```index.numStripedLocks```
@@ -37,7 +42,9 @@
     Defaults to 10
 
 ```index.synchronize```
-:   Whether to synchronize secondary indexes.
+:   Whether to synchronize secondary indexes. Set to false when using multi-site
+    on Gerrit replicas that do not have an index, or when using an external
+    service such as ElasticSearch.
     Defaults to true.
 
 ```index.threadPoolSize```
@@ -72,10 +79,19 @@
 :   Name of the topic to use for publishing cache eviction events
     Defaults to GERRIT.EVENT.PROJECT.LIST
 
+**NOTE**: All broker settings are ignored when all of the `cache`,
+`index` or `event` synchronization is disabled.
+
 ```ref-database.enabled```
 :   Enable the use of a shared ref-database
     Defaults: true
 
+```ref-database.replicationLagRefreshInterval```
+:   Enable the auto-refresh of the metrics to trace the auto-replication
+    lag by polling on a regular basis. Set to zero for disabling the polling
+    mechanism.
+    Defaults: 60 min
+
 ```ref-database.enforcementRules.<policy>```
 :   Level of consistency enforcement across sites on a project:refs basis.
     Supports two values for enforcing the policy on multiple projects or refs.
@@ -123,3 +139,80 @@
     the project `foo/bar`, but no other project.
 
     By default, all projects are matched.
+
+```replication.push-filter.minWaitBeforeReloadLocalVersionMs```
+:   Specifies the minimum amount of time in milliseconds replication plugin filter will
+    wait before retrying check for ref which is not up to date with global-refdb.
+
+    By default: 1000 milliseconds
+
+```replication.push-filter.maxRandomWaitBeforeReloadLocalVersionMs```
+:   Specifies the additional amount of time in milliseconds replication filter will
+    wait before retrying check for ref which is not up to date with global-refdb.
+
+    If maxRandomWaitBeforeReloadLocalVersionMs is set to zero random sleep for not in sync
+    refs is disabled.
+
+    By default: 1000 milliseconds
+
+```replication.fetch-filter.minWaitBeforeReloadLocalVersionMs```
+:   Specifies the minimum amount of time in milliseconds pull-replication filter wait
+    before retrying check for ref which is not up to date with global-refdb.
+
+    By default: 1000 milliseconds
+
+```replication.fetch-filter.maxRandomWaitBeforeReloadLocalVersionMs```
+:   Specifies the additional amount of time in milliseconds pull-replication filter will
+    wait before retrying check for ref which is not up to date with global-refdb.
+
+    If maxRandomWaitBeforeReloadLocalVersionMs is set to zero random sleep for not in sync
+    refs is disabled.
+
+    By default: 1000 milliseconds
+
+## Replication filters
+
+The @PLUGIN@ plugin is also responsible for filtering out replication events that may
+risk to create a split-brain situation.
+It integrates the push and pull replication filtering extension points for validating
+the refs to be replicated and dropping some of them.
+
+**Replication plugin**
+
+When using the Gerrit core replication plugin, also known as push-replication, link the
+`replication.jar` to the `$GERRIT_SITE/lib` directory and add the following libModule
+to `gerrit.config`:
+
+```
+[gerrit]
+  installModule = com.googlesource.gerrit.plugins.replication.ReplicationExtensionPointModule
+```
+
+The above configuration would be automatically detected by the @PLUGIN@ plugin which would then
+install the PushReplicationFilterModule for filtering outgoing replication refs based
+on their global-refdb status:
+
+- Outgoing replication of refs that are NOT up-to-date with the global-refdb will be
+  discarded, because they may cause split-brain on the remote replication endpoints.
+
+- All other refs will be pushed as normal to the remote replication ends.
+
+**Pull-replication plugin**
+
+When using the [pull-replication](https://gerrit.googlesource.com/plugins/pull-replication)
+plugin, link the `pull-replication.jar` to the `$GERRIT_SITE/lib` directory and add the following
+two libModules to `gerrit.config`:
+
+```
+[gerrit]
+        installModule = com.googlesource.gerrit.plugins.replication.pull.ReplicationExtensionPointModule
+```
+
+The above configuration would be automatically detected by the @PLUGIN@ plugin which would then
+install the PullReplicationFilterModule for filtering incoming fetch replication refs based
+on their global-refdb status.
+
+- Incoming replication of refs that locally are already up-to-date with the global-refdb will be
+  discarded, because they would not add anything more to the current status of the local refs.
+
+- All other refs will be fetched as normal from the replication sources.
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/ModuleTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/ModuleTest.java
deleted file mode 100644
index 2df60dd..0000000
--- a/src/test/java/com/googlesource/gerrit/plugins/multisite/ModuleTest.java
+++ /dev/null
@@ -1,65 +0,0 @@
-// Copyright (C) 2017 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.multisite;
-
-import static com.google.common.truth.Truth.assertThat;
-
-import com.google.gerrit.server.config.SitePaths;
-import java.io.File;
-import java.nio.file.Path;
-import java.nio.file.Paths;
-import java.util.UUID;
-import org.junit.Before;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
-import org.junit.runner.RunWith;
-import org.mockito.Answers;
-import org.mockito.Mock;
-import org.mockito.junit.MockitoJUnitRunner;
-
-@RunWith(MockitoJUnitRunner.class)
-public class ModuleTest {
-
-  @Mock(answer = Answers.RETURNS_DEEP_STUBS)
-  private Configuration configMock;
-
-  @Rule public TemporaryFolder tempFolder = new TemporaryFolder();
-
-  private Module module;
-
-  @Before
-  public void setup() {
-    module = new Module(configMock);
-  }
-
-  @Test
-  public void shouldGetInstanceId() throws Exception {
-    File tmpSitePath = tempFolder.newFolder();
-    File tmpPluginDataPath =
-        Paths.get(tmpSitePath.getPath(), "data", Configuration.PLUGIN_NAME).toFile();
-    tmpPluginDataPath.mkdirs();
-    Path path = Paths.get(tmpPluginDataPath.getPath(), Configuration.INSTANCE_ID_FILE);
-    SitePaths sitePaths = new SitePaths(Paths.get(tmpSitePath.getPath()));
-    assertThat(path.toFile().exists()).isFalse();
-
-    UUID gotUUID1 = module.getInstanceId(sitePaths);
-    assertThat(gotUUID1).isNotNull();
-    assertThat(path.toFile().exists()).isTrue();
-
-    UUID gotUUID2 = module.getInstanceId(sitePaths);
-    assertThat(gotUUID1).isEqualTo(gotUUID2);
-  }
-}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/broker/BrokerApiWrapperTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/broker/BrokerApiWrapperTest.java
index 92fa101..7d1751c 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/multisite/broker/BrokerApiWrapperTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/broker/BrokerApiWrapperTest.java
@@ -1,15 +1,19 @@
 package com.googlesource.gerrit.plugins.multisite.broker;
 
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.eq;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.only;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 import com.gerritforge.gerrit.eventbroker.BrokerApi;
+import com.google.common.util.concurrent.MoreExecutors;
+import com.google.common.util.concurrent.SettableFuture;
 import com.google.gerrit.extensions.registration.DynamicItem;
 import com.google.gerrit.server.events.Event;
+import com.google.gerrit.server.events.ProjectCreatedEvent;
 import com.googlesource.gerrit.plugins.multisite.MessageLogger;
-import java.util.UUID;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -22,28 +26,35 @@
   @Mock private BrokerApi brokerApi;
   @Mock Event event;
   @Mock MessageLogger msgLog;
-  private UUID instanceId = UUID.randomUUID();
   private String topic = "index";
 
   private BrokerApiWrapper objectUnderTest;
 
   @Before
   public void setUp() {
+    event.instanceId = "instance-id";
     objectUnderTest =
         new BrokerApiWrapper(
-            DynamicItem.itemOf(BrokerApi.class, brokerApi), brokerMetrics, msgLog, instanceId);
+            MoreExecutors.directExecutor(),
+            DynamicItem.itemOf(BrokerApi.class, brokerApi),
+            brokerMetrics,
+            msgLog);
   }
 
   @Test
   public void shouldIncrementBrokerMetricCounterWhenMessagePublished() {
-    when(brokerApi.send(any(), any())).thenReturn(true);
+    SettableFuture<Boolean> resultF = SettableFuture.create();
+    resultF.set(true);
+    when(brokerApi.send(any(), any())).thenReturn(resultF);
     objectUnderTest.send(topic, event);
     verify(brokerMetrics, only()).incrementBrokerPublishedMessage();
   }
 
   @Test
   public void shouldIncrementBrokerFailedMetricCounterWhenMessagePublishingFailed() {
-    when(brokerApi.send(any(), any())).thenReturn(false);
+    SettableFuture<Boolean> resultF = SettableFuture.create();
+    resultF.setException(new Exception("Force Future failure"));
+    when(brokerApi.send(any(), any())).thenReturn(resultF);
     objectUnderTest.send(topic, event);
     verify(brokerMetrics, only()).incrementBrokerFailedToPublishMessage();
   }
@@ -53,10 +64,26 @@
     when(brokerApi.send(any(), any()))
         .thenThrow(new RuntimeException("Unexpected runtime exception"));
     try {
-      objectUnderTest.send(topic, event);
+      objectUnderTest.sendSync(topic, event);
     } catch (RuntimeException e) {
       // expected
     }
     verify(brokerMetrics, only()).incrementBrokerFailedToPublishMessage();
   }
+
+  @Test
+  public void shouldSkipMessageSendingWhenInstanceIdIsNull() {
+    ProjectCreatedEvent event = new ProjectCreatedEvent();
+    event.instanceId = null;
+    objectUnderTest.send(topic, event);
+    verify(brokerApi, never()).send(any(), eq(event));
+  }
+
+  @Test
+  public void shouldSkipMessageSendingWhenInstanceIdIsEmpty() {
+    ProjectCreatedEvent event = new ProjectCreatedEvent();
+    event.instanceId = "";
+    objectUnderTest.send(topic, event);
+    verify(brokerApi, never()).send(any(), eq(event));
+  }
 }
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/cache/CacheEvictionHandlerTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/cache/CacheEvictionHandlerTest.java
index 67be583..ce222d0 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/multisite/cache/CacheEvictionHandlerTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/cache/CacheEvictionHandlerTest.java
@@ -36,9 +36,10 @@
 
   @Test
   public void shouldNotPublishAccountsCacheEvictions() {
-
+    String instanceId = "instance-id";
     final CacheEvictionHandler<String, String> handler =
-        new CacheEvictionHandler<>(DynamicSet.emptySet(), executorMock, defaultCacheMatcher);
+        new CacheEvictionHandler<>(
+            DynamicSet.emptySet(), executorMock, defaultCacheMatcher, instanceId);
 
     handler.onRemoval(
         "test", "accounts", RemovalNotification.create("test", "accounts", RemovalCause.EXPLICIT));
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/cache/ProjectListUpdateHandlerTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/cache/ProjectListUpdateHandlerTest.java
index 4263ddb..c8216bd 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/multisite/cache/ProjectListUpdateHandlerTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/cache/ProjectListUpdateHandlerTest.java
@@ -41,6 +41,8 @@
 @RunWith(MockitoJUnitRunner.class)
 public class ProjectListUpdateHandlerTest {
 
+  private static final String INSTANCE_ID = "instance-id";
+
   private ProjectListUpdateHandler handler;
 
   @Mock private ProjectListUpdateForwarder forwarder;
@@ -51,7 +53,7 @@
     when(projectsFilter.matches(any(String.class))).thenReturn(true);
     handler =
         new ProjectListUpdateHandler(
-            asDynamicSet(forwarder), MoreExecutors.directExecutor(), projectsFilter);
+            asDynamicSet(forwarder), MoreExecutors.directExecutor(), projectsFilter, INSTANCE_ID);
   }
 
   private DynamicSet<ProjectListUpdateForwarder> asDynamicSet(
@@ -69,7 +71,8 @@
     handler.onNewProjectCreated(event);
     verify(forwarder)
         .updateProjectList(
-            any(ProjectListUpdateTask.class), eq(new ProjectListUpdateEvent(projectName, false)));
+            any(ProjectListUpdateTask.class),
+            eq(new ProjectListUpdateEvent(projectName, false, INSTANCE_ID)));
   }
 
   @Test
@@ -80,7 +83,8 @@
     handler.onProjectDeleted(event);
     verify(forwarder)
         .updateProjectList(
-            any(ProjectListUpdateTask.class), eq(new ProjectListUpdateEvent(projectName, true)));
+            any(ProjectListUpdateTask.class),
+            eq(new ProjectListUpdateEvent(projectName, true, INSTANCE_ID)));
   }
 
   @Test
@@ -101,18 +105,22 @@
     handler.onNewProjectCreated(event);
     verify(forwarder, never())
         .updateProjectList(
-            any(ProjectListUpdateTask.class), eq(new ProjectListUpdateEvent(projectName, true)));
+            any(ProjectListUpdateTask.class),
+            eq(new ProjectListUpdateEvent(projectName, true, INSTANCE_ID)));
   }
 
   @Test
   public void testProjectUpdateTaskToString() throws Exception {
     String projectName = "someProjectName";
     ProjectListUpdateTask task =
-        handler.new ProjectListUpdateTask(new ProjectListUpdateEvent(projectName, false));
+        handler
+        .new ProjectListUpdateTask(new ProjectListUpdateEvent(projectName, false, INSTANCE_ID));
     assertThat(task.toString())
         .isEqualTo(String.format("Update project list in target instance: add '%s'", projectName));
 
-    task = handler.new ProjectListUpdateTask(new ProjectListUpdateEvent(projectName, true));
+    task =
+        handler
+        .new ProjectListUpdateTask(new ProjectListUpdateEvent(projectName, true, INSTANCE_ID));
     assertThat(task.toString())
         .isEqualTo(
             String.format("Update project list in target instance: remove '%s'", projectName));
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/AbstractSubscriberTestBase.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/AbstractSubscriberTestBase.java
new file mode 100644
index 0000000..cb1f78e
--- /dev/null
+++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/AbstractSubscriberTestBase.java
@@ -0,0 +1,162 @@
+// Copyright (C) 2021 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.multisite.consumer;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.reset;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import com.gerritforge.gerrit.globalrefdb.validation.ProjectsFilter;
+import com.google.gerrit.extensions.registration.DynamicSet;
+import com.google.gerrit.server.events.Event;
+import com.google.gerrit.server.permissions.PermissionBackendException;
+import com.googlesource.gerrit.plugins.multisite.Configuration;
+import com.googlesource.gerrit.plugins.multisite.Configuration.Broker;
+import com.googlesource.gerrit.plugins.multisite.MessageLogger;
+import com.googlesource.gerrit.plugins.multisite.forwarder.CacheNotFoundException;
+import com.googlesource.gerrit.plugins.multisite.forwarder.router.ForwardedEventRouter;
+import java.io.IOException;
+import java.util.List;
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+@RunWith(MockitoJUnitRunner.class)
+@Ignore
+public abstract class AbstractSubscriberTestBase {
+  protected static final String NODE_INSTANCE_ID = "node-instance-id";
+  protected static final String INSTANCE_ID = "other-node-instance-id";
+  protected static final String PROJECT_NAME = "project-name";
+
+  @Mock protected DroppedEventListener droppedEventListeners;
+  @Mock protected MessageLogger msgLog;
+  @Mock protected SubscriberMetrics subscriberMetrics;
+  @Mock protected Configuration cfg;
+  @Mock protected Broker brokerCfg;
+  @Mock protected ProjectsFilter projectsFilter;
+
+  @SuppressWarnings("rawtypes")
+  protected ForwardedEventRouter eventRouter;
+
+  protected AbstractSubcriber objectUnderTest;
+
+  @Before
+  public void setup() {
+    when(cfg.broker()).thenReturn(brokerCfg);
+    when(brokerCfg.getTopic(any(), any())).thenReturn("test-topic");
+    eventRouter = eventRouter();
+    objectUnderTest = objectUnderTest();
+  }
+
+  @Test
+  public void shouldConsumeEventsWhenNotFilteredByProjectName()
+      throws IOException, PermissionBackendException, CacheNotFoundException {
+    for (Event event : events()) {
+      when(projectsFilter.matches(any(String.class))).thenReturn(true);
+      objectUnderTest.getConsumer().accept(event);
+      verifyConsumed(event);
+    }
+  }
+
+  @Test
+  public void shouldSkipEventsWhenFilteredByProjectName()
+      throws IOException, PermissionBackendException, CacheNotFoundException {
+    for (Event event : events()) {
+      when(projectsFilter.matches(any(String.class))).thenReturn(false);
+      objectUnderTest.getConsumer().accept(event);
+      verifySkipped(event);
+    }
+  }
+
+  @SuppressWarnings("unchecked")
+  @Test
+  public void shouldSkipLocalEvents()
+      throws IOException, PermissionBackendException, CacheNotFoundException {
+    for (Event event : events()) {
+      event.instanceId = NODE_INSTANCE_ID;
+      when(projectsFilter.matches(any(String.class))).thenReturn(true);
+
+      objectUnderTest.getConsumer().accept(event);
+
+      verify(projectsFilter, never()).matches(PROJECT_NAME);
+      verify(eventRouter, never()).route(event);
+      verify(droppedEventListeners, times(1)).onEventDropped(event);
+      reset(projectsFilter, eventRouter, droppedEventListeners);
+    }
+  }
+
+  @Test
+  public void shouldUpdateReplicationMetricsWithLocalEvents() {
+    for (Event event : events()) {
+      event.instanceId = NODE_INSTANCE_ID;
+      when(projectsFilter.matches(any(String.class))).thenReturn(true);
+
+      objectUnderTest.getConsumer().accept(event);
+
+      verify(subscriberMetrics, times(1)).updateReplicationStatusMetrics(event);
+      reset(projectsFilter, eventRouter, droppedEventListeners, subscriberMetrics);
+    }
+  }
+
+  @Test
+  public void shouldUpdateReplicationMetricsWithNonLocalEvents() {
+    for (Event event : events()) {
+      event.instanceId = INSTANCE_ID;
+      when(projectsFilter.matches(any(String.class))).thenReturn(true);
+
+      objectUnderTest.getConsumer().accept(event);
+
+      verify(subscriberMetrics, times(1)).updateReplicationStatusMetrics(event);
+      reset(projectsFilter, eventRouter, droppedEventListeners, subscriberMetrics);
+    }
+  }
+
+  protected abstract AbstractSubcriber objectUnderTest();
+
+  protected abstract List<Event> events();
+
+  @SuppressWarnings("rawtypes")
+  protected abstract ForwardedEventRouter eventRouter();
+
+  @SuppressWarnings("unchecked")
+  protected void verifySkipped(Event event)
+      throws IOException, PermissionBackendException, CacheNotFoundException {
+    verify(projectsFilter, times(1)).matches(PROJECT_NAME);
+    verify(eventRouter, never()).route(event);
+    verify(droppedEventListeners, times(1)).onEventDropped(event);
+    reset(projectsFilter, eventRouter, droppedEventListeners);
+  }
+
+  @SuppressWarnings("unchecked")
+  protected void verifyConsumed(Event event)
+      throws IOException, PermissionBackendException, CacheNotFoundException {
+    verify(projectsFilter, times(1)).matches(PROJECT_NAME);
+    verify(eventRouter, times(1)).route(event);
+    verify(droppedEventListeners, never()).onEventDropped(event);
+    reset(projectsFilter, eventRouter, droppedEventListeners);
+  }
+
+  protected DynamicSet<DroppedEventListener> asDynamicSet(DroppedEventListener listener) {
+    DynamicSet<DroppedEventListener> result = new DynamicSet<>();
+    result.add("multi-site", listener);
+    return result;
+  }
+}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/BatchIndexEventSubscriberTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/BatchIndexEventSubscriberTest.java
new file mode 100644
index 0000000..5031f36
--- /dev/null
+++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/BatchIndexEventSubscriberTest.java
@@ -0,0 +1,77 @@
+// Copyright (C) 2021 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.multisite.consumer;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+import com.google.common.collect.ImmutableList;
+import com.google.gerrit.server.events.Event;
+import com.google.gerrit.server.permissions.PermissionBackendException;
+import com.googlesource.gerrit.plugins.multisite.forwarder.CacheNotFoundException;
+import com.googlesource.gerrit.plugins.multisite.forwarder.events.AccountIndexEvent;
+import com.googlesource.gerrit.plugins.multisite.forwarder.events.ChangeIndexEvent;
+import com.googlesource.gerrit.plugins.multisite.forwarder.events.IndexEvent;
+import com.googlesource.gerrit.plugins.multisite.forwarder.events.ProjectIndexEvent;
+import com.googlesource.gerrit.plugins.multisite.forwarder.router.ForwardedEventRouter;
+import com.googlesource.gerrit.plugins.multisite.forwarder.router.IndexEventRouter;
+import java.io.IOException;
+import java.util.List;
+import org.junit.Test;
+
+public class BatchIndexEventSubscriberTest extends AbstractSubscriberTestBase {
+  private static final boolean DELETED = false;
+  private static final int CHANGE_ID = 1;
+
+  @SuppressWarnings("unchecked")
+  @Test
+  public void shouldConsumeNonProjectAndNonChangeIndexingEventsTypes()
+      throws IOException, PermissionBackendException, CacheNotFoundException {
+    IndexEvent event = new AccountIndexEvent(1, INSTANCE_ID);
+
+    objectUnderTest.getConsumer().accept(event);
+
+    verify(projectsFilter, never()).matches(PROJECT_NAME);
+    verify(eventRouter, times(1)).route(event);
+    verify(droppedEventListeners, never()).onEventDropped(event);
+  }
+
+  @SuppressWarnings("rawtypes")
+  @Override
+  protected ForwardedEventRouter eventRouter() {
+    return mock(IndexEventRouter.class);
+  }
+
+  @Override
+  protected List<Event> events() {
+    return ImmutableList.of(
+        new ProjectIndexEvent(PROJECT_NAME, INSTANCE_ID),
+        new ChangeIndexEvent(PROJECT_NAME, CHANGE_ID, DELETED, INSTANCE_ID));
+  }
+
+  @Override
+  protected AbstractSubcriber objectUnderTest() {
+    return new BatchIndexEventSubscriber(
+        (IndexEventRouter) eventRouter,
+        asDynamicSet(droppedEventListeners),
+        NODE_INSTANCE_ID,
+        msgLog,
+        subscriberMetrics,
+        cfg,
+        projectsFilter);
+  }
+}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/CallbackMetricMaker.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/CallbackMetricMaker.java
new file mode 100644
index 0000000..a197c91
--- /dev/null
+++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/CallbackMetricMaker.java
@@ -0,0 +1,26 @@
+package com.googlesource.gerrit.plugins.multisite.consumer;
+
+import com.google.gerrit.metrics.Counter1;
+import com.google.gerrit.metrics.Description;
+import com.google.gerrit.metrics.DisabledMetricMaker;
+import com.google.gerrit.metrics.Field;
+import org.junit.Ignore;
+
+@Ignore
+public class CallbackMetricMaker extends DisabledMetricMaker {
+  private int callbackMetricCounter = 0;
+
+  public int getCallbackMetricCounter() {
+    return callbackMetricCounter;
+  }
+
+  @Override
+  public <F1> Counter1<F1> newCounter(String name, Description desc, Field<F1> field1) {
+    callbackMetricCounter += 1;
+    return super.newCounter(name, desc, field1);
+  }
+
+  public void resetCallbackMetricCounter() {
+    callbackMetricCounter = 0;
+  }
+}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/IndexEventSubscriberTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/IndexEventSubscriberTest.java
new file mode 100644
index 0000000..0e63528
--- /dev/null
+++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/IndexEventSubscriberTest.java
@@ -0,0 +1,133 @@
+// Copyright (C) 2021 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.multisite.consumer;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import com.google.common.collect.ImmutableList;
+import com.google.gerrit.entities.Account;
+import com.google.gerrit.entities.BranchNameKey;
+import com.google.gerrit.entities.Change;
+import com.google.gerrit.entities.Project;
+import com.google.gerrit.server.change.ChangeFinder;
+import com.google.gerrit.server.events.Event;
+import com.google.gerrit.server.notedb.ChangeNotes;
+import com.google.gerrit.server.permissions.PermissionBackendException;
+import com.google.gerrit.server.util.time.TimeUtil;
+import com.googlesource.gerrit.plugins.multisite.forwarder.CacheNotFoundException;
+import com.googlesource.gerrit.plugins.multisite.forwarder.events.AccountIndexEvent;
+import com.googlesource.gerrit.plugins.multisite.forwarder.events.ChangeIndexEvent;
+import com.googlesource.gerrit.plugins.multisite.forwarder.events.IndexEvent;
+import com.googlesource.gerrit.plugins.multisite.forwarder.events.ProjectIndexEvent;
+import com.googlesource.gerrit.plugins.multisite.forwarder.router.ForwardedEventRouter;
+import com.googlesource.gerrit.plugins.multisite.forwarder.router.IndexEventRouter;
+import java.io.IOException;
+import java.util.List;
+import java.util.Optional;
+import org.junit.Test;
+import org.mockito.Mock;
+
+public class IndexEventSubscriberTest extends AbstractSubscriberTestBase {
+  private static final boolean DELETED = false;
+  private static final int CHANGE_ID = 1;
+  private static final String EMPTY_PROJECT_NAME = "";
+
+  @Mock protected ChangeFinder changeFinderMock;
+
+  @SuppressWarnings("unchecked")
+  @Test
+  public void shouldConsumeNonProjectAndNonChangeIndexingEventsTypes()
+      throws IOException, PermissionBackendException, CacheNotFoundException {
+    IndexEvent event = new AccountIndexEvent(1, INSTANCE_ID);
+
+    objectUnderTest.getConsumer().accept(event);
+
+    verify(projectsFilter, never()).matches(PROJECT_NAME);
+    verify(eventRouter, times(1)).route(event);
+    verify(droppedEventListeners, never()).onEventDropped(event);
+  }
+
+  @SuppressWarnings("unchecked")
+  @Test
+  public void shouldConsumeDeleteChangeIndexEventWithEmptyProjectNameWhenFound()
+      throws IOException, PermissionBackendException, CacheNotFoundException {
+    ChangeIndexEvent event = new ChangeIndexEvent(EMPTY_PROJECT_NAME, CHANGE_ID, true, INSTANCE_ID);
+
+    ChangeNotes changeNotesMock = mock(ChangeNotes.class);
+    when(changeNotesMock.getChange()).thenReturn(newChange());
+    when(changeFinderMock.findOne(any(Change.Id.class))).thenReturn(Optional.of(changeNotesMock));
+    when(projectsFilter.matches(PROJECT_NAME)).thenReturn(true);
+
+    objectUnderTest.getConsumer().accept(event);
+
+    verify(projectsFilter, times(1)).matches(PROJECT_NAME);
+    verify(eventRouter, times(1)).route(event);
+  }
+
+  @SuppressWarnings("unchecked")
+  @Test
+  public void shouldNOTConsumeDeleteChangeIndexEventWithEmptyProjectNameWhenNotFound()
+      throws IOException, PermissionBackendException, CacheNotFoundException {
+    ChangeIndexEvent event = new ChangeIndexEvent("", CHANGE_ID, true, INSTANCE_ID);
+
+    when(changeFinderMock.findOne(any(Change.Id.class))).thenReturn(Optional.empty());
+    when(projectsFilter.matches(EMPTY_PROJECT_NAME)).thenReturn(false);
+
+    objectUnderTest.getConsumer().accept(event);
+
+    verify(projectsFilter, times(1)).matches(EMPTY_PROJECT_NAME);
+    verify(eventRouter, never()).route(event);
+  }
+
+  @SuppressWarnings("rawtypes")
+  @Override
+  protected ForwardedEventRouter eventRouter() {
+    return mock(IndexEventRouter.class);
+  }
+
+  @Override
+  protected List<Event> events() {
+    return ImmutableList.of(
+        new ProjectIndexEvent(PROJECT_NAME, INSTANCE_ID),
+        new ChangeIndexEvent(PROJECT_NAME, CHANGE_ID, DELETED, INSTANCE_ID));
+  }
+
+  @Override
+  protected AbstractSubcriber objectUnderTest() {
+    return new IndexEventSubscriber(
+        (IndexEventRouter) eventRouter,
+        asDynamicSet(droppedEventListeners),
+        NODE_INSTANCE_ID,
+        msgLog,
+        subscriberMetrics,
+        cfg,
+        projectsFilter,
+        changeFinderMock);
+  }
+
+  private Change newChange() {
+    return new Change(
+        Change.key(Integer.toString(CHANGE_ID)),
+        Change.id(CHANGE_ID),
+        Account.id(9999),
+        BranchNameKey.create(Project.nameKey(PROJECT_NAME), "refs/heads/master"),
+        TimeUtil.nowTs());
+  }
+}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/ProjectUpdateEventSubscriberTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/ProjectUpdateEventSubscriberTest.java
new file mode 100644
index 0000000..201ee81
--- /dev/null
+++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/ProjectUpdateEventSubscriberTest.java
@@ -0,0 +1,50 @@
+// Copyright (C) 2021 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.multisite.consumer;
+
+import static org.mockito.Mockito.mock;
+
+import com.google.common.collect.ImmutableList;
+import com.google.gerrit.server.events.Event;
+import com.googlesource.gerrit.plugins.multisite.forwarder.events.ProjectListUpdateEvent;
+import com.googlesource.gerrit.plugins.multisite.forwarder.router.ForwardedEventRouter;
+import com.googlesource.gerrit.plugins.multisite.forwarder.router.ProjectListUpdateRouter;
+import java.util.List;
+
+public class ProjectUpdateEventSubscriberTest extends AbstractSubscriberTestBase {
+
+  @SuppressWarnings("rawtypes")
+  @Override
+  protected ForwardedEventRouter eventRouter() {
+    return mock(ProjectListUpdateRouter.class);
+  }
+
+  @Override
+  protected List<Event> events() {
+    return ImmutableList.of(new ProjectListUpdateEvent(PROJECT_NAME, false, INSTANCE_ID));
+  }
+
+  @Override
+  protected AbstractSubcriber objectUnderTest() {
+    return new ProjectUpdateEventSubscriber(
+        (ProjectListUpdateRouter) eventRouter,
+        asDynamicSet(droppedEventListeners),
+        NODE_INSTANCE_ID,
+        msgLog,
+        subscriberMetrics,
+        cfg,
+        projectsFilter);
+  }
+}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/ReplicationStatusTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/ReplicationStatusTest.java
index ce82954..7828062 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/ReplicationStatusTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/ReplicationStatusTest.java
@@ -15,16 +15,27 @@
 package com.googlesource.gerrit.plugins.multisite.consumer;
 
 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.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 import com.google.common.cache.Cache;
 import com.google.common.cache.CacheBuilder;
 import com.google.common.collect.ImmutableSortedSet;
 import com.google.gerrit.entities.Project;
+import com.google.gerrit.extensions.events.ProjectDeletedListener;
+import com.google.gerrit.metrics.CallbackMetric1;
+import com.google.gerrit.metrics.DisabledMetricMaker;
 import com.google.gerrit.server.project.ProjectCache;
+import com.googlesource.gerrit.plugins.multisite.Configuration;
 import com.googlesource.gerrit.plugins.multisite.ProjectVersionLogger;
 import com.googlesource.gerrit.plugins.multisite.validation.ProjectVersionRefUpdate;
 import java.util.Optional;
+import java.util.concurrent.Executors;
+import org.eclipse.jgit.lib.Config;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -37,6 +48,7 @@
   @Mock private ProjectVersionLogger verLogger;
   @Mock private ProjectCache projectCache;
   @Mock private ProjectVersionRefUpdate projectVersionRefUpdate;
+  @Mock private CallbackMetric1<String, Long> perProjectReplicationLagMetricCallback;
   private ReplicationStatus objectUnderTest;
   private Cache<String, Long> replicationStatusCache;
 
@@ -48,7 +60,13 @@
     replicationStatusCache = CacheBuilder.newBuilder().build();
     objectUnderTest =
         new ReplicationStatus(
-            replicationStatusCache, Optional.of(projectVersionRefUpdate), verLogger, projectCache);
+            replicationStatusCache,
+            Optional.of(projectVersionRefUpdate),
+            verLogger,
+            projectCache,
+            Executors.newScheduledThreadPool(1),
+            new Configuration(new Config(), new Config()),
+            new DisabledMetricMaker());
   }
 
   @Test
@@ -57,6 +75,14 @@
     replicationStatusCache.put("projectB", 3L);
 
     objectUnderTest.start();
+    assertThat(objectUnderTest.getMaxLagMillis()).isEqualTo(10L);
+  }
+
+  @Test
+  public void shouldConvertMillisLagFromPersistedCacheOnStartToSecs() {
+    replicationStatusCache.put("projectA", 10000L);
+
+    objectUnderTest.start();
     assertThat(objectUnderTest.getMaxLag()).isEqualTo(10L);
   }
 
@@ -67,7 +93,7 @@
     objectUnderTest.start();
 
     objectUnderTest.doUpdateLag(Project.nameKey("projectA"), 20L);
-    assertThat(objectUnderTest.getMaxLag()).isEqualTo(20L);
+    assertThat(objectUnderTest.getMaxLagMillis()).isEqualTo(20L);
   }
 
   @Test
@@ -87,4 +113,127 @@
 
     assertThat(replicationStatusCache.getIfPresent("projectA")).isEqualTo(20L);
   }
+
+  @Test
+  public void shouldRemoveProjectFromPersistedCache() {
+    String projectName = "projectA";
+    long lag = 100;
+    setupReplicationLag(projectName, lag);
+    when(projectVersionRefUpdate.getProjectLocalVersion(projectName)).thenReturn(Optional.empty());
+
+    objectUnderTest.onProjectDeleted(projectDeletedEvent(projectName));
+
+    assertThat(replicationStatusCache.getIfPresent(projectName)).isNull();
+  }
+
+  @Test
+  public void shouldRemoveProjectFromReplicationLags() {
+    String projectName = "projectA";
+    long lag = 100;
+    setupReplicationLag(projectName, lag);
+    when(projectVersionRefUpdate.getProjectLocalVersion(projectName)).thenReturn(Optional.empty());
+
+    assertThat(objectUnderTest.getReplicationLags(1).keySet()).containsExactly(projectName);
+
+    objectUnderTest.onProjectDeleted(projectDeletedEvent(projectName));
+
+    assertThat(objectUnderTest.getReplicationLags(1).keySet()).isEmpty();
+  }
+
+  @Test
+  public void shouldNotRemoveProjectFromReplicationLagsIfLocalVersionStillExists() {
+    String projectName = "projectA";
+    long lag = 100;
+    setupReplicationLag(projectName, lag);
+    when(projectVersionRefUpdate.getProjectLocalVersion(projectName))
+        .thenReturn(Optional.of(System.currentTimeMillis()));
+
+    objectUnderTest.onProjectDeleted(projectDeletedEvent(projectName));
+
+    assertThat(objectUnderTest.getReplicationLags(1).keySet()).containsExactly(projectName);
+  }
+
+  @Test
+  public void shouldNotEvictFromPersistentCacheIfLocalVersionStillExists() {
+    String projectName = "projectA";
+    long lag = 100;
+    setupReplicationLag(projectName, lag);
+    when(projectVersionRefUpdate.getProjectLocalVersion(projectName))
+        .thenReturn(Optional.of(System.currentTimeMillis()));
+
+    objectUnderTest.onProjectDeleted(projectDeletedEvent(projectName));
+
+    assertThat(replicationStatusCache.getIfPresent(projectName)).isEqualTo(lag);
+  }
+
+  @Test
+  public void shouldUpdateReplicationLagForProject() {
+    String projectName = "projectA";
+    long projectLocalVersion = 10L;
+    long projectRemoteVersion = 20L;
+
+    when(projectVersionRefUpdate.getProjectLocalVersion(eq(projectName)))
+        .thenReturn(Optional.of(projectLocalVersion));
+    when(projectVersionRefUpdate.getProjectRemoteVersion(eq(projectName)))
+        .thenReturn(Optional.of(projectRemoteVersion));
+
+    objectUnderTest.updateReplicationLag(Project.nameKey(projectName));
+    objectUnderTest.replicationLagMetricPerProject(perProjectReplicationLagMetricCallback).run();
+
+    assertThat(replicationStatusCache.getIfPresent(projectName))
+        .isEqualTo(projectRemoteVersion - projectLocalVersion);
+    verify(perProjectReplicationLagMetricCallback)
+        .set(eq(projectName), eq(projectRemoteVersion - projectLocalVersion));
+  }
+
+  @Test
+  public void shouldNotGenerateCallbackMetricIfNoReplicationLag() {
+    String projectName = "projectA";
+    long projectLatestVersion = 10L;
+    when(projectVersionRefUpdate.getProjectLocalVersion(eq(projectName)))
+        .thenReturn(Optional.of(projectLatestVersion));
+
+    objectUnderTest.updateReplicationLag(Project.nameKey(projectName));
+
+    assertThat(replicationStatusCache.getIfPresent(projectName)).isNull();
+    verify(perProjectReplicationLagMetricCallback, never()).set(any(), any());
+  }
+
+  @SuppressWarnings("unchecked")
+  @Test
+  public void shouldAutoRefreshReplicationLagForProject() {
+    String projectName = "projectA";
+    long projectLocalVersion = 10L;
+    long projectRemoteVersion = 20L;
+
+    when(projectVersionRefUpdate.getProjectLocalVersion(eq(projectName)))
+        .thenReturn(Optional.of(projectLocalVersion), Optional.of(projectRemoteVersion));
+    when(projectVersionRefUpdate.getProjectRemoteVersion(eq(projectName)))
+        .thenReturn(Optional.of(projectRemoteVersion));
+
+    objectUnderTest.updateReplicationLag(Project.nameKey(projectName));
+
+    assertThat(replicationStatusCache.getIfPresent(projectName))
+        .isEqualTo(projectRemoteVersion - projectLocalVersion);
+    objectUnderTest.refreshProjectsWithLag();
+
+    assertThat(replicationStatusCache.getIfPresent(projectName)).isEqualTo(0);
+  }
+
+  private void setupReplicationLag(String projectName, long lag) {
+    long currentVersion = System.currentTimeMillis();
+    long newVersion = currentVersion + lag;
+    replicationStatusCache.put(projectName, 3L);
+    when(projectVersionRefUpdate.getProjectRemoteVersion(projectName))
+        .thenReturn(Optional.of(newVersion));
+    when(projectVersionRefUpdate.getProjectLocalVersion(projectName))
+        .thenReturn(Optional.of(currentVersion));
+    objectUnderTest.updateReplicationLag(Project.nameKey(projectName));
+  }
+
+  private ProjectDeletedListener.Event projectDeletedEvent(String projectName) {
+    ProjectDeletedListener.Event event = mock(ProjectDeletedListener.Event.class);
+    when(event.getProjectName()).thenReturn(projectName);
+    return event;
+  }
 }
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/StreamEventSubscriberTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/StreamEventSubscriberTest.java
new file mode 100644
index 0000000..71fd828
--- /dev/null
+++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/StreamEventSubscriberTest.java
@@ -0,0 +1,122 @@
+// Copyright (C) 2021 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.multisite.consumer;
+
+import static org.mockito.Mockito.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.reset;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import com.google.common.collect.ImmutableList;
+import com.google.gerrit.entities.Project;
+import com.google.gerrit.entities.Project.NameKey;
+import com.google.gerrit.server.events.Event;
+import com.google.gerrit.server.events.RefEvent;
+import com.google.gerrit.server.events.RefUpdatedEvent;
+import com.google.gerrit.server.permissions.PermissionBackendException;
+import com.googlesource.gerrit.plugins.multisite.forwarder.CacheNotFoundException;
+import com.googlesource.gerrit.plugins.multisite.forwarder.events.AccountIndexEvent;
+import com.googlesource.gerrit.plugins.multisite.forwarder.events.IndexEvent;
+import com.googlesource.gerrit.plugins.multisite.forwarder.router.ForwardedEventRouter;
+import com.googlesource.gerrit.plugins.multisite.forwarder.router.StreamEventRouter;
+import com.googlesource.gerrit.plugins.replication.events.RefReplicatedEvent;
+import com.googlesource.gerrit.plugins.replication.events.RefReplicationDoneEvent;
+import com.googlesource.gerrit.plugins.replication.events.ReplicationScheduledEvent;
+import java.io.IOException;
+import java.util.List;
+import org.junit.Test;
+import org.mockito.Mock;
+
+public class StreamEventSubscriberTest extends AbstractSubscriberTestBase {
+  private static final NameKey PROJECT_NAME_KEY = NameKey.parse(PROJECT_NAME);
+  private @Mock RefUpdatedEvent refUpdatedEvent;
+  private @Mock RefReplicationDoneEvent refReplicationDoneEvent;
+  private @Mock ReplicationScheduledEvent replicationScheduledEvent;
+  private @Mock RefReplicatedEvent refReplicatedEvent;
+
+  @SuppressWarnings("rawtypes")
+  @Override
+  protected ForwardedEventRouter eventRouter() {
+    return mock(StreamEventRouter.class);
+  }
+
+  @Override
+  protected List<Event> events() {
+    when(refUpdatedEvent.getProjectNameKey()).thenReturn(PROJECT_NAME_KEY);
+    refUpdatedEvent.instanceId = INSTANCE_ID;
+    when(refReplicationDoneEvent.getProjectNameKey()).thenReturn(PROJECT_NAME_KEY);
+    refReplicationDoneEvent.instanceId = INSTANCE_ID;
+    when(replicationScheduledEvent.getProjectNameKey()).thenReturn(PROJECT_NAME_KEY);
+    replicationScheduledEvent.instanceId = INSTANCE_ID;
+    when(refReplicatedEvent.getProjectNameKey()).thenReturn(PROJECT_NAME_KEY);
+    refReplicatedEvent.instanceId = INSTANCE_ID;
+
+    return ImmutableList.of(
+        refUpdatedEvent, refReplicationDoneEvent, replicationScheduledEvent, refReplicatedEvent);
+  }
+
+  @SuppressWarnings("unchecked")
+  @Test
+  public void shouldNotConsumeNonProjectEventTypeEvents()
+      throws IOException, PermissionBackendException, CacheNotFoundException {
+    IndexEvent event = new AccountIndexEvent(1, INSTANCE_ID);
+
+    objectUnderTest.getConsumer().accept(event);
+
+    verify(projectsFilter, never()).matches(PROJECT_NAME);
+    verify(eventRouter, times(1)).route(event);
+    verify(droppedEventListeners, never()).onEventDropped(event);
+  }
+
+  @Test
+  public void shouldUpdateReplicationMetricsWithRemoteRefEvent() {
+    Event event =
+        new RefEvent("ref-replicated") {
+
+          @Override
+          public String getRefName() {
+            return "foo-ref";
+          }
+
+          @Override
+          public NameKey getProjectNameKey() {
+            return Project.nameKey(PROJECT_NAME);
+          }
+        };
+
+    event.instanceId = INSTANCE_ID;
+    when(projectsFilter.matches(eq(PROJECT_NAME))).thenReturn(true);
+
+    objectUnderTest.getConsumer().accept(event);
+
+    verify(subscriberMetrics, times(1)).updateReplicationStatusMetrics(event);
+    reset(projectsFilter, eventRouter, droppedEventListeners, subscriberMetrics);
+  }
+
+  @Override
+  protected AbstractSubcriber objectUnderTest() {
+    return new StreamEventSubscriber(
+        (StreamEventRouter) eventRouter,
+        asDynamicSet(droppedEventListeners),
+        NODE_INSTANCE_ID,
+        msgLog,
+        subscriberMetrics,
+        cfg,
+        projectsFilter);
+  }
+}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/SubscriberMetricsTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/SubscriberMetricsTest.java
index 4c45e07..6413b18 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/SubscriberMetricsTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/SubscriberMetricsTest.java
@@ -14,23 +14,28 @@
 
 package com.googlesource.gerrit.plugins.multisite.consumer;
 
+import static com.google.common.truth.Truth.assertThat;
 import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyZeroInteractions;
 import static org.mockito.Mockito.when;
 
-import com.gerritforge.gerrit.eventbroker.EventMessage;
-import com.gerritforge.gerrit.globalrefdb.validation.SharedRefDatabaseWrapper;
 import com.google.common.base.Suppliers;
 import com.google.common.cache.CacheBuilder;
 import com.google.gerrit.entities.Project;
+import com.google.gerrit.metrics.DisabledMetricMaker;
 import com.google.gerrit.metrics.MetricMaker;
 import com.google.gerrit.server.data.RefUpdateAttribute;
+import com.google.gerrit.server.events.Event;
 import com.google.gerrit.server.events.RefUpdatedEvent;
-import com.google.gerrit.server.extensions.events.GitReferenceUpdated;
 import com.google.gerrit.server.project.ProjectCache;
 import com.googlesource.gerrit.plugins.multisite.ProjectVersionLogger;
 import com.googlesource.gerrit.plugins.multisite.validation.ProjectVersionRefUpdate;
+import com.googlesource.gerrit.plugins.replication.events.ProjectDeletionReplicationSucceededEvent;
+import java.net.URISyntaxException;
 import java.util.Optional;
-import java.util.UUID;
+import java.util.concurrent.Executors;
+import org.eclipse.jgit.lib.Config;
+import org.eclipse.jgit.transport.URIish;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -43,37 +48,36 @@
   private static final Project.NameKey A_TEST_PROJECT_NAME_KEY =
       Project.nameKey(A_TEST_PROJECT_NAME);
 
-  @Mock private SharedRefDatabaseWrapper sharedRefDb;
-  @Mock private GitReferenceUpdated gitReferenceUpdated;
   @Mock private MetricMaker metricMaker;
   @Mock private ProjectVersionLogger verLogger;
   @Mock private ProjectCache projectCache;
   @Mock private ProjectVersionRefUpdate projectVersionRefUpdate;
   private SubscriberMetrics metrics;
-  private EventMessage.Header msgHeader;
+  private ReplicationStatus replicationStatus;
 
   @Before
   public void setup() throws Exception {
-    msgHeader = new EventMessage.Header(UUID.randomUUID(), UUID.randomUUID());
-    metrics =
-        new SubscriberMetrics(
-            metricMaker,
-            new ReplicationStatus(
-                CacheBuilder.newBuilder().build(),
-                Optional.of(projectVersionRefUpdate),
-                verLogger,
-                projectCache));
+    replicationStatus =
+        new ReplicationStatus(
+            CacheBuilder.newBuilder().build(),
+            Optional.of(projectVersionRefUpdate),
+            verLogger,
+            projectCache,
+            Executors.newScheduledThreadPool(1),
+            new com.googlesource.gerrit.plugins.multisite.Configuration(new Config(), new Config()),
+            new DisabledMetricMaker());
+    metrics = new SubscriberMetrics(metricMaker, replicationStatus);
   }
 
   @Test
   public void shouldLogProjectVersionWhenReceivingRefUpdatedEventWithoutLag() {
-    Optional<Long> globalRefDbVersion = Optional.of(System.currentTimeMillis() / 1000);
+    Optional<Long> globalRefDbVersion = Optional.of(System.currentTimeMillis());
     when(projectVersionRefUpdate.getProjectRemoteVersion(A_TEST_PROJECT_NAME))
         .thenReturn(globalRefDbVersion);
     when(projectVersionRefUpdate.getProjectLocalVersion(A_TEST_PROJECT_NAME))
         .thenReturn(globalRefDbVersion);
 
-    EventMessage eventMessage = new EventMessage(msgHeader, newRefUpdateEvent());
+    Event eventMessage = newRefUpdateEvent();
 
     metrics.updateReplicationStatusMetrics(eventMessage);
 
@@ -82,20 +86,137 @@
 
   @Test
   public void shouldLogProjectVersionWhenReceivingRefUpdatedEventWithALag() {
-    Optional<Long> globalRefDbVersion = Optional.of(System.currentTimeMillis() / 1000);
+    Optional<Long> globalRefDbVersion = Optional.of(System.currentTimeMillis());
     long replicationLag = 60;
     when(projectVersionRefUpdate.getProjectRemoteVersion(A_TEST_PROJECT_NAME))
         .thenReturn(globalRefDbVersion.map(ts -> ts + replicationLag));
     when(projectVersionRefUpdate.getProjectLocalVersion(A_TEST_PROJECT_NAME))
         .thenReturn(globalRefDbVersion);
 
-    EventMessage eventMessage = new EventMessage(msgHeader, newRefUpdateEvent());
+    Event eventMessage = newRefUpdateEvent();
 
     metrics.updateReplicationStatusMetrics(eventMessage);
 
     verify(verLogger).log(A_TEST_PROJECT_NAME_KEY, globalRefDbVersion.get(), replicationLag);
   }
 
+  @Test
+  public void
+      shouldLogUponProjectDeletionSuccessWhenLocalVersionDoesNotExistAndSubscriberMetricsExist()
+          throws Exception {
+    long nowSecs = System.currentTimeMillis();
+    long replicationLagSecs = 60;
+    Optional<Long> globalRefDbVersion = Optional.of(nowSecs);
+    when(projectVersionRefUpdate.getProjectRemoteVersion(A_TEST_PROJECT_NAME))
+        .thenReturn(globalRefDbVersion.map(ts -> ts + replicationLagSecs));
+    when(projectVersionRefUpdate.getProjectLocalVersion(A_TEST_PROJECT_NAME))
+        .thenReturn(globalRefDbVersion);
+
+    Event refUpdateEventMessage = newRefUpdateEvent();
+    metrics.updateReplicationStatusMetrics(refUpdateEventMessage);
+
+    assertThat(replicationStatus.getReplicationStatus(A_TEST_PROJECT_NAME))
+        .isEqualTo(replicationLagSecs);
+    assertThat(replicationStatus.getLocalVersion(A_TEST_PROJECT_NAME)).isEqualTo(nowSecs);
+
+    when(projectVersionRefUpdate.getProjectLocalVersion(A_TEST_PROJECT_NAME))
+        .thenReturn(Optional.empty());
+
+    Event projectDeleteEventMessage = projectDeletionSuccess();
+    metrics.updateReplicationStatusMetrics(projectDeleteEventMessage);
+
+    verify(verLogger).logDeleted(A_TEST_PROJECT_NAME_KEY);
+  }
+
+  @Test
+  public void shouldNotLogUponProjectDeletionSuccessWhenSubscriberMetricsDoNotExist()
+      throws Exception {
+    Event eventMessage = projectDeletionSuccess();
+    when(projectVersionRefUpdate.getProjectLocalVersion(A_TEST_PROJECT_NAME))
+        .thenReturn(Optional.empty());
+
+    assertThat(replicationStatus.getReplicationStatus(A_TEST_PROJECT_NAME)).isNull();
+    assertThat(replicationStatus.getLocalVersion(A_TEST_PROJECT_NAME)).isNull();
+
+    metrics.updateReplicationStatusMetrics(eventMessage);
+
+    verifyZeroInteractions(verLogger);
+  }
+
+  @Test
+  public void shouldNotLogUponProjectDeletionSuccessWhenLocalVersionStillExists() throws Exception {
+    Event eventMessage = projectDeletionSuccess();
+    Optional<Long> anyRefVersionValue = Optional.of(System.currentTimeMillis());
+    when(projectVersionRefUpdate.getProjectLocalVersion(A_TEST_PROJECT_NAME))
+        .thenReturn(anyRefVersionValue);
+
+    metrics.updateReplicationStatusMetrics(eventMessage);
+
+    verifyZeroInteractions(verLogger);
+  }
+
+  @Test
+  public void shouldRemoveProjectMetricsUponProjectDeletionSuccess() throws Exception {
+    long nowSecs = System.currentTimeMillis();
+    long replicationLagSecs = 60;
+    Optional<Long> globalRefDbVersion = Optional.of(nowSecs);
+    when(projectVersionRefUpdate.getProjectRemoteVersion(A_TEST_PROJECT_NAME))
+        .thenReturn(globalRefDbVersion.map(ts -> ts + replicationLagSecs));
+    when(projectVersionRefUpdate.getProjectLocalVersion(A_TEST_PROJECT_NAME))
+        .thenReturn(globalRefDbVersion);
+
+    Event eventMessage = newRefUpdateEvent();
+
+    metrics.updateReplicationStatusMetrics(eventMessage);
+
+    assertThat(replicationStatus.getReplicationStatus(A_TEST_PROJECT_NAME))
+        .isEqualTo(replicationLagSecs);
+    assertThat(replicationStatus.getLocalVersion(A_TEST_PROJECT_NAME)).isEqualTo(nowSecs);
+
+    when(projectVersionRefUpdate.getProjectLocalVersion(A_TEST_PROJECT_NAME))
+        .thenReturn(Optional.empty());
+    Event projectDeleteEvent = projectDeletionSuccess();
+
+    metrics.updateReplicationStatusMetrics(projectDeleteEvent);
+
+    assertThat(replicationStatus.getReplicationStatus(A_TEST_PROJECT_NAME)).isNull();
+    assertThat(replicationStatus.getLocalVersion(A_TEST_PROJECT_NAME)).isNull();
+  }
+
+  @Test
+  public void shouldDoNothingIfNameIsValid() {
+    String validProjectName = "aValidProject-Name123";
+
+    assertThat(metrics.sanitizeProjectName(validProjectName)).isEqualTo(validProjectName);
+  }
+
+  @Test
+  public void shouldSanitizeNameWithDot() {
+    String validProjectName = "nameWithA.InTheMiddle";
+
+    assertThat(metrics.sanitizeProjectName(validProjectName)).isEqualTo("nameWithA_2eInTheMiddle");
+  }
+
+  @Test
+  public void shouldSanitizeNameWithSlash() {
+    String validProjectName = "nameWithA/InTheMiddle";
+
+    assertThat(metrics.sanitizeProjectName(validProjectName)).isEqualTo("nameWithA_2fInTheMiddle");
+  }
+
+  @Test
+  public void shouldDoubleUnderscoresInName() {
+    String validProjectName = "nameWithA_InTheMiddle";
+
+    assertThat(metrics.sanitizeProjectName(validProjectName)).isEqualTo("nameWithA__InTheMiddle");
+  }
+
+  private ProjectDeletionReplicationSucceededEvent projectDeletionSuccess()
+      throws URISyntaxException {
+    return new ProjectDeletionReplicationSucceededEvent(
+        A_TEST_PROJECT_NAME, new URIish("git://target"));
+  }
+
   private RefUpdatedEvent newRefUpdateEvent() {
     RefUpdateAttribute refUpdate = new RefUpdateAttribute();
     refUpdate.project = A_TEST_PROJECT_NAME;
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/event/CacheEvictionEventRouterTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/event/CacheEvictionEventRouterTest.java
index 76c001f..6762ba2 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/multisite/event/CacheEvictionEventRouterTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/event/CacheEvictionEventRouterTest.java
@@ -34,6 +34,7 @@
 @RunWith(MockitoJUnitRunner.class)
 public class CacheEvictionEventRouterTest {
 
+  private static final String INSTANCE_ID = "instance-id";
   private static Gson gson = new EventGsonProvider().get();
   private CacheEvictionEventRouter router;
   @Mock private ForwardedCacheEvictionHandler cacheEvictionHandler;
@@ -45,7 +46,7 @@
 
   @Test
   public void routerShouldSendEventsToTheAppropriateHandler_CacheEviction() throws Exception {
-    final CacheEvictionEvent event = new CacheEvictionEvent("cache", "key");
+    final CacheEvictionEvent event = new CacheEvictionEvent("cache", "key", INSTANCE_ID);
     router.route(event);
 
     verify(cacheEvictionHandler).evict(CacheEntry.from(event.cacheName, event.key));
@@ -54,7 +55,7 @@
   @Test
   public void routerShouldSendEventsToTheAppropriateHandler_CacheEvictionWithSlash()
       throws Exception {
-    final CacheEvictionEvent event = new CacheEvictionEvent("cache", "some/key");
+    final CacheEvictionEvent event = new CacheEvictionEvent("cache", "some/key", INSTANCE_ID);
     router.route(event);
 
     verify(cacheEvictionHandler).evict(CacheEntry.from(event.cacheName, event.key));
@@ -63,7 +64,8 @@
   @Test
   public void routerShouldSendEventsToTheAppropriateHandler_ProjectCacheEvictionWithSlash()
       throws Exception {
-    final CacheEvictionEvent event = new CacheEvictionEvent(Constants.PROJECTS, "some/project");
+    final CacheEvictionEvent event =
+        new CacheEvictionEvent(Constants.PROJECTS, "some/project", INSTANCE_ID);
     router.route(event);
 
     verify(cacheEvictionHandler)
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/event/IndexEventRouterTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/event/IndexEventRouterTest.java
index 1dc07bd..413b6c1 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/multisite/event/IndexEventRouterTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/event/IndexEventRouterTest.java
@@ -33,7 +33,7 @@
 import com.googlesource.gerrit.plugins.multisite.forwarder.events.ProjectIndexEvent;
 import com.googlesource.gerrit.plugins.multisite.forwarder.router.IndexEventRouter;
 import com.googlesource.gerrit.plugins.multisite.forwarder.router.StreamEventRouter;
-import com.googlesource.gerrit.plugins.replication.RefReplicationDoneEvent;
+import com.googlesource.gerrit.plugins.replication.events.RefReplicationDoneEvent;
 import java.util.Optional;
 import org.eclipse.jgit.lib.ObjectId;
 import org.junit.Before;
@@ -44,7 +44,7 @@
 
 @RunWith(MockitoJUnitRunner.class)
 public class IndexEventRouterTest {
-
+  private static final String INSTANCE_ID = "instance-id";
   private IndexEventRouter router;
   @Mock private ForwardedIndexAccountHandler indexAccountHandler;
   @Mock private ForwardedIndexChangeHandler indexChangeHandler;
@@ -61,12 +61,13 @@
             indexChangeHandler,
             indexGroupHandler,
             indexProjectHandler,
-            allUsersName);
+            allUsersName,
+            INSTANCE_ID);
   }
 
   @Test
   public void routerShouldSendEventsToTheAppropriateHandler_AccountIndex() throws Exception {
-    final AccountIndexEvent event = new AccountIndexEvent(1);
+    final AccountIndexEvent event = new AccountIndexEvent(1, INSTANCE_ID);
     router.route(event);
 
     verify(indexAccountHandler)
@@ -80,7 +81,7 @@
 
     StreamEventRouter streamEventRouter = new StreamEventRouter(forwardedEventHandler, router);
 
-    final AccountIndexEvent event = new AccountIndexEvent(1);
+    final AccountIndexEvent event = new AccountIndexEvent(1, INSTANCE_ID);
     router.route(event);
 
     verify(indexAccountHandler)
@@ -96,7 +97,7 @@
   @Test
   public void routerShouldSendEventsToTheAppropriateHandler_GroupIndex() throws Exception {
     final String groupId = "12";
-    final GroupIndexEvent event = new GroupIndexEvent(groupId, ObjectId.zeroId());
+    final GroupIndexEvent event = new GroupIndexEvent(groupId, ObjectId.zeroId(), INSTANCE_ID);
     router.route(event);
 
     verify(indexGroupHandler)
@@ -108,7 +109,7 @@
   @Test
   public void routerShouldSendEventsToTheAppropriateHandler_ProjectIndex() throws Exception {
     final String projectName = "projectName";
-    final ProjectIndexEvent event = new ProjectIndexEvent(projectName);
+    final ProjectIndexEvent event = new ProjectIndexEvent(projectName, INSTANCE_ID);
     router.route(event);
 
     verify(indexProjectHandler)
@@ -119,7 +120,7 @@
 
   @Test
   public void routerShouldSendEventsToTheAppropriateHandler_ChangeIndex() throws Exception {
-    final ChangeIndexEvent event = new ChangeIndexEvent("projectName", 3, false);
+    final ChangeIndexEvent event = new ChangeIndexEvent("projectName", 3, false, INSTANCE_ID);
     router.route(event);
 
     verify(indexChangeHandler)
@@ -133,7 +134,7 @@
 
   @Test
   public void routerShouldSendEventsToTheAppropriateHandler_ChangeIndexDelete() throws Exception {
-    final ChangeIndexEvent event = new ChangeIndexEvent("projectName", 3, true);
+    final ChangeIndexEvent event = new ChangeIndexEvent("projectName", 3, true, INSTANCE_ID);
     router.route(event);
 
     verify(indexChangeHandler)
@@ -147,7 +148,7 @@
 
   @Test
   public void routerShouldFailForNotRecognisedEvents() throws Exception {
-    final IndexEvent newEventType = new IndexEvent("new-type") {};
+    final IndexEvent newEventType = new IndexEvent("new-type", INSTANCE_ID) {};
 
     assertThrows(UnsupportedOperationException.class, () -> router.route(newEventType));
     verifyZeroInteractions(
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/event/ProjectListUpdateRouterTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/event/ProjectListUpdateRouterTest.java
index 93daf92..8a21c39 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/multisite/event/ProjectListUpdateRouterTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/event/ProjectListUpdateRouterTest.java
@@ -38,7 +38,8 @@
 
   @Test
   public void routerShouldSendEventsToTheAppropriateHandler_ProjectListUpdate() throws Exception {
-    final ProjectListUpdateEvent event = new ProjectListUpdateEvent("project", false);
+    String instanceId = "instance-id";
+    final ProjectListUpdateEvent event = new ProjectListUpdateEvent("project", false, instanceId);
     router.route(event);
 
     verify(projectListUpdateHandler).update(event);
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/BrokerForwarderTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/BrokerForwarderTest.java
index e3d1ae0..28bf56d 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/BrokerForwarderTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/BrokerForwarderTest.java
@@ -68,10 +68,10 @@
     }
   }
 
-  public class TestEvent extends MultiSiteEvent {
+  public static class TestEvent extends MultiSiteEvent {
 
     protected TestEvent() {
-      super("test");
+      super("test", "instance-id");
     }
   }
 
@@ -93,7 +93,7 @@
   @Test
   public void shouldSendEventToBrokerFromGenericSourceThread() {
     brokerForwarder.send(newForwarderTask(), testTopic, testEvent);
-    verify(brokerMock).send(eq(testTopicName), eq(testEvent));
+    verify(brokerMock).sendSync(eq(testTopicName), eq(testEvent));
   }
 
   @Test
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedCacheEvictionHandlerIT.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedCacheEvictionHandlerIT.java
index 826e154..75596ed 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedCacheEvictionHandlerIT.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedCacheEvictionHandlerIT.java
@@ -22,6 +22,7 @@
 import com.google.common.collect.Sets;
 import com.google.gerrit.acceptance.LightweightPluginDaemonTest;
 import com.google.gerrit.acceptance.TestPlugin;
+import com.google.gerrit.acceptance.config.GerritConfig;
 import com.google.gerrit.entities.Project;
 import com.google.gerrit.extensions.api.projects.ProjectInput;
 import com.google.gerrit.extensions.registration.DynamicSet;
@@ -123,8 +124,10 @@
   }
 
   @Test
+  @GerritConfig(name = "gerrit.instanceId", value = "testInstanceId")
   public void shouldEvictProjectCache() throws Exception {
-    objectUnderTest.route(new CacheEvictionEvent(ProjectCacheImpl.CACHE_NAME, project.get()));
+    objectUnderTest.route(
+        new CacheEvictionEvent(ProjectCacheImpl.CACHE_NAME, project.get(), "instance-id"));
     evictionsCacheTracker.waitForExpectedEvictions();
 
     assertThat(evictionsCacheTracker.trackedEvictionsFor(ProjectCacheImpl.CACHE_NAME))
@@ -132,6 +135,7 @@
   }
 
   @Test
+  @GerritConfig(name = "gerrit.instanceId", value = "instance-id")
   public void shouldEvictProjectCacheWithSlash() throws Exception {
     ProjectInput in = new ProjectInput();
     in.name = name("my/project");
@@ -141,7 +145,7 @@
     restartCacheEvictionsTracking();
 
     objectUnderTest.route(
-        new CacheEvictionEvent(ProjectCacheImpl.CACHE_NAME, projectNameKey.get()));
+        new CacheEvictionEvent(ProjectCacheImpl.CACHE_NAME, projectNameKey.get(), "instance-id"));
 
     evictionsCacheTracker.waitForExpectedEvictions();
     assertThat(evictionsCacheTracker.trackedEvictionsFor(ProjectCacheImpl.CACHE_NAME))
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedCacheEvictionHandlerTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedCacheEvictionHandlerTest.java
index 90ae8da..dbd358d 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedCacheEvictionHandlerTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedCacheEvictionHandlerTest.java
@@ -15,6 +15,7 @@
 package com.googlesource.gerrit.plugins.multisite.forwarder;
 
 import static com.google.common.truth.Truth.assertThat;
+import static com.google.gerrit.testing.GerritJUnit.assertThrows;
 import static org.mockito.Mockito.doReturn;
 
 import com.google.common.cache.Cache;
@@ -48,10 +49,12 @@
   public void shouldThrowAnExceptionWhenCacheNotFound() throws Exception {
     CacheEntry entry = new CacheEntry("somePlugin", "unexistingCache", null);
 
-    exception.expect(CacheNotFoundException.class);
-    exception.expectMessage(
-        String.format("cache %s.%s not found", entry.getPluginName(), entry.getCacheName()));
-    handler.evict(entry);
+    CacheNotFoundException thrown =
+        assertThrows(CacheNotFoundException.class, () -> handler.evict(entry));
+    assertThat(thrown)
+        .hasMessageThat()
+        .isEqualTo(
+            String.format("cache %s.%s not found", entry.getPluginName(), entry.getCacheName()));
   }
 
   @Test
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedIndexAccountHandlerTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedIndexAccountHandlerTest.java
index 32c6319..02efb77 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedIndexAccountHandlerTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedIndexAccountHandlerTest.java
@@ -61,9 +61,11 @@
 
   @Test
   public void deleteIsNotSupported() throws Exception {
-    exception.expect(UnsupportedOperationException.class);
-    exception.expectMessage("Delete from account index not supported");
-    handler.index(id, Operation.DELETE, Optional.empty());
+    UnsupportedOperationException thrown =
+        assertThrows(
+            UnsupportedOperationException.class,
+            () -> handler.index(id, Operation.DELETE, Optional.empty()));
+    assertThat(thrown).hasMessageThat().isEqualTo("Delete from account index not supported");
   }
 
   @Test
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedIndexChangeHandlerTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedIndexChangeHandlerTest.java
index 96470b6..b827cdf 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedIndexChangeHandlerTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedIndexChangeHandlerTest.java
@@ -17,6 +17,7 @@
 import static com.google.common.truth.Truth.assertThat;
 import static com.google.gerrit.testing.GerritJUnit.assertThrows;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.never;
@@ -49,6 +50,7 @@
 import org.mockito.Mock;
 import org.mockito.junit.MockitoJUnitRunner;
 import org.mockito.stubbing.Answer;
+import org.mockito.stubbing.OngoingStubbing;
 
 @RunWith(MockitoJUnitRunner.class)
 public class ForwardedIndexChangeHandlerTest {
@@ -62,6 +64,8 @@
   private static final boolean THROW_STORAGE_EXCEPTION = true;
   private static final boolean CHANGE_UP_TO_DATE = true;
   private static final boolean CHANGE_OUTDATED = false;
+  private static final boolean CHANGE_CONSISTENT = true;
+  private static final boolean CHANGE_INCONSISTENT = false;
 
   @Rule public ExpectedException exception = ExpectedException.none();
   @Mock private ChangeIndexer indexerMock;
@@ -88,6 +92,7 @@
     when(changeCheckerFactoryMock.create(any())).thenReturn(changeCheckerAbsentMock);
     when(configurationMock.index()).thenReturn(index);
     when(index.numStripedLocks()).thenReturn(10);
+    when(index.maxTries()).thenReturn(1);
     handler =
         new ForwardedIndexChangeHandler(
             indexerMock, configurationMock, indexExecutorMock, ctxMock, changeCheckerFactoryMock);
@@ -95,16 +100,40 @@
 
   @Test
   public void changeIsIndexedWhenUpToDate() throws Exception {
-    setupChangeAccessRelatedMocks(CHANGE_EXISTS, CHANGE_UP_TO_DATE);
+    setupChangeAccessRelatedMocks(CHANGE_EXISTS, CHANGE_UP_TO_DATE, CHANGE_CONSISTENT);
     handler.index(TEST_CHANGE_ID, Operation.INDEX, Optional.empty());
     verify(indexerMock, times(1)).index(any(Change.class));
   }
 
   @Test
   public void changeIsStillIndexedEvenWhenOutdated() throws Exception {
-    setupChangeAccessRelatedMocks(CHANGE_EXISTS, CHANGE_OUTDATED);
+    setupChangeAccessRelatedMocks(CHANGE_EXISTS, CHANGE_OUTDATED, CHANGE_CONSISTENT);
     handler.index(
-        TEST_CHANGE_ID, Operation.INDEX, Optional.of(new ChangeIndexEvent("foo", 1, false)));
+        TEST_CHANGE_ID,
+        Operation.INDEX,
+        Optional.of(new ChangeIndexEvent("foo", 1, false, "instance-id")));
+    verify(indexerMock, times(1)).index(any(Change.class));
+  }
+
+  @Test
+  public void changeIsIndexeAtFirstRetryWhenInitiallyInconsistent() throws Exception {
+    setupChangeAccessRelatedMocks(
+        CHANGE_EXISTS,
+        DO_NOT_THROW_STORAGE_EXCEPTION,
+        CHANGE_UP_TO_DATE,
+        CHANGE_INCONSISTENT,
+        CHANGE_CONSISTENT);
+    handler.index(
+        TEST_CHANGE_ID,
+        Operation.INDEX,
+        Optional.of(new ChangeIndexEvent("foo", 1, false, "instance-id")));
+    verify(indexerMock, never()).index(any(Change.class));
+    verify(indexExecutorMock, times(1)).schedule(any(Runnable.class), anyLong(), any());
+
+    handler.index(
+        TEST_CHANGE_ID,
+        Operation.INDEX,
+        Optional.of(new ChangeIndexEvent("foo", 1, false, "instance-id")));
     verify(indexerMock, times(1)).index(any(Change.class));
   }
 
@@ -124,9 +153,11 @@
 
   @Test
   public void indexerThrowsStorageExceptionTryingToIndexChange() throws Exception {
-    setupChangeAccessRelatedMocks(CHANGE_EXISTS, THROW_STORAGE_EXCEPTION, CHANGE_UP_TO_DATE);
-    exception.expect(StorageException.class);
-    handler.index(TEST_CHANGE_ID, Operation.INDEX, Optional.empty());
+    setupChangeAccessRelatedMocks(
+        CHANGE_EXISTS, THROW_STORAGE_EXCEPTION, CHANGE_UP_TO_DATE, CHANGE_CONSISTENT);
+    assertThrows(
+        StorageException.class,
+        () -> handler.index(TEST_CHANGE_ID, Operation.INDEX, Optional.empty()));
   }
 
   @Test
@@ -175,11 +206,21 @@
 
   private void setupChangeAccessRelatedMocks(boolean changeExist, boolean changeUpToDate)
       throws Exception {
-    setupChangeAccessRelatedMocks(changeExist, DO_NOT_THROW_STORAGE_EXCEPTION, changeUpToDate);
+    setupChangeAccessRelatedMocks(
+        changeExist, DO_NOT_THROW_STORAGE_EXCEPTION, changeUpToDate, CHANGE_CONSISTENT);
   }
 
   private void setupChangeAccessRelatedMocks(
-      boolean changeExists, boolean storageException, boolean changeIsUpToDate)
+      boolean changeExist, boolean changeUpToDate, boolean changeConsistent) throws Exception {
+    setupChangeAccessRelatedMocks(
+        changeExist, DO_NOT_THROW_STORAGE_EXCEPTION, changeUpToDate, changeConsistent);
+  }
+
+  private void setupChangeAccessRelatedMocks(
+      boolean changeExists,
+      boolean storageException,
+      boolean changeIsUpToDate,
+      boolean... changeConsistentReturnValues)
       throws StorageException {
     if (changeExists) {
       when(changeCheckerFactoryMock.create(TEST_CHANGE_ID)).thenReturn(changeCheckerPresentMock);
@@ -190,5 +231,11 @@
     }
 
     when(changeCheckerPresentMock.isUpToDate(any())).thenReturn(changeIsUpToDate);
+
+    OngoingStubbing<Boolean> changeConsistentCall =
+        when(changeCheckerPresentMock.isChangeConsistent());
+    for (boolean changeConsistent : changeConsistentReturnValues) {
+      changeConsistentCall = changeConsistentCall.thenReturn(changeConsistent);
+    }
   }
 }
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedIndexGroupHandlerTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedIndexGroupHandlerTest.java
index 982ac52..b49dbfa 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedIndexGroupHandlerTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedIndexGroupHandlerTest.java
@@ -74,9 +74,11 @@
 
   @Test
   public void deleteIsNotSupported() throws Exception {
-    exception.expect(UnsupportedOperationException.class);
-    exception.expectMessage("Delete from group index not supported");
-    handler.index(uuid, Operation.DELETE, Optional.empty());
+    UnsupportedOperationException thrown =
+        assertThrows(
+            UnsupportedOperationException.class,
+            () -> handler.index(uuid, Operation.DELETE, Optional.empty()));
+    assertThat(thrown).hasMessageThat().isEqualTo("Delete from group index not supported");
   }
 
   @Test
@@ -141,6 +143,6 @@
   }
 
   private Optional<GroupIndexEvent> groupIndexEvent(String uuid) {
-    return Optional.of(new GroupIndexEvent(uuid, null));
+    return Optional.of(new GroupIndexEvent(uuid, null, "instance-id"));
   }
 }
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedIndexProjectHandlerTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedIndexProjectHandlerTest.java
index 3ce5e14..72b9427 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedIndexProjectHandlerTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedIndexProjectHandlerTest.java
@@ -73,9 +73,11 @@
 
   @Test
   public void deleteIsNotSupported() throws Exception {
-    exception.expect(UnsupportedOperationException.class);
-    exception.expectMessage("Delete from project index not supported");
-    handler.index(nameKey, Operation.DELETE, Optional.empty());
+    UnsupportedOperationException thrown =
+        assertThrows(
+            UnsupportedOperationException.class,
+            () -> handler.index(nameKey, Operation.DELETE, Optional.empty()));
+    assertThat(thrown).hasMessageThat().isEqualTo("Delete from project index not supported");
   }
 
   @Test
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedProjectListUpdateHandlerTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedProjectListUpdateHandlerTest.java
index 9893ce7..2412c79 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedProjectListUpdateHandlerTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedProjectListUpdateHandlerTest.java
@@ -34,6 +34,7 @@
 @RunWith(MockitoJUnitRunner.class)
 public class ForwardedProjectListUpdateHandlerTest {
 
+  private static final String INSTANCE_ID = "instance-id";
   private static final String PROJECT_NAME = "someProject";
   private static final String SOME_MESSAGE = "someMessage";
   private static final Project.NameKey PROJECT_KEY = Project.nameKey(PROJECT_NAME);
@@ -48,13 +49,13 @@
 
   @Test
   public void testSuccessfulAdd() throws Exception {
-    handler.update(new ProjectListUpdateEvent(PROJECT_NAME, false));
+    handler.update(new ProjectListUpdateEvent(PROJECT_NAME, false, INSTANCE_ID));
     verify(projectCacheMock).onCreateProject(PROJECT_KEY);
   }
 
   @Test
   public void testSuccessfulRemove() throws Exception {
-    handler.update(new ProjectListUpdateEvent(PROJECT_NAME, true));
+    handler.update(new ProjectListUpdateEvent(PROJECT_NAME, true, INSTANCE_ID));
     verify(projectCacheMock).remove(PROJECT_KEY);
   }
 
@@ -72,7 +73,7 @@
         .onCreateProject(PROJECT_KEY);
 
     assertThat(Context.isForwardedEvent()).isFalse();
-    handler.update(new ProjectListUpdateEvent(PROJECT_NAME, false));
+    handler.update(new ProjectListUpdateEvent(PROJECT_NAME, false, INSTANCE_ID));
     assertThat(Context.isForwardedEvent()).isFalse();
 
     verify(projectCacheMock).onCreateProject(PROJECT_KEY);
@@ -92,7 +93,7 @@
         .remove(PROJECT_KEY);
 
     assertThat(Context.isForwardedEvent()).isFalse();
-    handler.update(new ProjectListUpdateEvent(PROJECT_NAME, true));
+    handler.update(new ProjectListUpdateEvent(PROJECT_NAME, true, INSTANCE_ID));
     assertThat(Context.isForwardedEvent()).isFalse();
 
     verify(projectCacheMock).remove(PROJECT_KEY);
@@ -113,7 +114,7 @@
     RuntimeException thrown =
         assertThrows(
             RuntimeException.class,
-            () -> handler.update(new ProjectListUpdateEvent(PROJECT_NAME, false)));
+            () -> handler.update(new ProjectListUpdateEvent(PROJECT_NAME, false, INSTANCE_ID)));
     assertThat(thrown).hasMessageThat().isEqualTo(SOME_MESSAGE);
     assertThat(Context.isForwardedEvent()).isFalse();
 
@@ -135,7 +136,7 @@
     RuntimeException thrown =
         assertThrows(
             RuntimeException.class,
-            () -> handler.update(new ProjectListUpdateEvent(PROJECT_NAME, true)));
+            () -> handler.update(new ProjectListUpdateEvent(PROJECT_NAME, true, INSTANCE_ID)));
     assertThat(thrown).hasMessageThat().isEqualTo(SOME_MESSAGE);
     assertThat(Context.isForwardedEvent()).isFalse();
 
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/http/ReplicationStatusServletIT.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/http/ReplicationStatusServletIT.java
index 0739eab..f940fc4 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/multisite/http/ReplicationStatusServletIT.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/http/ReplicationStatusServletIT.java
@@ -24,9 +24,12 @@
 import com.google.gerrit.acceptance.LightweightPluginDaemonTest;
 import com.google.gerrit.acceptance.RestResponse;
 import com.google.gerrit.acceptance.TestPlugin;
+import com.google.gerrit.acceptance.config.GerritConfig;
 import com.google.gerrit.entities.Project;
 import com.google.gerrit.httpd.restapi.RestApiServlet;
+import com.google.gerrit.server.git.WorkQueue;
 import com.google.inject.AbstractModule;
+import com.google.inject.Inject;
 import com.google.inject.Scopes;
 import com.google.inject.multibindings.OptionalBinder;
 import com.googlesource.gerrit.plugins.multisite.Log4jProjectVersionLogger;
@@ -56,13 +59,15 @@
   private ReplicationStatus replicationStatus;
 
   public static class TestModule extends AbstractModule {
+    @Inject WorkQueue workQueue;
+
     @Override
     protected void configure() {
       install(new ForwarderModule());
       install(new CacheModule());
       install(new RouterModule());
       install(new IndexModule());
-      install(new ReplicationStatusModule());
+      install(new ReplicationStatusModule(workQueue));
       SharedRefDbConfiguration sharedRefDbConfig =
           new SharedRefDbConfiguration(new Config(), "multi-site");
       bind(SharedRefDbConfiguration.class).toInstance(sharedRefDbConfig);
@@ -81,6 +86,7 @@
   }
 
   @Test
+  @GerritConfig(name = "gerrit.instanceId", value = "testInstanceId")
   public void shouldSucceedForAdminUsers() throws Exception {
     RestResponse result = adminRestSession.get(LAG_ENDPOINT);
 
@@ -89,6 +95,7 @@
   }
 
   @Test
+  @GerritConfig(name = "gerrit.instanceId", value = "testInstanceId")
   public void shouldFailWhenUserHasNoAdminServerCapability() throws Exception {
     RestResponse result = userRestSession.get(LAG_ENDPOINT);
     result.assertForbidden();
@@ -96,6 +103,7 @@
   }
 
   @Test
+  @GerritConfig(name = "gerrit.instanceId", value = "testInstanceId")
   public void shouldReturnCurrentProjectLag() throws Exception {
     replicationStatus.doUpdateLag(Project.nameKey("foo"), 123L);
 
@@ -106,6 +114,7 @@
   }
 
   @Test
+  @GerritConfig(name = "gerrit.instanceId", value = "testInstanceId")
   public void shouldReturnProjectsOrderedDescendinglyByLag() throws Exception {
     replicationStatus.doUpdateLag(Project.nameKey("bar"), 123L);
     replicationStatus.doUpdateLag(Project.nameKey("foo"), 3L);
@@ -118,6 +127,7 @@
   }
 
   @Test
+  @GerritConfig(name = "gerrit.instanceId", value = "testInstanceId")
   public void shouldHonourTheLimitParameter() throws Exception {
     replicationStatus.doUpdateLag(Project.nameKey("bar"), 1L);
     replicationStatus.doUpdateLag(Project.nameKey("foo"), 2L);
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/index/GroupCheckerImplTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/index/GroupCheckerImplTest.java
index 6403cce..09f8f59 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/multisite/index/GroupCheckerImplTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/index/GroupCheckerImplTest.java
@@ -101,7 +101,7 @@
   }
 
   private Optional<GroupIndexEvent> groupIndexEvent(String uuid, @Nullable ObjectId sha1) {
-    return Optional.of(new GroupIndexEvent(uuid, sha1));
+    return Optional.of(new GroupIndexEvent(uuid, sha1, "instance-id"));
   }
 
   private void setCommitExistsInRepo(boolean commitExists) {
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/index/GroupEventIndexTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/index/GroupEventIndexTest.java
index 93cec05..099f1dd 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/multisite/index/GroupEventIndexTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/index/GroupEventIndexTest.java
@@ -16,7 +16,7 @@
 
 import static com.google.common.truth.Truth.assertThat;
 
-import com.gerritforge.gerrit.eventbroker.EventGsonProvider;
+import com.google.gerrit.server.events.EventGsonProvider;
 import com.google.gson.Gson;
 import com.googlesource.gerrit.plugins.multisite.forwarder.events.GroupIndexEvent;
 import java.util.UUID;
@@ -24,13 +24,14 @@
 import org.junit.Test;
 
 public class GroupEventIndexTest {
+  private static final String INSTANCE_ID = "instance-id";
   private static final Gson gson = new EventGsonProvider().get();
 
   @Test
   public void groupEventIndexRoundTripWithSha1() {
     String aGroupUUID = UUID.randomUUID().toString();
     ObjectId anObjectId = ObjectId.fromString("deadbeefdeadbeefdeadbeefdeadbeefdeadbeef");
-    GroupIndexEvent original = new GroupIndexEvent(aGroupUUID, anObjectId);
+    GroupIndexEvent original = new GroupIndexEvent(aGroupUUID, anObjectId, INSTANCE_ID);
 
     assertThat(gson.fromJson(gson.toJson(original), GroupIndexEvent.class)).isEqualTo(original);
   }
@@ -38,7 +39,7 @@
   @Test
   public void groupEventIndexRoundTripWithoutSha1() {
     String aGroupUUID = UUID.randomUUID().toString();
-    GroupIndexEvent original = new GroupIndexEvent(aGroupUUID, null);
+    GroupIndexEvent original = new GroupIndexEvent(aGroupUUID, null, INSTANCE_ID);
 
     assertThat(gson.fromJson(gson.toJson(original), GroupIndexEvent.class)).isEqualTo(original);
   }
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/index/IndexEventHandlerTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/index/IndexEventHandlerTest.java
index 660a302..8cf6ea9 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/multisite/index/IndexEventHandlerTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/index/IndexEventHandlerTest.java
@@ -14,16 +14,21 @@
 
 package com.googlesource.gerrit.plugins.multisite.index;
 
+import static com.googlesource.gerrit.plugins.replication.pull.api.PullReplicationEndpoints.APPLY_OBJECT_API_ENDPOINT;
 import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.anyString;
 import static org.mockito.Mockito.eq;
+import static org.mockito.Mockito.lenient;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
 import static org.mockito.Mockito.verifyZeroInteractions;
 import static org.mockito.Mockito.when;
 
 import com.gerritforge.gerrit.globalrefdb.validation.ProjectsFilter;
 import com.google.common.util.concurrent.MoreExecutors;
 import com.google.gerrit.extensions.registration.DynamicSet;
+import com.googlesource.gerrit.plugins.multisite.forwarder.Context;
 import com.googlesource.gerrit.plugins.multisite.forwarder.IndexEventForwarder;
 import com.googlesource.gerrit.plugins.multisite.forwarder.events.ProjectIndexEvent;
 import com.googlesource.gerrit.plugins.multisite.index.IndexEventHandler.IndexProjectTask;
@@ -36,6 +41,8 @@
 @RunWith(MockitoJUnitRunner.class)
 public class IndexEventHandlerTest {
 
+  private static final String INSTANCE_ID = "instance-id";
+
   private IndexEventHandler eventHandler;
 
   @Mock private ProjectsFilter projectsFilter;
@@ -50,7 +57,8 @@
             asDynamicSet(forwarder),
             changeChecker,
             projectsFilter,
-            new TestGroupChecker(true));
+            new TestGroupChecker(true),
+            INSTANCE_ID);
   }
 
   private DynamicSet<IndexEventForwarder> asDynamicSet(IndexEventForwarder forwarder) {
@@ -65,7 +73,7 @@
 
     eventHandler.onProjectIndexed("test_project");
     verify(forwarder, never())
-        .index(any(IndexProjectTask.class), eq(new ProjectIndexEvent("test_project")));
+        .index(any(IndexProjectTask.class), eq(new ProjectIndexEvent("test_project", INSTANCE_ID)));
   }
 
   @Test
@@ -76,4 +84,25 @@
     eventHandler.onChangeIndexed("test_project", changeId);
     verifyZeroInteractions(changeChecker);
   }
+
+  @Test
+  public void shouldNotForwardIndexChangeIfCurrentThreadIsPullReplicationApplyObject()
+      throws Exception {
+    String currentThreadName = Thread.currentThread().getName();
+    try {
+      Thread.currentThread().setName("pull-replication~" + APPLY_OBJECT_API_ENDPOINT);
+      int changeId = 1;
+      Context.setForwardedEvent(false);
+      lenient().when(projectsFilter.matches(any(String.class))).thenReturn(true);
+      lenient()
+          .when(changeChecker.create(anyString()))
+          .thenThrow(
+              new IllegalStateException("Change indexing event should have not been triggered"));
+
+      eventHandler.onChangeIndexed("test_project", changeId);
+      verifyNoInteractions(changeChecker);
+    } finally {
+      Thread.currentThread().setName(currentThreadName);
+    }
+  }
 }
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/validation/DisabledSharedRefLogger.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/validation/DisabledSharedRefLogger.java
new file mode 100644
index 0000000..fc2259f
--- /dev/null
+++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/validation/DisabledSharedRefLogger.java
@@ -0,0 +1,42 @@
+// 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.multisite.validation;
+
+import com.gerritforge.gerrit.globalrefdb.validation.SharedRefLogger;
+import org.eclipse.jgit.lib.ObjectId;
+import org.eclipse.jgit.lib.Ref;
+import org.junit.Ignore;
+
+@Ignore
+public class DisabledSharedRefLogger implements SharedRefLogger {
+
+  @Override
+  public void logRefUpdate(String project, Ref currRef, ObjectId newRefValue) {}
+
+  @Override
+  public void logProjectDelete(String project) {}
+
+  @Override
+  public void logLockAcquisition(String project, String refName) {}
+
+  @Override
+  public void logLockRelease(String project, String refName) {}
+
+  @Override
+  public <T> void logRefUpdate(String project, String refName, T currRef, T newRefValue) {}
+
+  @Override
+  public <T> void logRefUpdate(String project, String refName, T newRefValue) {}
+}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/validation/FakeSharedRefDatabaseWrapper.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/validation/FakeSharedRefDatabaseWrapper.java
new file mode 100644
index 0000000..bf98a4b
--- /dev/null
+++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/validation/FakeSharedRefDatabaseWrapper.java
@@ -0,0 +1,84 @@
+// Copyright (C) 2023 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.googlesource.gerrit.plugins.multisite.validation;
+
+import com.gerritforge.gerrit.globalrefdb.GlobalRefDatabase;
+import com.gerritforge.gerrit.globalrefdb.GlobalRefDbLockException;
+import com.gerritforge.gerrit.globalrefdb.GlobalRefDbSystemError;
+import com.gerritforge.gerrit.globalrefdb.validation.SharedRefDBMetrics;
+import com.gerritforge.gerrit.globalrefdb.validation.SharedRefDatabaseWrapper;
+import com.google.gerrit.entities.Project;
+import com.google.gerrit.extensions.registration.DynamicItem;
+import com.google.gerrit.metrics.DisabledMetricMaker;
+import java.util.Arrays;
+import java.util.Optional;
+import org.eclipse.jgit.lib.ObjectId;
+import org.eclipse.jgit.lib.Ref;
+import org.junit.Ignore;
+
+@Ignore
+public class FakeSharedRefDatabaseWrapper extends SharedRefDatabaseWrapper {
+
+  public FakeSharedRefDatabaseWrapper(String... rejectedRefs) {
+
+    super(
+        DynamicItem.itemOf(
+            GlobalRefDatabase.class,
+            new GlobalRefDatabase() {
+
+              @Override
+              public boolean isUpToDate(Project.NameKey project, Ref ref)
+                  throws GlobalRefDbLockException {
+                return !Arrays.stream(rejectedRefs).anyMatch(r -> r.equals(ref.getName()));
+              }
+
+              @Override
+              public boolean exists(Project.NameKey project, String refName) {
+                return true;
+              }
+
+              @Override
+              public boolean compareAndPut(
+                  Project.NameKey project, Ref currRef, ObjectId newRefValue)
+                  throws GlobalRefDbSystemError {
+                return false;
+              }
+
+              @Override
+              public <T> boolean compareAndPut(
+                  Project.NameKey project, String refName, T currValue, T newValue)
+                  throws GlobalRefDbSystemError {
+                return false;
+              }
+
+              @Override
+              public AutoCloseable lockRef(Project.NameKey project, String refName)
+                  throws GlobalRefDbLockException {
+                return null;
+              }
+
+              @Override
+              public void remove(Project.NameKey project) throws GlobalRefDbSystemError {}
+
+              @Override
+              public <T> Optional<T> get(Project.NameKey project, String refName, Class<T> clazz)
+                  throws GlobalRefDbSystemError {
+                return Optional.empty();
+              }
+            }),
+        new DisabledSharedRefLogger(),
+        new SharedRefDBMetrics(new DisabledMetricMaker()));
+  }
+}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/validation/MultisiteReplicationFetchFilterTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/validation/MultisiteReplicationFetchFilterTest.java
new file mode 100644
index 0000000..e09f6f6
--- /dev/null
+++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/validation/MultisiteReplicationFetchFilterTest.java
@@ -0,0 +1,258 @@
+// Copyright (C) 2023 The Android Open Source Project
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.googlesource.gerrit.plugins.multisite.validation;
+
+import static com.google.common.truth.Truth.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+import com.gerritforge.gerrit.globalrefdb.validation.SharedRefDatabaseWrapper;
+import com.google.gerrit.entities.Project;
+import com.google.gerrit.testing.InMemoryRepositoryManager;
+import com.google.gerrit.testing.InMemoryTestEnvironment;
+import com.google.inject.Inject;
+import com.googlesource.gerrit.plugins.multisite.Configuration;
+import com.googlesource.gerrit.plugins.multisite.validation.dfsrefdb.RefFixture;
+import java.util.Optional;
+import java.util.Set;
+import org.eclipse.jgit.internal.storage.dfs.InMemoryRepository;
+import org.eclipse.jgit.junit.LocalDiskRepositoryTestCase;
+import org.eclipse.jgit.junit.TestRepository;
+import org.eclipse.jgit.lib.ObjectId;
+import org.eclipse.jgit.revwalk.RevCommit;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+@RunWith(MockitoJUnitRunner.class)
+public class MultisiteReplicationFetchFilterTest extends LocalDiskRepositoryTestCase
+    implements RefFixture {
+
+  @Rule public InMemoryTestEnvironment testEnvironment = new InMemoryTestEnvironment();
+
+  @Mock SharedRefDatabaseWrapper sharedRefDatabaseMock;
+  @Mock Configuration config;
+  @Mock Configuration.ReplicationFilter replicationFilterConfig;
+
+  @Inject private InMemoryRepositoryManager gitRepositoryManager;
+
+  String project = A_TEST_PROJECT_NAME;
+  Project.NameKey projectName = A_TEST_PROJECT_NAME_KEY;
+
+  private TestRepository<InMemoryRepository> repo;
+
+  @Before
+  public void setupTestRepo() throws Exception {
+    InMemoryRepository inMemoryRepo =
+        gitRepositoryManager.createRepository(A_TEST_PROJECT_NAME_KEY);
+    repo = new TestRepository<>(inMemoryRepo);
+    doReturn(replicationFilterConfig).when(config).replicationFilter();
+  }
+
+  @Test
+  public void shouldReturnEmptyRefsWhenAllUpToDate() throws Exception {
+    String fooRefName = "refs/heads/foo";
+    ObjectId fooObjectId = newRef(fooRefName).getId();
+    String barRefName = "refs/heads/bar";
+    ObjectId barObjectId = newRef(barRefName).getId();
+    Set<String> refs = Set.of(fooRefName, barRefName);
+
+    doReturn(Optional.of(fooObjectId.getName()))
+        .when(sharedRefDatabaseMock)
+        .get(eq(projectName), eq(fooRefName), eq(String.class));
+    doReturn(Optional.of(barObjectId.getName()))
+        .when(sharedRefDatabaseMock)
+        .get(eq(projectName), eq(barRefName), eq(String.class));
+
+    MultisiteReplicationFetchFilter fetchFilter =
+        new MultisiteReplicationFetchFilter(sharedRefDatabaseMock, gitRepositoryManager, config);
+    Set<String> filteredRefs = fetchFilter.filter(project, refs);
+
+    assertThat(filteredRefs).isEmpty();
+  }
+
+  @Test
+  public void shouldFilterOutOneUpToDateRef() throws Exception {
+    String refUpToDate = "refs/heads/uptodate";
+    String outdatedRef = "refs/heads/outdated";
+    ObjectId upToDateObjectId = newRef(refUpToDate).getId();
+    newRef(outdatedRef).getId();
+    Set<String> refsToFetch = Set.of(refUpToDate, outdatedRef);
+
+    doReturn(Optional.of(upToDateObjectId.getName()))
+        .when(sharedRefDatabaseMock)
+        .get(eq(projectName), eq(refUpToDate), eq(String.class));
+    doReturn(Optional.of(AN_OUTDATED_OBJECT_ID.getName()))
+        .when(sharedRefDatabaseMock)
+        .get(eq(projectName), eq(outdatedRef), eq(String.class));
+
+    MultisiteReplicationFetchFilter fetchFilter =
+        new MultisiteReplicationFetchFilter(sharedRefDatabaseMock, gitRepositoryManager, config);
+    Set<String> filteredRefsToFetch = fetchFilter.filter(project, refsToFetch);
+
+    assertThat(filteredRefsToFetch).containsExactly(outdatedRef);
+  }
+
+  @Test
+  public void shouldLoadLocalVersionAndFilterOut() throws Exception {
+    String temporaryOutdated = "refs/heads/temporaryOutdated";
+    RevCommit localRef = newRef(temporaryOutdated);
+
+    Set<String> refsToFetch = Set.of(temporaryOutdated);
+    doReturn(Optional.empty())
+        .doReturn(Optional.of(localRef.getId().getName()))
+        .when(sharedRefDatabaseMock)
+        .get(eq(projectName), eq(temporaryOutdated), eq(String.class));
+
+    MultisiteReplicationFetchFilter fetchFilter =
+        new MultisiteReplicationFetchFilter(sharedRefDatabaseMock, gitRepositoryManager, config);
+    Set<String> filteredRefsToFetch = fetchFilter.filter(project, refsToFetch);
+
+    assertThat(filteredRefsToFetch).isEmpty();
+
+    verify(sharedRefDatabaseMock, times(2)).get(any(), any(), any());
+  }
+
+  @Test
+  public void shouldLoadLocalVersionAndNotFilter() throws Exception {
+    String temporaryOutdated = "refs/heads/temporaryOutdated";
+    newRef(temporaryOutdated);
+
+    Set<String> refsToFetch = Set.of(temporaryOutdated);
+    doReturn(Optional.of(AN_OUTDATED_OBJECT_ID.getName()))
+        .doReturn(Optional.of(AN_OUTDATED_OBJECT_ID.getName()))
+        .when(sharedRefDatabaseMock)
+        .get(eq(projectName), eq(temporaryOutdated), eq(String.class));
+
+    MultisiteReplicationFetchFilter fetchFilter =
+        new MultisiteReplicationFetchFilter(sharedRefDatabaseMock, gitRepositoryManager, config);
+    Set<String> filteredRefsToFetch = fetchFilter.filter(project, refsToFetch);
+
+    assertThat(filteredRefsToFetch).hasSize(1);
+    verify(sharedRefDatabaseMock, times(3))
+        .get(eq(projectName), eq(temporaryOutdated), eq(String.class));
+  }
+
+  @Test
+  public void shouldNotFilterOutWhenMissingInTheSharedRefDb() throws Exception {
+    String temporaryOutdated = "refs/heads/temporaryOutdated";
+    newRef(temporaryOutdated);
+
+    doReturn(Optional.empty())
+        .when(sharedRefDatabaseMock)
+        .get(eq(projectName), eq(temporaryOutdated), eq(String.class));
+
+    Set<String> refsToFetch = Set.of(temporaryOutdated);
+
+    MultisiteReplicationFetchFilter fetchFilter =
+        new MultisiteReplicationFetchFilter(sharedRefDatabaseMock, gitRepositoryManager, config);
+    Set<String> filteredRefsToFetch = fetchFilter.filter(project, refsToFetch);
+
+    assertThat(filteredRefsToFetch).hasSize(1);
+  }
+
+  @Test
+  public void shouldNotFilterOutWhenRefsMultisiteVersionIsPresentInSharedRefDb() throws Exception {
+    String refsMultisiteVersionRef = ProjectVersionRefUpdate.MULTI_SITE_VERSIONING_REF;
+    RevCommit multiSiteVersionRef = newRef(refsMultisiteVersionRef);
+
+    doReturn(Optional.of(multiSiteVersionRef.getId().getName()))
+        .when(sharedRefDatabaseMock)
+        .get(eq(projectName), eq(refsMultisiteVersionRef), eq(String.class));
+
+    Set<String> refsToFetch = Set.of(refsMultisiteVersionRef);
+
+    MultisiteReplicationFetchFilter fetchFilter =
+        new MultisiteReplicationFetchFilter(sharedRefDatabaseMock, gitRepositoryManager, config);
+    Set<String> filteredRefsToFetch = fetchFilter.filter(project, refsToFetch);
+
+    assertThat(filteredRefsToFetch).hasSize(1);
+  }
+
+  @Test
+  public void shouldFilterOutWhenRefIsDeletedInTheSharedRefDb() throws Exception {
+    String temporaryOutdated = "refs/heads/temporaryOutdated";
+    newRef(temporaryOutdated);
+
+    Set<String> refsToFetch = Set.of(temporaryOutdated);
+    doReturn(Optional.of(ObjectId.zeroId().getName()))
+        .when(sharedRefDatabaseMock)
+        .get(eq(projectName), eq(temporaryOutdated), eq(String.class));
+
+    MultisiteReplicationFetchFilter fetchFilter =
+        new MultisiteReplicationFetchFilter(sharedRefDatabaseMock, gitRepositoryManager, config);
+    Set<String> filteredRefsToFetch = fetchFilter.filter(project, refsToFetch);
+
+    assertThat(filteredRefsToFetch).hasSize(0);
+    verify(sharedRefDatabaseMock).get(eq(projectName), any(), any());
+  }
+
+  @Test
+  public void shouldNotFilterOutWhenRefIsMissingOnlyInTheLocalRepository() throws Exception {
+    String refObjectId = "0000000000000000000000000000000000000001";
+    String nonExistingLocalRef = "refs/heads/temporaryOutdated";
+
+    Set<String> refsToFetch = Set.of(nonExistingLocalRef);
+    doReturn(Optional.of(refObjectId))
+        .when(sharedRefDatabaseMock)
+        .get(eq(projectName), any(), any());
+
+    MultisiteReplicationFetchFilter fetchFilter =
+        new MultisiteReplicationFetchFilter(sharedRefDatabaseMock, gitRepositoryManager, config);
+    Set<String> filteredRefsToFetch = fetchFilter.filter(project, refsToFetch);
+
+    assertThat(filteredRefsToFetch).hasSize(1);
+  }
+
+  @Test
+  public void shouldNotFilterOutRefThatDoesntExistLocallyOrInSharedRefDb() throws Exception {
+    String nonExisting = "refs/heads/non-existing-ref";
+
+    Set<String> refsToFetch = Set.of(nonExisting);
+
+    MultisiteReplicationFetchFilter fetchFilter =
+        new MultisiteReplicationFetchFilter(sharedRefDatabaseMock, gitRepositoryManager, config);
+    Set<String> filteredRefsToFetch = fetchFilter.filter(project, refsToFetch);
+
+    assertThat(filteredRefsToFetch).hasSize(1);
+  }
+
+  @Test
+  public void shouldFilterOutRefMissingInTheLocalRepositoryAndDeletedInSharedRefDb()
+      throws Exception {
+    String nonExistingLocalRef = "refs/heads/temporaryOutdated";
+
+    Set<String> refsToFetch = Set.of(nonExistingLocalRef);
+    doReturn(Optional.of(ObjectId.zeroId().getName()))
+        .when(sharedRefDatabaseMock)
+        .get(eq(projectName), any(), any());
+
+    MultisiteReplicationFetchFilter fetchFilter =
+        new MultisiteReplicationFetchFilter(sharedRefDatabaseMock, gitRepositoryManager, config);
+    Set<String> filteredRefsToFetch = fetchFilter.filter(project, refsToFetch);
+
+    assertThat(filteredRefsToFetch).hasSize(0);
+  }
+
+  private RevCommit newRef(String refName) throws Exception {
+    return repo.branch(refName).commit().create();
+  }
+}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/validation/MultisiteReplicationPushFilterTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/validation/MultisiteReplicationPushFilterTest.java
new file mode 100644
index 0000000..b4a5e5d
--- /dev/null
+++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/validation/MultisiteReplicationPushFilterTest.java
@@ -0,0 +1,159 @@
+// 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.multisite.validation;
+
+import static com.google.common.truth.Truth.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+import com.gerritforge.gerrit.globalrefdb.validation.SharedRefDatabaseWrapper;
+import com.google.gerrit.entities.Project;
+import com.google.gerrit.testing.InMemoryRepositoryManager;
+import com.google.gerrit.testing.InMemoryTestEnvironment;
+import com.google.inject.Inject;
+import com.googlesource.gerrit.plugins.multisite.Configuration;
+import com.googlesource.gerrit.plugins.multisite.validation.dfsrefdb.RefFixture;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import org.eclipse.jgit.internal.storage.dfs.InMemoryRepository;
+import org.eclipse.jgit.junit.LocalDiskRepositoryTestCase;
+import org.eclipse.jgit.junit.TestRepository;
+import org.eclipse.jgit.lib.ObjectId;
+import org.eclipse.jgit.lib.ObjectIdRef;
+import org.eclipse.jgit.lib.Ref;
+import org.eclipse.jgit.transport.RemoteRefUpdate;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+@RunWith(MockitoJUnitRunner.class)
+public class MultisiteReplicationPushFilterTest extends LocalDiskRepositoryTestCase
+    implements RefFixture {
+
+  @Rule public InMemoryTestEnvironment testEnvironment = new InMemoryTestEnvironment();
+
+  @Mock SharedRefDatabaseWrapper sharedRefDatabaseMock;
+  @Mock Configuration config;
+  @Mock Configuration.ReplicationFilter replicationFilterConfig;
+
+  @Inject private InMemoryRepositoryManager gitRepositoryManager;
+
+  String project = A_TEST_PROJECT_NAME;
+  Project.NameKey projectName = A_TEST_PROJECT_NAME_KEY;
+
+  private TestRepository<InMemoryRepository> repo;
+
+  @Before
+  public void setupTestRepo() throws Exception {
+    InMemoryRepository inMemoryRepo =
+        gitRepositoryManager.createRepository(A_TEST_PROJECT_NAME_KEY);
+    doReturn(replicationFilterConfig).when(config).replicationFilter();
+    repo = new TestRepository<>(inMemoryRepo);
+  }
+
+  @Test
+  public void shouldReturnAllRefUpdatesWhenAllUpToDate() throws Exception {
+    List<RemoteRefUpdate> refUpdates =
+        Arrays.asList(refUpdate("refs/heads/foo"), refUpdate("refs/heads/bar"));
+    doReturn(true).when(sharedRefDatabaseMock).isUpToDate(eq(projectName), any());
+
+    MultisiteReplicationPushFilter pushFilter =
+        new MultisiteReplicationPushFilter(sharedRefDatabaseMock, gitRepositoryManager, config);
+    List<RemoteRefUpdate> filteredRefUpdates = pushFilter.filter(project, refUpdates);
+
+    assertThat(filteredRefUpdates).containsExactlyElementsIn(refUpdates);
+  }
+
+  @Test
+  public void shouldFilterOutOneOutdatedRef() throws Exception {
+    RemoteRefUpdate refUpToDate = refUpdate("refs/heads/uptodate");
+    RemoteRefUpdate outdatedRef = refUpdate("refs/heads/outdated");
+    List<RemoteRefUpdate> refUpdates = Arrays.asList(refUpToDate, outdatedRef);
+    SharedRefDatabaseWrapper sharedRefDatabase =
+        new FakeSharedRefDatabaseWrapper(outdatedRef.getSrcRef());
+
+    MultisiteReplicationPushFilter pushFilter =
+        new MultisiteReplicationPushFilter(sharedRefDatabase, gitRepositoryManager, config);
+    List<RemoteRefUpdate> filteredRefUpdates = pushFilter.filter(project, refUpdates);
+
+    assertThat(filteredRefUpdates).containsExactly(refUpToDate);
+  }
+
+  @Test
+  public void shouldLoadLocalVersionAndNotFilter() throws Exception {
+    String refName = "refs/heads/temporaryOutdated";
+    RemoteRefUpdate temporaryOutdated = refUpdate(refName);
+    ObjectId latestObjectId = repo.getRepository().exactRef(refName).getObjectId();
+
+    List<RemoteRefUpdate> refUpdates = Collections.singletonList(temporaryOutdated);
+    doReturn(false).doReturn(true).when(sharedRefDatabaseMock).isUpToDate(eq(projectName), any());
+
+    MultisiteReplicationPushFilter pushFilter =
+        new MultisiteReplicationPushFilter(sharedRefDatabaseMock, gitRepositoryManager, config);
+    List<RemoteRefUpdate> filteredRefUpdates = pushFilter.filter(project, refUpdates);
+
+    assertThat(filteredRefUpdates).hasSize(1);
+    assertThat(filteredRefUpdates.get(0).getNewObjectId()).isEqualTo(latestObjectId);
+
+    verify(sharedRefDatabaseMock, times(2)).isUpToDate(any(), any());
+  }
+
+  @Test
+  public void shouldLoadLocalVersionAndFilter() throws Exception {
+    RemoteRefUpdate temporaryOutdated = refUpdate("refs/heads/temporaryOutdated");
+    repo.branch("refs/heads/temporaryOutdated").commit().create();
+    List<RemoteRefUpdate> refUpdates = Collections.singletonList(temporaryOutdated);
+    doReturn(false).doReturn(false).when(sharedRefDatabaseMock).isUpToDate(eq(projectName), any());
+
+    MultisiteReplicationPushFilter pushFilter =
+        new MultisiteReplicationPushFilter(sharedRefDatabaseMock, gitRepositoryManager, config);
+    List<RemoteRefUpdate> filteredRefUpdates = pushFilter.filter(project, refUpdates);
+
+    assertThat(filteredRefUpdates).isEmpty();
+    verify(sharedRefDatabaseMock, times(2)).isUpToDate(any(), any());
+  }
+
+  @Test
+  public void shouldFilterOutAllOutdatedChangesRef() throws Exception {
+    RemoteRefUpdate refUpToDate = refUpdate("refs/heads/uptodate");
+    RemoteRefUpdate refChangeUpToDate = refUpdate("refs/changes/25/1225/2");
+    RemoteRefUpdate changeMetaRef = refUpdate("refs/changes/12/4512/meta");
+    RemoteRefUpdate changeRef = refUpdate("refs/changes/12/4512/1");
+    List<RemoteRefUpdate> refUpdates =
+        Arrays.asList(refUpToDate, refChangeUpToDate, changeMetaRef, changeRef);
+    SharedRefDatabaseWrapper sharedRefDatabase =
+        new FakeSharedRefDatabaseWrapper(changeMetaRef.getSrcRef());
+
+    MultisiteReplicationPushFilter pushFilter =
+        new MultisiteReplicationPushFilter(sharedRefDatabase, gitRepositoryManager, config);
+    List<RemoteRefUpdate> filteredRefUpdates = pushFilter.filter(project, refUpdates);
+
+    assertThat(filteredRefUpdates).containsExactly(refUpToDate, refChangeUpToDate);
+  }
+
+  private RemoteRefUpdate refUpdate(String refName) throws Exception {
+    ObjectId srcObjId = ObjectId.fromString("0000000000000000000000000000000000000001");
+    Ref srcRef = new ObjectIdRef.Unpeeled(Ref.Storage.NEW, refName, srcObjId);
+    repo.branch(refName).commit().create();
+    return new RemoteRefUpdate(null, srcRef, "origin", false, "origin", srcObjId);
+  }
+}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/validation/ProjectVersionRefUpdateTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/validation/ProjectVersionRefUpdateTest.java
index 88d2270..d99fb4b 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/multisite/validation/ProjectVersionRefUpdateTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/validation/ProjectVersionRefUpdateTest.java
@@ -18,16 +18,21 @@
 import static com.googlesource.gerrit.plugins.multisite.validation.ProjectVersionRefUpdate.MULTI_SITE_VERSIONING_REF;
 import static com.googlesource.gerrit.plugins.multisite.validation.ProjectVersionRefUpdate.MULTI_SITE_VERSIONING_VALUE_REF;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.ArgumentMatchers.isNull;
 import static org.mockito.Mockito.atMost;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyZeroInteractions;
 import static org.mockito.Mockito.when;
 
 import com.gerritforge.gerrit.globalrefdb.validation.ProjectsFilter;
 import com.gerritforge.gerrit.globalrefdb.validation.SharedRefDatabaseWrapper;
+import com.google.common.base.Suppliers;
 import com.google.gerrit.entities.Project;
 import com.google.gerrit.entities.RefNames;
+import com.google.gerrit.server.data.RefUpdateAttribute;
 import com.google.gerrit.server.events.Event;
 import com.google.gerrit.server.events.RefUpdatedEvent;
 import com.google.gerrit.server.extensions.events.GitReferenceUpdated;
@@ -95,16 +100,9 @@
     Context.setForwardedEvent(false);
     when(sharedRefDb.get(
             A_TEST_PROJECT_NAME_KEY,
-            ProjectVersionRefUpdate.MULTI_SITE_VERSIONING_REF,
-            String.class))
-        .thenReturn(Optional.of("26f7ee61bf0e470e8393c884526eec8a9b943a63"));
-    when(sharedRefDb.get(
-            A_TEST_PROJECT_NAME_KEY,
             ProjectVersionRefUpdate.MULTI_SITE_VERSIONING_VALUE_REF,
             String.class))
         .thenReturn(Optional.of("" + (masterCommit.getCommitTime() - 1)));
-    when(sharedRefDb.compareAndPut(any(Project.NameKey.class), any(Ref.class), any(ObjectId.class)))
-        .thenReturn(true);
     when(sharedRefDb.compareAndPut(any(Project.NameKey.class), any(String.class), any(), any()))
         .thenReturn(true);
     when(refUpdatedEvent.getProjectNameKey()).thenReturn(A_TEST_PROJECT_NAME_KEY);
@@ -129,6 +127,35 @@
   }
 
   @Test
+  public void producerShouldUsePutInsteadOfCompareAndPutWhenExtendedGlobalRefDb()
+      throws IOException {
+    when(sharedRefDb.isSetOperationSupported()).thenReturn(true);
+    RefUpdatedEvent refUpdatedEvent = new RefUpdatedEvent();
+    RefUpdateAttribute refUpdatedAttribute = new RefUpdateAttribute();
+    refUpdatedAttribute.project = A_TEST_PROJECT_NAME_KEY.get();
+    refUpdatedAttribute.refName = A_TEST_REF_NAME;
+    refUpdatedEvent.refUpdate = Suppliers.memoize(() -> refUpdatedAttribute);
+
+    new ProjectVersionRefUpdateImpl(
+            repoManager, sharedRefDb, gitReferenceUpdated, verLogger, projectsFilter)
+        .onEvent(refUpdatedEvent);
+
+    Ref ref = repo.getRepository().findRef(MULTI_SITE_VERSIONING_REF);
+
+    verify(sharedRefDb, never())
+        .compareAndPut(any(Project.NameKey.class), anyString(), anyLong(), anyLong());
+
+    verify(sharedRefDb).put(any(Project.NameKey.class), any(String.class), any(String.class));
+    assertThat(ref).isNotNull();
+
+    ObjectLoader loader = repo.getRepository().open(ref.getObjectId());
+    long storedVersion = readLongObject(loader);
+    assertThat(storedVersion).isGreaterThan((long) masterCommit.getCommitTime());
+
+    verify(verLogger).log(A_TEST_PROJECT_NAME_KEY, storedVersion, 0);
+  }
+
+  @Test
   public void producerShouldUpdateProjectVersionUponForcedPushRefUpdatedEvent() throws Exception {
     Context.setForwardedEvent(false);
 
@@ -137,27 +164,20 @@
 
     Thread.sleep(1000L);
     repo.branch("master").update(masterCommit);
-
-    when(sharedRefDb.get(
-            A_TEST_PROJECT_NAME_KEY,
-            ProjectVersionRefUpdate.MULTI_SITE_VERSIONING_REF,
-            String.class))
-        .thenReturn(Optional.of("26f7ee61bf0e470e8393c884526eec8a9b943a63"));
     when(sharedRefDb.get(
             A_TEST_PROJECT_NAME_KEY,
             ProjectVersionRefUpdate.MULTI_SITE_VERSIONING_VALUE_REF,
             String.class))
         .thenReturn(Optional.of("" + (masterCommit.getCommitTime() - 1)));
-    when(sharedRefDb.compareAndPut(any(Project.NameKey.class), any(Ref.class), any(ObjectId.class)))
-        .thenReturn(true);
     when(sharedRefDb.compareAndPut(any(Project.NameKey.class), any(String.class), any(), any()))
         .thenReturn(true);
     when(refUpdatedEvent.getProjectNameKey()).thenReturn(A_TEST_PROJECT_NAME_KEY);
     when(refUpdatedEvent.getRefName()).thenReturn(A_TEST_REF_NAME);
 
-    new ProjectVersionRefUpdateImpl(
-            repoManager, sharedRefDb, gitReferenceUpdated, verLogger, projectsFilter)
-        .onEvent(refUpdatedEvent);
+    ProjectVersionRefUpdateImpl projectVersion =
+        new ProjectVersionRefUpdateImpl(
+            repoManager, sharedRefDb, gitReferenceUpdated, verLogger, projectsFilter);
+    projectVersion.onEvent(refUpdatedEvent);
 
     Ref ref = repo.getRepository().findRef(MULTI_SITE_VERSIONING_REF);
 
@@ -168,6 +188,10 @@
 
     ObjectLoader loader = repo.getRepository().open(ref.getObjectId());
     long storedVersion = readLongObject(loader);
+
+    Optional<Long> localStoredVersion = projectVersion.getProjectLocalVersion(A_TEST_PROJECT_NAME);
+    assertThat(localStoredVersion).isEqualTo(Optional.of(storedVersion));
+
     assertThat(storedVersion).isGreaterThan((long) masterPlusOneCommit.getCommitTime());
 
     verify(verLogger).log(A_TEST_PROJECT_NAME_KEY, storedVersion, 0);
@@ -179,17 +203,10 @@
     Context.setForwardedEvent(false);
     when(sharedRefDb.get(
             A_TEST_PROJECT_NAME_KEY,
-            ProjectVersionRefUpdate.MULTI_SITE_VERSIONING_REF,
-            String.class))
-        .thenReturn(Optional.empty());
-    when(sharedRefDb.get(
-            A_TEST_PROJECT_NAME_KEY,
             ProjectVersionRefUpdate.MULTI_SITE_VERSIONING_VALUE_REF,
             String.class))
         .thenReturn(Optional.empty());
 
-    when(sharedRefDb.compareAndPut(any(Project.NameKey.class), any(Ref.class), any(ObjectId.class)))
-        .thenReturn(true);
     when(sharedRefDb.compareAndPut(any(Project.NameKey.class), any(String.class), any(), any()))
         .thenReturn(true);
     when(refUpdatedEvent.getProjectNameKey()).thenReturn(A_TEST_PROJECT_NAME_KEY);
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/validation/dfsrefdb/RefFixture.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/validation/dfsrefdb/RefFixture.java
index baba94b..1885d27 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/multisite/validation/dfsrefdb/RefFixture.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/validation/dfsrefdb/RefFixture.java
@@ -33,6 +33,8 @@
   static final ObjectId AN_OBJECT_ID_1 = new ObjectId(1, 2, 3, 4, 5);
   static final ObjectId AN_OBJECT_ID_2 = new ObjectId(1, 2, 3, 4, 6);
   static final ObjectId AN_OBJECT_ID_3 = new ObjectId(1, 2, 3, 4, 7);
+  static final ObjectId AN_OUTDATED_OBJECT_ID =
+      ObjectId.fromString("da37cb098dd7df4c8662ae8afc1acae5f7567775");
   static final String A_TEST_REF_NAME = "refs/heads/master";
   static final String A_REF_NAME_OF_A_PATCHSET = "refs/changes/01/1/1";