@Override
    public void run() {
        try {
            acceptorThread.start();
            while (serverSocketChannel.isOpen()) {
                selector.poll(100);
                synchronized (newChannels) {
                    for (SocketChannel socketChannel : newChannels) {
                        selector.register(id(socketChannel), socketChannel);
                        socketChannels.add(socketChannel);
                    }
                    newChannels.clear();
                }
                if (closeKafkaChannels) {
                    for (KafkaChannel channel : selector.channels())
                        selector.close(channel.id());
                }

                Collection<NetworkReceive> completedReceives = selector.completedReceives();
                for (NetworkReceive rcv : completedReceives) {
                    KafkaChannel channel = channel(rcv.source());
                    if (!maybeBeginServerReauthentication(channel, rcv, time)) {
                        String channelId = channel.id();
                        selector.mute(channelId);
                        NetworkSend send = new NetworkSend(rcv.source(), ByteBufferSend.sizePrefixed(rcv.payload()));
                        if (outputChannel == null)
                            selector.send(send);
                        else {
                            send.writeTo(outputChannel);
                            selector.unmute(channelId);
                        }
                    }
                }
                for (NetworkSend send : selector.completedSends()) {
                    selector.unmute(send.destinationId());
                    numSent += 1;
                }
            }
        } catch (IOException e) {
            LOG.warn(e.getMessage(), e);
        }
    }
Non-blocking EchoServer implementation that uses ChannelBuilder to create channels
with the configured security protocol.
clients/src/test/java/org/apache/kafka/common/network/NioEchoServer.java