#vulnerable code 
                                protected void handleConnect(IoSession session, ConnectMessage msg) {
        LOG.info("handleConnect invoked");
        if (msg.getProcotolVersion() != 0x03) {
            ConnAckMessage badProto = new ConnAckMessage();
            badProto.setReturnCode(ConnAckMessage.UNNACEPTABLE_PROTOCOL_VERSION);
            session.write(badProto);
            session.close(false);
            return;
        }

        if (msg.getClientID() == null || msg.getClientID().length() > 23) {
            ConnAckMessage okResp = new ConnAckMessage();
            okResp.setReturnCode(ConnAckMessage.IDENTIFIER_REJECTED);
            session.write(okResp);
            return;
        }

        m_clientIDsLock.lock();
        try {
            //if an old client with the same ID already exists close its session.
            if (m_clientIDs.containsKey(msg.getClientID())) {
                //clean the subscriptions if the old used a cleanSession = true
                IoSession oldSession = m_clientIDs.get(msg.getClientID()).getSession();
                boolean cleanSession = (Boolean) oldSession.getAttribute(Constants.CLEAN_SESSION);
                if (cleanSession) {
                    //cleanup topic subscriptions
                    m_messaging.removeSubscriptions(msg.getClientID());
                }

                m_clientIDs.get(msg.getClientID()).getSession().close(false);
            }

            ConnectionDescriptor connDescr = new ConnectionDescriptor(msg.getClientID(), session, msg.isCleanSession());
            m_clientIDs.put(msg.getClientID(), connDescr);
        } finally {
            m_clientIDsLock.unlock();
        }

        int keepAlive = msg.getKeepAlive();
        session.setAttribute("keepAlive", keepAlive);
        session.setAttribute(Constants.CLEAN_SESSION, msg.isCleanSession());
        //used to track the client in the subscription and publishing phases. 
        session.setAttribute(Constants.ATTR_CLIENTID, msg.getClientID());

        session.getConfig().setIdleTime(IdleStatus.READER_IDLE, Math.round(keepAlive * 1.5f));

        //Handle will flag
        if (msg.isWillFlag()) {
            QOSType willQos = QOSType.values()[msg.getWillQos()];
            m_messaging.publish(msg.getWillTopic(), msg.getWillMessage().getBytes(),
                    willQos, msg.isWillRetain(), msg.getClientID(), session);
        }

        //handle user authentication
        if (msg.isUserFlag()) {
            String pwd = null;
            if (msg.isPasswordFlag()) {
                pwd = msg.getPassword();
            }
            if (!m_authenticator.checkValid(msg.getUsername(), pwd)) {
                ConnAckMessage okResp = new ConnAckMessage();
                okResp.setReturnCode(ConnAckMessage.BAD_USERNAME_OR_PASSWORD);
                session.write(okResp);
                return;
            }
        }

        //handle clean session flag
        if (msg.isCleanSession()) {
            //remove all prev subscriptions
            //cleanup topic subscriptions
            m_messaging.removeSubscriptions(msg.getClientID());
        }  else {
            //force the republish of stored QoS1 and QoS2
            m_messaging.republishStored(msg.getClientID());
        }

        ConnAckMessage okResp = new ConnAckMessage();
        okResp.setReturnCode(ConnAckMessage.CONNECTION_ACCEPTED);
        session.write(okResp);
    } 
                            #location 75 
                            #vulnerability type THREAD_SAFETY_VIOLATION
#fixed code
                                protected void handleConnect(IoSession session, ConnectMessage msg) {
        LOG.info("handleConnect invoked");

        m_messaging.connect(session, msg);
    }