Add acknowledgement-aware consumer API to BrokerApi Introduce MessageAcknowledgement and AckAwareConsumer in the events-broker API. Allow consumers to receive messages together with an acknowledgement handle and explicitly acknowledge successful processing when the broker implementation requires it. Implementations are not required to be thread-safe, callers can inspect whether messages are auto-acknowledged, and acknowledgement failures are reported through a dedicated runtime exception. This change is NOT backwards-compatible: broker implementations must adapt their receiveAsync implementations to provide the new acknowledgement-aware API. Bug: Issue 454827132 Change-Id: I226ce4601527b167e9370cbff29b4dcadde4e3e6
diff --git a/src/main/java/com/gerritforge/gerrit/eventbroker/AckAwareConsumer.java b/src/main/java/com/gerritforge/gerrit/eventbroker/AckAwareConsumer.java new file mode 100644 index 0000000..2dd872d --- /dev/null +++ b/src/main/java/com/gerritforge/gerrit/eventbroker/AckAwareConsumer.java
@@ -0,0 +1,32 @@ +// 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; + +/** + * A consumer that accepts a message and its acknowledgement handle, allowing for explicit + * acknowledgement. + * + * @param <T> the type of the event input to the operation + */ +@FunctionalInterface +public interface AckAwareConsumer<T> { + /** + * Performs this operation on the given argument. + * + * @param t the input argument + * @param acknowledgement the message acknowledgement handle + */ + void accept(T t, MessageAcknowledgement acknowledgement); +}
diff --git a/src/main/java/com/gerritforge/gerrit/eventbroker/BrokerApi.java b/src/main/java/com/gerritforge/gerrit/eventbroker/BrokerApi.java index e8c336a..093cfd0 100644 --- a/src/main/java/com/gerritforge/gerrit/eventbroker/BrokerApi.java +++ b/src/main/java/com/gerritforge/gerrit/eventbroker/BrokerApi.java
@@ -18,7 +18,6 @@ import com.google.gerrit.common.Nullable; import com.google.gerrit.server.events.Event; import java.util.Set; -import java.util.function.Consumer; /** API for sending/receiving events through a message Broker. */ public interface BrokerApi { @@ -33,12 +32,14 @@ ListenableFuture<Boolean> send(String topic, Event message); /** - * Receive asynchronously a message from a topic. + * Receive asynchronously a message from a topic, using an acknowledgement-aware consumer. * * @param topic topic name - * @param consumer an operation that accepts and process a single message + * @param consumer an operation that accepts and processes a single message with acknowledgement + * support + * @since 3.15 */ - void receiveAsync(String topic, Consumer<Event> consumer); + void receiveAsync(String topic, AckAwareConsumer<Event> consumer); /** * Get the active subscribers @@ -66,14 +67,16 @@ void replayAllEvents(String topic); /** - * Receive asynchronously a message from a topic using a consumer's group id. + * Receive asynchronously a message from a topic using a consumer's group id, using an + * acknowledgement-aware consumer. * * @param topic topic name * @param groupId the group identifier that consumer belongs to for that topic - * @param consumer an operation that accepts and process a single message - * @since 3.10 + * @param consumer an operation that accepts and processes a single message with acknowledgement + * support + * @since 3.15 */ - void receiveAsync(String topic, String groupId, Consumer<Event> consumer); + void receiveAsync(String topic, String groupId, AckAwareConsumer<Event> consumer); /** * Get the active subscribers with their consumer's group id.
diff --git a/src/main/java/com/gerritforge/gerrit/eventbroker/InProcessBrokerApi.java b/src/main/java/com/gerritforge/gerrit/eventbroker/InProcessBrokerApi.java index 6f0692a..8e7d44e 100644 --- a/src/main/java/com/gerritforge/gerrit/eventbroker/InProcessBrokerApi.java +++ b/src/main/java/com/gerritforge/gerrit/eventbroker/InProcessBrokerApi.java
@@ -24,7 +24,6 @@ import com.google.gerrit.server.events.Event; import java.util.HashSet; import java.util.Set; -import java.util.function.Consumer; import java.util.stream.Collectors; public class InProcessBrokerApi implements BrokerApi { @@ -43,12 +42,12 @@ } @Override - public void receiveAsync(String topic, Consumer<Event> eventConsumer) { + public void receiveAsync(String topic, AckAwareConsumer<Event> eventConsumer) { topicSubscribers.add(topicSubscriber(topic, eventConsumer)); } @Override - public void receiveAsync(String topic, String groupId, Consumer<Event> eventConsumer) { + public void receiveAsync(String topic, String groupId, AckAwareConsumer<Event> eventConsumer) { topicSubscribersWithGroupId.add( topicSubscriberWithGroupId(groupId, topicSubscriber(topic, eventConsumer))); }
diff --git a/src/main/java/com/gerritforge/gerrit/eventbroker/MessageAcknowledgement.java b/src/main/java/com/gerritforge/gerrit/eventbroker/MessageAcknowledgement.java new file mode 100644 index 0000000..9d6f402 --- /dev/null +++ b/src/main/java/com/gerritforge/gerrit/eventbroker/MessageAcknowledgement.java
@@ -0,0 +1,49 @@ +// 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; + +/** + * Acknowledgement handle associated with the processing of a single message. + * + * <p>The {@link MessageAcknowledgement} allows a consumer to explicitly acknowledge that the + * currently processed message has been handled successfully. + */ +public interface MessageAcknowledgement { + + /** + * Explicitly acknowledges successful processing progress for the current message. + * + * <p>This method performs an immediate acknowledgement attempt. Implementations are expected to + * honour it without introducing extra batching or delay. Any batching policy belongs to the + * caller, which can decide when to invoke {@code ack()}. + * + * <p>Implementations are not required to be thread-safe. Unless documented otherwise by the + * concrete broker implementation, callers should invoke {@code ack()} from the same thread that + * received the message and must not call it concurrently. + * + * <p>Calling {@code ack()} when {@link #isAutoAck()} is {@code true} will fail with {@link + * IllegalStateException}. Failures during acknowledgement should surface as {@link + * MessageAcknowledgementException}. + */ + void ack(); + + /** + * Returns whether the current message is acknowledged automatically by the implementation. + * + * <p>When this method returns {@code true}, callers should not invoke {@link #ack()} because the + * implementation is already handling acknowledgement automatically. + */ + boolean isAutoAck(); +}
diff --git a/src/main/java/com/gerritforge/gerrit/eventbroker/MessageAcknowledgementException.java b/src/main/java/com/gerritforge/gerrit/eventbroker/MessageAcknowledgementException.java new file mode 100644 index 0000000..3d34557 --- /dev/null +++ b/src/main/java/com/gerritforge/gerrit/eventbroker/MessageAcknowledgementException.java
@@ -0,0 +1,26 @@ +// 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; + +/** Unchecked exception thrown when a broker acknowledgement attempt fails. */ +public class MessageAcknowledgementException extends RuntimeException { + public MessageAcknowledgementException(String message) { + super(message); + } + + public MessageAcknowledgementException(String message, Throwable cause) { + super(message, cause); + } +}
diff --git a/src/main/java/com/gerritforge/gerrit/eventbroker/TopicSubscriber.java b/src/main/java/com/gerritforge/gerrit/eventbroker/TopicSubscriber.java index 3fecc48..65aa4cc 100644 --- a/src/main/java/com/gerritforge/gerrit/eventbroker/TopicSubscriber.java +++ b/src/main/java/com/gerritforge/gerrit/eventbroker/TopicSubscriber.java
@@ -1,4 +1,4 @@ -// Copyright (C) 2019 The Android Open Source Project +// 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. @@ -16,15 +16,14 @@ import com.google.auto.value.AutoValue; import com.google.gerrit.server.events.Event; -import java.util.function.Consumer; @AutoValue public abstract class TopicSubscriber { - public static TopicSubscriber topicSubscriber(String topic, Consumer<Event> consumer) { + public static TopicSubscriber topicSubscriber(String topic, AckAwareConsumer<Event> consumer) { return new AutoValue_TopicSubscriber(topic, consumer); } public abstract String topic(); - public abstract Consumer<Event> consumer(); + public abstract AckAwareConsumer<Event> consumer(); }
diff --git a/src/test/java/com/gerritforge/gerrit/eventbroker/InProcessBrokerApiTest.java b/src/test/java/com/gerritforge/gerrit/eventbroker/InProcessBrokerApiTest.java index eee2bdf..5ddf44f 100644 --- a/src/test/java/com/gerritforge/gerrit/eventbroker/InProcessBrokerApiTest.java +++ b/src/test/java/com/gerritforge/gerrit/eventbroker/InProcessBrokerApiTest.java
@@ -21,14 +21,13 @@ import com.google.gerrit.server.events.Event; import java.util.Set; import java.util.UUID; -import java.util.function.Consumer; import org.junit.Before; import org.junit.Test; public class InProcessBrokerApiTest { public static final int SEND_FUTURE_TIMEOUT = 1; - Consumer<Event> eventConsumer; + AckAwareConsumer<Event> eventConsumer; BrokerApi brokerApiUnderTest; UUID instanceId = UUID.randomUUID(); @@ -46,7 +45,7 @@ @Test public void shouldRegisterConsumerPerTopic() { - Consumer<Event> secondConsumer = mockEventConsumer(); + AckAwareConsumer<Event> secondConsumer = mockEventConsumer(); brokerApiUnderTest.receiveAsync("topic", eventConsumer); brokerApiUnderTest.receiveAsync("topic2", secondConsumer); assertThat(brokerApiUnderTest.topicSubscribers().size()).isEqualTo(2); @@ -54,10 +53,10 @@ @Test public void shouldReturnMapOfConsumersPerTopic() { - Consumer<Event> firstConsumerTopicA = mockEventConsumer(); + AckAwareConsumer<Event> firstConsumerTopicA = mockEventConsumer(); - Consumer<Event> secondConsumerTopicA = mockEventConsumer(); - Consumer<Event> thirdConsumerTopicB = mockEventConsumer(); + AckAwareConsumer<Event> secondConsumerTopicA = mockEventConsumer(); + AckAwareConsumer<Event> thirdConsumerTopicB = mockEventConsumer(); brokerApiUnderTest.receiveAsync("TopicA", firstConsumerTopicA); brokerApiUnderTest.receiveAsync("TopicA", secondConsumerTopicA); @@ -76,7 +75,7 @@ @Test public void shouldDeliverAsynchronouslyEventToAllRegisteredConsumers() { - Consumer<Event> secondConsumer = mockEventConsumer(); + AckAwareConsumer<Event> secondConsumer = mockEventConsumer(); brokerApiUnderTest.receiveAsync("topic", eventConsumer); brokerApiUnderTest.receiveAsync("topic", secondConsumer); assertThat(brokerApiUnderTest.topicSubscribers().size()).isEqualTo(2); @@ -95,7 +94,7 @@ brokerApiUnderTest.receiveAsync("topic", eventConsumer); assertThat(brokerApiUnderTest.topicSubscribers()).isNotEmpty(); - Consumer<Event> newConsumer = mockEventConsumer(); + AckAwareConsumer<Event> newConsumer = mockEventConsumer(); brokerApiUnderTest.disconnect(); assertThat(brokerApiUnderTest.topicSubscribers()).isEmpty(); @@ -117,13 +116,13 @@ UnsupportedOperationException.class, () -> brokerApiUnderTest.replayAllEvents("topic")); } - private static class Subscriber<T> implements Consumer<T> { + private static class Subscriber<T> implements AckAwareConsumer<T> { @Override - public void accept(T eventMessage) {} + public void accept(T t, MessageAcknowledgement acknowledgement) {} } - private <T> Consumer<T> mockEventConsumer() { + private <T> AckAwareConsumer<T> mockEventConsumer() { return new Subscriber<>(); } }