This is an automated email from the ASF dual-hosted git repository. chibenwa pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/james-project.git
commit cd70ea729d4cafea9751443232d538b03610bb51 Author: Benoit TELLIER <[email protected]> AuthorDate: Wed Sep 2 15:34:50 2026 +0200 [PERF] eventbus.publishOnNoGroups: skip group publishing when no group is registered --- .../backends/rabbitmq/RabbitMQConfiguration.java | 27 ++++++-- .../rabbitmq/RabbitMQConfigurationTest.java | 26 ++++++++ .../servers/partials/configure/rabbitmq.adoc | 6 +- .../org/apache/james/events/EventDispatcher.java | 16 ++++- .../org/apache/james/events/RabbitMQEventBus.java | 4 +- .../apache/james/events/RabbitMQEventBusTest.java | 75 ++++++++++++++++++++++ 6 files changed, 146 insertions(+), 8 deletions(-) diff --git a/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/RabbitMQConfiguration.java b/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/RabbitMQConfiguration.java index 645ea32648..42715eeb63 100644 --- a/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/RabbitMQConfiguration.java +++ b/backends-common/rabbitmq/src/main/java/org/apache/james/backends/rabbitmq/RabbitMQConfiguration.java @@ -313,6 +313,7 @@ public class RabbitMQConfiguration { private static final String EVENT_BUS_NOTIFICATION_QUEUE_AUTO_DELETE = "notification.queue.autoDelete"; private static final String EVENT_BUS_PUBLISH_CONFIRM_ENABLED = "event.bus.publish.confirm.enabled"; private static final String EVENT_BUS_PROPAGATE_DISPATCH_ERROR = "event.bus.propagate.dispatch.error"; + private static final String EVENT_BUS_PUBLISH_ON_NO_GROUPS = "eventbus.publishOnNoGroups"; private static final String TASK_QUEUE_CONSUMER_TIMEOUT = "task.queue.consumer.timeout"; private static final String VHOST = "vhost"; @@ -415,6 +416,7 @@ public class RabbitMQConfiguration { private Optional<Boolean> eventBusNotificationDurabilityEnabled; private Optional<Boolean> eventBusNotificationQueueAutoDelete; private Optional<Boolean> eventBusPropagateDispatchError; + private Optional<Boolean> eventBusPublishOnNoGroups; private Optional<String> vhost; private Optional<Duration> taskQueueConsumerTimeout; @@ -443,6 +445,7 @@ public class RabbitMQConfiguration { this.vhost = Optional.empty(); this.taskQueueConsumerTimeout = Optional.empty(); this.eventBusPropagateDispatchError = Optional.empty(); + this.eventBusPublishOnNoGroups = Optional.empty(); } public Builder maxRetries(int maxRetries) { @@ -535,6 +538,11 @@ public class RabbitMQConfiguration { return this; } + public Builder eventBusPublishOnNoGroups(Boolean eventBusPublishOnNoGroups) { + this.eventBusPublishOnNoGroups = Optional.ofNullable(eventBusPublishOnNoGroups); + return this; + } + public Builder eventBusPropagateDispatchError(Boolean eventBusPropagateDispatchError) { this.eventBusPropagateDispatchError = Optional.ofNullable(eventBusPropagateDispatchError); return this; @@ -598,7 +606,8 @@ public class RabbitMQConfiguration { eventBusNotificationQueueAutoDelete.orElse(Constants.AUTO_DELETE), vhost, taskQueueConsumerTimeout.orElse(DEFAULT_TASK_QUEUE_CONSUMER_TIMEOUT), - eventBusPropagateDispatchError.orElse(true)); + eventBusPropagateDispatchError.orElse(true), + eventBusPublishOnNoGroups.orElse(true)); } private List<Host> hostsDefaultingToUri() { @@ -675,6 +684,7 @@ public class RabbitMQConfiguration { .eventBusNotificationQueueAutoDelete(configuration.getBoolean(EVENT_BUS_NOTIFICATION_QUEUE_AUTO_DELETE, null)) .eventBusPublishConfirmEnabled(configuration.getBoolean(EVENT_BUS_PUBLISH_CONFIRM_ENABLED, null)) .eventBusPropagateDispatchError(configuration.getBoolean(EVENT_BUS_PROPAGATE_DISPATCH_ERROR, null)) + .eventBusPublishOnNoGroups(configuration.getBoolean(EVENT_BUS_PUBLISH_ON_NO_GROUPS, null)) .vhost(vhost) .taskQueueConsumerTimeout(taskQueueConsumerTimeout) .build(); @@ -752,6 +762,7 @@ public class RabbitMQConfiguration { private final Optional<String> vhost; private final Duration taskQueueConsumerTimeout; private final boolean eventBusPropagateDispatchError; + private final boolean eventBusPublishOnNoGroups; private RabbitMQConfiguration(URI uri, URI managementUri, ManagementCredentials managementCredentials, int maxRetries, int minDelayInMs, int connectionTimeoutInMs, int channelRpcTimeoutInMs, int handshakeTimeoutInMs, int shutdownTimeoutInMs, @@ -759,7 +770,8 @@ public class RabbitMQConfiguration { boolean useQuorumQueues, Optional<Integer> quorumQueueDeliveryLimit, int quorumQueueReplicationFactor, List<Host> hosts, Optional<Long> queueTTL, boolean eventBusPublishConfirmEnabled, boolean eventBusNotificationDurabilityEnabled, boolean eventBusNotificationQueueAutoDelete, - Optional<String> vhost, Duration taskQueueConsumerTimeout, boolean eventBusPropagateDispatchError) { + Optional<String> vhost, Duration taskQueueConsumerTimeout, boolean eventBusPropagateDispatchError, + boolean eventBusPublishOnNoGroups) { this.uri = uri; this.managementUri = managementUri; this.managementCredentials = managementCredentials; @@ -784,6 +796,7 @@ public class RabbitMQConfiguration { this.vhost = vhost; this.taskQueueConsumerTimeout = taskQueueConsumerTimeout; this.eventBusPropagateDispatchError = eventBusPropagateDispatchError; + this.eventBusPublishOnNoGroups = eventBusPublishOnNoGroups; } public URI getUri() { @@ -904,6 +917,10 @@ public class RabbitMQConfiguration { return quorumQueueReplicationFactor; } + public boolean eventBusPublishOnNoGroups() { + return eventBusPublishOnNoGroups; + } + public boolean eventBusPropagateDispatchError() { return eventBusPropagateDispatchError; } @@ -936,7 +953,8 @@ public class RabbitMQConfiguration { && Objects.equals(this.eventBusNotificationQueueAutoDelete, that.eventBusNotificationQueueAutoDelete) && Objects.equals(this.vhost, that.vhost) && Objects.equals(this.taskQueueConsumerTimeout, that.taskQueueConsumerTimeout) - && Objects.equals(this.eventBusPropagateDispatchError, that.eventBusPropagateDispatchError); + && Objects.equals(this.eventBusPropagateDispatchError, that.eventBusPropagateDispatchError) + && Objects.equals(this.eventBusPublishOnNoGroups, that.eventBusPublishOnNoGroups); } return false; } @@ -945,6 +963,7 @@ public class RabbitMQConfiguration { public final int hashCode() { return Objects.hash(uri, managementUri, maxRetries, minDelayInMs, connectionTimeoutInMs, quorumQueueReplicationFactor, quorumQueueDeliveryLimit, useQuorumQueues, hosts, channelRpcTimeoutInMs, handshakeTimeoutInMs, shutdownTimeoutInMs, networkRecoveryIntervalInMs, managementCredentials, useSsl, useSslManagement, - sslConfiguration, queueTTL, eventBusPublishConfirmEnabled, eventBusNotificationDurabilityEnabled, eventBusNotificationQueueAutoDelete, vhost, taskQueueConsumerTimeout, eventBusPropagateDispatchError); + sslConfiguration, queueTTL, eventBusPublishConfirmEnabled, eventBusNotificationDurabilityEnabled, eventBusNotificationQueueAutoDelete, vhost, taskQueueConsumerTimeout, eventBusPropagateDispatchError, + eventBusPublishOnNoGroups); } } diff --git a/backends-common/rabbitmq/src/test/java/org/apache/james/backends/rabbitmq/RabbitMQConfigurationTest.java b/backends-common/rabbitmq/src/test/java/org/apache/james/backends/rabbitmq/RabbitMQConfigurationTest.java index df747723d1..2aebe342dc 100644 --- a/backends-common/rabbitmq/src/test/java/org/apache/james/backends/rabbitmq/RabbitMQConfigurationTest.java +++ b/backends-common/rabbitmq/src/test/java/org/apache/james/backends/rabbitmq/RabbitMQConfigurationTest.java @@ -598,6 +598,32 @@ class RabbitMQConfigurationTest { .isFalse(); } + @Test + void eventBusPublishOnNoGroupsShouldBeTrueByDefault() { + PropertiesConfiguration configuration = new PropertiesConfiguration(); + configuration.addProperty("uri", "amqp://james:james@rabbitmqhost:5672"); + configuration.addProperty("management.uri", "http://james:james@rabbitmqhost:15672/api/";); + configuration.addProperty("management.user", DEFAULT_USER); + configuration.addProperty("management.password", DEFAULT_PASSWORD_STRING); + + assertThat(RabbitMQConfiguration.from(configuration).eventBusPublishOnNoGroups()) + .isTrue(); + } + + @Test + void eventBusPublishOnNoGroupsShouldBeDisabledWhenConfiguredFalse() { + PropertiesConfiguration configuration = new PropertiesConfiguration(); + configuration.addProperty("uri", "amqp://james:james@rabbitmqhost:5672"); + configuration.addProperty("management.uri", "http://james:james@rabbitmqhost:15672/api/";); + configuration.addProperty("management.user", DEFAULT_USER); + configuration.addProperty("management.password", DEFAULT_PASSWORD_STRING); + + configuration.addProperty("eventbus.publishOnNoGroups", "false"); + + assertThat(RabbitMQConfiguration.from(configuration).eventBusPublishOnNoGroups()) + .isFalse(); + } + @Nested class ManagementCredentialsTest { @Test diff --git a/docs/modules/servers/partials/configure/rabbitmq.adoc b/docs/modules/servers/partials/configure/rabbitmq.adoc index 3408aba3a2..7d105d1f53 100644 --- a/docs/modules/servers/partials/configure/rabbitmq.adoc +++ b/docs/modules/servers/partials/configure/rabbitmq.adoc @@ -117,6 +117,10 @@ collected once it has stayed unused for the TTL, which also cleans up the queues | Whether to propagate errors back to the callers when eventbus fails to dispatch group events to RabbitMQ (then store the failed events in the event dead letters). Optional boolean, defaults to true. +| eventbus.publishOnNoGroups +| Whether to publish group events to RabbitMQ even when this node has no group listener registered. Optional boolean, +defaults to true. + | vhost | Optional string. This parameter is only a workaround to support invalid URIs containing character like '_'. You still need to specify the vhost in the uri parameter. @@ -167,4 +171,4 @@ Required at least RabbitMQ version 3.12 to have effect. This is used to avoid the task queue consumer (which could run very long tasks) being disconnected by RabbitMQ after the default acknowledgement timeout 30 minutes. References: https://www.rabbitmq.com/consumers.html#acknowledgement-timeout. -|=== \ No newline at end of file +|=== diff --git a/event-bus/distributed/src/main/java/org/apache/james/events/EventDispatcher.java b/event-bus/distributed/src/main/java/org/apache/james/events/EventDispatcher.java index 92864f705a..75f232d958 100644 --- a/event-bus/distributed/src/main/java/org/apache/james/events/EventDispatcher.java +++ b/event-bus/distributed/src/main/java/org/apache/james/events/EventDispatcher.java @@ -69,13 +69,15 @@ public class EventDispatcher { private final ListenerExecutor listenerExecutor; private final EventDeadLetters deadLetters; private final RabbitMQConfiguration configuration; + private final GroupRegistrationHandler groupRegistrationHandler; private final DispatchingFailureGroup dispatchingFailureGroup; EventDispatcher(NamingStrategy namingStrategy, EventBusId eventBusId, EventSerializer eventSerializer, Sender sender, LocalListenerRegistry localListenerRegistry, ListenerExecutor listenerExecutor, - EventDeadLetters deadLetters, RabbitMQConfiguration configuration) { + EventDeadLetters deadLetters, RabbitMQConfiguration configuration, + GroupRegistrationHandler groupRegistrationHandler) { this.namingStrategy = namingStrategy; this.eventSerializer = eventSerializer; this.sender = sender; @@ -89,6 +91,7 @@ public class EventDispatcher { this.listenerExecutor = listenerExecutor; this.deadLetters = deadLetters; this.configuration = configuration; + this.groupRegistrationHandler = groupRegistrationHandler; this.dispatchingFailureGroup = new DispatchingFailureGroup(namingStrategy.getEventBusName()); } @@ -188,6 +191,9 @@ public class EventDispatcher { } private Mono<Void> remoteGroupsDispatch(byte[] serializedEvent, Event event) { + if (shouldSkipGroupsDispatch()) { + return Mono.empty(); + } return remoteDispatchWithAcks(serializedEvent) .doOnError(ex -> LOGGER.error( "cannot dispatch event of type '{}' belonging '{}' with id '{}' to remote groups, store it into dead letter", @@ -200,6 +206,9 @@ public class EventDispatcher { } private Mono<Void> remoteGroupsDispatch(byte[] serializedEvent, List<Event> events) { + if (shouldSkipGroupsDispatch()) { + return Mono.empty(); + } return remoteDispatchWithAcks(serializedEvent) .onErrorResume(ex -> Flux.fromIterable(events) .map(event -> { @@ -214,6 +223,11 @@ public class EventDispatcher { .then(propagateErrorIfNeeded(ex))); } + private boolean shouldSkipGroupsDispatch() { + return !configuration.eventBusPublishOnNoGroups() + && groupRegistrationHandler.registeredGroups().isEmpty(); + } + private Mono<Void> propagateErrorIfNeeded(Throwable throwable) { if (configuration.eventBusPropagateDispatchError()) { return Mono.error(throwable); diff --git a/event-bus/distributed/src/main/java/org/apache/james/events/RabbitMQEventBus.java b/event-bus/distributed/src/main/java/org/apache/james/events/RabbitMQEventBus.java index 558866651d..7ee1adb22f 100644 --- a/event-bus/distributed/src/main/java/org/apache/james/events/RabbitMQEventBus.java +++ b/event-bus/distributed/src/main/java/org/apache/james/events/RabbitMQEventBus.java @@ -116,7 +116,7 @@ public class RabbitMQEventBus implements EventBus, Startable { LocalListenerRegistry localListenerRegistry = new LocalListenerRegistry(); keyRegistrationHandler = new KeyRegistrationHandler(namingStrategy, eventBusId, eventSerializer, sender, receiverProvider, routingKeyConverter, localListenerRegistry, listenerExecutor, configurations, metricFactory); groupRegistrationHandler = new GroupRegistrationHandler(namingStrategy, eventSerializer, channelPool, sender, receiverProvider, eventDeadLetters, listenerExecutor, configurations); - eventDispatcher = new EventDispatcher(namingStrategy, eventBusId, eventSerializer, sender, localListenerRegistry, listenerExecutor, eventDeadLetters, configurations.rabbitMQConfiguration()); + eventDispatcher = new EventDispatcher(namingStrategy, eventBusId, eventSerializer, sender, localListenerRegistry, listenerExecutor, eventDeadLetters, configurations.rabbitMQConfiguration(), groupRegistrationHandler); eventDispatcher.start(); keyRegistrationHandler.start(); @@ -137,7 +137,7 @@ public class RabbitMQEventBus implements EventBus, Startable { LocalListenerRegistry localListenerRegistry = new LocalListenerRegistry(); keyRegistrationHandler = new KeyRegistrationHandler(namingStrategy, eventBusId, eventSerializer, sender, receiverProvider, routingKeyConverter, localListenerRegistry, listenerExecutor, configurations, metricFactory); groupRegistrationHandler = new GroupRegistrationHandler(namingStrategy, eventSerializer, channelPool, sender, receiverProvider, eventDeadLetters, listenerExecutor, configurations); - eventDispatcher = new EventDispatcher(namingStrategy, eventBusId, eventSerializer, sender, localListenerRegistry, listenerExecutor, eventDeadLetters, configurations.rabbitMQConfiguration()); + eventDispatcher = new EventDispatcher(namingStrategy, eventBusId, eventSerializer, sender, localListenerRegistry, listenerExecutor, eventDeadLetters, configurations.rabbitMQConfiguration(), groupRegistrationHandler); keyRegistrationHandler.declareQueue(); diff --git a/event-bus/distributed/src/test/java/org/apache/james/events/RabbitMQEventBusTest.java b/event-bus/distributed/src/test/java/org/apache/james/events/RabbitMQEventBusTest.java index b1e531cd8a..ce12c5d628 100644 --- a/event-bus/distributed/src/test/java/org/apache/james/events/RabbitMQEventBusTest.java +++ b/event-bus/distributed/src/test/java/org/apache/james/events/RabbitMQEventBusTest.java @@ -53,6 +53,8 @@ import java.io.Closeable; import java.nio.charset.StandardCharsets; import java.time.Duration; import java.util.NoSuchElementException; +import java.util.Optional; +import java.util.UUID; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.IntStream; @@ -434,6 +436,79 @@ class RabbitMQEventBusTest implements GroupContract.SingleEventBusGroupContract, return eventSerializer.asEvent(new String(eventInBytes, StandardCharsets.UTF_8)).event(); } } + + } + + @Nested + class PublishOnNoGroupsTest { + private String probeQueueName; + private RabbitMQEventBus optimizedEventBus; + + @BeforeEach + void setUp() throws Exception { + probeQueueName = "test-publishOnNoGroups-" + UUID.randomUUID(); + Sender sender = rabbitMQExtension.getSender(); + sender.declareQueue(QueueSpecification.queue(probeQueueName) + .durable(!DURABLE) + .exclusive(!EXCLUSIVE) + .autoDelete(!AUTO_DELETE) + .arguments(NO_ARGUMENTS)) + .block(); + // Group dispatches are the ones published with the empty routing key + sender.bind(BindingSpecification.binding() + .exchange(TEST_NAMING_STRATEGY.exchange()) + .queue(probeQueueName) + .routingKey(EMPTY_ROUTING_KEY)) + .block(); + + optimizedEventBus = new RabbitMQEventBus(TEST_NAMING_STRATEGY, sender, rabbitMQExtension.getReceiverProvider(), + eventSerializer, routingKeyConverter, memoryEventDeadLetters, new RecordingMetricFactory(), + rabbitMQExtension.getRabbitChannelPool(), EventBusId.random(), + new RabbitMQEventBus.Configurations(rabbitMQExtension.getRabbitMQ().getConfigurationBuilder() + .eventBusPublishOnNoGroups(false) + .build(), EventBusTestFixture.RETRY_BACKOFF_CONFIGURATION)); + optimizedEventBus.start(); + } + + @AfterEach + void tearDown() { + optimizedEventBus.stop(); + rabbitMQExtension.getSender().delete(QueueSpecification.queue(probeQueueName)).block(); + } + + @Test + void dispatchShouldNotPublishToGroupsWhenNoGroupIsRegistered() { + optimizedEventBus.dispatch(EVENT, NO_KEYS).block(); + + assertThat(dequeueEventWithin(Duration.ofSeconds(2))).isEmpty(); + } + + @Test + void dispatchShouldPublishToGroupsWhenAGroupIsRegistered() { + optimizedEventBus.register(newListener(), GROUP_A); + + optimizedEventBus.dispatch(EVENT, NO_KEYS).block(); + + assertThat(dequeueEventWithin(Duration.ofSeconds(10))).contains(EVENT); + } + + @Test + void dispatchShouldPublishToGroupsWithoutRegistrationWhenOptionIsEnabled() { + // eventBus runs with the default configuration: `eventbus.publishOnNoGroups` is true + eventBus.dispatch(EVENT, NO_KEYS).block(); + + assertThat(dequeueEventWithin(Duration.ofSeconds(10))).contains(EVENT); + } + + private Optional<Event> dequeueEventWithin(Duration timeout) { + try (Receiver receiver = rabbitMQExtension.getReceiverProvider().createReceiver()) { + return Optional.ofNullable(receiver.consumeAutoAck(probeQueueName) + .next() + .timeout(timeout, Mono.empty()) + .block()) + .map(delivery -> eventSerializer.asEvent(new String(delivery.getBody(), StandardCharsets.UTF_8)).event()); + } + } } @Nested --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
