@Override
    public void notifyTopicPublished(final MqttPublishMessage msg, final String clientID, final String username) {
        msg.retain();

        executor.execute(() -> {
                try {
                    int messageId = msg.variableHeader().messageId();
                    String topic = msg.variableHeader().topicName();
                    for (InterceptHandler handler : handlers.get(InterceptPublishMessage.class)) {
                        LOG.debug("Notifying MQTT PUBLISH message to interceptor. CId={}, messageId={}, topic={}, "
                                + "interceptorId={}", clientID, messageId, topic, handler.getID());
                        // Sending to the outside, make a retainedDuplicate.
                        handler.onPublish(new InterceptPublishMessage(msg.retainedDuplicate(), clientID, username));
                    }
                } finally {
                    ReferenceCountUtil.release(msg);
                }
        });
    }
@Test
    public void testNotifyTopicPublished() throws Exception {
        final ByteBuf payload = Unpooled.copiedBuffer("Hello".getBytes(UTF_8));
        // Internal function call, will not release buffers.
        interceptor.notifyTopicPublished(
                MqttMessageBuilders.publish().qos(MqttQoS.AT_MOST_ONCE)
                    .payload(payload).build(),
                "cli1234",
                "cli1234");
        interval();
        assertEquals(60, n.get());
        payload.release();
    }