diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/Manager.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/Manager.java
index 9ed5165..7123d7f 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/kafka/Manager.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/Manager.java
@@ -15,7 +15,6 @@
 package com.googlesource.gerrit.plugins.kafka;
 
 import com.gerritforge.gerrit.eventbroker.BrokerApi;
-import com.gerritforge.gerrit.eventbroker.ExtendedBrokerApi;
 import com.gerritforge.gerrit.eventbroker.TopicSubscriber;
 import com.gerritforge.gerrit.eventbroker.TopicSubscriberWithGroupId;
 import com.google.gerrit.extensions.events.LifecycleListener;
@@ -30,7 +29,7 @@
   private final KafkaPublisher publisher;
   private final Set<TopicSubscriber> consumers;
   private final Set<TopicSubscriberWithGroupId> consumersWithGroupId;
-  private final ExtendedBrokerApi brokerApi;
+  private final BrokerApi brokerApi;
 
   @Inject
   public Manager(
@@ -40,7 +39,7 @@
       BrokerApi brokerApi) {
     this.publisher = publisher;
     this.consumers = consumers;
-    this.brokerApi = (ExtendedBrokerApi) brokerApi;
+    this.brokerApi = brokerApi;
     this.consumersWithGroupId = consumersWithGroupId;
   }
 
diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/api/KafkaApiModule.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/api/KafkaApiModule.java
index d3ee3f1..e566204 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/kafka/api/KafkaApiModule.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/api/KafkaApiModule.java
@@ -15,7 +15,6 @@
 package com.googlesource.gerrit.plugins.kafka.api;
 
 import com.gerritforge.gerrit.eventbroker.BrokerApi;
-import com.gerritforge.gerrit.eventbroker.ExtendedBrokerApi;
 import com.gerritforge.gerrit.eventbroker.TopicSubscriber;
 import com.gerritforge.gerrit.eventbroker.TopicSubscriberWithGroupId;
 import com.google.common.collect.Sets;
@@ -57,9 +56,7 @@
   public void setPreviousBrokerApi(DynamicItem<BrokerApi> previousBrokerApi) {
     if (previousBrokerApi != null && previousBrokerApi.get() != null) {
       BrokerApi api = previousBrokerApi.get();
-      if (api instanceof ExtendedBrokerApi) {
-        this.activeConsumersWithGroupId = ((ExtendedBrokerApi) api).topicSubscribersWithGroupId();
-      }
+      this.activeConsumersWithGroupId = api.topicSubscribersWithGroupId();
       this.activeConsumers = api.topicSubscribers();
     }
   }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerApi.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerApi.java
index d1d6961..06f1ac6 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerApi.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerApi.java
@@ -14,22 +14,24 @@
 
 package com.googlesource.gerrit.plugins.kafka.api;
 
-import com.gerritforge.gerrit.eventbroker.ExtendedBrokerApi;
+import com.gerritforge.gerrit.eventbroker.BrokerApi;
 import com.gerritforge.gerrit.eventbroker.TopicSubscriber;
 import com.gerritforge.gerrit.eventbroker.TopicSubscriberWithGroupId;
 import com.google.common.util.concurrent.ListenableFuture;
+import com.google.gerrit.common.Nullable;
 import com.google.gerrit.server.events.Event;
 import com.google.inject.Inject;
 import com.googlesource.gerrit.plugins.kafka.publish.KafkaPublisher;
 import com.googlesource.gerrit.plugins.kafka.subscribe.KafkaEventSubscriber;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
 import java.util.Set;
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
 
-public class KafkaBrokerApi implements ExtendedBrokerApi {
+public class KafkaBrokerApi implements BrokerApi {
 
   private final KafkaPublisher publisher;
   private final KafkaEventSubscriber.Factory kafkaEventSubscriberFactory;
@@ -40,7 +42,7 @@
       KafkaPublisher publisher, KafkaEventSubscriber.Factory kafkaEventSubscriberFactory) {
     this.publisher = publisher;
     this.kafkaEventSubscriberFactory = kafkaEventSubscriberFactory;
-    subscribers = new ArrayList<>();
+    subscribers = Collections.synchronizedList(new ArrayList<>());
   }
 
   @Override
@@ -67,6 +69,18 @@
   }
 
   @Override
+  public void disconnect(String topic, @Nullable String groupId) {
+    Set<KafkaEventSubscriber> subscribersToDisconnect =
+        subscribers.stream()
+            .filter(s -> topic.equals(s.getTopic()))
+            .filter(
+                s -> groupId == null || s.getExternalGroupId().stream().anyMatch(groupId::equals))
+            .collect(Collectors.toSet());
+    subscribersToDisconnect.forEach(KafkaEventSubscriber::shutdown);
+    subscribers.removeAll(subscribersToDisconnect);
+  }
+
+  @Override
   public Set<TopicSubscriber> topicSubscribers() {
     return subscribers.stream()
         .filter(s -> !s.getExternalGroupId().isPresent())
diff --git a/src/test/java/com/googlesource/gerrit/plugins/kafka/EventConsumerIT.java b/src/test/java/com/googlesource/gerrit/plugins/kafka/EventConsumerIT.java
index 028d589..d9e82e3 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/kafka/EventConsumerIT.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/kafka/EventConsumerIT.java
@@ -19,7 +19,6 @@
 import static org.junit.Assert.fail;
 
 import com.gerritforge.gerrit.eventbroker.BrokerApi;
-import com.gerritforge.gerrit.eventbroker.ExtendedBrokerApi;
 import com.google.common.base.Stopwatch;
 import com.google.common.collect.Iterables;
 import com.google.gerrit.acceptance.LightweightPluginDaemonTest;
@@ -143,7 +142,7 @@
     eventMessage.instanceId = "test-instance-id-1";
     List<Event> receivedEventsWithGroupId1 = new ArrayList<>();
 
-    ExtendedBrokerApi kafkaBrokerApi = ((ExtendedBrokerApi) kafkaBrokerApi());
+    BrokerApi kafkaBrokerApi = kafkaBrokerApi();
     kafkaBrokerApi.send(topic, eventMessage);
     kafkaBrokerApi.receiveAsync(topic, consumerGroup1, receivedEventsWithGroupId1::add);