[FIX] GroupRegistrationHandler's scheduler should be final
And be initialized in the constructor.
Given the RabbitMQEventBus beans are limited and the group use case is soon or late being used, we do not need lazy initialization for the scheduler IMO.
Otherwise, James could restart the `GroupRegistrationHandler` while it is not started yet (e.g. a RabbitMQ node is down before any group listener is registered) which leads to `java.lang.NullPointerException: scheduler` and fail the `RabbitEventBusConsumerHealthCheck`.
diff --git a/event-bus/distributed/src/main/java/org/apache/james/events/GroupRegistrationHandler.java b/event-bus/distributed/src/main/java/org/apache/james/events/GroupRegistrationHandler.java
index 98b006d..6381851 100644
--- a/event-bus/distributed/src/main/java/org/apache/james/events/GroupRegistrationHandler.java
+++ b/event-bus/distributed/src/main/java/org/apache/james/events/GroupRegistrationHandler.java
@@ -77,7 +77,7 @@
private final ListenerExecutor listenerExecutor;
private final RabbitMQConfiguration configuration;
private final GroupRegistration.WorkQueueName queueName;
- private Scheduler scheduler;
+ private final Scheduler scheduler;
private Optional<Disposable> consumer;
GroupRegistrationHandler(NamingStrategy namingStrategy, EventSerializer eventSerializer, ReactorRabbitMQChannelPool channelPool, Sender sender, ReceiverProvider receiverProvider,
@@ -94,8 +94,8 @@
this.configuration = configuration;
this.groupRegistrations = new ConcurrentHashMap<>();
this.queueName = namingStrategy.workQueue(GROUP);
+ this.scheduler = Schedulers.newBoundedElastic(EventBus.EXECUTION_RATE, ReactorUtils.DEFAULT_BOUNDED_ELASTIC_QUEUESIZE, "groups-handler");
this.consumer = Optional.empty();
-
}
GroupRegistration retrieveGroupRegistration(Group group) {
@@ -104,7 +104,6 @@
}
public void start() {
- scheduler = Schedulers.newBoundedElastic(EventBus.EXECUTION_RATE, ReactorUtils.DEFAULT_BOUNDED_ELASTIC_QUEUESIZE, "groups-handler");
channelPool.createWorkQueue(
QueueSpecification.queue(queueName.asString())
.durable(DURABLE)
