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 3226a42..7348137 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/Configuration.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/Configuration.java
@@ -18,28 +18,20 @@
 import static com.google.common.base.Suppliers.ofInstance;
 
 import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Strings;
 import com.google.common.base.Supplier;
 import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Multimap;
 import com.google.common.collect.MultimapBuilder;
 import com.google.gerrit.server.config.SitePaths;
 import com.google.inject.Inject;
 import com.google.inject.Singleton;
 import com.google.inject.spi.Message;
-import com.googlesource.gerrit.plugins.multisite.forwarder.events.EventFamily;
 import com.googlesource.gerrit.plugins.multisite.validation.dfsrefdb.SharedRefEnforcement.EnforcePolicy;
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.UUID;
-import org.apache.kafka.common.serialization.StringSerializer;
 import org.eclipse.jgit.errors.ConfigInvalidException;
 import org.eclipse.jgit.lib.Config;
 import org.eclipse.jgit.storage.file.FileBasedConfig;
@@ -72,7 +64,6 @@
   private final Supplier<SharedRefDatabase> sharedRefDb;
   private final Supplier<Collection<Message>> replicationConfigValidation;
   private final Config multiSiteConfig;
-  private final KafkaConfiguration kafkaConfig;
 
   @Inject
   Configuration(SitePaths sitePaths) {
@@ -83,7 +74,6 @@
   public Configuration(Config multiSiteConfig, Config replicationConfig) {
     Supplier<Config> lazyMultiSiteCfg = lazyLoad(multiSiteConfig);
     this.multiSiteConfig = multiSiteConfig;
-    this.kafkaConfig = new KafkaConfiguration(multiSiteConfig);
     replicationConfigValidation = lazyValidateReplicatioConfig(replicationConfig);
     cache = memoize(() -> new Cache(lazyMultiSiteCfg));
     event = memoize(() -> new Event(lazyMultiSiteCfg));
@@ -99,14 +89,6 @@
     return sharedRefDb.get();
   }
 
-  public Kafka getKafka() {
-    return kafkaConfig.getKafka();
-  }
-
-  public KafkaPublisher kafkaPublisher() {
-    return kafkaConfig.kafkaPublisher();
-  }
-
   public Cache cache() {
     return cache.get();
   }
@@ -119,10 +101,6 @@
     return index.get();
   }
 
-  public KafkaSubscriber kafkaSubscriber() {
-    return kafkaConfig.kafkaSubscriber();
-  }
-
   public Collection<Message> validate() {
     return replicationConfigValidation.get();
   }
@@ -182,179 +160,6 @@
     }
   }
 
-  public static class Kafka {
-    private final Map<EventFamily, String> eventTopics;
-    private final String bootstrapServers;
-
-    private static final ImmutableMap<EventFamily, String> EVENT_TOPICS =
-        ImmutableMap.of(
-            EventFamily.INDEX_EVENT,
-            "GERRIT.EVENT.INDEX",
-            EventFamily.STREAM_EVENT,
-            "GERRIT.EVENT.STREAM",
-            EventFamily.CACHE_EVENT,
-            "GERRIT.EVENT.CACHE",
-            EventFamily.PROJECT_LIST_EVENT,
-            "GERRIT.EVENT.PROJECT.LIST");
-
-    Kafka(Supplier<Config> config) {
-      this.bootstrapServers =
-          getString(
-              config,
-              KafkaConfiguration.KAFKA_SECTION,
-              null,
-              "bootstrapServers",
-              KafkaConfiguration.DEFAULT_KAFKA_BOOTSTRAP_SERVERS);
-
-      this.eventTopics = new HashMap<>();
-      for (Map.Entry<EventFamily, String> topicDefault : EVENT_TOPICS.entrySet()) {
-        String topicConfigKey = topicDefault.getKey().lowerCamelName() + "Topic";
-        eventTopics.put(
-            topicDefault.getKey(),
-            getString(
-                config,
-                KafkaConfiguration.KAFKA_SECTION,
-                null,
-                topicConfigKey,
-                topicDefault.getValue()));
-      }
-    }
-
-    public String getTopic(EventFamily eventType) {
-      return eventTopics.get(eventType);
-    }
-
-    public String getBootstrapServers() {
-      return bootstrapServers;
-    }
-
-    private static String getString(
-        Supplier<Config> cfg, String section, String subsection, String name, String defaultValue) {
-      String value = cfg.get().getString(section, subsection, name);
-      if (!Strings.isNullOrEmpty(value)) {
-        return value;
-      }
-      return defaultValue;
-    }
-  }
-
-  public static class KafkaPublisher extends Properties {
-    private static final long serialVersionUID = 0L;
-
-    public static final String KAFKA_STRING_SERIALIZER = StringSerializer.class.getName();
-
-    public static final String KAFKA_PUBLISHER_SUBSECTION = "publisher";
-    public static final boolean DEFAULT_BROKER_ENABLED = false;
-
-    private final boolean enabled;
-    private final Map<EventFamily, Boolean> eventsEnabled;
-
-    KafkaPublisher(Supplier<Config> cfg) {
-      enabled =
-          cfg.get()
-              .getBoolean(
-                  KafkaConfiguration.KAFKA_SECTION,
-                  KAFKA_PUBLISHER_SUBSECTION,
-                  KafkaConfiguration.ENABLE_KEY,
-                  DEFAULT_BROKER_ENABLED);
-
-      eventsEnabled = KafkaConfiguration.eventsEnabled(cfg, KAFKA_PUBLISHER_SUBSECTION);
-
-      if (enabled) {
-        setDefaults();
-        KafkaConfiguration.applyKafkaConfig(cfg, KAFKA_PUBLISHER_SUBSECTION, this);
-      }
-    }
-
-    private void setDefaults() {
-      put("acks", "all");
-      put("retries", 0);
-      put("batch.size", 16384);
-      put("linger.ms", 1);
-      put("buffer.memory", 33554432);
-      put("key.serializer", KAFKA_STRING_SERIALIZER);
-      put("value.serializer", KAFKA_STRING_SERIALIZER);
-      put("reconnect.backoff.ms", 5000L);
-    }
-
-    public boolean enabled() {
-      return enabled;
-    }
-
-    public boolean enabledEvent(EventFamily eventType) {
-      return eventsEnabled.get(eventType);
-    }
-  }
-
-  public static class KafkaSubscriber extends Properties {
-    private static final long serialVersionUID = 1L;
-
-    static final String KAFKA_SUBSCRIBER_SUBSECTION = "subscriber";
-
-    private final boolean enabled;
-    private final Integer pollingInterval;
-    private Map<EventFamily, Boolean> eventsEnabled;
-    private final Config cfg;
-
-    public KafkaSubscriber(Supplier<Config> configSupplier) {
-      this.cfg = configSupplier.get();
-
-      this.pollingInterval =
-          cfg.getInt(
-              KafkaConfiguration.KAFKA_SECTION,
-              KAFKA_SUBSCRIBER_SUBSECTION,
-              "pollingIntervalMs",
-              KafkaConfiguration.DEFAULT_POLLING_INTERVAL_MS);
-
-      enabled =
-          cfg.getBoolean(
-              KafkaConfiguration.KAFKA_SECTION,
-              KAFKA_SUBSCRIBER_SUBSECTION,
-              KafkaConfiguration.ENABLE_KEY,
-              false);
-
-      eventsEnabled = KafkaConfiguration.eventsEnabled(configSupplier, KAFKA_SUBSCRIBER_SUBSECTION);
-
-      if (enabled) {
-        KafkaConfiguration.applyKafkaConfig(configSupplier, KAFKA_SUBSCRIBER_SUBSECTION, this);
-      }
-    }
-
-    public boolean enabled() {
-      return enabled;
-    }
-
-    public boolean enabledEvent(EventFamily eventFamily) {
-      return eventsEnabled.get(eventFamily);
-    }
-
-    public Properties initPropsWith(UUID instanceId) {
-      String groupId =
-          getString(
-              cfg,
-              KafkaConfiguration.KAFKA_SECTION,
-              KAFKA_SUBSCRIBER_SUBSECTION,
-              "groupId",
-              instanceId.toString());
-      this.put("group.id", groupId);
-
-      return this;
-    }
-
-    public Integer getPollingInterval() {
-      return pollingInterval;
-    }
-
-    private String getString(
-        Config cfg, String section, String subsection, String name, String defaultValue) {
-      String value = cfg.getString(section, subsection, name);
-      if (!Strings.isNullOrEmpty(value)) {
-        return value;
-      }
-      return defaultValue;
-    }
-  }
-
   public static class SharedRefDatabase {
     public static final String SECTION = "ref-database";
     public static final String SUBSECTION_ENFORCEMENT_RULES = "enforcementRules";
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/KafkaConfiguration.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/KafkaConfiguration.java
index 3376b49..b88699a 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/KafkaConfiguration.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/KafkaConfiguration.java
@@ -21,34 +21,44 @@
 import com.google.common.base.CaseFormat;
 import com.google.common.base.Strings;
 import com.google.common.base.Supplier;
-import com.googlesource.gerrit.plugins.multisite.Configuration.Kafka;
-import com.googlesource.gerrit.plugins.multisite.Configuration.KafkaPublisher;
-import com.googlesource.gerrit.plugins.multisite.Configuration.KafkaSubscriber;
+import com.google.common.collect.ImmutableMap;
+import com.google.gerrit.server.config.SitePaths;
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
 import com.googlesource.gerrit.plugins.multisite.forwarder.events.EventFamily;
 import java.io.IOException;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Properties;
+import java.util.UUID;
+import org.apache.kafka.common.serialization.StringSerializer;
 import org.eclipse.jgit.errors.ConfigInvalidException;
 import org.eclipse.jgit.lib.Config;
 import org.eclipse.jgit.storage.file.FileBasedConfig;
+import org.eclipse.jgit.util.FS;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+@Singleton
 public class KafkaConfiguration {
-  private static final Logger log = LoggerFactory.getLogger(KafkaConfiguration.class);
 
-  static final String KAFKA_PROPERTY_PREFIX = "KafkaProp-";
+  private static final Logger log = LoggerFactory.getLogger(KafkaConfiguration.class);
+  public static final String KAFKA_PROPERTY_PREFIX = "KafkaProp-";
   static final String KAFKA_SECTION = "kafka";
   static final String ENABLE_KEY = "enabled";
   static final String DEFAULT_KAFKA_BOOTSTRAP_SERVERS = "localhost:9092";
   static final boolean DEFAULT_ENABLE_PROCESSING = true;
-  static final int DEFAULT_POLLING_INTERVAL_MS = 1000;
+  private static final int DEFAULT_POLLING_INTERVAL_MS = 1000;
 
   private final Supplier<KafkaSubscriber> subscriber;
   private final Supplier<Kafka> kafka;
   private final Supplier<KafkaPublisher> publisher;
 
+  @Inject
+  KafkaConfiguration(SitePaths sitePaths) {
+    this(getConfigFile(sitePaths, Configuration.MULTI_SITE_CONFIG));
+  }
+
   @VisibleForTesting
   public KafkaConfiguration(Config kafkaConfig) {
     Supplier<Config> lazyCfg = lazyLoad(kafkaConfig);
@@ -57,6 +67,10 @@
     subscriber = memoize(() -> new KafkaSubscriber(lazyCfg));
   }
 
+  private static FileBasedConfig getConfigFile(SitePaths sitePaths, String configFileName) {
+    return new FileBasedConfig(sitePaths.etc_dir.resolve(configFileName).toFile(), FS.DETECTED);
+  }
+
   public Kafka getKafka() {
     return kafka.get();
   }
@@ -65,7 +79,7 @@
     return subscriber.get();
   }
 
-  static void applyKafkaConfig(
+  private static void applyKafkaConfig(
       Supplier<Config> configSupplier, String subsectionName, Properties target) {
     Config config = configSupplier.get();
     for (String section : config.getSubsections(KAFKA_SECTION)) {
@@ -103,7 +117,8 @@
     return defaultValue;
   }
 
-  static Map<EventFamily, Boolean> eventsEnabled(Supplier<Config> config, String subsection) {
+  private static Map<EventFamily, Boolean> eventsEnabled(
+      Supplier<Config> config, String subsection) {
     Map<EventFamily, Boolean> eventsEnabled = new HashMap<>();
     for (EventFamily eventFamily : EventFamily.values()) {
       String enabledConfigKey = eventFamily.lowerCamelName() + "Enabled";
@@ -112,11 +127,7 @@
           eventFamily,
           config
               .get()
-              .getBoolean(
-                  KafkaConfiguration.KAFKA_SECTION,
-                  subsection,
-                  enabledConfigKey,
-                  KafkaConfiguration.DEFAULT_ENABLE_PROCESSING));
+              .getBoolean(KAFKA_SECTION, subsection, enabledConfigKey, DEFAULT_ENABLE_PROCESSING));
     }
     return eventsEnabled;
   }
@@ -142,4 +153,156 @@
     }
     return ofInstance(config);
   }
+
+  public static class Kafka {
+    private final Map<EventFamily, String> eventTopics;
+    private final String bootstrapServers;
+
+    private static final ImmutableMap<EventFamily, String> EVENT_TOPICS =
+        ImmutableMap.of(
+            EventFamily.INDEX_EVENT,
+            "GERRIT.EVENT.INDEX",
+            EventFamily.STREAM_EVENT,
+            "GERRIT.EVENT.STREAM",
+            EventFamily.CACHE_EVENT,
+            "GERRIT.EVENT.CACHE",
+            EventFamily.PROJECT_LIST_EVENT,
+            "GERRIT.EVENT.PROJECT.LIST");
+
+    Kafka(Supplier<Config> config) {
+      this.bootstrapServers =
+          getString(
+              config, KAFKA_SECTION, null, "bootstrapServers", DEFAULT_KAFKA_BOOTSTRAP_SERVERS);
+
+      this.eventTopics = new HashMap<>();
+      for (Map.Entry<EventFamily, String> topicDefault : EVENT_TOPICS.entrySet()) {
+        String topicConfigKey = topicDefault.getKey().lowerCamelName() + "Topic";
+        eventTopics.put(
+            topicDefault.getKey(),
+            getString(config, KAFKA_SECTION, null, topicConfigKey, topicDefault.getValue()));
+      }
+    }
+
+    public String getTopic(EventFamily eventType) {
+      return eventTopics.get(eventType);
+    }
+
+    public String getBootstrapServers() {
+      return bootstrapServers;
+    }
+
+    private static String getString(
+        Supplier<Config> cfg, String section, String subsection, String name, String defaultValue) {
+      String value = cfg.get().getString(section, subsection, name);
+      if (!Strings.isNullOrEmpty(value)) {
+        return value;
+      }
+      return defaultValue;
+    }
+  }
+
+  public static class KafkaPublisher extends Properties {
+    private static final long serialVersionUID = 0L;
+
+    public static final String KAFKA_STRING_SERIALIZER = StringSerializer.class.getName();
+
+    public static final String KAFKA_PUBLISHER_SUBSECTION = "publisher";
+    public static final boolean DEFAULT_BROKER_ENABLED = false;
+
+    private final boolean enabled;
+    private final Map<EventFamily, Boolean> eventsEnabled;
+
+    private KafkaPublisher(Supplier<Config> cfg) {
+      enabled =
+          cfg.get()
+              .getBoolean(
+                  KAFKA_SECTION, KAFKA_PUBLISHER_SUBSECTION, ENABLE_KEY, DEFAULT_BROKER_ENABLED);
+
+      eventsEnabled = eventsEnabled(cfg, KAFKA_PUBLISHER_SUBSECTION);
+
+      if (enabled) {
+        setDefaults();
+        applyKafkaConfig(cfg, KAFKA_PUBLISHER_SUBSECTION, this);
+      }
+    }
+
+    private void setDefaults() {
+      put("acks", "all");
+      put("retries", 0);
+      put("batch.size", 16384);
+      put("linger.ms", 1);
+      put("buffer.memory", 33554432);
+      put("key.serializer", KAFKA_STRING_SERIALIZER);
+      put("value.serializer", KAFKA_STRING_SERIALIZER);
+      put("reconnect.backoff.ms", 5000L);
+    }
+
+    public boolean enabled() {
+      return enabled;
+    }
+
+    public boolean enabledEvent(EventFamily eventType) {
+      return eventsEnabled.get(eventType);
+    }
+  }
+
+  public static class KafkaSubscriber extends Properties {
+    private static final long serialVersionUID = 1L;
+
+    static final String KAFKA_SUBSCRIBER_SUBSECTION = "subscriber";
+
+    private final boolean enabled;
+    private final Integer pollingInterval;
+    private Map<EventFamily, Boolean> eventsEnabled;
+    private final Config cfg;
+
+    public KafkaSubscriber(Supplier<Config> configSupplier) {
+      this.cfg = configSupplier.get();
+
+      this.pollingInterval =
+          cfg.getInt(
+              KAFKA_SECTION,
+              KAFKA_SUBSCRIBER_SUBSECTION,
+              "pollingIntervalMs",
+              DEFAULT_POLLING_INTERVAL_MS);
+
+      enabled = cfg.getBoolean(KAFKA_SECTION, KAFKA_SUBSCRIBER_SUBSECTION, ENABLE_KEY, false);
+
+      eventsEnabled = eventsEnabled(configSupplier, KAFKA_SUBSCRIBER_SUBSECTION);
+
+      if (enabled) {
+        applyKafkaConfig(configSupplier, KAFKA_SUBSCRIBER_SUBSECTION, this);
+      }
+    }
+
+    public boolean enabled() {
+      return enabled;
+    }
+
+    public boolean enabledEvent(EventFamily eventFamily) {
+      return eventsEnabled.get(eventFamily);
+    }
+
+    public Properties initPropsWith(UUID instanceId) {
+      String groupId =
+          getString(
+              cfg, KAFKA_SECTION, KAFKA_SUBSCRIBER_SUBSECTION, "groupId", instanceId.toString());
+      this.put("group.id", groupId);
+
+      return this;
+    }
+
+    public Integer getPollingInterval() {
+      return pollingInterval;
+    }
+
+    private String getString(
+        Config cfg, String section, String subsection, String name, String defaultValue) {
+      String value = cfg.getString(section, subsection, name);
+      if (!Strings.isNullOrEmpty(value)) {
+        return value;
+      }
+      return defaultValue;
+    }
+  }
 }
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 620d658..c9d7c4b 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/Module.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/Module.java
@@ -14,6 +14,7 @@
 
 package com.googlesource.gerrit.plugins.multisite;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.gerrit.extensions.events.ProjectDeletedListener;
 import com.google.gerrit.extensions.registration.DynamicSet;
 import com.google.gerrit.lifecycle.LifecycleModule;
@@ -47,11 +48,31 @@
 
 public class Module extends LifecycleModule {
   private static final Logger log = LoggerFactory.getLogger(Module.class);
-  private final Configuration config;
+  private Configuration config;
+  private final boolean disableGitRepositoryValidation;
+  private KafkaConfiguration kafkaConfig;
 
   @Inject
-  public Module(Configuration config) {
+  public Module(Configuration config, KafkaConfiguration kafkaConfig) {
+    this(config, kafkaConfig, false);
+  }
+
+  // TODO: It is not possible to properly test the libModules in Gerrit.
+  // Disable the Git repository validation during integration test and then build the necessary
+  // support
+  // in Gerrit for it.
+  @VisibleForTesting
+  public Module(
+      Configuration config,
+      KafkaConfiguration kafkaConfig,
+      boolean disableGitRepositoryValidation) {
+    init(config, kafkaConfig);
+    this.disableGitRepositoryValidation = disableGitRepositoryValidation;
+  }
+
+  private void init(Configuration config, KafkaConfiguration kafkaConfig) {
     this.config = config;
+    this.kafkaConfig = kafkaConfig;
   }
 
   @Override
@@ -77,12 +98,12 @@
       install(new IndexModule());
     }
 
-    if (config.kafkaSubscriber().enabled()) {
-      install(new KafkaConsumerModule(config.kafkaSubscriber()));
+    if (kafkaConfig.kafkaSubscriber().enabled()) {
+      install(new KafkaConsumerModule(kafkaConfig.kafkaSubscriber()));
       install(new ForwardedEventRouterModule());
     }
-    if (config.kafkaPublisher().enabled()) {
-      install(new BrokerForwarderModule(config.kafkaPublisher()));
+    if (kafkaConfig.kafkaPublisher().enabled()) {
+      install(new BrokerForwarderModule(kafkaConfig.kafkaPublisher()));
     }
 
     if (config.getSharedRefDb().isEnabled()) {
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/broker/BrokerMetrics.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/broker/BrokerMetrics.java
new file mode 100644
index 0000000..f6be65a
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/broker/BrokerMetrics.java
@@ -0,0 +1,58 @@
+// Copyright (C) 2019 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.metrics.Counter1;
+import com.google.gerrit.metrics.Description;
+import com.google.gerrit.metrics.Field;
+import com.google.gerrit.metrics.MetricMaker;
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+
+@Singleton
+public class BrokerMetrics {
+  private static final String PUBLISHER_SUCCESS_COUNTER = "broker_msg_publisher_counter";
+  private static final String PUBLISHER_FAILURE_COUNTER = "broker_msg_publisher_failure_counter";
+
+  private final Counter1<String> brokerPublisherSuccessCounter;
+  private final Counter1<String> brokerPublisherFailureCounter;
+
+  @Inject
+  public BrokerMetrics(MetricMaker metricMaker) {
+
+    this.brokerPublisherSuccessCounter =
+        metricMaker.newCounter(
+            "multi_site/broker/broker_message_publisher_counter",
+            new Description("Number of messages published by the broker publisher")
+                .setRate()
+                .setUnit("messages"),
+            Field.ofString(PUBLISHER_SUCCESS_COUNTER, "Broker message published count"));
+    this.brokerPublisherFailureCounter =
+        metricMaker.newCounter(
+            "multi_site/broker/broker_message_publisher_failure_counter",
+            new Description("Number of messages failed to publish by the broker publisher")
+                .setRate()
+                .setUnit("errors"),
+            Field.ofString(PUBLISHER_FAILURE_COUNTER, "Broker failed to publish message count"));
+  }
+
+  public void incrementBrokerPublishedMessage() {
+    brokerPublisherSuccessCounter.increment(PUBLISHER_SUCCESS_COUNTER);
+  }
+
+  public void incrementBrokerFailedToPublishMessage() {
+    brokerPublisherFailureCounter.increment(PUBLISHER_FAILURE_COUNTER);
+  }
+}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/broker/BrokerPublisher.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/broker/BrokerPublisher.java
index 4dca9e9..62716e6 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/broker/BrokerPublisher.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/broker/BrokerPublisher.java
@@ -39,17 +39,20 @@
   private final Gson gson;
   private final UUID instanceId;
   private final MessageLogger msgLog;
+  private final BrokerMetrics brokerMetrics;
 
   @Inject
   public BrokerPublisher(
       BrokerSession session,
       @BrokerGson Gson gson,
       @InstanceId UUID instanceId,
-      MessageLogger msgLog) {
+      MessageLogger msgLog,
+      BrokerMetrics brokerMetrics) {
     this.session = session;
     this.gson = gson;
     this.instanceId = instanceId;
     this.msgLog = msgLog;
+    this.brokerMetrics = brokerMetrics;
   }
 
   @Override
@@ -73,7 +76,13 @@
 
     SourceAwareEventWrapper brokerEvent = toBrokerEvent(event);
     msgLog.log(Direction.PUBLISH, brokerEvent);
-    return session.publishEvent(eventType, getPayload(brokerEvent));
+    Boolean eventPublished = session.publishEvent(eventType, getPayload(brokerEvent));
+    if (eventPublished) {
+      brokerMetrics.incrementBrokerPublishedMessage();
+    } else {
+      brokerMetrics.incrementBrokerFailedToPublishMessage();
+    }
+    return eventPublished;
   }
 
   private String getPayload(SourceAwareEventWrapper event) {
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/broker/kafka/KafkaSession.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/broker/kafka/KafkaSession.java
index 6c3c3cf..7c83346 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/broker/kafka/KafkaSession.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/broker/kafka/KafkaSession.java
@@ -15,8 +15,8 @@
 package com.googlesource.gerrit.plugins.multisite.broker.kafka;
 
 import com.google.inject.Inject;
-import com.googlesource.gerrit.plugins.multisite.Configuration;
 import com.googlesource.gerrit.plugins.multisite.InstanceId;
+import com.googlesource.gerrit.plugins.multisite.KafkaConfiguration;
 import com.googlesource.gerrit.plugins.multisite.broker.BrokerSession;
 import com.googlesource.gerrit.plugins.multisite.forwarder.events.EventFamily;
 import java.util.UUID;
@@ -31,13 +31,13 @@
 
 public class KafkaSession implements BrokerSession {
   private static final Logger LOGGER = LoggerFactory.getLogger(KafkaSession.class);
-  private final Configuration properties;
+  private KafkaConfiguration properties;
   private final UUID instanceId;
   private volatile Producer<String, String> producer;
 
   @Inject
-  public KafkaSession(Configuration configuration, @InstanceId UUID instanceId) {
-    this.properties = configuration;
+  public KafkaSession(KafkaConfiguration kafkaConfig, @InstanceId UUID instanceId) {
+    this.properties = kafkaConfig;
     this.instanceId = instanceId;
   }
 
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/broker/BrokerForwarderModule.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/broker/BrokerForwarderModule.java
index 7c6bef3..b7ee20b 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/broker/BrokerForwarderModule.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/broker/BrokerForwarderModule.java
@@ -16,7 +16,7 @@
 
 import com.google.gerrit.extensions.registration.DynamicSet;
 import com.google.gerrit.lifecycle.LifecycleModule;
-import com.googlesource.gerrit.plugins.multisite.Configuration.KafkaPublisher;
+import com.googlesource.gerrit.plugins.multisite.KafkaConfiguration.KafkaPublisher;
 import com.googlesource.gerrit.plugins.multisite.broker.BrokerPublisher;
 import com.googlesource.gerrit.plugins.multisite.broker.BrokerSession;
 import com.googlesource.gerrit.plugins.multisite.broker.kafka.KafkaSession;
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/kafka/consumer/AbstractKafkaSubcriber.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/kafka/consumer/AbstractKafkaSubcriber.java
index e6c73f1..4116c06 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/kafka/consumer/AbstractKafkaSubcriber.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/kafka/consumer/AbstractKafkaSubcriber.java
@@ -23,8 +23,8 @@
 import com.google.gerrit.server.util.ManualRequestContext;
 import com.google.gerrit.server.util.OneOffRequestContext;
 import com.google.gson.Gson;
-import com.googlesource.gerrit.plugins.multisite.Configuration;
 import com.googlesource.gerrit.plugins.multisite.InstanceId;
+import com.googlesource.gerrit.plugins.multisite.KafkaConfiguration;
 import com.googlesource.gerrit.plugins.multisite.MessageLogger;
 import com.googlesource.gerrit.plugins.multisite.MessageLogger.Direction;
 import com.googlesource.gerrit.plugins.multisite.broker.BrokerGson;
@@ -52,12 +52,12 @@
   private final UUID instanceId;
   private final AtomicBoolean closed = new AtomicBoolean(false);
   private final Deserializer<SourceAwareEventWrapper> valueDeserializer;
-  private final Configuration configuration;
+  private final KafkaConfiguration configuration;
   private final OneOffRequestContext oneOffCtx;
   private final MessageLogger msgLog;
 
   public AbstractKafkaSubcriber(
-      Configuration configuration,
+      KafkaConfiguration configuration,
       Deserializer<byte[]> keyDeserializer,
       Deserializer<SourceAwareEventWrapper> valueDeserializer,
       ForwardedEventRouter eventRouter,
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/kafka/consumer/CacheEvictionEventSubscriber.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/kafka/consumer/CacheEvictionEventSubscriber.java
index 8e724be..12b2790 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/kafka/consumer/CacheEvictionEventSubscriber.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/kafka/consumer/CacheEvictionEventSubscriber.java
@@ -19,8 +19,8 @@
 import com.google.gson.Gson;
 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.KafkaConfiguration;
 import com.googlesource.gerrit.plugins.multisite.MessageLogger;
 import com.googlesource.gerrit.plugins.multisite.broker.BrokerGson;
 import com.googlesource.gerrit.plugins.multisite.forwarder.events.EventFamily;
@@ -32,7 +32,7 @@
 public class CacheEvictionEventSubscriber extends AbstractKafkaSubcriber {
   @Inject
   public CacheEvictionEventSubscriber(
-      Configuration configuration,
+      KafkaConfiguration configuration,
       Deserializer<byte[]> keyDeserializer,
       Deserializer<SourceAwareEventWrapper> valueDeserializer,
       StreamEventRouter eventRouter,
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/kafka/consumer/IndexEventSubscriber.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/kafka/consumer/IndexEventSubscriber.java
index e52b624..9020486 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/kafka/consumer/IndexEventSubscriber.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/kafka/consumer/IndexEventSubscriber.java
@@ -19,8 +19,8 @@
 import com.google.gson.Gson;
 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.KafkaConfiguration;
 import com.googlesource.gerrit.plugins.multisite.MessageLogger;
 import com.googlesource.gerrit.plugins.multisite.broker.BrokerGson;
 import com.googlesource.gerrit.plugins.multisite.forwarder.events.EventFamily;
@@ -32,7 +32,7 @@
 public class IndexEventSubscriber extends AbstractKafkaSubcriber {
   @Inject
   public IndexEventSubscriber(
-      Configuration configuration,
+      KafkaConfiguration configuration,
       Deserializer<byte[]> keyDeserializer,
       Deserializer<SourceAwareEventWrapper> valueDeserializer,
       IndexEventRouter eventRouter,
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/kafka/consumer/KafkaConsumerModule.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/kafka/consumer/KafkaConsumerModule.java
index 5bc7669..3e0e5d7 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/kafka/consumer/KafkaConsumerModule.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/kafka/consumer/KafkaConsumerModule.java
@@ -17,7 +17,7 @@
 import com.google.gerrit.extensions.registration.DynamicSet;
 import com.google.gerrit.lifecycle.LifecycleModule;
 import com.google.inject.TypeLiteral;
-import com.googlesource.gerrit.plugins.multisite.Configuration.KafkaSubscriber;
+import com.googlesource.gerrit.plugins.multisite.KafkaConfiguration.KafkaSubscriber;
 import com.googlesource.gerrit.plugins.multisite.forwarder.events.EventFamily;
 import com.googlesource.gerrit.plugins.multisite.forwarder.events.MultiSiteEvent;
 import java.util.concurrent.Executor;
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/kafka/consumer/ProjectUpdateEventSubscriber.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/kafka/consumer/ProjectUpdateEventSubscriber.java
index a949cd2..9e5db3a 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/kafka/consumer/ProjectUpdateEventSubscriber.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/kafka/consumer/ProjectUpdateEventSubscriber.java
@@ -19,8 +19,8 @@
 import com.google.gson.Gson;
 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.KafkaConfiguration;
 import com.googlesource.gerrit.plugins.multisite.MessageLogger;
 import com.googlesource.gerrit.plugins.multisite.broker.BrokerGson;
 import com.googlesource.gerrit.plugins.multisite.forwarder.events.EventFamily;
@@ -32,7 +32,7 @@
 public class ProjectUpdateEventSubscriber extends AbstractKafkaSubcriber {
   @Inject
   public ProjectUpdateEventSubscriber(
-      Configuration configuration,
+      KafkaConfiguration configuration,
       Deserializer<byte[]> keyDeserializer,
       Deserializer<SourceAwareEventWrapper> valueDeserializer,
       ProjectListUpdateRouter eventRouter,
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/kafka/consumer/StreamEventSubscriber.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/kafka/consumer/StreamEventSubscriber.java
index d43a3c3..66070b6 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/multisite/kafka/consumer/StreamEventSubscriber.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/kafka/consumer/StreamEventSubscriber.java
@@ -19,8 +19,8 @@
 import com.google.gson.Gson;
 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.KafkaConfiguration;
 import com.googlesource.gerrit.plugins.multisite.MessageLogger;
 import com.googlesource.gerrit.plugins.multisite.broker.BrokerGson;
 import com.googlesource.gerrit.plugins.multisite.forwarder.events.EventFamily;
@@ -32,7 +32,7 @@
 public class StreamEventSubscriber extends AbstractKafkaSubcriber {
   @Inject
   public StreamEventSubscriber(
-      Configuration configuration,
+      KafkaConfiguration configuration,
       Deserializer<byte[]> keyDeserializer,
       Deserializer<SourceAwareEventWrapper> valueDeserializer,
       StreamEventRouter eventRouter,
diff --git a/src/main/resources/Documentation/about.md b/src/main/resources/Documentation/about.md
index 5c5d75a..83082b2 100644
--- a/src/main/resources/Documentation/about.md
+++ b/src/main/resources/Documentation/about.md
@@ -86,3 +86,16 @@
 
 For further information and supported options, refer to [config](config.md)
 documentation.
+
+## Metrics
+
+@PLUGIN@ plugin exposes following metrics:
+
+### Broker message publisher
+* Broker message published count
+
+`metric=multi_site/broker/broker_message_publisher_counter/broker_msg_publisher_counter, type=com.codahale.metrics.Meter`
+
+* Broker failed to publish message count
+
+`metric=multi_site/broker/broker_message_publisher_failure_counter/broker_msg_publisher_failure_counter, type=com.codahale.metrics.Meter`
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/ConfigurationTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/ConfigurationTest.java
index ec3a431..0863f55 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/multisite/ConfigurationTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/ConfigurationTest.java
@@ -18,16 +18,11 @@
 import static com.googlesource.gerrit.plugins.multisite.Configuration.Cache.CACHE_SECTION;
 import static com.googlesource.gerrit.plugins.multisite.Configuration.Cache.PATTERN_KEY;
 import static com.googlesource.gerrit.plugins.multisite.Configuration.DEFAULT_THREAD_POOL_SIZE;
-import static com.googlesource.gerrit.plugins.multisite.Configuration.ENABLE_KEY;
 import static com.googlesource.gerrit.plugins.multisite.Configuration.Event.EVENT_SECTION;
 import static com.googlesource.gerrit.plugins.multisite.Configuration.Forwarding.DEFAULT_SYNCHRONIZE;
 import static com.googlesource.gerrit.plugins.multisite.Configuration.Forwarding.SYNCHRONIZE_KEY;
 import static com.googlesource.gerrit.plugins.multisite.Configuration.Index.INDEX_SECTION;
-import static com.googlesource.gerrit.plugins.multisite.Configuration.KafkaPublisher.KAFKA_PUBLISHER_SUBSECTION;
-import static com.googlesource.gerrit.plugins.multisite.Configuration.KafkaSubscriber.KAFKA_SUBSCRIBER_SUBSECTION;
 import static com.googlesource.gerrit.plugins.multisite.Configuration.THREAD_POOL_SIZE_KEY;
-import static com.googlesource.gerrit.plugins.multisite.KafkaConfiguration.KAFKA_PROPERTY_PREFIX;
-import static com.googlesource.gerrit.plugins.multisite.KafkaConfiguration.KAFKA_SECTION;
 
 import com.google.common.collect.ImmutableList;
 import org.eclipse.jgit.lib.Config;
@@ -129,84 +124,6 @@
   }
 
   @Test
-  public void kafkaSubscriberPropertiesAreSetWhenSectionIsEnabled() {
-    final String kafkaPropertyName = KAFKA_PROPERTY_PREFIX + "fooBarBaz";
-    final String kafkaPropertyValue = "aValue";
-    globalPluginConfig.setBoolean(KAFKA_SECTION, KAFKA_SUBSCRIBER_SUBSECTION, ENABLE_KEY, true);
-    globalPluginConfig.setString(
-        KAFKA_SECTION, KAFKA_SUBSCRIBER_SUBSECTION, kafkaPropertyName, kafkaPropertyValue);
-
-    final String property = getConfiguration().kafkaSubscriber().getProperty("foo.bar.baz");
-
-    assertThat(property.equals(kafkaPropertyValue)).isTrue();
-  }
-
-  @Test
-  public void kafkaSubscriberPropertiesAreNotSetWhenSectionIsDisabled() {
-    final String kafkaPropertyName = KAFKA_PROPERTY_PREFIX + "fooBarBaz";
-    final String kafkaPropertyValue = "aValue";
-    globalPluginConfig.setBoolean(KAFKA_SECTION, KAFKA_SUBSCRIBER_SUBSECTION, ENABLE_KEY, false);
-    globalPluginConfig.setString(
-        KAFKA_SECTION, KAFKA_SUBSCRIBER_SUBSECTION, kafkaPropertyName, kafkaPropertyValue);
-
-    final String property = getConfiguration().kafkaSubscriber().getProperty("foo.bar.baz");
-
-    assertThat(property).isNull();
-  }
-
-  @Test
-  public void kafkaSubscriberPropertiesAreIgnoredWhenPrefixIsNotSet() {
-    final String kafkaPropertyName = "fooBarBaz";
-    final String kafkaPropertyValue = "aValue";
-    globalPluginConfig.setBoolean(KAFKA_SECTION, KAFKA_SUBSCRIBER_SUBSECTION, ENABLE_KEY, true);
-    globalPluginConfig.setString(
-        KAFKA_SECTION, KAFKA_SUBSCRIBER_SUBSECTION, kafkaPropertyName, kafkaPropertyValue);
-
-    final String property = getConfiguration().kafkaSubscriber().getProperty("foo.bar.baz");
-
-    assertThat(property).isNull();
-  }
-
-  @Test
-  public void kafkaPublisherPropertiesAreSetWhenSectionIsEnabled() {
-    final String kafkaPropertyName = KAFKA_PROPERTY_PREFIX + "fooBarBaz";
-    final String kafkaPropertyValue = "aValue";
-    globalPluginConfig.setBoolean(KAFKA_SECTION, KAFKA_PUBLISHER_SUBSECTION, ENABLE_KEY, true);
-    globalPluginConfig.setString(
-        KAFKA_SECTION, KAFKA_PUBLISHER_SUBSECTION, kafkaPropertyName, kafkaPropertyValue);
-
-    final String property = getConfiguration().kafkaPublisher().getProperty("foo.bar.baz");
-
-    assertThat(property.equals(kafkaPropertyValue)).isTrue();
-  }
-
-  @Test
-  public void kafkaPublisherPropertiesAreIgnoredWhenPrefixIsNotSet() {
-    final String kafkaPropertyName = "fooBarBaz";
-    final String kafkaPropertyValue = "aValue";
-    globalPluginConfig.setBoolean(KAFKA_SECTION, KAFKA_PUBLISHER_SUBSECTION, ENABLE_KEY, true);
-    globalPluginConfig.setString(
-        KAFKA_SECTION, KAFKA_PUBLISHER_SUBSECTION, kafkaPropertyName, kafkaPropertyValue);
-
-    final String property = getConfiguration().kafkaPublisher().getProperty("foo.bar.baz");
-
-    assertThat(property).isNull();
-  }
-
-  @Test
-  public void kafkaPublisherPropertiesAreNotSetWhenSectionIsDisabled() {
-    final String kafkaPropertyName = KAFKA_PROPERTY_PREFIX + "fooBarBaz";
-    final String kafkaPropertyValue = "aValue";
-    globalPluginConfig.setBoolean(KAFKA_SECTION, KAFKA_PUBLISHER_SUBSECTION, ENABLE_KEY, false);
-    globalPluginConfig.setString(
-        KAFKA_SECTION, KAFKA_PUBLISHER_SUBSECTION, kafkaPropertyName, kafkaPropertyValue);
-
-    final String property = getConfiguration().kafkaPublisher().getProperty("foo.bar.baz");
-
-    assertThat(property).isNull();
-  }
-
-  @Test
   public void shouldReturnValidationErrorsWhenReplicationOnStartupIsEnabled() throws Exception {
     Config replicationConfig = new Config();
     replicationConfig.setBoolean("gerrit", null, "replicateOnStartup", true);
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/KafkaConfigurationTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/KafkaConfigurationTest.java
new file mode 100644
index 0000000..ab0d3bc
--- /dev/null
+++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/KafkaConfigurationTest.java
@@ -0,0 +1,121 @@
+// Copyright (C) 2019 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 static com.googlesource.gerrit.plugins.multisite.Configuration.ENABLE_KEY;
+import static com.googlesource.gerrit.plugins.multisite.KafkaConfiguration.KAFKA_PROPERTY_PREFIX;
+import static com.googlesource.gerrit.plugins.multisite.KafkaConfiguration.KAFKA_SECTION;
+import static com.googlesource.gerrit.plugins.multisite.KafkaConfiguration.KafkaPublisher.KAFKA_PUBLISHER_SUBSECTION;
+import static com.googlesource.gerrit.plugins.multisite.KafkaConfiguration.KafkaSubscriber.KAFKA_SUBSCRIBER_SUBSECTION;
+
+import org.eclipse.jgit.lib.Config;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.junit.MockitoJUnitRunner;
+
+@RunWith(MockitoJUnitRunner.class)
+public class KafkaConfigurationTest {
+
+  private Config globalPluginConfig;
+
+  @Before
+  public void setUp() {
+    globalPluginConfig = new Config();
+  }
+
+  private KafkaConfiguration getConfiguration() {
+    return new KafkaConfiguration(globalPluginConfig);
+  }
+
+  @Test
+  public void kafkaSubscriberPropertiesAreSetWhenSectionIsEnabled() {
+    final String kafkaPropertyName = KAFKA_PROPERTY_PREFIX + "fooBarBaz";
+    final String kafkaPropertyValue = "aValue";
+    globalPluginConfig.setBoolean(KAFKA_SECTION, KAFKA_SUBSCRIBER_SUBSECTION, ENABLE_KEY, true);
+    globalPluginConfig.setString(
+        KAFKA_SECTION, KAFKA_SUBSCRIBER_SUBSECTION, kafkaPropertyName, kafkaPropertyValue);
+
+    final String property = getConfiguration().kafkaSubscriber().getProperty("foo.bar.baz");
+
+    assertThat(property.equals(kafkaPropertyValue)).isTrue();
+  }
+
+  @Test
+  public void kafkaSubscriberPropertiesAreNotSetWhenSectionIsDisabled() {
+    final String kafkaPropertyName = KAFKA_PROPERTY_PREFIX + "fooBarBaz";
+    final String kafkaPropertyValue = "aValue";
+    globalPluginConfig.setBoolean(KAFKA_SECTION, KAFKA_SUBSCRIBER_SUBSECTION, ENABLE_KEY, false);
+    globalPluginConfig.setString(
+        KAFKA_SECTION, KAFKA_SUBSCRIBER_SUBSECTION, kafkaPropertyName, kafkaPropertyValue);
+
+    final String property = getConfiguration().kafkaSubscriber().getProperty("foo.bar.baz");
+
+    assertThat(property).isNull();
+  }
+
+  @Test
+  public void kafkaSubscriberPropertiesAreIgnoredWhenPrefixIsNotSet() {
+    final String kafkaPropertyName = "fooBarBaz";
+    final String kafkaPropertyValue = "aValue";
+    globalPluginConfig.setBoolean(KAFKA_SECTION, KAFKA_SUBSCRIBER_SUBSECTION, ENABLE_KEY, true);
+    globalPluginConfig.setString(
+        KAFKA_SECTION, KAFKA_SUBSCRIBER_SUBSECTION, kafkaPropertyName, kafkaPropertyValue);
+
+    final String property = getConfiguration().kafkaSubscriber().getProperty("foo.bar.baz");
+
+    assertThat(property).isNull();
+  }
+
+  @Test
+  public void kafkaPublisherPropertiesAreSetWhenSectionIsEnabled() {
+    final String kafkaPropertyName = KAFKA_PROPERTY_PREFIX + "fooBarBaz";
+    final String kafkaPropertyValue = "aValue";
+    globalPluginConfig.setBoolean(KAFKA_SECTION, KAFKA_PUBLISHER_SUBSECTION, ENABLE_KEY, true);
+    globalPluginConfig.setString(
+        KAFKA_SECTION, KAFKA_PUBLISHER_SUBSECTION, kafkaPropertyName, kafkaPropertyValue);
+
+    final String property = getConfiguration().kafkaPublisher().getProperty("foo.bar.baz");
+
+    assertThat(property.equals(kafkaPropertyValue)).isTrue();
+  }
+
+  @Test
+  public void kafkaPublisherPropertiesAreIgnoredWhenPrefixIsNotSet() {
+    final String kafkaPropertyName = "fooBarBaz";
+    final String kafkaPropertyValue = "aValue";
+    globalPluginConfig.setBoolean(KAFKA_SECTION, KAFKA_PUBLISHER_SUBSECTION, ENABLE_KEY, true);
+    globalPluginConfig.setString(
+        KAFKA_SECTION, KAFKA_PUBLISHER_SUBSECTION, kafkaPropertyName, kafkaPropertyValue);
+
+    final String property = getConfiguration().kafkaPublisher().getProperty("foo.bar.baz");
+
+    assertThat(property).isNull();
+  }
+
+  @Test
+  public void kafkaPublisherPropertiesAreNotSetWhenSectionIsDisabled() {
+    final String kafkaPropertyName = KAFKA_PROPERTY_PREFIX + "fooBarBaz";
+    final String kafkaPropertyValue = "aValue";
+    globalPluginConfig.setBoolean(KAFKA_SECTION, KAFKA_PUBLISHER_SUBSECTION, ENABLE_KEY, false);
+    globalPluginConfig.setString(
+        KAFKA_SECTION, KAFKA_PUBLISHER_SUBSECTION, kafkaPropertyName, kafkaPropertyValue);
+
+    final String property = getConfiguration().kafkaPublisher().getProperty("foo.bar.baz");
+
+    assertThat(property).isNull();
+  }
+}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/ModuleTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/ModuleTest.java
index 5281371..556917b 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/multisite/ModuleTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/ModuleTest.java
@@ -36,13 +36,16 @@
   @Mock(answer = Answers.RETURNS_DEEP_STUBS)
   private Configuration configMock;
 
+  @Mock(answer = Answers.RETURNS_DEEP_STUBS)
+  private KafkaConfiguration kafkaConfig;
+
   @Rule public TemporaryFolder tempFolder = new TemporaryFolder();
 
   private Module module;
 
   @Before
   public void setUp() {
-    module = new Module(configMock);
+    module = new Module(configMock, kafkaConfig);
   }
 
   @Test
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/broker/kafka/BrokerPublisherTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/broker/kafka/BrokerPublisherTest.java
index 8131a3f..36408ba 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/multisite/broker/kafka/BrokerPublisherTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/broker/kafka/BrokerPublisherTest.java
@@ -15,6 +15,10 @@
 package com.googlesource.gerrit.plugins.multisite.broker.kafka;
 
 import static com.google.common.truth.Truth.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.only;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 
 import com.google.gerrit.extensions.client.ChangeKind;
 import com.google.gerrit.reviewdb.client.Account;
@@ -23,12 +27,14 @@
 import com.google.gerrit.server.data.AccountAttribute;
 import com.google.gerrit.server.data.ApprovalAttribute;
 import com.google.gerrit.server.events.CommentAddedEvent;
+import com.google.gerrit.server.events.Event;
 import com.google.gerrit.server.util.time.TimeUtil;
 import com.google.gson.Gson;
 import com.google.gson.JsonElement;
 import com.google.gson.JsonObject;
 import com.googlesource.gerrit.plugins.multisite.DisabledMessageLogger;
 import com.googlesource.gerrit.plugins.multisite.MessageLogger;
+import com.googlesource.gerrit.plugins.multisite.broker.BrokerMetrics;
 import com.googlesource.gerrit.plugins.multisite.broker.BrokerPublisher;
 import com.googlesource.gerrit.plugins.multisite.broker.BrokerSession;
 import com.googlesource.gerrit.plugins.multisite.broker.GsonProvider;
@@ -36,65 +42,44 @@
 import java.util.UUID;
 import org.junit.Before;
 import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
 
+@RunWith(MockitoJUnitRunner.class)
 public class BrokerPublisherTest {
+
+  @Mock private BrokerMetrics brokerMetrics;
+  @Mock private BrokerSession session;
   private BrokerPublisher publisher;
+
   private MessageLogger NO_MSG_LOG = new DisabledMessageLogger();
   private Gson gson = new GsonProvider().get();
 
+  private String accountName = "Foo Bar";
+  private String accountEmail = "foo@bar.com";
+  private String accountUsername = "foobar";
+  private String approvalType = ChangeKind.REWORK.toString();
+
+  private String approvalDescription = "ApprovalDescription";
+  private String approvalValue = "+2";
+  private String oldApprovalValue = "+1";
+  private Long approvalGrantedOn = 123L;
+  private String commentDescription = "Patch Set 1: Code-Review+2";
+  private String projectName = "project";
+  private String refName = "refs/heads/master";
+  private String changeId = "Iabcd1234abcd1234abcd1234abcd1234abcd1234";
+  private Long eventCreatedOn = 123L;
+
   @Before
   public void setUp() {
-    publisher = new BrokerPublisher(new TestBrokerSession(), gson, UUID.randomUUID(), NO_MSG_LOG);
+    publisher = new BrokerPublisher(session, gson, UUID.randomUUID(), NO_MSG_LOG, brokerMetrics);
   }
 
   @Test
   public void shouldSerializeCommentAddedEvent() {
 
-    final String accountName = "Foo Bar";
-    final String accountEmail = "foo@bar.com";
-    final String accountUsername = "foobar";
-    final String approvalType = ChangeKind.REWORK.toString();
-
-    final String approvalDescription = "ApprovalDescription";
-    final String approvalValue = "+2";
-    final String oldApprovalValue = "+1";
-    final Long approvalGrantedOn = 123L;
-    final String commentDescription = "Patch Set 1: Code-Review+2";
-    final String projectName = "project";
-    final String refName = "refs/heads/master";
-    final String changeId = "Iabcd1234abcd1234abcd1234abcd1234abcd1234";
-    final Long eventCreatedOn = 123L;
-
-    final Change change =
-        new Change(
-            new Change.Key(changeId),
-            new Change.Id(1),
-            new Account.Id(1),
-            new Branch.NameKey(projectName, refName),
-            TimeUtil.nowTs());
-
-    CommentAddedEvent event = new CommentAddedEvent(change);
-    AccountAttribute accountAttribute = new AccountAttribute();
-    accountAttribute.email = accountEmail;
-    accountAttribute.name = accountName;
-    accountAttribute.username = accountUsername;
-
-    event.eventCreatedOn = eventCreatedOn;
-    event.approvals =
-        () -> {
-          ApprovalAttribute approvalAttribute = new ApprovalAttribute();
-          approvalAttribute.value = approvalValue;
-          approvalAttribute.oldValue = oldApprovalValue;
-          approvalAttribute.description = approvalDescription;
-          approvalAttribute.by = accountAttribute;
-          approvalAttribute.type = ChangeKind.REWORK.toString();
-          approvalAttribute.grantedOn = approvalGrantedOn;
-
-          return new ApprovalAttribute[] {approvalAttribute};
-        };
-
-    event.author = () -> accountAttribute;
-    event.comment = commentDescription;
+    Event event = createSampleEvent();
 
     String expectedSerializedCommentEvent =
         "{\"author\": {\"name\": \""
@@ -140,22 +125,55 @@
     assertThat(publisher.eventToJson(event).equals(expectedCommentEventJsonObject)).isTrue();
   }
 
-  private static class TestBrokerSession implements BrokerSession {
+  @Test
+  public void shouldIncrementBrokerMetricCounterWhenMessagePublished() {
+    Event event = createSampleEvent();
+    when(session.publishEvent(any(), any())).thenReturn(true);
+    publisher.publishEvent(EventFamily.INDEX_EVENT, event);
+    verify(brokerMetrics, only()).incrementBrokerPublishedMessage();
+  }
 
-    @Override
-    public boolean isOpen() {
-      return false;
-    }
+  @Test
+  public void shouldIncrementBrokerFailedMetricCounterWhenMessagePublished() {
+    Event event = createSampleEvent();
+    when(session.publishEvent(any(), any())).thenReturn(false);
 
-    @Override
-    public void connect() {}
+    publisher.publishEvent(EventFamily.INDEX_EVENT, event);
+    verify(brokerMetrics, only()).incrementBrokerFailedToPublishMessage();
+  }
 
-    @Override
-    public void disconnect() {}
+  private Event createSampleEvent() {
+    final Change change =
+        new Change(
+            new Change.Key(changeId),
+            new Change.Id(1),
+            new Account.Id(1),
+            new Branch.NameKey(projectName, refName),
+            TimeUtil.nowTs());
 
-    @Override
-    public boolean publishEvent(EventFamily eventFamily, String payload) {
-      return false;
-    }
+    CommentAddedEvent event = new CommentAddedEvent(change);
+    AccountAttribute accountAttribute = new AccountAttribute();
+    accountAttribute.email = accountEmail;
+    accountAttribute.name = accountName;
+    accountAttribute.username = accountUsername;
+
+    event.eventCreatedOn = eventCreatedOn;
+    event.approvals =
+        () -> {
+          ApprovalAttribute approvalAttribute = new ApprovalAttribute();
+          approvalAttribute.value = approvalValue;
+          approvalAttribute.oldValue = oldApprovalValue;
+          approvalAttribute.description = approvalDescription;
+          approvalAttribute.by = accountAttribute;
+          approvalAttribute.type = ChangeKind.REWORK.toString();
+          approvalAttribute.grantedOn = approvalGrantedOn;
+
+          return new ApprovalAttribute[] {approvalAttribute};
+        };
+
+    event.author = () -> accountAttribute;
+    event.comment = commentDescription;
+
+    return event;
   }
 }
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/kafka/consumer/EventConsumerIT.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/kafka/consumer/EventConsumerIT.java
index 6f7aa74..a65a455 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/multisite/kafka/consumer/EventConsumerIT.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/kafka/consumer/EventConsumerIT.java
@@ -38,6 +38,7 @@
 import com.google.inject.Key;
 import com.google.inject.TypeLiteral;
 import com.googlesource.gerrit.plugins.multisite.Configuration;
+import com.googlesource.gerrit.plugins.multisite.KafkaConfiguration;
 import com.googlesource.gerrit.plugins.multisite.Module;
 import com.googlesource.gerrit.plugins.multisite.broker.BrokerGson;
 import com.googlesource.gerrit.plugins.multisite.forwarder.events.ChangeIndexEvent;
@@ -96,7 +97,8 @@
       this.config =
           new FileBasedConfig(
               sitePaths.etc_dir.resolve(Configuration.MULTI_SITE_CONFIG).toFile(), FS.DETECTED);
-      this.multiSiteModule = new Module(new Configuration(config, new Config()));
+      this.multiSiteModule =
+          new Module(new Configuration(config, new Config()), new KafkaConfiguration(config), true);
     }
 
     @Override