No PublishInboundInterceptor added in extension for clients connected before startup (restart)

[globz-eu]

Expected behavior

When the HiveMQ CE broker starts (or restarts) with the artcom/hivemq-retained-message-query-plugin extension, the extension should add a PublishInboundInterceptor for all clients connected to the broker.

Actual behavior

When the HiveMQ CE broker starts (or restarts) with the artcom/hivemq-retained-message-query-plugin extension:

• The extension does not add a PublishInboundInterceptor for clients that were connected to the broker before the extension is initialized.
• The extension does add a PublishInboundInterceptor for clients that connect to the broker after the extension is initialized.

To Reproduce

Steps

• Start the broker with the artcom/hivemq-retained-message-query-plugin extension and a client service simultaneously.
• Stop the broker
• Start the broker

Reproducer code

git clone https://github.com/artcom/retained-message-query-plugin-test.git
cd retained-message-query-plugin-test
docker compose -f compose-ce.yml up --build
docker compose -f compose-ce.yml  stop broker
docker compose -f compose-ce.yml start broker

The following message is logged when a PublishInboundInterceptor is added to a client:
DEBUG - Adding a PublishInboundInterceptor to client id UnknownApp-<randomId>

Details

• When the client service is restarted after the extension was iniitalized the PublishInboundInterceptor is added for that client (as seen in the logs). This can be reproduced with:

  docker compose -f compose-ce.yml stop retained-message-generator
  docker compose -f compose-ce.yml start retained-message-generator

• When using the enterprise edition of the broker a PublishInboundInterceptor is added to all connected clients regardless of when they connected. This can be verified by using the compose-ee.yml file.

• Affected HiveMQ CE version(s): 2023.9

• Used JVM version: 11

[MicWalter]

Hello @globz-eu,

can you explain what you mean with connected clients?

Because the HiveMQ broker does not open the MQTT listeners before the extension system has been run through, aka all extension completed.

As you can see the broker waits before doing anything else, when I have a sleep in my extensionStart

2023-12-20 18:33:36,485 DEBUG - Starting extension with id "test-extension" at /opt/hivemq/extensions/test-extension
2023-12-20 18:33:36,491 INFO  - start
2023-12-20 18:33:46,492 INFO  - stop
2023-12-20 18:33:46,492 INFO  - Extension "test-extension" version 1.0.0 started successfully.

Greetings,
Michael from the HiveMQ team

[MicWalter]

I had a look at your code. You asynchronously add the interceptor, so I tested it and you are correct that the interceptor are not applied to already connected clients.

Though I'm not sure what the intended behaviour is, as one could argue when an extension said it started that it allocated all necessary resources so it can operate. I'll start a discussion internally.

As I don't know when this will be tackled, can I ask you why you need all this async steps during the extensionStart phase?
IMHO you should make steps in a blocking manner so you can stop the extension from starting in case anything goes wrong (for example HTTP server doesn't start). This way you also do not need to wait for a fix.

EDIT: Simple reproducer code:

    @Override
    public void extensionStart(final ExtensionStartInput input, final ExtensionStartOutput output) {
        try {
            Services.extensionExecutorService().schedule(() ->
                    Services.initializerRegistry().setClientInitializer((initializerInput, clientContext) ->
                            log.info("adding pub interceptor for client {}", initializerInput.getClientInformation().getClientId()))
                    , 10, TimeUnit.SECONDS);
        } catch (final Exception ignore) {
        }
    }

Greetings,
Michael from the HiveMQ team