SessionCreationResult createOrReopenSession(MqttConnectMessage msg, String clientId, String username) {
        SessionCreationResult postConnectAction;
        final Session oldSession = retrieve(clientId);
        if (oldSession == null) {
            // case 1, no existing session with given clientId.
            final Session newSession = createNewSession(msg, clientId);
            postConnectAction = new SessionCreationResult(newSession, CreationModeEnum.CREATED_CLEAN_NEW, false);

            // publish the session
            final Session previous = pool.put(clientId, newSession);
            if (previous != null) {
                // if this happens mean that another Session Event Loop thread processed a CONNECT message
                // with the same clientId. This is a bug because all messages for the same clientId should
                // be handled by the same event loop thread.
                LOG.error("Another thread added a Session for our clientId {}, this is a bug!", clientId);
            }

            LOG.trace("case 1, not existing session with CId {}", clientId);
        } else {
            postConnectAction = reopenExistingSession(msg, clientId, oldSession, username);
        }
        return postConnectAction;
    }
@Test
    public void testConnAckContainsSessionPresentFlag() {
        LOG.info("testConnAckContainsSessionPresentFlag invoked");
        MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID)
                                        .protocolVersion(MqttVersion.MQTT_3_1_1)
                                        .build();
        NettyUtils.clientID(channel, FAKE_CLIENT_ID);
        NettyUtils.cleanSession(channel, false);

        // Connect a first time
        final SessionRegistry.SessionCreationResult res = sut.createOrReopenSession(msg, FAKE_CLIENT_ID, connection.getUsername());
        // disconnect
        res.session.disconnect();
//        sut.disconnect(FAKE_CLIENT_ID);

        // Exercise, reconnect
        EmbeddedChannel anotherChannel = new EmbeddedChannel();
        MQTTConnection anotherConnection = createMQTTConnection(ALLOW_ANONYMOUS_AND_ZEROBYTE_CLIENT_ID, anotherChannel);
        final SessionRegistry.SessionCreationResult result = sut.createOrReopenSession(msg, FAKE_CLIENT_ID, anotherConnection.getUsername());

        // Verify
        assertEquals(SessionRegistry.CreationModeEnum.CREATED_CLEAN_NEW, result.mode);
        assertTrue(anotherChannel.isOpen(), "Connection is accepted and therefore should remain open");
    }