The official example code is misleading and may cause Java OutOfMemoryError.

[xiaokxluoshu]

public final class EmbeddedLauncher {

static class PublisherListener extends AbstractInterceptHandler {

    @Override
    public String getID() {
        return "EmbeddedLauncherPublishListener";
    }

    @Override
    public void onPublish(InterceptPublishMessage msg) {
        final String decodedPayload = msg.getPayload().toString(UTF_8);
        System.out.println("Received on topic: " + msg.getTopicName() + " content: " + decodedPayload);
    }
}

public static void main(String[] args) throws InterruptedException, IOException {
    IResourceLoader classpathLoader = new ClasspathResourceLoader();
    final IConfig classPathConfig = new ResourceLoaderConfig(classpathLoader);

    final Server mqttBroker = new Server();
    List<? extends InterceptHandler> userHandlers = Collections.singletonList(new PublisherListener());
    mqttBroker.startServer(classPathConfig, userHandlers);

    System.out.println("Broker started press [CTRL+C] to stop");
    //Bind  a shutdown hook
    Runtime.getRuntime().addShutdownHook(new Thread(() -> {
        System.out.println("Stopping broker");
        mqttBroker.stopServer();
        System.out.println("Broker stopped");
    }));

    Thread.sleep(20000);
    System.out.println("Before self publish");
    MqttPublishMessage message = MqttMessageBuilders.publish()
        .topicName("/exit")
        .retained(true)
        .qos(MqttQoS.EXACTLY_ONCE)
        .payload(Unpooled.copiedBuffer("Hello World!!".getBytes(UTF_8)))
        .build();

    mqttBroker.internalPublish(message, "INTRLPUB");
    System.out.println("After self publish");
}

private EmbeddedLauncher() {
}
}

The class PublisherListener inherits from the AbstractInterceptHandler class, but the onPublish method in the implementation class does not call msg.getPayload().release(). If all implementations follow this method, it will cause OutOfMemoryError after the MQTT broker runs for a period of time.

Actual behavior

Steps to reproduce

Minimal yet complete reproducer code (or URL to code) or complete log file

Moquette MQTT version

0.17

JVM version (e.g. java -version)

1.8.0_331

OS version (e.g. uname -a)

win11

[xiaokxluoshu]

I suggest making the following modifications, providing developers with proper guidance.

 static class PublisherListener extends AbstractInterceptHandler {

    @Override
    public String getID() {
        return "EmbeddedLauncherPublishListener";
    }

    @Override
    public void onPublish(InterceptPublishMessage msg) {
        final String decodedPayload = msg.getPayload().toString(UTF_8);
        System.out.println("Received on topic: " + msg.getTopicName() + " content: " + decodedPayload);
        super.onPublish(msg);
    }
 }