@Override
    public void poll(long timeout) throws IOException {
        if (timeout < 0)
            throw new IllegalArgumentException("timeout should be >= 0");

        boolean madeReadProgressLastCall = madeReadProgressLastPoll;
        clear();

        boolean dataInBuffers = !keysWithBufferedRead.isEmpty();

        if (!immediatelyConnectedKeys.isEmpty() || (madeReadProgressLastCall && dataInBuffers))
            timeout = 0;

        if (!memoryPool.isOutOfMemory() && outOfMemory) {
            //we have recovered from memory pressure. unmute any channel not explicitly muted for other reasons
            log.trace("Broker no longer low on memory - unmuting incoming sockets");
            for (KafkaChannel channel : channels.values()) {
                if (channel.isInMutableState() && !explicitlyMutedChannels.contains(channel)) {
                    channel.maybeUnmute();
                }
            }
            outOfMemory = false;
        }

        /* check ready keys */
        long startSelect = time.nanoseconds();
        int numReadyKeys = select(timeout);
        long endSelect = time.nanoseconds();
        this.sensors.selectTime.record(endSelect - startSelect, time.milliseconds(), false);

        if (numReadyKeys > 0 || !immediatelyConnectedKeys.isEmpty() || dataInBuffers) {
            Set<SelectionKey> readyKeys = this.nioSelector.selectedKeys();

            // Poll from channels that have buffered data (but nothing more from the underlying socket)
            if (dataInBuffers) {
                keysWithBufferedRead.removeAll(readyKeys); //so no channel gets polled twice
                Set<SelectionKey> toPoll = keysWithBufferedRead;
                keysWithBufferedRead = new HashSet<>(); //poll() calls will repopulate if needed
                pollSelectionKeys(toPoll, false, endSelect);
            }

            // Poll from channels where the underlying socket has more data
            pollSelectionKeys(readyKeys, false, endSelect);
            // Clear all selected keys so that they are excluded from the ready count for the next select
            readyKeys.clear();

            pollSelectionKeys(immediatelyConnectedKeys, true, endSelect);
            immediatelyConnectedKeys.clear();
        } else {
            madeReadProgressLastPoll = true; //no work is also "progress"
        }

        long endIo = time.nanoseconds();
        this.sensors.ioTime.record(endIo - endSelect, time.milliseconds(), false);

        // Close channels that were delayed and are now ready to be closed
        completeDelayedChannelClose(endIo);

        // we use the time at the end of select to ensure that we don't close any connections that
        // have just been processed in pollSelectionKeys
        maybeCloseOldestConnection(endSelect);
    }
Do whatever I/O can be done on each connection without blocking. This includes completing connections, completing
disconnections, initiating new sends, or making progress on in-progress sends or receives.

When this call is completed the user can check for completed sends, receives, connections or disconnects using
{@link #completedSends()}, {@link #completedReceives()}, {@link #connected()}, {@link #disconnected()}. These
lists will be cleared at the beginning of each `poll` call and repopulated by the call if there is
any completed I/O.

In the "Plaintext" setting, we are using socketChannel to read & write to the network. But for the "SSL" setting,
we encrypt the data before we use socketChannel to write data to the network, and decrypt before we return the responses.
This requires additional buffers to be maintained as we are reading from network, since the data on the wire is encrypted
we won't be able to read exact no.of bytes as kafka protocol requires. We read as many bytes as we can, up to SSLEngine's
application buffer size. This means we might be reading additional bytes than the requested size.
If there is no further data to read from socketChannel selector won't invoke that channel and we have additional bytes
in the buffer. To overcome this issue we added "keysWithBufferedRead" map which tracks channels which have data in the SSL
buffers. If there are channels with buffered data that can by processed, we set "timeout" to 0 and process the data even
if there is no more data to read from the socket.

At most one entry is added to "completedReceives" for a channel in each poll. This is necessary to guarantee that
requests from a channel are processed on the broker in the order they are sent. Since outstanding requests added
by SocketServer to the request queue may be processed by different request handler threads, requests on each
channel must be processed one-at-a-time to guarantee ordering.

@param timeout The amount of time to wait, in milliseconds, which must be non-negative
@throws IllegalArgumentException If `timeout` is negative
@throws IllegalStateException If a send is given for which we have no existing connection or for which there is
        already an in-progress send
clients/src/main/java/org/apache/kafka/common/network/Selector.java