Merge branch 'stable-3.4' into stable-3.5 * stable-3.4: Allow subscribing to the topic with custom group id Consume events-broker from source Clone also events-broker for build validation Change-Id: I1212ce369d127a015d1d203090e706fb6c079860
diff --git a/BUILD b/BUILD index 972f0d8..dbbc7e9 100644 --- a/BUILD +++ b/BUILD
@@ -16,10 +16,10 @@ ], resources = glob(["src/main/resources/**/*"]), deps = [ + ":events-broker-neverlink", "//lib/commons:io", "//plugins/replication", "@commons-lang3//jar", - "@events-broker//jar:neverlink", ], ) @@ -35,7 +35,7 @@ ":pull-replication__plugin", ":pull_replication_util", "//plugins/replication", - "@events-broker//jar", + "//plugins/events-broker", ], ) @@ -73,6 +73,12 @@ visibility = ["//visibility:public"], exports = PLUGIN_DEPS + PLUGIN_TEST_DEPS + [ ":pull-replication__plugin", - "@events-broker//jar", + "//plugins/events-broker", ], ) + +java_library( + name = "events-broker-neverlink", + neverlink = 1, + exports = ["//plugins/events-broker"], +)
diff --git a/Jenkinsfile b/Jenkinsfile index 448692b..a4fff37 100644 --- a/Jenkinsfile +++ b/Jenkinsfile
@@ -1,2 +1,3 @@ pluginPipeline(formatCheckId: 'gerritforge:pull-replication-format-3852e64366bb37d13b8baf8af9b15cfd38eb9227', - buildCheckId: 'gerritforge:pull-replication-3852e64366bb37d13b8baf8af9b15cfd38eb9227') \ No newline at end of file + buildCheckId: 'gerritforge:pull-replication-3852e64366bb37d13b8baf8af9b15cfd38eb9227', + extraModules: ['events-broker'])
diff --git a/external_plugin_deps.bzl b/external_plugin_deps.bzl deleted file mode 100644 index 5d2ba13..0000000 --- a/external_plugin_deps.bzl +++ /dev/null
@@ -1,8 +0,0 @@ -load("//tools/bzl:maven_jar.bzl", "maven_jar") - -def external_plugin_deps(): - maven_jar( - name = "events-broker", - artifact = "com.gerritforge:events-broker:3.5.0.1", - sha1 = "af192a8bceaf7ff54d19356f9bfe1f1e83634b40", - )
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationModule.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationModule.java index 5308294..72a6b74 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationModule.java +++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/PullReplicationModule.java
@@ -152,7 +152,7 @@ if (replicationConfig.getBoolean("replication", "consumeStreamEvents", false)) { install(new StreamEventModule()); } else if (eventBrokerTopic != null) { - install(new EventsBrokerConsumerModule(eventBrokerTopic)); + install(new EventsBrokerConsumerModule(eventBrokerTopic, replicationConfig)); } DynamicSet.setOf(binder(), ReplicationStateListener.class);
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/event/EventsBrokerConsumerModule.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/event/EventsBrokerConsumerModule.java index ce1a076..a8d5cf4 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/event/EventsBrokerConsumerModule.java +++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/event/EventsBrokerConsumerModule.java
@@ -17,21 +17,28 @@ import com.google.gerrit.lifecycle.LifecycleModule; import com.google.inject.Scopes; import com.google.inject.name.Names; +import com.google.inject.util.Providers; +import org.eclipse.jgit.lib.Config; public class EventsBrokerConsumerModule extends LifecycleModule { public static final String STREAM_EVENTS_TOPIC_NAME = "stream_events_topic_name"; + public static final String STREAM_EVENTS_GROUP_ID = "stream_events_group_id"; private final String topicName; + private final Config config; - public EventsBrokerConsumerModule(String topicName) { + public EventsBrokerConsumerModule(String topicName, Config config) { this.topicName = topicName; + this.config = config; } @Override protected void configure() { bind(EventsBrokerMessageConsumer.class).in(Scopes.SINGLETON); bind(String.class).annotatedWith(Names.named(STREAM_EVENTS_TOPIC_NAME)).toInstance(topicName); - + bind(String.class) + .annotatedWith(Names.named(STREAM_EVENTS_GROUP_ID)) + .toProvider(Providers.of(config.getString("replication", null, "eventBrokerGroupId"))); listener().to(EventsBrokerMessageConsumer.class); } }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/event/EventsBrokerMessageConsumer.java b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/event/EventsBrokerMessageConsumer.java index d06681b..13b038e 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/replication/pull/event/EventsBrokerMessageConsumer.java +++ b/src/main/java/com/googlesource/gerrit/plugins/replication/pull/event/EventsBrokerMessageConsumer.java
@@ -14,9 +14,13 @@ package com.googlesource.gerrit.plugins.replication.pull.event; +import static com.googlesource.gerrit.plugins.replication.pull.event.EventsBrokerConsumerModule.STREAM_EVENTS_GROUP_ID; import static com.googlesource.gerrit.plugins.replication.pull.event.EventsBrokerConsumerModule.STREAM_EVENTS_TOPIC_NAME; import com.gerritforge.gerrit.eventbroker.BrokerApi; +import com.gerritforge.gerrit.eventbroker.ExtendedBrokerApi; +import com.google.common.flogger.FluentLogger; +import com.google.gerrit.common.Nullable; import com.google.gerrit.extensions.events.LifecycleListener; import com.google.gerrit.extensions.registration.DynamicItem; import com.google.gerrit.extensions.restapi.AuthException; @@ -29,22 +33,26 @@ public class EventsBrokerMessageConsumer implements Consumer<Event>, LifecycleListener { + private static final FluentLogger logger = FluentLogger.forEnclosingClass(); private final DynamicItem<BrokerApi> eventsBroker; private final StreamEventListener eventListener; private final ShutdownState shutdownState; private final String eventsTopicName; + private final String groupId; @Inject public EventsBrokerMessageConsumer( DynamicItem<BrokerApi> eventsBroker, StreamEventListener eventListener, ShutdownState shutdownState, - @Named(STREAM_EVENTS_TOPIC_NAME) String eventsTopicName) { + @Named(STREAM_EVENTS_TOPIC_NAME) String eventsTopicName, + @Nullable @Named(STREAM_EVENTS_GROUP_ID) String groupId) { this.eventsBroker = eventsBroker; this.eventListener = eventListener; this.shutdownState = shutdownState; this.eventsTopicName = eventsTopicName; + this.groupId = groupId; } @Override @@ -59,7 +67,21 @@ @Override public void start() { - eventsBroker.get().receiveAsync(eventsTopicName, this); + BrokerApi brokerApi = eventsBroker.get(); + if (groupId == null) { + brokerApi.receiveAsync(eventsTopicName, this); + return; + } + + if (!(eventsBroker instanceof ExtendedBrokerApi)) { + throw new IllegalArgumentException( + String.format( + "Failed to load the pull-replication plugin: %s does not support the custom group-id '%s'.\n" + + "Remove replication.eventBrokerGroupId from replication.config or install a different event-broker plugin.", + eventsBroker.getClass(), groupId)); + } + + ((ExtendedBrokerApi) brokerApi).receiveAsync(eventsTopicName, groupId, this); } @Override
diff --git a/src/main/resources/Documentation/config.md b/src/main/resources/Documentation/config.md index 6287773..0c3e02c 100644 --- a/src/main/resources/Documentation/config.md +++ b/src/main/resources/Documentation/config.md
@@ -222,6 +222,16 @@ Default: unset +replication.eventBrokerGroupId +: Optional group id for consumers. Used for keeping the offset of the + the last acked message on the topic with a specific group id; supported + only if the event-broker plugin supports the ExtendedBrokerApi + interface. + When not set, the messages received are acked using a global group id + shared by all subscribers in Gerrit. + + Default: unset + replication.maxConnectionsPerRoute : Maximum number of HTTP connections per one HTTP route.
diff --git a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/event/EventsBrokerMessageConsumerTest.java b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/event/EventsBrokerMessageConsumerTest.java index 9640b82..8739d4b 100644 --- a/src/test/java/com/googlesource/gerrit/plugins/replication/pull/event/EventsBrokerMessageConsumerTest.java +++ b/src/test/java/com/googlesource/gerrit/plugins/replication/pull/event/EventsBrokerMessageConsumerTest.java
@@ -49,7 +49,7 @@ shutdownState = new ShutdownState(); objectUnderTest = new EventsBrokerMessageConsumer( - eventsBrokerDynamicItem, eventListener, shutdownState, "topicName"); + eventsBrokerDynamicItem, eventListener, shutdownState, "topicName", null); } @Test