Move kafka related code from multi-site plugin to kafka-events plugin
Kafka broker implementation is provided as a plugin.
Multi-site plugin is message broker implementation agnostic.
This separation allows replacing message broker implementation.
Feature: Issue 11599
Change-Id: Id35f0a8fe6e1d43cd893e7b9521c2c67f0160ed9
diff --git a/BUILD b/BUILD
index db2462f..6b11272 100644
--- a/BUILD
+++ b/BUILD
@@ -17,7 +17,8 @@
],
resources = glob(["src/main/resources/**/*"]),
deps = [
- "@kafka_client//jar",
+ "@kafka-client//jar",
+ "@events-broker//jar",
],
)
@@ -28,7 +29,8 @@
deps = [
":kafka-events__plugin_test_deps",
"//lib/testcontainers",
- "@kafka_client//jar",
+ "@kafka-client//jar",
+ "@events-broker//jar",
"@testcontainers-kafka//jar",
],
)
diff --git a/external_plugin_deps.bzl b/external_plugin_deps.bzl
index bebcc4e..bfbf487 100644
--- a/external_plugin_deps.bzl
+++ b/external_plugin_deps.bzl
@@ -2,7 +2,7 @@
def external_plugin_deps():
maven_jar(
- name = "kafka_client",
+ name = "kafka-client",
artifact = "org.apache.kafka:kafka-clients:2.1.0",
sha1 = "34d9983705c953b97abb01e1cd04647f47272fe5",
)
@@ -12,3 +12,9 @@
artifact = "org.testcontainers:kafka:1.10.6",
sha1 = "5984e31306bd6c84a36092cdd19e0ef7e2268d98",
)
+
+ maven_jar(
+ name = "events-broker",
+ artifact = "com.gerritforge:events-broker:3.0.3",
+ sha1 = "efdc5bf6897563e2f6f85bfc1b8a5d65e3393424",
+ )
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 b20b312..4dc394b 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/kafka/Manager.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/Manager.java
@@ -14,28 +14,39 @@
package com.googlesource.gerrit.plugins.kafka;
+import com.gerritforge.gerrit.eventbroker.BrokerApi;
+import com.gerritforge.gerrit.eventbroker.TopicSubscriber;
import com.google.gerrit.extensions.events.LifecycleListener;
import com.google.inject.Inject;
import com.google.inject.Singleton;
import com.googlesource.gerrit.plugins.kafka.publish.KafkaPublisher;
+import java.util.Set;
@Singleton
public class Manager implements LifecycleListener {
private final KafkaPublisher publisher;
+ private final Set<TopicSubscriber> consumers;
+ private final BrokerApi brokerApi;
@Inject
- public Manager(KafkaPublisher publisher) {
+ public Manager(KafkaPublisher publisher, Set<TopicSubscriber> consumers, BrokerApi brokerApi) {
this.publisher = publisher;
+ this.consumers = consumers;
+ this.brokerApi = brokerApi;
}
@Override
public void start() {
publisher.start();
+ consumers.forEach(
+ topicSubscriber ->
+ brokerApi.receiveAsync(topicSubscriber.topic(), topicSubscriber.consumer()));
}
@Override
public void stop() {
publisher.stop();
+ brokerApi.disconnect();
}
}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/Module.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/Module.java
index 7f219d3..c9a28ba 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/kafka/Module.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/Module.java
@@ -14,21 +14,32 @@
package com.googlesource.gerrit.plugins.kafka;
+import com.gerritforge.gerrit.eventbroker.EventGsonProvider;
import com.google.gerrit.extensions.events.LifecycleListener;
import com.google.gerrit.extensions.registration.DynamicSet;
import com.google.gerrit.server.events.EventListener;
import com.google.gson.Gson;
import com.google.inject.AbstractModule;
+import com.google.inject.Inject;
import com.google.inject.Singleton;
-import com.googlesource.gerrit.plugins.kafka.publish.GsonProvider;
+import com.googlesource.gerrit.plugins.kafka.api.KafkaApiModule;
import com.googlesource.gerrit.plugins.kafka.publish.KafkaPublisher;
class Module extends AbstractModule {
+ private final KafkaApiModule kafkaBrokerModule;
+
+ @Inject
+ public Module(KafkaApiModule kafkaBrokerModule) {
+ this.kafkaBrokerModule = kafkaBrokerModule;
+ }
+
@Override
protected void configure() {
- bind(Gson.class).toProvider(GsonProvider.class).in(Singleton.class);
+ bind(Gson.class).toProvider(EventGsonProvider.class).in(Singleton.class);
DynamicSet.bind(binder(), LifecycleListener.class).to(Manager.class);
DynamicSet.bind(binder(), EventListener.class).to(KafkaPublisher.class);
+
+ install(kafkaBrokerModule);
}
}
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
new file mode 100644
index 0000000..73c7509
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/api/KafkaApiModule.java
@@ -0,0 +1,69 @@
+// 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.kafka.api;
+
+import com.gerritforge.gerrit.eventbroker.BrokerApi;
+import com.gerritforge.gerrit.eventbroker.EventMessage;
+import com.gerritforge.gerrit.eventbroker.TopicSubscriber;
+import com.google.common.collect.Sets;
+import com.google.gerrit.extensions.registration.DynamicItem;
+import com.google.gerrit.lifecycle.LifecycleModule;
+import com.google.gerrit.server.git.WorkQueue;
+import com.google.inject.Inject;
+import com.google.inject.Scopes;
+import com.google.inject.Singleton;
+import com.google.inject.TypeLiteral;
+import com.googlesource.gerrit.plugins.kafka.broker.ConsumerExecutor;
+import com.googlesource.gerrit.plugins.kafka.config.KafkaSubscriberProperties;
+import com.googlesource.gerrit.plugins.kafka.subscribe.KafkaEventDeserializer;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+import org.apache.kafka.common.serialization.Deserializer;
+
+@Singleton
+public class KafkaApiModule extends LifecycleModule {
+ private Set<TopicSubscriber> activeConsumers = Sets.newHashSet();
+ private WorkQueue workQueue;
+ private KafkaSubscriberProperties configuration;
+
+ @Inject
+ public KafkaApiModule(WorkQueue workQueue, KafkaSubscriberProperties configuration) {
+ this.workQueue = workQueue;
+ this.configuration = configuration;
+ }
+
+ @Inject(optional = true)
+ public void setPreviousBrokerApi(DynamicItem<BrokerApi> previousBrokerApi) {
+ if (previousBrokerApi != null && previousBrokerApi.get() != null) {
+ this.activeConsumers = previousBrokerApi.get().topicSubscribers();
+ }
+ }
+
+ @Override
+ protected void configure() {
+
+ bind(ExecutorService.class)
+ .annotatedWith(ConsumerExecutor.class)
+ .toInstance(
+ workQueue.createQueue(configuration.getNumberOfSubscribers(), "kafka-subscriber"));
+
+ bind(new TypeLiteral<Deserializer<byte[]>>() {}).toInstance(new ByteArrayDeserializer());
+ bind(new TypeLiteral<Deserializer<EventMessage>>() {}).to(KafkaEventDeserializer.class);
+ bind(new TypeLiteral<Set<TopicSubscriber>>() {}).toInstance(activeConsumers);
+
+ DynamicItem.bind(binder(), BrokerApi.class).to(KafkaBrokerApi.class).in(Scopes.SINGLETON);
+ }
+}
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
new file mode 100644
index 0000000..78df908
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerApi.java
@@ -0,0 +1,72 @@
+// 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.kafka.api;
+
+import com.gerritforge.gerrit.eventbroker.BrokerApi;
+import com.gerritforge.gerrit.eventbroker.EventMessage;
+import com.gerritforge.gerrit.eventbroker.TopicSubscriber;
+import com.google.inject.Inject;
+import com.google.inject.Provider;
+import com.googlesource.gerrit.plugins.kafka.publish.KafkaPublisher;
+import com.googlesource.gerrit.plugins.kafka.subscribe.KafkaEventSubscriber;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.function.Consumer;
+import java.util.stream.Collectors;
+
+public class KafkaBrokerApi implements BrokerApi {
+
+ private final KafkaPublisher publisher;
+ private final Provider<KafkaEventSubscriber> subscriberProvider;
+ private List<KafkaEventSubscriber> subscribers;
+
+ @Inject
+ public KafkaBrokerApi(
+ KafkaPublisher publisher, Provider<KafkaEventSubscriber> subscriberProvider) {
+ this.publisher = publisher;
+ this.subscriberProvider = subscriberProvider;
+ subscribers = new ArrayList<>();
+ }
+
+ @Override
+ public boolean send(String topic, EventMessage event) {
+ return publisher.publish(topic, event);
+ }
+
+ @Override
+ public void receiveAsync(String topic, Consumer<EventMessage> eventConsumer) {
+ KafkaEventSubscriber subscriber = subscriberProvider.get();
+ synchronized (subscribers) {
+ subscribers.add(subscriber);
+ }
+ subscriber.subscribe(topic, eventConsumer);
+ }
+
+ @Override
+ public void disconnect() {
+ for (KafkaEventSubscriber subscriber : subscribers) {
+ subscriber.shutdown();
+ }
+ subscribers.clear();
+ }
+
+ @Override
+ public Set<TopicSubscriber> topicSubscribers() {
+ return subscribers.stream()
+ .map(s -> TopicSubscriber.topicSubscriber(s.getTopic(), s.getMessageProcessor()))
+ .collect(Collectors.toSet());
+ }
+}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/broker/ConsumerExecutor.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/broker/ConsumerExecutor.java
new file mode 100644
index 0000000..eb878fa
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/broker/ConsumerExecutor.java
@@ -0,0 +1,24 @@
+// 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.kafka.broker;
+
+import static java.lang.annotation.RetentionPolicy.RUNTIME;
+
+import com.google.inject.BindingAnnotation;
+import java.lang.annotation.Retention;
+
+@Retention(RUNTIME)
+@BindingAnnotation
+public @interface ConsumerExecutor {}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/config/KafkaProperties.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/config/KafkaProperties.java
index c01e34e..7bb1f4f 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/kafka/config/KafkaProperties.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/config/KafkaProperties.java
@@ -79,4 +79,8 @@
public String getTopic() {
return topic;
}
+
+ public String getBootstrapServers() {
+ return getProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG);
+ }
}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/config/KafkaPublisherProperties.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/config/KafkaPublisherProperties.java
new file mode 100644
index 0000000..17f3ebd
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/config/KafkaPublisherProperties.java
@@ -0,0 +1,29 @@
+// 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.kafka.config;
+
+import com.google.gerrit.extensions.annotations.PluginName;
+import com.google.gerrit.server.config.PluginConfigFactory;
+import com.google.inject.Inject;
+
+public class KafkaPublisherProperties extends KafkaProperties {
+ private static final long serialVersionUID = 0L;
+
+ @Inject
+ public KafkaPublisherProperties(
+ PluginConfigFactory configFactory, @PluginName String pluginName) {
+ super(configFactory, pluginName);
+ }
+}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/config/KafkaSubscriberProperties.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/config/KafkaSubscriberProperties.java
new file mode 100644
index 0000000..b50053f
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/config/KafkaSubscriberProperties.java
@@ -0,0 +1,53 @@
+// 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.kafka.config;
+
+import com.google.gerrit.extensions.annotations.PluginName;
+import com.google.gerrit.server.config.PluginConfigFactory;
+import com.google.inject.Inject;
+
+public class KafkaSubscriberProperties extends KafkaProperties {
+ private static final long serialVersionUID = 1L;
+ private static final String DEFAULT_POLLING_INTERVAL_MS = "1000";
+ private static final String DEFAULT_NUMBER_OF_SUBSCRIBERS = "4";
+
+ private final Integer pollingInterval;
+ private final String groupId;
+ private final Integer numberOfSubscribers;
+
+ @Inject
+ public KafkaSubscriberProperties(
+ PluginConfigFactory configFactory, @PluginName String pluginName) {
+ super(configFactory, pluginName);
+
+ this.pollingInterval =
+ Integer.parseInt(getProperty("pollingIntervalMs", DEFAULT_POLLING_INTERVAL_MS));
+ this.groupId = getProperty("groupId");
+ this.numberOfSubscribers =
+ Integer.parseInt(getProperty("numberOfSubscribers", DEFAULT_NUMBER_OF_SUBSCRIBERS));
+ }
+
+ public Integer getPollingInterval() {
+ return pollingInterval;
+ }
+
+ public String getGroupId() {
+ return groupId;
+ }
+
+ public Integer getNumberOfSubscribers() {
+ return numberOfSubscribers;
+ }
+}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/publish/GsonProvider.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/publish/GsonProvider.java
deleted file mode 100644
index 2c5c1e7..0000000
--- a/src/main/java/com/googlesource/gerrit/plugins/kafka/publish/GsonProvider.java
+++ /dev/null
@@ -1,29 +0,0 @@
-// Copyright (C) 2016 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.kafka.publish;
-
-import com.google.common.base.Supplier;
-import com.google.gerrit.server.events.SupplierSerializer;
-import com.google.gson.Gson;
-import com.google.gson.GsonBuilder;
-import com.google.inject.Provider;
-
-public class GsonProvider implements Provider<Gson> {
-
- @Override
- public Gson get() {
- return new GsonBuilder().registerTypeAdapter(Supplier.class, new SupplierSerializer()).create();
- }
-}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/publish/KafkaPublisher.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/publish/KafkaPublisher.java
index 21b89ce..cc271b5 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/kafka/publish/KafkaPublisher.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/publish/KafkaPublisher.java
@@ -14,9 +14,12 @@
package com.googlesource.gerrit.plugins.kafka.publish;
+import com.gerritforge.gerrit.eventbroker.EventMessage;
+import com.google.common.annotations.VisibleForTesting;
import com.google.gerrit.server.events.Event;
import com.google.gerrit.server.events.EventListener;
import com.google.gson.Gson;
+import com.google.gson.JsonObject;
import com.google.inject.Inject;
import com.google.inject.Singleton;
import com.googlesource.gerrit.plugins.kafka.session.KafkaSession;
@@ -49,4 +52,17 @@
session.publish(gson.toJson(event));
}
}
+
+ public boolean publish(String topic, EventMessage event) {
+ return session.publish(topic, getPayload(event));
+ }
+
+ private String getPayload(EventMessage event) {
+ return gson.toJson(event);
+ }
+
+ @VisibleForTesting
+ public JsonObject eventToJson(Event event) {
+ return gson.toJsonTree(event).getAsJsonObject();
+ }
}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/session/KafkaSession.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/session/KafkaSession.java
index fcb5263..52b242a 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/kafka/session/KafkaSession.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/session/KafkaSession.java
@@ -16,9 +16,12 @@
import com.google.inject.Inject;
import com.googlesource.gerrit.plugins.kafka.config.KafkaProperties;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -70,6 +73,19 @@
}
public void publish(String messageBody) {
- producer.send(new ProducerRecord<>(properties.getTopic(), "" + System.nanoTime(), messageBody));
+ publish(properties.getTopic(), messageBody);
+ }
+
+ public boolean publish(String topic, String messageBody) {
+ Future<RecordMetadata> future =
+ producer.send(new ProducerRecord<>(topic, "" + System.nanoTime(), messageBody));
+ try {
+ RecordMetadata metadata = future.get();
+ LOGGER.debug("The offset of the record we just sent is: {}", metadata.offset());
+ return true;
+ } catch (InterruptedException | ExecutionException e) {
+ LOGGER.error("Cannot send the message", e);
+ return false;
+ }
}
}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaConsumerFactory.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaConsumerFactory.java
new file mode 100644
index 0000000..9ec109d
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaConsumerFactory.java
@@ -0,0 +1,37 @@
+// 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.kafka.subscribe;
+
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+import com.googlesource.gerrit.plugins.kafka.config.KafkaSubscriberProperties;
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+import org.apache.kafka.common.serialization.Deserializer;
+
+@Singleton
+class KafkaConsumerFactory {
+ private KafkaSubscriberProperties config;
+
+ @Inject
+ public KafkaConsumerFactory(KafkaSubscriberProperties configuration) {
+ this.config = configuration;
+ }
+
+ public Consumer<byte[], byte[]> create(Deserializer<byte[]> keyDeserializer) {
+ return new KafkaConsumer<>(config, keyDeserializer, new ByteArrayDeserializer());
+ }
+}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventDeserializer.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventDeserializer.java
new file mode 100644
index 0000000..bab2ad0
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventDeserializer.java
@@ -0,0 +1,54 @@
+// 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.kafka.subscribe;
+
+import com.gerritforge.gerrit.eventbroker.EventMessage;
+import com.google.gson.Gson;
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+import java.util.Map;
+import org.apache.kafka.common.serialization.Deserializer;
+import org.apache.kafka.common.serialization.StringDeserializer;
+
+@Singleton
+public class KafkaEventDeserializer implements Deserializer<EventMessage> {
+
+ private final StringDeserializer stringDeserializer = new StringDeserializer();
+ private Gson gson;
+
+ // To be used when providing this deserializer with class name (then need to add a configuration
+ // entry to set the gson.provider
+ public KafkaEventDeserializer() {}
+
+ @Inject
+ public KafkaEventDeserializer(Gson gson) {
+ this.gson = gson;
+ }
+
+ @Override
+ public void configure(Map<String, ?> configs, boolean isKey) {}
+
+ @Override
+ public EventMessage deserialize(String topic, byte[] data) {
+ final EventMessage result =
+ gson.fromJson(stringDeserializer.deserialize(topic, data), EventMessage.class);
+ result.validate();
+
+ return result;
+ }
+
+ @Override
+ public void close() {}
+}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventSubscriber.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventSubscriber.java
new file mode 100644
index 0000000..6548f2d
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventSubscriber.java
@@ -0,0 +1,134 @@
+// 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.kafka.subscribe;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+
+import com.gerritforge.gerrit.eventbroker.EventMessage;
+import com.google.common.flogger.FluentLogger;
+import com.google.gerrit.server.util.ManualRequestContext;
+import com.google.gerrit.server.util.OneOffRequestContext;
+import com.google.inject.Inject;
+import com.googlesource.gerrit.plugins.kafka.broker.ConsumerExecutor;
+import com.googlesource.gerrit.plugins.kafka.config.KafkaSubscriberProperties;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.common.errors.WakeupException;
+import org.apache.kafka.common.serialization.Deserializer;
+
+public class KafkaEventSubscriber {
+ private static final FluentLogger logger = FluentLogger.forEnclosingClass();
+
+ private final Consumer<byte[], byte[]> consumer;
+ private final OneOffRequestContext oneOffCtx;
+ private final AtomicBoolean closed = new AtomicBoolean(false);
+
+ private final Deserializer<EventMessage> valueDeserializer;
+ private final KafkaSubscriberProperties configuration;
+ private final ExecutorService executor;
+ private final KafkaEventSubscriberMetrics subscriberMetrics;
+
+ private java.util.function.Consumer<EventMessage> messageProcessor;
+
+ private String topic;
+
+ @Inject
+ public KafkaEventSubscriber(
+ KafkaSubscriberProperties configuration,
+ KafkaConsumerFactory consumerFactory,
+ Deserializer<byte[]> keyDeserializer,
+ Deserializer<EventMessage> valueDeserializer,
+ OneOffRequestContext oneOffCtx,
+ @ConsumerExecutor ExecutorService executor,
+ KafkaEventSubscriberMetrics subscriberMetrics) {
+
+ this.configuration = configuration;
+ this.oneOffCtx = oneOffCtx;
+ this.executor = executor;
+ this.subscriberMetrics = subscriberMetrics;
+
+ final ClassLoader previousClassLoader = Thread.currentThread().getContextClassLoader();
+ try {
+ Thread.currentThread().setContextClassLoader(KafkaEventSubscriber.class.getClassLoader());
+ this.consumer = consumerFactory.create(keyDeserializer);
+ } finally {
+ Thread.currentThread().setContextClassLoader(previousClassLoader);
+ }
+ this.valueDeserializer = valueDeserializer;
+ }
+
+ public void subscribe(String topic, java.util.function.Consumer<EventMessage> messageProcessor) {
+ this.topic = topic;
+ this.messageProcessor = messageProcessor;
+ logger.atInfo().log(
+ "Kafka consumer subscribing to topic alias [%s] for event topic [%s]", topic, topic);
+ consumer.subscribe(Collections.singleton(topic));
+ executor.execute(new ReceiverJob());
+ }
+
+ public void shutdown() {
+ closed.set(true);
+ consumer.wakeup();
+ }
+
+ public java.util.function.Consumer<EventMessage> getMessageProcessor() {
+ return messageProcessor;
+ }
+
+ public String getTopic() {
+ return topic;
+ }
+
+ private class ReceiverJob implements Runnable {
+
+ @Override
+ public void run() {
+ consume();
+ }
+
+ private void consume() {
+ try {
+ while (!closed.get()) {
+ ConsumerRecords<byte[], byte[]> consumerRecords =
+ consumer.poll(Duration.ofMillis(configuration.getPollingInterval()));
+ consumerRecords.forEach(
+ consumerRecord -> {
+ try (ManualRequestContext ctx = oneOffCtx.open()) {
+ EventMessage event =
+ valueDeserializer.deserialize(consumerRecord.topic(), consumerRecord.value());
+ messageProcessor.accept(event);
+ } catch (Exception e) {
+ logger.atSevere().withCause(e).log(
+ "Malformed event '%s': [Exception: %s]",
+ new String(consumerRecord.value(), UTF_8));
+ subscriberMetrics.incrementSubscriberFailedToConsumeMessage();
+ }
+ });
+ }
+ } catch (WakeupException e) {
+ // Ignore exception if closing
+ if (!closed.get()) throw e;
+ } catch (Exception e) {
+ subscriberMetrics.incrementSubscriberFailedToPollMessages();
+ throw e;
+ } finally {
+ consumer.close();
+ }
+ }
+ }
+}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventSubscriberMetrics.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventSubscriberMetrics.java
new file mode 100644
index 0000000..81174f9
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventSubscriberMetrics.java
@@ -0,0 +1,48 @@
+package com.googlesource.gerrit.plugins.kafka.subscribe;
+
+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
+class KafkaEventSubscriberMetrics {
+
+ private static final String SUBSCRIBER_POLL_FAILURE_COUNTER =
+ "subscriber_msg_consumer_poll_failure_counter";
+ private static final String SUBSCRIBER_FAILURE_COUNTER =
+ "subscriber_msg_consumer_failure_counter";
+
+ private final Counter1<String> subscriberPollFailureCounter;
+ private final Counter1<String> subscriberFailureCounter;
+
+ @Inject
+ public KafkaEventSubscriberMetrics(MetricMaker metricMaker) {
+ this.subscriberPollFailureCounter =
+ metricMaker.newCounter(
+ "kafka/subscriber/subscriber_message_consumer_poll_failure_counter",
+ new Description("Number of failed attempts to poll messages by the subscriber")
+ .setRate()
+ .setUnit("errors"),
+ Field.ofString(
+ SUBSCRIBER_POLL_FAILURE_COUNTER, "Subscriber failed to poll messages count"));
+ this.subscriberFailureCounter =
+ metricMaker.newCounter(
+ "kafka/subscriber/subscriber_message_consumer_failure_counter",
+ new Description("Number of messages failed to consume by the subscriber consumer")
+ .setRate()
+ .setUnit("errors"),
+ Field.ofString(
+ SUBSCRIBER_FAILURE_COUNTER, "Subscriber failed to consume messages count"));
+ }
+
+ public void incrementSubscriberFailedToPollMessages() {
+ subscriberPollFailureCounter.increment(SUBSCRIBER_POLL_FAILURE_COUNTER);
+ }
+
+ public void incrementSubscriberFailedToConsumeMessage() {
+ subscriberFailureCounter.increment(SUBSCRIBER_FAILURE_COUNTER);
+ }
+}
diff --git a/src/main/resources/Documentation/about.md b/src/main/resources/Documentation/about.md
index 685c195..c6dc18e 100644
--- a/src/main/resources/Documentation/about.md
+++ b/src/main/resources/Documentation/about.md
@@ -1 +1,27 @@
-This plugin can publish gerrit stream events to an Apache Kafka topic.
+This plugin publishes gerrit stream events to an Apache Kafka topic.
+
+It also provides a Kafka-based implementation of a generic
+[Events Broker Api](https://github.com/GerritForge/events-broker) which can be used by
+Gerrit and other plugins.
+
+Use-cases
+=========
+
+CI/CD Validation
+----------------
+
+Gerrit stream events can be published to the internal network where other subscribers
+can trigger automated jobs (e.g. CI/CD validation) for fetching the changes and validating
+them through build and testing.
+
+__NOTE__: This use-case would require a CI/CD system (e.g. Jenkins, Zuul or other) and
+the development of a Kafka-based subscriber to receive the event and trigger the build.
+
+Events replication
+------------------
+
+Multiple Gerrit masters in a multi-site setup can be informed on the stream events
+happening on every node thanks to the notification to a Kafka pub/sub topic.
+
+__NOTE__: This use-case would require the [multi-site plugin](https://gerrit.googlesource.com/plugins/multi-site)
+on each of the Gerrit masters that are part of the same multi-site cluster.
\ No newline at end of file
diff --git a/src/main/resources/Documentation/config.md b/src/main/resources/Documentation/config.md
index 96e4b94..faa5bf1 100644
--- a/src/main/resources/Documentation/config.md
+++ b/src/main/resources/Documentation/config.md
@@ -31,4 +31,15 @@
| lingerMs | 1
| bufferMemory | 33554432
| keySerializer | org.apache.kafka.common.serialization.StringSerializer
-| valueSerializer | org.apache.kafka.common.serialization.StringSerializer
\ No newline at end of file
+| valueSerializer | org.apache.kafka.common.serialization.StringSerializer
+
+Additional properties
+---------------------
+
+`plugin.kafka-events.groupId`
+: Kafka consumer group for receiving messages.
+ Default: Gerrit instance-id
+
+`plugin.kafka-events.pollingIntervalMs`
+: Polling interval in msec for receiving messages from Kafka topic subscription.
+ Default: 1000
\ No newline at end of file
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 12a448c..95d2603 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/kafka/EventConsumerIT.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/kafka/EventConsumerIT.java
@@ -16,7 +16,7 @@
import static com.google.common.truth.Truth.assertThat;
-import com.google.common.base.Supplier;
+import com.gerritforge.gerrit.eventbroker.EventGsonProvider;
import com.google.common.collect.Iterables;
import com.google.gerrit.acceptance.GerritConfig;
import com.google.gerrit.acceptance.LightweightPluginDaemonTest;
@@ -28,10 +28,7 @@
import com.google.gerrit.extensions.common.ChangeMessageInfo;
import com.google.gerrit.server.events.CommentAddedEvent;
import com.google.gerrit.server.events.Event;
-import com.google.gerrit.server.events.EventDeserializer;
-import com.google.gerrit.server.events.SupplierDeserializer;
import com.google.gson.Gson;
-import com.google.gson.GsonBuilder;
import com.googlesource.gerrit.plugins.kafka.config.KafkaProperties;
import java.util.ArrayList;
import java.util.Collections;
@@ -115,11 +112,7 @@
assertThat(events).hasSize(6);
String commentAddedEventJson = Iterables.getLast(events);
- Gson gson =
- new GsonBuilder()
- .registerTypeAdapter(Event.class, new EventDeserializer())
- .registerTypeAdapter(Supplier.class, new SupplierDeserializer())
- .create();
+ Gson gson = new EventGsonProvider().get();
Event event = gson.fromJson(commentAddedEventJson, Event.class);
assertThat(event).isInstanceOf(CommentAddedEvent.class);
diff --git a/src/test/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventDeserializerTest.java b/src/test/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventDeserializerTest.java
new file mode 100644
index 0000000..e456a2a
--- /dev/null
+++ b/src/test/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventDeserializerTest.java
@@ -0,0 +1,64 @@
+// 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.kafka.subscribe;
+
+import static com.google.common.truth.Truth.assertThat;
+import static java.nio.charset.StandardCharsets.UTF_8;
+
+import com.gerritforge.gerrit.eventbroker.EventGsonProvider;
+import com.gerritforge.gerrit.eventbroker.EventMessage;
+import com.google.gson.Gson;
+import java.util.UUID;
+import org.junit.Before;
+import org.junit.Test;
+
+public class KafkaEventDeserializerTest {
+ private KafkaEventDeserializer deserializer;
+
+ @Before
+ public void setUp() {
+ final Gson gson = new EventGsonProvider().get();
+ deserializer = new KafkaEventDeserializer(gson);
+ }
+
+ @Test
+ public void kafkaEventDeserializerShouldParseAKafkaEvent() {
+ final UUID eventId = UUID.randomUUID();
+ final String eventType = "event-type";
+ final UUID sourceInstanceId = UUID.randomUUID();
+ final long eventCreatedOn = 10L;
+ final String eventJson =
+ String.format(
+ "{ "
+ + "\"header\": { \"eventId\": \"%s\", \"eventType\": \"%s\", \"sourceInstanceId\": \"%s\", \"eventCreatedOn\": %d },"
+ + "\"body\": { \"type\": \"project-created\" }"
+ + "}",
+ eventId, eventType, sourceInstanceId, eventCreatedOn);
+ final EventMessage event = deserializer.deserialize("ignored", eventJson.getBytes(UTF_8));
+
+ assertThat(event.getHeader().eventId).isEqualTo(eventId);
+ assertThat(event.getHeader().sourceInstanceId).isEqualTo(sourceInstanceId);
+ }
+
+ @Test(expected = RuntimeException.class)
+ public void kafkaEventDeserializerShouldFailForInvalidJson() {
+ deserializer.deserialize("ignored", "this is not a JSON string".getBytes(UTF_8));
+ }
+
+ @Test(expected = RuntimeException.class)
+ public void kafkaEventDeserializerShouldFailForInvalidObjectButValidJSON() {
+ deserializer.deserialize("ignored", "{}".getBytes(UTF_8));
+ }
+}
