Allow subscribing to the topic with custom group id In the past, when utilizing pull-replication as a standalone solution in conjunction with a broker (inclusive of plugins such as events-kafka, events-aws-kinesis, events-gcloud-pubsub), the consumer's groupId was defined either within the events-* plugin configuration or, in the absence of configuration, as "gerrit.instance-id." Thanks to a new feature in events-broker, since v3.4.8-4-gd9f859f, it is possible to define a different group id by configuring "replication.eventBrokerGroupId" in the replication.conf. NOTE: This feature is applicable only if the the events-* plugin implements the ExtendedBrokerApi interface, which supports the ability to subscribe to a topic with a custom group id. Bug: Issue 299327285 Change-Id: I33d056ebd3e80b6a4c9d8eb02a0c2c1b591e8eb5

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 5308294..72a6b74 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

@@ -152,7 +152,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 6287773..0c3e02c 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