Google Git
Sign in
gerrit / modules / events-broker / a7466df891a2883ac2a0b40c8c6e1fedcd251fac
commita7466df891a2883ac2a0b40c8c6e1fedcd251fac[log] [tgz]
authorLuca Milanesio <luca.milanesio@gmail.com>Tue Feb 27 20:04:24 2024 +0000
committerLuca Milanesio <luca.milanesio@gmail.com>Fri Mar 01 00:21:17 2024 +0000
tree891b1220b76132a3ce640b52f37ff3250bf2e69a
parent013c23661e375c81932a2d781f9ba6c003b9c4e1 [diff]
Introduce new interface method for disconnecting consumers by groupId

Allow plugins to isolate their consumers using groupId, so that
multiple consumers can work independently.

Allow the broker API to disconnect only specific consumers on
topics by groupId, so that unloading one plugin would not impact
the functionality of other plugins consuming messages on the
same topics.

Bug:Issue 327226782
Change-Id: I859cddd012295e576fc4df4a66af65eedfd24d29
2 files changed
tree: 891b1220b76132a3ce640b52f37ff3250bf2e69a
  1. src/
  2. .gitignore
  3. BUILD
  4. Jenkinsfile
  5. LICENSE
  6. README.md
README.md

Events Broker API for Gerrit Code Review

API of a generic events broker for use with Gerrit Code Review.

Enables the de-coupling between Gerrit, plugins and the different implementations of a generic events broker.

Stream Events Publisher

It is a quite common use case for consumers of this library to listen for Gerrit events and to stream them on a specific topic.

Since the implementation of such logic is always the same, this library provides a generic stream events publisher which will perform the relevant operations.

In order to listen and stream gerrit events, consumers of this API need to provide a binding for the StreamEventPublisherConfig configuration and java.util.concurrent.Executor binding annotated with StreamEventPublisherExecutor annotation. A default single threaded implementation (StreamEventPublisherExecutor) is provided by the library. The last step is to explicitly bind the Stream Events Publisher, as such:

import java.util.concurrent.Executor;

import com.gerritforge.gerrit.eventbroker.publisher.StreamEventPublisher;
import com.gerritforge.gerrit.eventbroker.publisher.StreamEventPublisherConfig;
import com.gerritforge.gerrit.eventbroker.publisher.executor.StreamEventPublisherExecutor;
import com.gerritforge.gerrit.eventbroker.publisher.executor.StreamEventPublisherExecutorProvider;
import com.google.gerrit.extensions.registration.DynamicSet;
import com.google.gerrit.server.events.EventListener;
import com.google.inject.AbstractModule;

public class SomeModule extends AbstractModule {
    @Override
    protected void configure() {
        long messagePublishingTimeout = 1000L;

        bind(StreamEventPublisherConfig.class)
                .toInstance(new StreamEventPublisherConfig(
                    "name_of_the_stream_events_topic",
                    messagePublishingTimeout));

        bind(Executor.class).annotatedWith(StreamEventPublisherExecutor.class).toProvider(StreamEventPublisherExecutorProvider.class);
        DynamicSet.bind(binder(), EventListener.class).to(StreamEventPublisher.class);
    }
}

Alternative way to setup Stream Event Publisher is to use default Guice module StreamEventPublisherModule:

import com.gerritforge.gerrit.eventbroker.publisher.StreamEventPublisherConfig;
import com.gerritforge.gerrit.eventbroker.publisher.StreamEventPublisherModule;
import com.google.inject.AbstractModule;
import com.google.inject.TypeLiteral;
import com.google.inject.name.Names;

public class SomeModule extends AbstractModule {
    @Override
    protected void configure() {
        long messagePublishingTimeout = 1000L;
        bind(StreamEventPublisherConfig.class)
                .toInstance(new StreamEventPublisherConfig(
                    "name_of_the_stream_events_topic",
                    messagePublishingTimeout));
        install(new StreamEventPublisherModule());
    }
}

Note: To avoid message duplication Stream Events Publisher uses gerrit.instanceId and Event.instanceId to filter out forwarded events.

Broker Metrics

When StreamEventPublisher is used user can optionally bind an implementation of the BrokerMetrics` interface. This will allow to collect metrics about successful/failure stream events publishing. If no binding is provided default implementation will skip collecting metrics:

import com.google.gerrit.extensions.registration.DynamicItem;
import com.google.inject.AbstractModule;

public class SomeModule extends AbstractModule {
    @Override
    protected void configure() {
        DynamicItem.bind(binder(), BrokerMetrics.class)
            .to(BrokerMetricsImpl.class)
            .in(Scopes.SINGLETON);
    }
}

Note: BrokerMetrics implementation must be bound in a plugin module.