@Override
    public IoSession newSession(SocketAddress remoteAddress,
            SocketAddress localAddress) {
        if (remoteAddress == null) {
            throw new NullPointerException("remoteAddress");
        }
        if (localAddress == null) {
            throw new NullPointerException("localAddress");
        }

        Selector selector = this.selector;
        DatagramChannel ch = channels.get(localAddress);
        if (selector == null || ch == null) {
            throw new IllegalArgumentException("Unknown localAddress: "
                    + localAddress);
        }

        SelectionKey key = ch.keyFor(selector);
        if (key == null) {
            throw new IllegalArgumentException("Unknown localAddress: "
                    + localAddress);
        }

        RegistrationRequest req = (RegistrationRequest) key.attachment();
        IoSession session;
        IoSessionRecycler sessionRecycler = getSessionRecycler(req);
        synchronized (sessionRecycler) {
            session = sessionRecycler.recycle(localAddress, remoteAddress);
            if (session != null) {
                return session;
            }

            // If a new session needs to be created.
            // Note that the local address is the service address in the
            // acceptor side, and I didn't call getLocalSocketAddress().
            // This avoids strange cases where getLocalSocketAddress() on the
            // underlying socket would return an IPv6 address while the
            // specified service address is an IPv4 address.
            DatagramSessionImpl datagramSession = new DatagramSessionImpl(
                    wrapper, this, req.config, ch, req.handler, req.address,
                    req.address);
            datagramSession.setRemoteAddress(remoteAddress);
            datagramSession.setSelectionKey(key);

            getSessionRecycler(req).put(datagramSession);
            session = datagramSession;
        }

        try {
            buildFilterChain(req, session);
            getListeners().fireSessionCreated(session);
        } catch (Throwable t) {
            ExceptionMonitor.getInstance().exceptionCaught(t);
        }

        return session;
    }