private void attemptRead(KafkaChannel channel) throws IOException {
        String nodeId = channel.id();

        long bytesReceived = channel.read();
        if (bytesReceived != 0) {
            long currentTimeMs = time.milliseconds();
            sensors.recordBytesReceived(nodeId, bytesReceived, currentTimeMs);
            madeReadProgressLastPoll = true;

            NetworkReceive receive = channel.maybeCompleteReceive();
            if (receive != null) {
                addToCompletedReceives(channel, receive, currentTimeMs);
            }
        }
        if (channel.isMuted()) {
            outOfMemory = true; //channel has muted itself due to memory pressure.
        } else {
            madeReadProgressLastPoll = true;
        }
    }
handle any ready I/O on a set of selection keys
@param selectionKeys set of keys to handle
@param isImmediatelyConnected true if running over a set of keys for just-connected sockets
@param currentTimeNanos time at which set of keys was determined
clients/src/main/java/org/apache/kafka/common/network/Selector.java