Initial stable-3.0 commit from kafka-events repo
Change-Id: I4b774850a16849456d5a1ca9558667dce76a1f85
diff --git a/BUILD b/BUILD
index ba9f702..c0eab0c 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 = [
":events-kafka__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 197cbed..dbcf708 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.15.0",
sha1 = "d34760b11ab656e08b72c1e2e9b852f037a89f90",
)
+
+ maven_jar(
+ name = "events-broker",
+ artifact = "com.gerritforge:events-broker:3.0.5",
+ sha1 = "7abf72d2252f975baff666fbbf28b7036767aa81",
+ )
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..1be52cb 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,38 @@
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.google.inject.TypeLiteral;
+import com.googlesource.gerrit.plugins.kafka.api.KafkaApiModule;
import com.googlesource.gerrit.plugins.kafka.publish.KafkaPublisher;
+import com.googlesource.gerrit.plugins.kafka.session.KafkaProducerProvider;
+import org.apache.kafka.clients.producer.KafkaProducer;
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);
+
+ bind(new TypeLiteral<KafkaProducer<String, String>>() {})
+ .toProvider(KafkaProducerProvider.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..9a7c66a
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerApi.java
@@ -0,0 +1,79 @@
+// 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());
+ }
+
+ @Override
+ public void replayAllEvents(String topic) {
+ subscribers.stream()
+ .filter(subscriber -> topic.equals(subscriber.getTopic()))
+ .forEach(subscriber -> subscriber.resetOffset());
+ }
+}
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..72d7f91 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
@@ -14,6 +14,7 @@
package com.googlesource.gerrit.plugins.kafka.config;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.CaseFormat;
import com.google.common.base.Strings;
import com.google.gerrit.extensions.annotations.PluginName;
@@ -33,6 +34,7 @@
public static final String KAFKA_STRING_SERIALIZER = StringSerializer.class.getName();
private final String topic;
+ private final boolean sendAsync;
@Inject
public KafkaProperties(PluginConfigFactory configFactory, @PluginName String pluginName) {
@@ -40,10 +42,20 @@
setDefaults();
PluginConfig fromGerritConfig = configFactory.getFromGerritConfig(pluginName);
topic = fromGerritConfig.getString("topic", "gerrit");
+ sendAsync = fromGerritConfig.getBoolean("sendAsync", true);
applyConfig(fromGerritConfig);
initDockerizedKafkaServer();
}
+ @VisibleForTesting
+ public KafkaProperties(boolean sendAsync) {
+ super();
+ setDefaults();
+ topic = "gerrit";
+ this.sendAsync = sendAsync;
+ initDockerizedKafkaServer();
+ }
+
private void setDefaults() {
put("acks", "all");
put("retries", 0);
@@ -79,4 +91,12 @@
public String getTopic() {
return topic;
}
+
+ public boolean isSendAsync() {
+ return sendAsync;
+ }
+
+ 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..52d4726
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/config/KafkaSubscriberProperties.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.config;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.gerrit.extensions.annotations.PluginName;
+import com.google.gerrit.server.config.PluginConfigFactory;
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+
+@Singleton
+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 = "6";
+
+ 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("polling.interval.ms", DEFAULT_POLLING_INTERVAL_MS));
+ this.groupId = getProperty("group.id");
+ this.numberOfSubscribers =
+ Integer.parseInt(getProperty("number.of.subscribers", DEFAULT_NUMBER_OF_SUBSCRIBERS));
+ }
+
+ @VisibleForTesting
+ public KafkaSubscriberProperties(int pollingInterval, String groupId, int numberOfSubscribers) {
+ super(true);
+ this.pollingInterval = pollingInterval;
+ this.groupId = groupId;
+ this.numberOfSubscribers = numberOfSubscribers;
+ }
+
+ 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/KafkaEventsPublisherMetrics.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/publish/KafkaEventsPublisherMetrics.java
new file mode 100644
index 0000000..083d032
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/publish/KafkaEventsPublisherMetrics.java
@@ -0,0 +1,58 @@
+// Copyright (C) 2020 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.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 KafkaEventsPublisherMetrics {
+ private static final String PUBLISHER_SUCCESS_COUNTER = "broker_msg_publisher_success_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 KafkaEventsPublisherMetrics(MetricMaker metricMaker) {
+
+ this.brokerPublisherSuccessCounter =
+ metricMaker.newCounter(
+ "kafka/broker/broker_message_publisher_counter",
+ new Description("Number of successfully published messages by the broker publisher")
+ .setRate()
+ .setUnit("messages"),
+ Field.ofString(PUBLISHER_SUCCESS_COUNTER, "Broker message published count"));
+ this.brokerPublisherFailureCounter =
+ metricMaker.newCounter(
+ "kafka/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/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/KafkaProducerProvider.java b/src/main/java/com/googlesource/gerrit/plugins/kafka/session/KafkaProducerProvider.java
new file mode 100644
index 0000000..b1f11f7
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/session/KafkaProducerProvider.java
@@ -0,0 +1,34 @@
+// Copyright (C) 2020 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.session;
+
+import com.google.inject.Inject;
+import com.google.inject.Provider;
+import com.googlesource.gerrit.plugins.kafka.config.KafkaProperties;
+import org.apache.kafka.clients.producer.KafkaProducer;
+
+public class KafkaProducerProvider implements Provider<KafkaProducer<String, String>> {
+ private final KafkaProperties properties;
+
+ @Inject
+ public KafkaProducerProvider(KafkaProperties properties) {
+ this.properties = properties;
+ }
+
+ @Override
+ public KafkaProducer<String, String> get() {
+ return new KafkaProducer<>(properties);
+ }
+}
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..bb79cb5 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
@@ -15,10 +15,14 @@
package com.googlesource.gerrit.plugins.kafka.session;
import com.google.inject.Inject;
+import com.google.inject.Provider;
import com.googlesource.gerrit.plugins.kafka.config.KafkaProperties;
+import com.googlesource.gerrit.plugins.kafka.publish.KafkaEventsPublisherMetrics;
+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;
@@ -26,11 +30,18 @@
private static final Logger LOGGER = LoggerFactory.getLogger(KafkaSession.class);
private final KafkaProperties properties;
+ private final Provider<KafkaProducer<String, String>> producerProvider;
+ private final KafkaEventsPublisherMetrics publisherMetrics;
private volatile Producer<String, String> producer;
@Inject
- public KafkaSession(KafkaProperties properties) {
+ public KafkaSession(
+ Provider<KafkaProducer<String, String>> producerProvider,
+ KafkaProperties properties,
+ KafkaEventsPublisherMetrics publisherMetrics) {
+ this.producerProvider = producerProvider;
this.properties = properties;
+ this.publisherMetrics = publisherMetrics;
}
public boolean isOpen() {
@@ -52,7 +63,7 @@
* ClassNotFoundExceptions
*/
setConnectionClassLoader();
- producer = new KafkaProducer<>(properties);
+ producer = producerProvider.get();
LOGGER.info("Connection established.");
}
@@ -70,6 +81,51 @@
}
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) {
+ if (properties.isSendAsync()) {
+ return publishAsync(topic, messageBody);
+ }
+ return publishSync(topic, messageBody);
+ }
+
+ private boolean publishSync(String topic, String messageBody) {
+
+ try {
+ Future<RecordMetadata> future =
+ producer.send(new ProducerRecord<>(topic, "" + System.nanoTime(), messageBody));
+ RecordMetadata metadata = future.get();
+ LOGGER.debug("The offset of the record we just sent is: {}", metadata.offset());
+ publisherMetrics.incrementBrokerPublishedMessage();
+ return true;
+ } catch (Throwable e) {
+ LOGGER.error("Cannot send the message", e);
+ publisherMetrics.incrementBrokerFailedToPublishMessage();
+ return false;
+ }
+ }
+
+ private boolean publishAsync(String topic, String messageBody) {
+ try {
+ Future<RecordMetadata> future =
+ producer.send(
+ new ProducerRecord<>(topic, Long.toString(System.nanoTime()), messageBody),
+ (metadata, e) -> {
+ if (metadata != null && e == null) {
+ LOGGER.debug("The offset of the record we just sent is: {}", metadata.offset());
+ publisherMetrics.incrementBrokerPublishedMessage();
+ } else {
+ LOGGER.error("Cannot send the message", e);
+ publisherMetrics.incrementBrokerFailedToPublishMessage();
+ }
+ });
+ return future != null;
+ } catch (Throwable e) {
+ LOGGER.error("Cannot send the message", e);
+ publisherMetrics.incrementBrokerFailedToPublishMessage();
+ 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..40415a4
--- /dev/null
+++ b/src/main/java/com/googlesource/gerrit/plugins/kafka/subscribe/KafkaEventSubscriber.java
@@ -0,0 +1,180 @@
+// 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.Random;
+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 static final int DELAY_RECONNECT_AFTER_FAILURE_MSEC = 1000;
+
+ 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 final KafkaConsumerFactory consumerFactory;
+ private final Deserializer<byte[]> keyDeserializer;
+
+ private java.util.function.Consumer<EventMessage> messageProcessor;
+ private String topic;
+ private AtomicBoolean resetOffset = new AtomicBoolean(false);
+
+ private volatile ReceiverJob receiver;
+
+ @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;
+ this.consumerFactory = consumerFactory;
+ this.keyDeserializer = keyDeserializer;
+ 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);
+ runReceiver();
+ }
+
+ private void runReceiver() {
+ final ClassLoader previousClassLoader = Thread.currentThread().getContextClassLoader();
+ try {
+ Thread.currentThread().setContextClassLoader(KafkaEventSubscriber.class.getClassLoader());
+ Consumer<byte[], byte[]> consumer = consumerFactory.create(keyDeserializer);
+ consumer.subscribe(Collections.singleton(topic));
+ receiver = new ReceiverJob(consumer);
+ executor.execute(receiver);
+ } finally {
+ Thread.currentThread().setContextClassLoader(previousClassLoader);
+ }
+ }
+
+ public void shutdown() {
+ closed.set(true);
+ receiver.wakeup();
+ }
+
+ public java.util.function.Consumer<EventMessage> getMessageProcessor() {
+ return messageProcessor;
+ }
+
+ public String getTopic() {
+ return topic;
+ }
+
+ public void resetOffset() {
+ resetOffset.set(true);
+ }
+
+ private class ReceiverJob implements Runnable {
+ private final Consumer<byte[], byte[]> consumer;
+
+ public ReceiverJob(Consumer<byte[], byte[]> consumer) {
+ this.consumer = consumer;
+ }
+
+ public void wakeup() {
+ consumer.wakeup();
+ }
+
+ @Override
+ public void run() {
+ try {
+ consume();
+ } catch (Exception e) {
+ logger.atSevere().withCause(e).log("Consumer loop of topic %s ended", topic);
+ }
+ }
+
+ private void consume() throws InterruptedException {
+ try {
+ while (!closed.get()) {
+ if (resetOffset.getAndSet(false)) {
+ consumer.seekToBeginning(consumer.assignment());
+ }
+ 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()) {
+ logger.atSevere().withCause(e).log("Consumer loop of topic %s interrupted", topic);
+ reconnectAfterFailure();
+ }
+ } catch (Exception e) {
+ subscriberMetrics.incrementSubscriberFailedToPollMessages();
+ logger.atSevere().withCause(e).log(
+ "Existing consumer loop of topic %s because of a non-recoverable exception", topic);
+ reconnectAfterFailure();
+ } finally {
+ consumer.close();
+ }
+ }
+
+ private void reconnectAfterFailure() throws InterruptedException {
+ // Random delay with average of DELAY_RECONNECT_AFTER_FAILURE_MSEC
+ // for avoiding hammering exactly at the same interval in case of failure
+ long reconnectDelay =
+ DELAY_RECONNECT_AFTER_FAILURE_MSEC / 2
+ + new Random().nextInt(DELAY_RECONNECT_AFTER_FAILURE_MSEC);
+ Thread.sleep(reconnectDelay);
+ runReceiver();
+ }
+ }
+}
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 4db7b38..727b7e5 100644
--- a/src/main/resources/Documentation/config.md
+++ b/src/main/resources/Documentation/config.md
@@ -7,7 +7,7 @@
---------------------
```
-[plugin "events-kafka"]
+[plugin "@PLUGIN@"]
bootstrapServers = localhost:9092
```
@@ -31,4 +31,20 @@
| 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.@PLUGIN@.groupId`
+: Kafka consumer group for receiving messages.
+ Default: Gerrit instance-id
+
+`plugin.@PLUGIN@.pollingIntervalMs`
+: Polling interval in msec for receiving messages from Kafka topic subscription.
+ Default: 1000
+
+`plugin.@PLUGIN@.sendAsync`
+: Send messages to Kafka asynchronously, detaching the calling process from the
+ acknowledge of the message being sent.
+ Default: true
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 0cafbdd..fbae0c6 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;
@@ -104,29 +101,18 @@
}
}
- // TODO(davido): Remove special ReviewDb case when it is killed
- // In ReviewDb case 3 events are received in the following order:
- // 1. refUpdate: ref: refs/changes/01/1/1
- // 2. patchset-created: ref: refs/changes/01/1/1
- // 3. comment-added: ref: refs/heads/master
- int numberOfEvents = 3;
- if (notesMigration.commitChangeWrites()) {
- // In NoteDb case the 4 events are received in the following order:
- // 1. refUpdate: ref: refs/sequences/changes
- // 2. refUpdate: ref: refs/changes/01/1/1
- // 3. patchset-created: ref: refs/changes/01/1/1
- // 4. comment-added: ref: refs/heads/master
- numberOfEvents = 4;
- }
+ // There are 6 events are received in the following order:
+ // 1. refUpdate: ref: refs/sequences/changes
+ // 2. refUpdate: ref: refs/changes/01/1/1
+ // 3. refUpdate: ref: refs/changes/01/1/meta
+ // 4. patchset-created: ref: refs/changes/01/1/1
+ // 5. refUpdate: ref: refs/changes/01/1/meta
+ // 6. comment-added: ref: refs/heads/master
- assertThat(events).hasSize(numberOfEvents);
+ 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/api/KafkaBrokerApiTest.java b/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerApiTest.java
new file mode 100644
index 0000000..eabc833
--- /dev/null
+++ b/src/test/java/com/googlesource/gerrit/plugins/kafka/api/KafkaBrokerApiTest.java
@@ -0,0 +1,216 @@
+// Copyright (C) 2020 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 static com.google.common.truth.Truth.assertThat;
+import static org.mockito.Mockito.mock;
+
+import com.gerritforge.gerrit.eventbroker.EventGsonProvider;
+import com.gerritforge.gerrit.eventbroker.EventMessage;
+import com.gerritforge.gerrit.eventbroker.EventMessage.Header;
+import com.google.gerrit.metrics.MetricMaker;
+import com.google.gerrit.server.events.ProjectCreatedEvent;
+import com.google.gerrit.server.git.WorkQueue;
+import com.google.gerrit.server.util.IdGenerator;
+import com.google.gerrit.server.util.OneOffRequestContext;
+import com.google.gson.Gson;
+import com.google.inject.AbstractModule;
+import com.google.inject.Guice;
+import com.google.inject.Inject;
+import com.google.inject.Injector;
+import com.google.inject.Scopes;
+import com.google.inject.Singleton;
+import com.google.inject.TypeLiteral;
+import com.googlesource.gerrit.plugins.kafka.config.KafkaProperties;
+import com.googlesource.gerrit.plugins.kafka.config.KafkaSubscriberProperties;
+import com.googlesource.gerrit.plugins.kafka.session.KafkaProducerProvider;
+import com.googlesource.gerrit.plugins.kafka.session.KafkaSession;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Answers;
+import org.mockito.junit.MockitoJUnitRunner;
+import org.testcontainers.containers.KafkaContainer;
+
+@RunWith(MockitoJUnitRunner.class)
+public class KafkaBrokerApiTest {
+ private static KafkaContainer kafka;
+
+ private static final int TEST_NUM_SUBSCRIBERS = 1;
+ private static final String TEST_GROUP_ID = KafkaBrokerApiTest.class.getName();
+ private static final int TEST_POLLING_INTERVAL_MSEC = 100;
+ private static final int TEST_THREAD_POOL_SIZE = 10;
+ private static final UUID TEST_INSTANCE_ID = UUID.randomUUID();
+ private static final TimeUnit TEST_TIMOUT_UNIT = TimeUnit.SECONDS;
+ private static final int TEST_TIMEOUT = 30;
+
+ private Injector injector;
+ private KafkaSession session;
+ private Gson gson;
+
+ public static class TestKafkaContainer extends KafkaContainer {
+ public TestKafkaContainer() {
+ addFixedExposedPort(KAFKA_PORT, KAFKA_PORT);
+ addFixedExposedPort(ZOOKEEPER_PORT, ZOOKEEPER_PORT);
+ }
+
+ @Override
+ public String getBootstrapServers() {
+ return String.format("PLAINTEXT://%s:%s", getContainerIpAddress(), KAFKA_PORT);
+ }
+ }
+
+ public static class TestWorkQueue extends WorkQueue {
+
+ @Inject
+ public TestWorkQueue(IdGenerator idGenerator, MetricMaker metrics) {
+ super(idGenerator, TEST_THREAD_POOL_SIZE, metrics);
+ }
+ }
+
+ public static class TestModule extends AbstractModule {
+ private KafkaProperties kafkaProperties;
+
+ public TestModule(KafkaProperties kafkaProperties) {
+ this.kafkaProperties = kafkaProperties;
+ }
+
+ @Override
+ protected void configure() {
+ bind(Gson.class).toProvider(EventGsonProvider.class).in(Singleton.class);
+ bind(MetricMaker.class).toInstance(mock(MetricMaker.class, Answers.RETURNS_DEEP_STUBS));
+ bind(OneOffRequestContext.class)
+ .toInstance(mock(OneOffRequestContext.class, Answers.RETURNS_DEEP_STUBS));
+
+ bind(KafkaProperties.class).toInstance(kafkaProperties);
+ bind(KafkaSession.class).in(Scopes.SINGLETON);
+ KafkaSubscriberProperties kafkaSubscriberProperties =
+ new KafkaSubscriberProperties(
+ TEST_POLLING_INTERVAL_MSEC, TEST_GROUP_ID, TEST_NUM_SUBSCRIBERS);
+ bind(KafkaSubscriberProperties.class).toInstance(kafkaSubscriberProperties);
+ bind(new TypeLiteral<KafkaProducer<String, String>>() {})
+ .toProvider(KafkaProducerProvider.class);
+
+ bind(WorkQueue.class).to(TestWorkQueue.class);
+ }
+ }
+
+ public static class TestConsumer implements Consumer<EventMessage> {
+ public final List<EventMessage> messages = new ArrayList<>();
+ private final CountDownLatch lock;
+
+ public TestConsumer(int numMessagesExpected) {
+ lock = new CountDownLatch(numMessagesExpected);
+ }
+
+ @Override
+ public void accept(EventMessage message) {
+ messages.add(message);
+ lock.countDown();
+ }
+
+ public boolean await() {
+ try {
+ return lock.await(TEST_TIMEOUT, TEST_TIMOUT_UNIT);
+ } catch (InterruptedException e) {
+ return false;
+ }
+ }
+ }
+
+ public static class TestHeader extends Header {
+
+ public TestHeader() {
+ super(UUID.randomUUID(), TEST_INSTANCE_ID);
+ }
+ }
+
+ @BeforeClass
+ public static void beforeClass() throws Exception {
+ kafka = new TestKafkaContainer();
+ kafka.start();
+ System.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers());
+ }
+
+ @AfterClass
+ public static void afterClass() {
+ if (kafka != null) {
+ kafka.stop();
+ }
+ }
+
+ public void connectToKafka(KafkaProperties kafkaProperties) {
+ Injector baseInjector = Guice.createInjector(new TestModule(kafkaProperties));
+ WorkQueue testWorkQueue = baseInjector.getInstance(WorkQueue.class);
+ KafkaSubscriberProperties kafkaSubscriberProperties =
+ baseInjector.getInstance(KafkaSubscriberProperties.class);
+ injector =
+ baseInjector.createChildInjector(
+ new KafkaApiModule(testWorkQueue, kafkaSubscriberProperties));
+ session = injector.getInstance(KafkaSession.class);
+ gson = injector.getInstance(Gson.class);
+
+ session.connect();
+ }
+
+ @After
+ public void teardown() {
+ if (session != null) {
+ session.disconnect();
+ }
+ }
+
+ @Test
+ public void shouldSendSyncAndReceiveToTopic() {
+ connectToKafka(new KafkaProperties(false));
+ KafkaBrokerApi kafkaBrokerApi = injector.getInstance(KafkaBrokerApi.class);
+ String testTopic = "test_topic_sync";
+ TestConsumer testConsumer = new TestConsumer(1);
+ EventMessage testEventMessage = new EventMessage(new TestHeader(), new ProjectCreatedEvent());
+
+ kafkaBrokerApi.receiveAsync(testTopic, testConsumer);
+ kafkaBrokerApi.send(testTopic, testEventMessage);
+
+ assertThat(testConsumer.await()).isTrue();
+ assertThat(testConsumer.messages).hasSize(1);
+ assertThat(gson.toJson(testConsumer.messages.get(0))).isEqualTo(gson.toJson(testEventMessage));
+ }
+
+ @Test
+ public void shouldSendAsyncAndReceiveToTopic() {
+ connectToKafka(new KafkaProperties(true));
+ KafkaBrokerApi kafkaBrokerApi = injector.getInstance(KafkaBrokerApi.class);
+ String testTopic = "test_topic_async";
+ TestConsumer testConsumer = new TestConsumer(1);
+ EventMessage testEventMessage = new EventMessage(new TestHeader(), new ProjectCreatedEvent());
+
+ kafkaBrokerApi.send(testTopic, testEventMessage);
+ kafkaBrokerApi.receiveAsync(testTopic, testConsumer);
+
+ assertThat(testConsumer.await()).isTrue();
+ assertThat(testConsumer.messages).hasSize(1);
+ assertThat(gson.toJson(testConsumer.messages.get(0))).isEqualTo(gson.toJson(testEventMessage));
+ }
+}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/kafka/publish/KafkaSessionTest.java b/src/test/java/com/googlesource/gerrit/plugins/kafka/publish/KafkaSessionTest.java
new file mode 100644
index 0000000..5aa9ca8
--- /dev/null
+++ b/src/test/java/com/googlesource/gerrit/plugins/kafka/publish/KafkaSessionTest.java
@@ -0,0 +1,127 @@
+// Copyright (C) 2020 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 static org.mockito.Mockito.any;
+import static org.mockito.Mockito.only;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import com.google.common.util.concurrent.Futures;
+import com.googlesource.gerrit.plugins.kafka.config.KafkaProperties;
+import com.googlesource.gerrit.plugins.kafka.session.KafkaProducerProvider;
+import com.googlesource.gerrit.plugins.kafka.session.KafkaSession;
+import org.apache.kafka.clients.producer.Callback;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.TopicPartition;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Captor;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+@RunWith(MockitoJUnitRunner.class)
+public class KafkaSessionTest {
+ KafkaSession objectUnderTest;
+ @Mock KafkaProducer<String, String> kafkaProducer;
+ @Mock KafkaProducerProvider producerProvider;
+ @Mock KafkaProperties properties;
+ @Mock KafkaEventsPublisherMetrics publisherMetrics;
+ @Captor ArgumentCaptor<Callback> callbackCaptor;
+
+ RecordMetadata recordMetadata;
+ String message = "sample_message";
+ private String topic = "index";
+
+ @Before
+ public void setUp() {
+ when(producerProvider.get()).thenReturn(kafkaProducer);
+ when(properties.getTopic()).thenReturn(topic);
+
+ recordMetadata = new RecordMetadata(new TopicPartition(topic, 0), 0L, 0L, 0L, 0L, 0, 0);
+
+ objectUnderTest = new KafkaSession(producerProvider, properties, publisherMetrics);
+ objectUnderTest.connect();
+ }
+
+ @Test
+ public void shouldIncrementBrokerMetricCounterWhenMessagePublishedInSyncMode() {
+ when(properties.isSendAsync()).thenReturn(false);
+ when(kafkaProducer.send(any())).thenReturn(Futures.immediateFuture(recordMetadata));
+ objectUnderTest.publish(message);
+ verify(publisherMetrics, only()).incrementBrokerPublishedMessage();
+ }
+
+ @Test
+ public void shouldIncrementBrokerFailedMetricCounterWhenMessagePublishingFailedInSyncMode() {
+ when(properties.isSendAsync()).thenReturn(false);
+ when(kafkaProducer.send(any())).thenReturn(Futures.immediateFailedFuture(new Exception()));
+ objectUnderTest.publish(message);
+ verify(publisherMetrics, only()).incrementBrokerFailedToPublishMessage();
+ }
+
+ @Test
+ public void shouldIncrementBrokerFailedMetricCounterWhenUnexpectedExceptionInSyncMode() {
+ when(properties.isSendAsync()).thenReturn(false);
+ when(kafkaProducer.send(any())).thenThrow(new RuntimeException("Unexpected runtime exception"));
+ try {
+ objectUnderTest.publish(message);
+ } catch (RuntimeException e) {
+ // expected
+ }
+ verify(publisherMetrics, only()).incrementBrokerFailedToPublishMessage();
+ }
+
+ @Test
+ public void shouldIncrementBrokerMetricCounterWhenMessagePublishedInAsyncMode() {
+ when(properties.isSendAsync()).thenReturn(true);
+ when(kafkaProducer.send(any(), any())).thenReturn(Futures.immediateFuture(recordMetadata));
+
+ objectUnderTest.publish(message);
+
+ verify(kafkaProducer).send(any(), callbackCaptor.capture());
+ callbackCaptor.getValue().onCompletion(recordMetadata, null);
+ verify(publisherMetrics, only()).incrementBrokerPublishedMessage();
+ }
+
+ @Test
+ public void shouldIncrementBrokerFailedMetricCounterWhenMessagePublishingFailedInAsyncMode() {
+ when(properties.isSendAsync()).thenReturn(true);
+ when(kafkaProducer.send(any(), any()))
+ .thenReturn(Futures.immediateFailedFuture(new Exception()));
+
+ objectUnderTest.publish(message);
+
+ verify(kafkaProducer).send(any(), callbackCaptor.capture());
+ callbackCaptor.getValue().onCompletion(null, new Exception());
+ verify(publisherMetrics, only()).incrementBrokerFailedToPublishMessage();
+ }
+
+ @Test
+ public void shouldIncrementBrokerFailedMetricCounterWhenUnexpectedExceptionInAsyncMode() {
+ when(properties.isSendAsync()).thenReturn(true);
+ when(kafkaProducer.send(any(), any()))
+ .thenThrow(new RuntimeException("Unexpected runtime exception"));
+ try {
+ objectUnderTest.publish(message);
+ } catch (RuntimeException e) {
+ // expected
+ }
+ verify(publisherMetrics, only()).incrementBrokerFailedToPublishMessage();
+ }
+}
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));
+ }
+}
