Merge of ExtendedBrokerApi into BrokerApi
Remove all references to the legacy ExtendedBrokerApi and
use the new updated BrokerApi, including the implementation
of the disconnect by topic/groupId which allows to isolate
the disconnection of a plugin's subscribers without impacting
all other subscribers.
Bug:Issue 327226782
Change-Id: Id3ca8efb9e5432b53b954bead0ff5dea82f3d041
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);
