public void close() {
        sessionExpirationService.shutdown();
        // Update all not clean session with the proper expiry date
        updateNotCleanSessionsWithProperExpire();
        queueRepository.close();
    }
@Test
    public void testSerializabilityOfPublishedMessage() {
        LOG.info("testSerializabilityOfPublishedMessage");
        MVStore mvStore = new MVStore.Builder()
            .fileName(BrokerConstants.DEFAULT_PERSISTENT_PATH)
            .autoCommitDisabled()
            .open();
        final MVMap.Builder<String, SessionRegistry.PublishedMessage> builder =
            new MVMap.Builder<String, SessionRegistry.PublishedMessage>()
                .valueType(new EnqueuedMessageValueType());

        final ByteBuf payload = Unpooled.wrappedBuffer("Hello World!".getBytes(StandardCharsets.UTF_8));
        SessionRegistry.PublishedMessage msg = new SessionRegistry.PublishedMessage(Topic.asTopic("/say"),
            MqttQoS.AT_LEAST_ONCE, payload, false, Instant.MAX);
        try {
            // store a message in the MVStore
            final String mapName = "test_map";
            MVMap<String, SessionRegistry.PublishedMessage> persistentMap = mvStore.openMap(mapName, builder);
            String key = "message";
            persistentMap.put(key, msg);
            mvStore.close();

            // reopen the MVStore and read it
            mvStore = new MVStore.Builder()
                .fileName(BrokerConstants.DEFAULT_PERSISTENT_PATH)
                .autoCommitDisabled()
                .open();
            final SessionRegistry.PublishedMessage reloadedMsg = mvStore.openMap(mapName, builder).get(key);

            // Verify
            assertEquals("/say", reloadedMsg.topic.toString());
        } finally {
            mvStore.close();
            File dbFile = new File(BrokerConstants.DEFAULT_PERSISTENT_PATH);
            if (dbFile.exists()) {
                dbFile.delete();
            }
            assertFalse(dbFile.exists());
        }
    }