diff --git a/BUILD b/BUILD
index c139c5a..53b55dc 100644
--- a/BUILD
+++ b/BUILD
@@ -16,11 +16,11 @@
     ],
     resources = glob(["src/main/resources/**/*"]),
     deps = [
+        ":events-broker-neverlink",
         "//lib/commons:io",
         "//plugins/delete-project",
         "//plugins/replication",
         "@commons-lang3//jar",
-        "@events-broker//jar:neverlink",
     ],
 )
 
@@ -37,7 +37,7 @@
         ":pull_replication_util",
         "//plugins/delete-project",
         "//plugins/replication",
-        "@events-broker//jar",
+        "//plugins/events-broker",
     ],
 )
 
@@ -76,6 +76,12 @@
     visibility = ["//visibility:public"],
     exports = PLUGIN_DEPS + PLUGIN_TEST_DEPS + [
         ":pull-replication__plugin",
-        "@events-broker//jar",
+        "//plugins/events-broker",
     ],
 )
+
+java_library(
+    name = "events-broker-neverlink",
+    neverlink = 1,
+    exports = ["//plugins/events-broker"],
+)
diff --git a/Jenkinsfile b/Jenkinsfile
index 448692b..a4fff37 100644
--- a/Jenkinsfile
+++ b/Jenkinsfile
@@ -1,2 +1,3 @@
 pluginPipeline(formatCheckId: 'gerritforge:pull-replication-format-3852e64366bb37d13b8baf8af9b15cfd38eb9227',
-               buildCheckId: 'gerritforge:pull-replication-3852e64366bb37d13b8baf8af9b15cfd38eb9227')
\ No newline at end of file
+               buildCheckId: 'gerritforge:pull-replication-3852e64366bb37d13b8baf8af9b15cfd38eb9227',
+               extraModules: ['events-broker'])
diff --git a/external_plugin_deps.bzl b/external_plugin_deps.bzl
deleted file mode 100644
index 3bf728d..0000000
--- a/external_plugin_deps.bzl
+++ /dev/null
@@ -1,8 +0,0 @@
-load("//tools/bzl:maven_jar.bzl", "maven_jar")
-
-def external_plugin_deps():
-    maven_jar(
-        name = "events-broker",
-        artifact = "com.gerritforge:events-broker:3.6.0-rc3",
-        sha1 = "cb398afa4f76367be5c62b99a7ffce74ae1d3d8b",
-    )
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationModule.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationModule.java
index 93dfeb8..cbdfe1b 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationModule.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationModule.java
@@ -153,7 +153,7 @@
     if (replicationConfig.getBoolean("replication", "consumeStreamEvents", false)) {
       install(new StreamEventModule());
     } else if (eventBrokerTopic != null) {
-      install(new EventsBrokerConsumerModule(eventBrokerTopic));
+      install(new EventsBrokerConsumerModule(eventBrokerTopic, replicationConfig));
     }
 
     DynamicSet.setOf(binder(), ReplicationStateListener.class);
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/event/EventsBrokerConsumerModule.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/event/EventsBrokerConsumerModule.java
index ce1a076..a8d5cf4 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/event/EventsBrokerConsumerModule.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/event/EventsBrokerConsumerModule.java
@@ -17,21 +17,28 @@
 import com.google.gerrit.lifecycle.LifecycleModule;
 import com.google.inject.Scopes;
 import com.google.inject.name.Names;
+import com.google.inject.util.Providers;
+import org.eclipse.jgit.lib.Config;
 
 public class EventsBrokerConsumerModule extends LifecycleModule {
   public static final String STREAM_EVENTS_TOPIC_NAME = "stream_events_topic_name";
+  public static final String STREAM_EVENTS_GROUP_ID = "stream_events_group_id";
 
   private final String topicName;
+  private final Config config;
 
-  public EventsBrokerConsumerModule(String topicName) {
+  public EventsBrokerConsumerModule(String topicName, Config config) {
     this.topicName = topicName;
+    this.config = config;
   }
 
   @Override
   protected void configure() {
     bind(EventsBrokerMessageConsumer.class).in(Scopes.SINGLETON);
     bind(String.class).annotatedWith(Names.named(STREAM_EVENTS_TOPIC_NAME)).toInstance(topicName);
-
+    bind(String.class)
+        .annotatedWith(Names.named(STREAM_EVENTS_GROUP_ID))
+        .toProvider(Providers.of(config.getString("replication", null, "eventBrokerGroupId")));
     listener().to(EventsBrokerMessageConsumer.class);
   }
 }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/event/EventsBrokerMessageConsumer.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/event/EventsBrokerMessageConsumer.java
index d06681b..13b038e 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/event/EventsBrokerMessageConsumer.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/event/EventsBrokerMessageConsumer.java
@@ -14,9 +14,13 @@
 
 package com.googlesource.gerrit.plugins.replication.pull.event;
 
+import static com.googlesource.gerrit.plugins.replication.pull.event.EventsBrokerConsumerModule.STREAM_EVENTS_GROUP_ID;
 import static com.googlesource.gerrit.plugins.replication.pull.event.EventsBrokerConsumerModule.STREAM_EVENTS_TOPIC_NAME;
 
 import com.gerritforge.gerrit.eventbroker.BrokerApi;
+import com.gerritforge.gerrit.eventbroker.ExtendedBrokerApi;
+import com.google.common.flogger.FluentLogger;
+import com.google.gerrit.common.Nullable;
 import com.google.gerrit.extensions.events.LifecycleListener;
 import com.google.gerrit.extensions.registration.DynamicItem;
 import com.google.gerrit.extensions.restapi.AuthException;
@@ -29,22 +33,26 @@
 
 public class EventsBrokerMessageConsumer implements Consumer<Event>, LifecycleListener {
 
+  private static final FluentLogger logger = FluentLogger.forEnclosingClass();
   private final DynamicItem<BrokerApi> eventsBroker;
   private final StreamEventListener eventListener;
   private final ShutdownState shutdownState;
   private final String eventsTopicName;
+  private final String groupId;
 
   @Inject
   public EventsBrokerMessageConsumer(
       DynamicItem<BrokerApi> eventsBroker,
       StreamEventListener eventListener,
       ShutdownState shutdownState,
-      @Named(STREAM_EVENTS_TOPIC_NAME) String eventsTopicName) {
+      @Named(STREAM_EVENTS_TOPIC_NAME) String eventsTopicName,
+      @Nullable @Named(STREAM_EVENTS_GROUP_ID) String groupId) {
 
     this.eventsBroker = eventsBroker;
     this.eventListener = eventListener;
     this.shutdownState = shutdownState;
     this.eventsTopicName = eventsTopicName;
+    this.groupId = groupId;
   }
 
   @Override
@@ -59,7 +67,21 @@
 
   @Override
   public void start() {
-    eventsBroker.get().receiveAsync(eventsTopicName, this);
+    BrokerApi brokerApi = eventsBroker.get();
+    if (groupId == null) {
+      brokerApi.receiveAsync(eventsTopicName, this);
+      return;
+    }
+
+    if (!(eventsBroker instanceof ExtendedBrokerApi)) {
+      throw new IllegalArgumentException(
+          String.format(
+              "Failed to load the pull-replication plugin: %s does not support the custom group-id '%s'.\n"
+                  + "Remove replication.eventBrokerGroupId from replication.config or install a different event-broker plugin.",
+              eventsBroker.getClass(), groupId));
+    }
+
+    ((ExtendedBrokerApi) brokerApi).receiveAsync(eventsTopicName, groupId, this);
   }
 
   @Override
diff --git a/src/main/resources/Documentation/config.md b/src/main/resources/Documentation/config.md
index 128179c..0c31995 100644
--- a/src/main/resources/Documentation/config.md
+++ b/src/main/resources/Documentation/config.md
@@ -222,6 +222,16 @@
 
 	Default: unset
 
+replication.eventBrokerGroupId
+:	Optional group id for consumers. Used for keeping the offset of the
+	the last acked message on the topic with a specific group id; supported
+	only if the event-broker plugin supports the ExtendedBrokerApi
+	interface.
+	When not set, the messages received are acked using a global group id
+	shared by all subscribers in Gerrit.
+
+	Default: unset
+
 replication.maxConnectionsPerRoute
 :	Maximum number of HTTP connections per one HTTP route.
 
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/event/EventsBrokerMessageConsumerTest.java b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/event/EventsBrokerMessageConsumerTest.java
index 9640b82..8739d4b 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/event/EventsBrokerMessageConsumerTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/event/EventsBrokerMessageConsumerTest.java
@@ -49,7 +49,7 @@
     shutdownState = new ShutdownState();
     objectUnderTest =
         new EventsBrokerMessageConsumer(
-            eventsBrokerDynamicItem, eventListener, shutdownState, "topicName");
+            eventsBrokerDynamicItem, eventListener, shutdownState, "topicName", null);
   }
 
   @Test