Add partition-aware event broker API Extend BrokerApi with partition-specific receiveAsync method. Add EventsBrokerConfiguration to read per-topic partition values and the event property used for partition selection, defaulting to the event type when the property is not configured. Bug: Issue 514246373 Change-Id: I12c98a338298133e226fb927c998791d267a8d1c
diff --git a/README.md b/README.md index 3ada0cc..a28bc03 100644 --- a/README.md +++ b/README.md
@@ -73,6 +73,47 @@ Note: To avoid message duplication Stream Events Publisher uses [gerrit.instanceId](https://gerrit-review.googlesource.com/Documentation/config-gerrit.html) and Event.instanceId to filter out forwarded events. +### Partition-aware Topics + +Broker clients can use partition-aware subscriptions through +`BrokerApi.receiveAsyncWithPartition(...)`, passing one of the configured +logical partition values for the topic. + +The partitions available for a topic, and the event property used to choose a +partition, are read from the plugin configuration file, for example +`$site_path/etc/events-broker.config`: + +```ini +[topic "stream-events"] + partitionValue = change-index + partitionValue = account-index + partitionEventProperty = eventType +``` + +The supported settings are: + +* `topic.<topic-name>.partitionValue`: zero or more partition values for the + topic. Repeat the setting to configure multiple partitions. +* `topic.<topic-name>.partitionEventProperty`: optional event property used by + the broker implementation to select the partition. When omitted, it defaults + to `type`. + +The order of `partitionValue` entries matters. Broker implementations may use +each value's position when mapping logical partitions to backend-specific +routing, so changing the order can change where events are published or +consumed. + +The target broker topic is expected to have at least the partitions configured +through `partitionValue`, so events can be published to the matching partition +accordingly. + +Topics without a matching `[topic "<topic-name>"]` subsection have no configured +partition metadata. Topics with a subsection but no `partitionValue` configured +have an empty partition list. +In both cases, implementations should treat the topic as non-partition-aware: +publishing should fall back to the normal broker behavior, while partition-specific subscription +cannot resolve a logical partition and fails if requested. + ### Broker Metrics When `StreamEventPublisher` is used user can optionally bind an implementation of
diff --git a/src/main/java/com/gerritforge/gerrit/eventbroker/BrokerApi.java b/src/main/java/com/gerritforge/gerrit/eventbroker/BrokerApi.java index 2e7d4de..aa36aa5 100644 --- a/src/main/java/com/gerritforge/gerrit/eventbroker/BrokerApi.java +++ b/src/main/java/com/gerritforge/gerrit/eventbroker/BrokerApi.java
@@ -25,6 +25,9 @@ /** * Send a message to a topic. * + * <p>When publishing to a partition-aware topic, implementations are expected to resolve the + * partition for {@code message} and honor that partition when sending the event to the broker. + * * @param topic topic name * @param message to be send to the topic * @return a future that returns when the message has been sent. @@ -79,6 +82,22 @@ void receiveAsync(String topic, String groupId, AckAwareConsumer<Event> consumer); /** + * Receive asynchronously messages from a specific partition of a topic using a consumer's group + * id, using an acknowledgement-aware consumer. + * + * <p>The supplied {@code partition} is a logical partition value from the topic's configured + * partition values. Broker implementations map it to the broker-native partition identifier. + * + * @param topic topic name + * @param partition logical partition value to consume from + * @param groupId the group identifier that consumer belongs to for that topic + * @param consumer an operation that accepts and processes a single message with acknowledgement + * support + */ + void receiveAsyncWithPartition( + String topic, String partition, String groupId, AckAwareConsumer<Event> consumer); + + /** * Get the active subscribers with their consumer's group id. * * @return {@link Set} of the topics subscribers using a consumer's group id.
diff --git a/src/main/java/com/gerritforge/gerrit/eventbroker/EventsBrokerConfiguration.java b/src/main/java/com/gerritforge/gerrit/eventbroker/EventsBrokerConfiguration.java new file mode 100644 index 0000000..d4cb48c --- /dev/null +++ b/src/main/java/com/gerritforge/gerrit/eventbroker/EventsBrokerConfiguration.java
@@ -0,0 +1,84 @@ +// Copyright (C) 2026 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.gerritforge.gerrit.eventbroker; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import com.google.common.collect.ImmutableSet; +import com.google.gerrit.server.config.PluginConfigFactory; +import com.google.inject.Inject; +import com.google.inject.Singleton; +import java.util.List; +import java.util.Optional; +import org.eclipse.jgit.lib.Config; + +@Singleton +public class EventsBrokerConfiguration { + static final String EVENTS_BROKER_FILE_NAME = "events-broker"; + static final String PARTITION_EVENT_PROPERTY_FIELD = "partitionEventProperty"; + static final String PARTITION_VALUE_FIELD = "partitionValue"; + static final String TOPIC_SECTION = "topic"; + + static final String DEFAULT_PARTITION_EVENT_PROPERTY = "type"; + + private final ImmutableMap<String, ImmutableList<String>> topicToPartitions; + private final ImmutableMap<String, String> topicToEventProperty; + + @Inject + public EventsBrokerConfiguration(PluginConfigFactory configFactory) { + // This class is reused by broker implementations, so @PluginName would resolve to the + // implementation plugin name (for example events-kafka) instead of the shared broker config. + Config config = configFactory.getGlobalPluginConfig(EVENTS_BROKER_FILE_NAME); + ImmutableMap.Builder<String, ImmutableList<String>> partitionsByTopic = ImmutableMap.builder(); + ImmutableMap.Builder<String, String> eventPropertyByTopic = ImmutableMap.builder(); + + for (String subsection : config.getSubsections(TOPIC_SECTION)) { + ImmutableList<String> partitionValue = + ImmutableList.copyOf( + config.getStringList(TOPIC_SECTION, subsection, PARTITION_VALUE_FIELD)); + if (partitionValue.size() != ImmutableSet.copyOf(partitionValue).size()) { + throw new IllegalArgumentException( + String.format("Duplicate partitionValue entries configured for topic %s", subsection)); + } + + partitionsByTopic.put(subsection, partitionValue); + eventPropertyByTopic.put( + subsection, + Optional.ofNullable( + config.getString(TOPIC_SECTION, subsection, PARTITION_EVENT_PROPERTY_FIELD)) + .orElse(DEFAULT_PARTITION_EVENT_PROPERTY)); + } + topicToPartitions = partitionsByTopic.build(); + topicToEventProperty = eventPropertyByTopic.build(); + } + + /** + * Returns the configured logical partition values for the topic. + * + * <p>Returns an empty list when the topic has no configured partition values. + */ + public List<String> getPartitionsForTopic(String topic) { + return topicToPartitions.getOrDefault(topic, ImmutableList.of()); + } + + /** + * Returns the event property used to pick a logical partition for the topic. + * + * <p>Returns {@link Optional#empty()} when the topic has no partition metadata. + */ + public Optional<String> getEventPropertyForTopic(String topic) { + return Optional.ofNullable(topicToEventProperty.get(topic)); + } +}
diff --git a/src/main/java/com/gerritforge/gerrit/eventbroker/InProcessBrokerApi.java b/src/main/java/com/gerritforge/gerrit/eventbroker/InProcessBrokerApi.java index 5ad4ac7..65e3887 100644 --- a/src/main/java/com/gerritforge/gerrit/eventbroker/InProcessBrokerApi.java +++ b/src/main/java/com/gerritforge/gerrit/eventbroker/InProcessBrokerApi.java
@@ -23,6 +23,7 @@ import com.google.gerrit.common.Nullable; import com.google.gerrit.server.events.Event; import java.util.HashSet; +import java.util.Optional; import java.util.Set; import java.util.stream.Collectors; @@ -53,6 +54,14 @@ } @Override + public void receiveAsyncWithPartition( + String topic, String partition, String groupId, AckAwareConsumer<Event> consumer) { + topicSubscribersWithGroupId.add( + topicSubscriberWithGroupId( + groupId, topicSubscriber(topic, consumer), Optional.of(partition))); + } + + @Override public Set<TopicSubscriber> topicSubscribers() { return ImmutableSet.copyOf(topicSubscribers); }
diff --git a/src/main/java/com/gerritforge/gerrit/eventbroker/TopicSubscriberWithGroupId.java b/src/main/java/com/gerritforge/gerrit/eventbroker/TopicSubscriberWithGroupId.java index 3ce2d56..829705e 100644 --- a/src/main/java/com/gerritforge/gerrit/eventbroker/TopicSubscriberWithGroupId.java +++ b/src/main/java/com/gerritforge/gerrit/eventbroker/TopicSubscriberWithGroupId.java
@@ -15,15 +15,24 @@ package com.gerritforge.gerrit.eventbroker; import com.google.auto.value.AutoValue; +import java.util.Optional; @AutoValue public abstract class TopicSubscriberWithGroupId { public static TopicSubscriberWithGroupId topicSubscriberWithGroupId( String groupId, TopicSubscriber topicSubscriber) { - return new AutoValue_TopicSubscriberWithGroupId(groupId, topicSubscriber); + return topicSubscriberWithGroupId(groupId, topicSubscriber, Optional.empty()); + } + + public static TopicSubscriberWithGroupId topicSubscriberWithGroupId( + String groupId, TopicSubscriber topicSubscriber, Optional<String> partition) { + return new AutoValue_TopicSubscriberWithGroupId(groupId, topicSubscriber, partition); } public abstract String groupId(); public abstract TopicSubscriber topicSubscriber(); + + /** Returns the logical partition value when this is a partition-aware subscription. */ + public abstract Optional<String> partition(); }
diff --git a/src/test/java/com/gerritforge/gerrit/eventbroker/EventsBrokerConfigurationTest.java b/src/test/java/com/gerritforge/gerrit/eventbroker/EventsBrokerConfigurationTest.java new file mode 100644 index 0000000..9dea335 --- /dev/null +++ b/src/test/java/com/gerritforge/gerrit/eventbroker/EventsBrokerConfigurationTest.java
@@ -0,0 +1,158 @@ +// Copyright (C) 2026 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.gerritforge.gerrit.eventbroker; + +import static com.google.common.truth.Truth.assertThat; +import static com.google.gerrit.testing.GerritJUnit.assertThrows; +import static org.mockito.Mockito.when; + +import com.google.gerrit.server.config.PluginConfigFactory; +import java.util.Arrays; +import java.util.List; +import java.util.Optional; +import org.eclipse.jgit.lib.Config; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnitRunner; + +@RunWith(MockitoJUnitRunner.class) +public class EventsBrokerConfigurationTest { + private static final String TOPIC = "gerrit-index"; + private static final String EVENT_TYPE = "eventType"; + + @Mock private PluginConfigFactory configFactory; + + @Test + public void shouldReturnConfiguredPartitionsForTopic() { + Config config = new Config(); + config.setStringList( + EventsBrokerConfiguration.TOPIC_SECTION, + TOPIC, + EventsBrokerConfiguration.PARTITION_VALUE_FIELD, + Arrays.asList("change-index", "account-index")); + + EventsBrokerConfiguration configuration = configuration(config); + + assertThat(configuration.getPartitionsForTopic(TOPIC)) + .containsExactly("change-index", "account-index") + .inOrder(); + } + + @Test + public void shouldReturnConfiguredEventPropertyForTopic() { + Config config = new Config(); + config.setString( + EventsBrokerConfiguration.TOPIC_SECTION, + TOPIC, + EventsBrokerConfiguration.PARTITION_EVENT_PROPERTY_FIELD, + EVENT_TYPE); + + EventsBrokerConfiguration configuration = configuration(config); + + assertThat(configuration.getEventPropertyForTopic(TOPIC)).isEqualTo(Optional.of(EVENT_TYPE)); + } + + @Test + public void shouldReturnDefaultEventPropertyWhenNotConfiguredForTopic() { + Config config = new Config(); + config.setStringList( + EventsBrokerConfiguration.TOPIC_SECTION, + TOPIC, + EventsBrokerConfiguration.PARTITION_VALUE_FIELD, + List.of("change-index")); + + EventsBrokerConfiguration configuration = configuration(config); + + assertThat(configuration.getEventPropertyForTopic(TOPIC)) + .isEqualTo(Optional.of(EventsBrokerConfiguration.DEFAULT_PARTITION_EVENT_PROPERTY)); + } + + @Test + public void shouldReturnEmptyPartitionsWhenTopicHasNoPartitionValues() { + Config config = new Config(); + config.setString( + EventsBrokerConfiguration.TOPIC_SECTION, + TOPIC, + EventsBrokerConfiguration.PARTITION_EVENT_PROPERTY_FIELD, + EVENT_TYPE); + + EventsBrokerConfiguration configuration = configuration(config); + + assertThat(configuration.getPartitionsForTopic(TOPIC)).isEmpty(); + } + + @Test + public void shouldReturnEmptyValuesForUnknownTopic() { + EventsBrokerConfiguration configuration = configuration(new Config()); + + assertThat(configuration.getPartitionsForTopic(TOPIC)).isEmpty(); + assertThat(configuration.getEventPropertyForTopic(TOPIC)).isEmpty(); + } + + @Test + public void shouldRejectDuplicatePartitionValues() { + Config config = new Config(); + config.setStringList( + EventsBrokerConfiguration.TOPIC_SECTION, + TOPIC, + EventsBrokerConfiguration.PARTITION_VALUE_FIELD, + Arrays.asList("change-index", "change-index")); + + IllegalArgumentException thrown = + assertThrows(IllegalArgumentException.class, () -> configuration(config)); + + assertThat(thrown) + .hasMessageThat() + .isEqualTo("Duplicate partitionValue entries configured for topic " + TOPIC); + } + + @Test + public void shouldReturnConfigurationForMultipleTopics() { + String otherTopic = "stream-events"; + Config config = new Config(); + config.setStringList( + EventsBrokerConfiguration.TOPIC_SECTION, + TOPIC, + EventsBrokerConfiguration.PARTITION_VALUE_FIELD, + List.of("change-index", "account-index")); + config.setString( + EventsBrokerConfiguration.TOPIC_SECTION, + TOPIC, + EventsBrokerConfiguration.PARTITION_EVENT_PROPERTY_FIELD, + EVENT_TYPE); + config.setStringList( + EventsBrokerConfiguration.TOPIC_SECTION, + otherTopic, + EventsBrokerConfiguration.PARTITION_VALUE_FIELD, + List.of("project-updated")); + + EventsBrokerConfiguration configuration = configuration(config); + + assertThat(configuration.getPartitionsForTopic(TOPIC)) + .containsExactly("change-index", "account-index") + .inOrder(); + assertThat(configuration.getEventPropertyForTopic(TOPIC)).isEqualTo(Optional.of(EVENT_TYPE)); + assertThat(configuration.getPartitionsForTopic(otherTopic)).containsExactly("project-updated"); + assertThat(configuration.getEventPropertyForTopic(otherTopic)) + .isEqualTo(Optional.of(EventsBrokerConfiguration.DEFAULT_PARTITION_EVENT_PROPERTY)); + } + + private EventsBrokerConfiguration configuration(Config config) { + when(configFactory.getGlobalPluginConfig(EventsBrokerConfiguration.EVENTS_BROKER_FILE_NAME)) + .thenReturn(config); + return new EventsBrokerConfiguration(configFactory); + } +}
diff --git a/src/test/java/com/gerritforge/gerrit/eventbroker/InProcessBrokerApiTest.java b/src/test/java/com/gerritforge/gerrit/eventbroker/InProcessBrokerApiTest.java index bc7281c..4497672 100644 --- a/src/test/java/com/gerritforge/gerrit/eventbroker/InProcessBrokerApiTest.java +++ b/src/test/java/com/gerritforge/gerrit/eventbroker/InProcessBrokerApiTest.java
@@ -15,10 +15,12 @@ package com.gerritforge.gerrit.eventbroker; import static com.gerritforge.gerrit.eventbroker.TopicSubscriber.topicSubscriber; +import static com.gerritforge.gerrit.eventbroker.TopicSubscriberWithGroupId.topicSubscriberWithGroupId; import static com.google.common.truth.Truth.assertThat; import static com.google.gerrit.testing.GerritJUnit.assertThrows; import com.google.gerrit.server.events.Event; +import java.util.Optional; import java.util.Set; import org.junit.Before; import org.junit.Test; @@ -85,6 +87,16 @@ } @Test + public void shouldRegisterPartitionConsumerWithGroupId() { + brokerApiUnderTest.receiveAsyncWithPartition("topic", "partition", "group", eventConsumer); + + assertThat(brokerApiUnderTest.topicSubscribersWithGroupId()) + .containsExactly( + topicSubscriberWithGroupId( + "group", topicSubscriber("topic", eventConsumer), Optional.of("partition"))); + } + + @Test public void shouldReconnectSubscribers() { brokerApiUnderTest.receiveAsync("topic", eventConsumer); assertThat(brokerApiUnderTest.topicSubscribers()).isNotEmpty();