EventBus consumer in QueueStatisticsCollector never unregistered on shutdown - causes stale subscriptions in clustered mode

[mcweba]

When running redisques in a Vert.x clustered environment (e.g., with Hazelcast), the EventBus consumer registered by QueueStatisticsCollector is never unregistered during shutdown. This causes stale entries to accumulate in the Hazelcast __vertx.subs multimap, leading to persistent "Not a member of the cluster" errors when other nodes try to send messages to the dead member.

Root Cause

In QueueStatisticsCollector.java (lines 103-112) (https://github.com/swisspost/vertx-redisques/blob/cb72c1e/src/main/java/org/swisspush/redisques/util/QueueStatisticsCollector.java#L103-L112), an EventBus consumer is registered in the constructor:

vertx.eventBus().consumer(keyspaceHelper.getQueueStatisticQueueSizeSyncKey(),
        (Handler<Message<QueueSizeInfoMap>>) event ->
               event.body().getValue().forEach((key, value) -> {
                    if (!keyspaceHelper.getVerticleUid().equals(key)) {
                        approximateQueueSize.put(key, value);
                    }
                }));

Problems:

1. The MessageConsumer reference is not stored
2. QueueStatisticsCollector has no stop() or close() method
3. The consumer is not included in QueueRegistryService.stop() or gracefulStop() cleanup

Impact

In a Kubernetes environment with pod scaling/restarts:

1. Pod A registers EventBus consumer for address redisques-addr-queueStateQueueItemCounter
2. Pod A terminates (scaled down, crashed, or restarted)
3. Subscription remains in Hazelcast __vertx.subs multimap with Pod A's member UUID
4. Other pods continuously try to send messages to the dead UUID
5. Logs fill with warnings: WARN ConnectionHolder - Connecting to server <dead-uuid> failed io.vertx.core.impl.NoStackTraceThrowable: Not a member of the cluster
6. The stale entry persists indefinitely until ALL cluster nodes are restarted simultaneously

Proposed Fix

Store the MessageConsumer reference and unregister it during shutdown:

// Add field
private MessageConsumer<QueueSizeInfoMap> queueSizeConsumer;

// In constructor, store reference
this.queueSizeConsumer = vertx.eventBus().consumer(
    keyspaceHelper.getQueueStatisticQueueSizeSyncKey(),
    (Handler<Message<QueueSizeInfoMap>>) event -> { ... }
);

// Add stop method
public void stop() {
    if (queueSizeConsumer != null) {
        queueSizeConsumer.unregister();
    }
}

Then integrate into QueueRegistryService.stop():

public void stop() {
    queueStatisticsCollector.stop();  // Add this line
    unregisterConsumers(UnregisterConsumerType.FORCE);
}

[ZhengXinCN]

internal issue id: SDCISA-23374