Add stream event publishing timeout

Add timeout configuration for publishing stream events.

Also use StreamEventPublisherModule to bind StreamEventPublisher.

Bug: Issue 14907
Change-Id: I0f452eec502589a2ec18a748d72eb14cf8304b15

external_plugin_deps.bzl

diff --git a/external_plugin_deps.bzl b/external_plugin_deps.bzl
index 06cb71c..bfffbfe 100644
--- a/external_plugin_deps.bzl
+++ b/external_plugin_deps.bzl
@@ -9,7 +9,7 @@
 
     maven_jar(
         name = "events-broker",
-        artifact = "com.gerritforge:events-broker:3.5.0-alpha-202108041529",
-        sha1 = "309fe8cc08c46593d9990d4e5c448cc85e5a62b0",
+        artifact = "com.gerritforge:events-broker:3.5.0-alpha-202108301155",
+        sha1 = "ef4d94bb4ba1d136cd90ea901776f03a25bcb517",
     )
 

src/main/java/com/googlesource/gerrit/plugins/multisite/Configuration.java

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 81e1c12..8497882 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/Configuration.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/Configuration.java
@@ -57,6 +57,8 @@
   private static final String NUM_STRIPED_LOCKS = "numStripedLocks";
   private static final int DEFAULT_NUM_STRIPED_LOCKS = 10;
 
+  private static final long DEFALULT_STREAM_EVENT_PUBLISH_TIMEOUT = 30000;
+
   private final Supplier<Cache> cache;
   private final Supplier<Event> event;
   private final Supplier<Index> index;
@@ -190,6 +192,17 @@
     }
   }
 
+  private static long getLong(
+      Supplier<Config> cfg, String section, String subSection, String name, long defaultValue) {
+    try {
+      return cfg.get().getLong(section, subSection, name, defaultValue);
+    } catch (IllegalArgumentException e) {
+      log.error("invalid value for {}; using default value {}", name, defaultValue);
+      log.debug("Failed to retrieve long value: {}", e.getMessage(), e);
+      return defaultValue;
+    }
+  }
+
   public static class Projects {
     public static final String SECTION = "projects";
     public static final String PATTERN_KEY = "pattern";
@@ -293,15 +306,28 @@
 
   public static class Broker {
     static final String BROKER_SECTION = "broker";
+    static final String STREAM_EVENT_PUBLISH_TIMEOUT = "streamEventPublishTimeoutMs";
     private final Config cfg;
+    private long streamEventPublishTimeout;
 
     Broker(Supplier<Config> cfgSupplier) {
       cfg = cfgSupplier.get();
+      streamEventPublishTimeout =
+          getLong(
+              cfgSupplier,
+              BROKER_SECTION,
+              null,
+              STREAM_EVENT_PUBLISH_TIMEOUT,
+              DEFALULT_STREAM_EVENT_PUBLISH_TIMEOUT);
     }
 
     public String getTopic(String topicKey, String defValue) {
       return MoreObjects.firstNonNull(cfg.getString(BROKER_SECTION, null, topicKey), defValue);
     }
+
+    public long getStreamEventPublishTimeout() {
+      return streamEventPublishTimeout;
+    }
   }
 
   static boolean getBoolean(

src/main/java/com/googlesource/gerrit/plugins/multisite/event/EventModule.java

diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/event/EventModule.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/event/EventModule.java
index 94d7867..b055f26 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/event/EventModule.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/event/EventModule.java
@@ -14,13 +14,12 @@
 
 package com.googlesource.gerrit.plugins.multisite.event;
 
-import com.gerritforge.gerrit.eventbroker.StreamEventPublisher;
+import com.gerritforge.gerrit.eventbroker.publisher.StreamEventPublisherConfig;
+import com.gerritforge.gerrit.eventbroker.publisher.StreamEventPublisherModule;
 import com.google.gerrit.extensions.registration.DynamicSet;
 import com.google.gerrit.lifecycle.LifecycleModule;
 import com.google.gerrit.server.events.EventListener;
 import com.google.inject.Inject;
-import com.google.inject.TypeLiteral;
-import com.google.inject.name.Names;
 import com.googlesource.gerrit.plugins.multisite.Configuration;
 import com.googlesource.gerrit.plugins.multisite.forwarder.events.EventTopic;
 import com.googlesource.gerrit.plugins.multisite.validation.ProjectVersionRefUpdate;
@@ -36,10 +35,13 @@
   @Override
   protected void configure() {
     DynamicSet.bind(binder(), EventListener.class).to(ProjectVersionRefUpdate.class);
-    bind(new TypeLiteral<String>() {})
-        .annotatedWith(Names.named(StreamEventPublisher.STREAM_EVENTS_TOPIC))
-        .toInstance(EventTopic.STREAM_EVENT_TOPIC.topic(configuration));
 
-    DynamicSet.bind(binder(), EventListener.class).to(StreamEventPublisher.class);
+    bind(StreamEventPublisherConfig.class)
+        .toInstance(
+            new StreamEventPublisherConfig(
+                EventTopic.STREAM_EVENT_TOPIC.topic(configuration),
+                configuration.broker().getStreamEventPublishTimeout()));
+
+    install(new StreamEventPublisherModule());
   }
 }

src/main/resources/Documentation/config.md

diff --git a/src/main/resources/Documentation/config.md b/src/main/resources/Documentation/config.md
index 7eedc6a..e217994 100644
--- a/src/main/resources/Documentation/config.md
+++ b/src/main/resources/Documentation/config.md
@@ -70,6 +70,10 @@
 :   Name of the topic to use for publishing cache eviction events
     Defaults to GERRIT.EVENT.PROJECT.LIST
 
+```broker.streamEventPublishTimeoutMs```
+:   The timeout in milliseconds for publishing stream events.
+    Defaults to 30000 (30 seconds).
+
 ```ref-database.enabled```
 :   Enable the use of a shared ref-database
     Defaults: true