kafka源码分析---NIO_kafka 的poll 是nio吗

1

2

3

4

5

6

7

8

9

10

11

12

>

The network server for kafka. No application specific code here, just general network server stuff.

The classes Receive and Send encapsulate the incoming and outgoing transmission of bytes. A Handler

is a mapping between a Receive and a Send, and represents the users hook to add logic for mapping requests

to actual processing code. Any uncaught exceptions in the reading or writing of the transmissions will result in

the server logging an error and closing the offending socket. As a result it is the duty of the Handler

implementation to catch and serialize any application-level errors that should be sent to the client.

This slightly lower-level interface that models sending and receiving rather than requests and responses

is necessary in order to allow the send or receive to be overridden with a non-user-space writing of bytes

using FileChannel.transferTo.

1

2

3

4

5

6

7

8

9

10

11

12

# The number of threads that the server uses for receiving requests from the network and sending responses to the network

num.network.threads=3

# The number of threads that the server uses for processing requests, which may include disk I/O

num.io.threads=8

# The number of queued request allowed before blocking the network threads

#queued.max.requests

# The send buffer (SO_SNDBUF) used by the socket server

socket.send.buffer.bytes=102400

# The receive buffer (SO_RCVBUF) used by the socket server

socket.receive.buffer.bytes=102400

# The maximum size of a request that the socket server will accept (protection against OOM)

socket.request.max.bytes=104857600

1

2

3

4

5

6

7

8

这个类上的注释阐述了kafka server的io线程模型

/**

 * An NIO socket server. The threading model is

 *   1 Acceptor thread that handles new connections

 *   Acceptor has N Processor threads that each have their own selector and read requests from sockets

 *   M Handler threads that handle requests and produce responses back to the processor threads for writing.

 */

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

private val endpoints = config.listeners.map(l => l.listenerName -> l).toMap

private val numProcessorThreads = config.numNetworkThreads

private val maxQueuedRequests = config.queuedMaxRequests

private val totalProcessorThreads = numProcessorThreads * endpoints.size

private val maxConnectionsPerIp = config.maxConnectionsPerIp

private val maxConnectionsPerIpOverrides = config.maxConnectionsPerIpOverrides

this.logIdent = "[Socket Server on Broker " + config.brokerId + "], "

private val memoryPoolSensor = metrics.sensor("MemoryPoolUtilization")

private val memoryPoolDepletedPercentMetricName = metrics.metricName("MemoryPoolAvgDepletedPercent", "socket-server-metrics")

memoryPoolSensor.add(memoryPoolDepletedPercentMetricName, new Rate(TimeUnit.MILLISECONDS))

private val memoryPool = if (config.queuedMaxBytes > 0) new SimpleMemoryPool(config.queuedMaxBytes, config.socketRequestMaxBytes, false, memoryPoolSensor) else MemoryPool.NONE

val requestChannel = new RequestChannel(totalProcessorThreads, maxQueuedRequests)

private val processors = new Array[Processor](totalProcessorThreads)

private[network] val acceptors = mutable.Map[EndPoint, Acceptor]()

private var connectionQuotas: ConnectionQuotas = _

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

  val sendBufferSize = config.socketSendBufferBytes

  val recvBufferSize = config.socketReceiveBufferBytes

  val brokerId = config.brokerId

  var processorBeginIndex = 0

  config.listeners.foreach { endpoint =>

    val listenerName = endpoint.listenerName

    val securityProtocol = endpoint.securityProtocol

    val processorEndIndex = processorBeginIndex + numProcessorThreads

    for (i <- processorBeginIndex until processorEndIndex)

      processors(i) = newProcessor(i, connectionQuotas, listenerName, securityProtocol, memoryPool)

    val acceptor = new Acceptor(endpoint, sendBufferSize, recvBufferSize, brokerId,

      processors.slice(processorBeginIndex, processorEndIndex), connectionQuotas)

    acceptors.put(endpoint, acceptor)

    KafkaThread.nonDaemon(s"kafka-socket-acceptor-$listenerName-$securityProtocol-${endpoint.port}", acceptor).start()

    acceptor.awaitStartup()

    processorBeginIndex = processorEndIndex

  }

}

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

/**

 * Thread that accepts and configures new connections. There is one of these per endpoint.

 */

private[kafka] class Acceptor(val endPoint: EndPoint,

                              val sendBufferSize: Int,

                              val recvBufferSize: Int,

                              brokerId: Int,

                              processors: Array[Processor],

                              connectionQuotas: ConnectionQuotas) extends AbstractServerThread(connectionQuotas) with KafkaMetricsGroup {

  private val nioSelector = NSelector.open()

  val serverChannel = openServerSocket(endPoint.host, endPoint.port)

  this.synchronized {

    processors.foreach { processor =>

      KafkaThread.nonDaemon(s"kafka-network-thread-$brokerId-${endPoint.listenerName}-${endPoint.securityProtocol}-${processor.id}",

        processor).start()

    }

  }

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

/**

  * Accept loop that checks for new connection attempts

  */

 def run() {

   serverChannel.register(nioSelector, SelectionKey.OP_ACCEPT)

   startupComplete()

   try {

     var currentProcessor = 0

     while (isRunning) {

       try {

         val ready = nioSelector.select(500)

         if (ready > 0) {

           val keys = nioSelector.selectedKeys()

           val iter = keys.iterator()

           while (iter.hasNext && isRunning) {

             try {

               val key = iter.next

               iter.remove()

               if (key.isAcceptable)

                 accept(key, processors(currentProcessor))

               else

                 throw new IllegalStateException("Unrecognized key state for acceptor thread.")

               // round robin to the next processor thread

               currentProcessor = (currentProcessor + 1) % processors.length

             } catch {

               case e: Throwable => error("Error while accepting connection", e)

             }

           }

         }

       }

       catch {

         // We catch all the throwables to prevent the acceptor thread from exiting on exceptions due

         // to a select operation on a specific channel or a bad request. We don't want

         // the broker to stop responding to requests from other clients in these scenarios.

         case e: ControlThrowable => throw e

         case e: Throwable => error("Error occurred", e)

       }

     }

   } finally {

     debug("Closing server socket and selector.")

     swallowError(serverChannel.close())

     swallowError(nioSelector.close())

     shutdownComplete()

   }

 }

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

/*

   * Accept a new connection

   */

  def accept(key: SelectionKey, processor: Processor) {

    val serverSocketChannel = key.channel().asInstanceOf[ServerSocketChannel]

    val socketChannel = serverSocketChannel.accept()

    try {

      connectionQuotas.inc(socketChannel.socket().getInetAddress)

      socketChannel.configureBlocking(false)

      socketChannel.socket().setTcpNoDelay(true)

      socketChannel.socket().setKeepAlive(true)

      if (sendBufferSize != Selectable.USE_DEFAULT_BUFFER_SIZE)

        socketChannel.socket().setSendBufferSize(sendBufferSize)

      debug("Accepted connection from %s on %s and assigned it to processor %d, sendBufferSize [actual|requested]: [%d|%d] recvBufferSize [actual|requested]: [%d|%d]"

            .format(socketChannel.socket.getRemoteSocketAddress, socketChannel.socket.getLocalSocketAddress, processor.id,

                  socketChannel.socket.getSendBufferSize, sendBufferSize,

                  socketChannel.socket.getReceiveBufferSize, recvBufferSize))

      processor.accept(socketChannel)

    } catch {

      case e: TooManyConnectionsException =>

        info("Rejected connection from %s, address already has the configured maximum of %d connections.".format(e.ip, e.count))

        close(socketChannel)

    }

  }

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

override def run() {

   startupComplete()

   while (isRunning) {

     try {

       // setup any new connections that have been queued up

       configureNewConnections()

       // register any new responses for writing

       processNewResponses()

       poll()

       processCompletedReceives()

       processCompletedSends()

       processDisconnected()

     } catch {

       // We catch all the throwables here to prevent the processor thread from exiting. We do this because

       // letting a processor exit might cause a bigger impact on the broker. Usually the exceptions thrown would

       // be either associated with a specific socket channel or a bad request. We just ignore the bad socket channel

       // or request. This behavior might need to be reviewed if we see an exception that need the entire broker to stop.

       case e: ControlThrowable => throw e

       case e: Throwable =>

         error("Processor got uncaught exception.", e)

     }

   }

   debug("Closing selector - processor " + id)

   swallowError(closeAll())

   shutdownComplete()

 }

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

/**

   * Register any new connections that have been queued up

   */

  private def configureNewConnections() {

    while (!newConnections.isEmpty) {

      val channel = newConnections.poll()

      try {

        debug(s"Processor $id listening to new connection from ${channel.socket.getRemoteSocketAddress}")

        val localHost = channel.socket().getLocalAddress.getHostAddress

        val localPort = channel.socket().getLocalPort

        val remoteHost = channel.socket().getInetAddress.getHostAddress

        val remotePort = channel.socket().getPort

        val connectionId = ConnectionId(localHost, localPort, remoteHost, remotePort).toString

        selector.register(connectionId, channel)

      } catch {

        // We explicitly catch all non fatal exceptions and close the socket to avoid a socket leak. The other

        // throwables will be caught in processor and logged as uncaught exceptions.

        case NonFatal(e) =>

          val remoteAddress = channel.getRemoteAddress

          // need to close the channel here to avoid a socket leak.

          close(channel)

          error(s"Processor $id closed connection from $remoteAddress", e)

      }

    }

  }

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

private def processNewResponses() {

    var curr = requestChannel.receiveResponse(id)

    while (curr != null) {

      try {

        curr.responseAction match {

          case RequestChannel.NoOpAction =>

            // There is no response to send to the client, we need to read more pipelined requests

            // that are sitting in the server's socket buffer

            updateRequestMetrics(curr.request)

            trace("Socket server received empty response to send, registering for read: " + curr)

            val channelId = curr.request.connectionId

            if (selector.channel(channelId) != null || selector.closingChannel(channelId) != null)

                selector.unmute(channelId)

          case RequestChannel.SendAction =>

            val responseSend = curr.responseSend.getOrElse(

              throw new IllegalStateException(s"responseSend must be defined for SendAction, response: $curr"))

            sendResponse(curr, responseSend)

          case RequestChannel.CloseConnectionAction =>

            updateRequestMetrics(curr.request)

            trace("Closing socket connection actively according to the response code.")

            close(selector, curr.request.connectionId)

        }

      } finally {

        curr = requestChannel.receiveResponse(id)

      }

    }

  }

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

/**

    * Queue the given request for sending in the subsequent {@link #poll(long)} calls

    * @param send The request to send

    */

   public void send(Send send) {

       String connectionId = send.destination();

       if (closingChannels.containsKey(connectionId))

           this.failedSends.add(connectionId);

       else {

           KafkaChannel channel = channelOrFail(connectionId, false);

           try {

               channel.setSend(send);

           } catch (CancelledKeyException e) {

               this.failedSends.add(connectionId);

               close(channel, false);

           }

       }

   }

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

private def processCompletedReceives() {

    selector.completedReceives.asScala.foreach { receive =>

      try {

        val openChannel = selector.channel(receive.source)

        // Only methods that are safe to call on a disconnected channel should be invoked on 'openOrClosingChannel'.

        val openOrClosingChannel = if (openChannel != null) openChannel else selector.closingChannel(receive.source)

        val session = RequestChannel.Session(new KafkaPrincipal(KafkaPrincipal.USER_TYPE, openOrClosingChannel.principal.getName), openOrClosingChannel.socketAddress)

        val req = new RequestChannel.Request(processor = id, connectionId = receive.source, session = session,

          startTimeNanos = time.nanoseconds, listenerName = listenerName, securityProtocol = securityProtocol,

          memoryPool, receive.payload)

        requestChannel.sendRequest(req)

        selector.mute(receive.source)

      } catch {

        case e @ (_: InvalidRequestException | _: SchemaException) =>

          // note that even though we got an exception, we can assume that receive.source is valid. Issues with constructing a valid receive object were handled earlier

          error(s"Closing socket for ${receive.source} because of error", e)

          close(selector, receive.source)

      }

    }

  }

1

2

3

4

5

6

7

8

9

private def processCompletedSends() {

  selector.completedSends.asScala.foreach { send =>

    val resp = inflightResponses.remove(send.destination).getOrElse {

      throw new IllegalStateException(s"Send for ${send.destination} completed, but not in `inflightResponses`")

    }

    updateRequestMetrics(resp.request)

    selector.unmute(send.destination)

  }

}

1

2

3

4

5

6

7

8

9

10

private def processDisconnected() {

    selector.disconnected.keySet.asScala.foreach { connectionId =>

      val remoteHost = ConnectionId.fromString(connectionId).getOrElse {

        throw new IllegalStateException(s"connectionId has unexpected format: $connectionId")

      }.remoteHost

      inflightResponses.remove(connectionId).foreach(response => updateRequestMetrics(response.request))

      // the channel has been closed by the selector but the quotas still need to be updated

      connectionQuotas.dec(InetAddress.getByName(remoteHost))

    }

  }

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

/**

 * An interface for asynchronous, multi-channel network I/O

 */

public interface Selectable {

    /**

     * See {@link #connect(String, InetSocketAddress, int, int) connect()}

     */

    public static final int USE_DEFAULT_BUFFER_SIZE = -1;

    /**

     * Begin establishing a socket connection to the given address identified by the given address

     * @param id The id for this connection

     * @param address The address to connect to

     * @param sendBufferSize The send buffer for the socket

     * @param receiveBufferSize The receive buffer for the socket

     * @throws IOException If we cannot begin connecting

     */

    public void connect(String id, InetSocketAddress address, int sendBufferSize, int receiveBufferSize) throws IOException;

    /**

     * Wakeup this selector if it is blocked on I/O

     */

    public void wakeup();

    /**

     * Close this selector

     */

    public void close();

    /**

     * Close the connection identified by the given id

     */

    public void close(String id);

    /**

     * Queue the given request for sending in the subsequent {@link #poll(long) poll()} calls

     * @param send The request to send

     */

    public void send(Send send);

    /**

     * Do I/O. Reads, writes, connection establishment, etc.

     * @param timeout The amount of time to block if there is nothing to do

     * @throws IOException

     */

    public void poll(long timeout) throws IOException;

    /**

     * The list of sends that completed on the last {@link #poll(long) poll()} call.

     */

    public List<Send> completedSends();

    /**

     * The list of receives that completed on the last {@link #poll(long) poll()} call.

     */

    public List<NetworkReceive> completedReceives();

    /**

     * The connections that finished disconnecting on the last {@link #poll(long) poll()}

     * call. Channel state indicates the local channel state at the time of disconnection.

     */

    public Map<String, ChannelState> disconnected();

    /**

     * The list of connections that completed their connection on the last {@link #poll(long) poll()}

     * call.

     */

    public List<String> connected();

    ...

}

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

/**

 * This interface models the in-progress sending of data to a destination identified by an integer id.

 */

public interface Send {

    /**

     * The numeric id for the destination of this send

     */

    String destination();

    /**

     * Is this send complete?

     */

    boolean completed();

    /**

     * Write some as-yet unwritten bytes from this send to the provided channel. It may take multiple calls for the send

     * to be completely written

     * @param channel The Channel to write to

     * @return The number of bytes written

     * @throws IOException If the write fails

     */

    long writeTo(GatheringByteChannel channel) throws IOException;

    /**

     * Size of the send

     */

    long size();

}

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

public class ByteBufferSend implements Send {

    private final String destination;

    private final int size;

    protected final ByteBuffer[] buffers;

    private int remaining;

    private boolean pending = false;

    public ByteBufferSend(String destination, ByteBuffer... buffers) {

        this.destination = destination;

        this.buffers = buffers;

        for (ByteBuffer buffer : buffers)

            remaining += buffer.remaining();

        this.size = remaining;

    }

    @Override

    public String destination() {

        return destination;

    }

    @Override

    public boolean completed() {

        return remaining <= 0 && !pending;

    }

    @Override

    public long size() {

        return this.size;

    }

    @Override

    public long writeTo(GatheringByteChannel channel) throws IOException {

        long written = channel.write(buffers);

        if (written < 0)

            throw new EOFException("Wrote negative bytes to channel. This shouldn't happen.");

        remaining -= written;

        pending = TransportLayers.hasPendingWrites(channel);

        return written;

    }

}

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

/**

 * A size delimited Send that consists of a 4 byte network-ordered size N followed by N bytes of content

 */

public class NetworkSend extends ByteBufferSend {

    public NetworkSend(String destination, ByteBuffer buffer) {

        super(destination, sizeDelimit(buffer));

    }

    private static ByteBuffer[] sizeDelimit(ByteBuffer buffer) {

        return new ByteBuffer[] {sizeBuffer(buffer.remaining()), buffer};

    }

    private static ByteBuffer sizeBuffer(int size) {

        ByteBuffer sizeBuffer = ByteBuffer.allocate(4);

        sizeBuffer.putInt(size);

        sizeBuffer.rewind();

        return sizeBuffer;

    }

}

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

public interface Receive extends Closeable {

    /**

     * The numeric id of the source from which we are receiving data.

     */

    String source();

    /**

     * Are we done receiving data?

     */

    boolean complete();

    /**

     * Read bytes into this receive from the given channel

     * @param channel The channel to read from

     * @return The number of bytes read

     * @throws IOException If the reading fails

     */

    long readFrom(ScatteringByteChannel channel) throws IOException;

    /**

     * Do we know yet how much memory we require to fully read this

     */

    boolean requiredMemoryAmountKnown();

    /**

     * Has the underlying memory required to complete reading been allocated yet?

     */

    boolean memoryAllocated();

}

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

public void connect(String id, InetSocketAddress address, int sendBufferSize, int receiveBufferSize) throws IOException {

        if (this.channels.containsKey(id))

            throw new IllegalStateException("There is already a connection for id " + id);

        SocketChannel socketChannel = SocketChannel.open();

        socketChannel.configureBlocking(false);

        Socket socket = socketChannel.socket();

        socket.setKeepAlive(true);

        if (sendBufferSize != Selectable.USE_DEFAULT_BUFFER_SIZE)

            socket.setSendBufferSize(sendBufferSize);

        if (receiveBufferSize != Selectable.USE_DEFAULT_BUFFER_SIZE)

            socket.setReceiveBufferSize(receiveBufferSize);

        socket.setTcpNoDelay(true);

        boolean connected;

        try {

            connected = socketChannel.connect(address);

        } catch (UnresolvedAddressException e) {

            socketChannel.close();

            throw new IOException("Can't resolve address: " + address, e);

        } catch (IOException e) {

            socketChannel.close();

            throw e;

        }

        SelectionKey key = socketChannel.register(nioSelector, SelectionKey.OP_CONNECT);

        KafkaChannel channel;

        try {

            channel = channelBuilder.buildChannel(id, key, maxReceiveSize, memoryPool);

        } catch (Exception e) {

            try {

                socketChannel.close();

            } finally {

                key.cancel();

            }

            throw new IOException("Channel could not be created for socket " + socketChannel, e);

        }

        key.attach(channel);

        this.channels.put(id, channel);

        if (connected) {

            // OP_CONNECT won't trigger for immediately connected channels

            log.debug("Immediately connected to node {}", channel.id());

            immediatelyConnectedKeys.add(key);

            key.interestOps(0);

        }

    }

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

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 (hasStagedReceives() || !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.unmute();

             }

         }

         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());

     if (numReadyKeys > 0 || !immediatelyConnectedKeys.isEmpty() || dataInBuffers) {

         Set<SelectionKey> readyKeys = this.nioSelector.selectedKeys();

         keysWithBufferedRead.removeAll(readyKeys); //so no channel gets polled twice

         //poll from channels that have buffered data (but nothing more from the underlying socket)

         if (!keysWithBufferedRead.isEmpty()) {

             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);

         pollSelectionKeys(immediatelyConnectedKeys, true, endSelect);

     } else {

         madeReadProgressLastPoll = true; //no work is also "progress"

     }

     long endIo = time.nanoseconds();

     this.sensors.ioTime.record(endIo - endSelect, time.milliseconds());

     // 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);

     // Add to completedReceives after closing expired connections to avoid removing

     // channels with completed receives until all staged receives are completed.

     addToCompletedReceives();

 }

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

/**

     * 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

     */

    private void pollSelectionKeys(Set<SelectionKey> selectionKeys,

                                   boolean isImmediatelyConnected,

                                   long currentTimeNanos) {

        Iterator<SelectionKey> iterator = determineHandlingOrder(selectionKeys).iterator();

        while (iterator.hasNext()) {

            SelectionKey key = iterator.next();

            iterator.remove();

            KafkaChannel channel = channel(key);

            long channelStartTimeNanos = recordTimePerConnection ? time.nanoseconds() : 0;

            // register all per-connection metrics at once

            sensors.maybeRegisterConnectionMetrics(channel.id());

            if (idleExpiryManager != null)

                idleExpiryManager.update(channel.id(), currentTimeNanos);

            try {

                /* complete any connections that have finished their handshake (either normally or immediately) */

                if (isImmediatelyConnected || key.isConnectable()) {

                    if (channel.finishConnect()) {

                        this.connected.add(channel.id());

                        this.sensors.connectionCreated.record();

                        SocketChannel socketChannel = (SocketChannel) key.channel();

                        log.debug("Created socket with SO_RCVBUF = {}, SO_SNDBUF = {}, SO_TIMEOUT = {} to node {}",

                                socketChannel.socket().getReceiveBufferSize(),

                                socketChannel.socket().getSendBufferSize(),

                                socketChannel.socket().getSoTimeout(),

                                channel.id());

                    } else

                        continue;

                }

                /* if channel is not ready finish prepare */

                if (channel.isConnected() && !channel.ready()) {

                    channel.prepare();

                }

                attemptRead(key, channel);

                if (channel.hasBytesBuffered()) {

                    //this channel has bytes enqueued in intermediary buffers that we could not read

                    //(possibly because no memory). it may be the case that the underlying socket will

                    //not come up in the next poll() and so we need to remember this channel for the

                    //next poll call otherwise data may be stuck in said buffers forever.

                    keysWithBufferedRead.add(key);

                }

                /* if channel is ready write to any sockets that have space in their buffer and for which we have data */

                if (channel.ready() && key.isWritable()) {

                    Send send = channel.write();

                    if (send != null) {

                        this.completedSends.add(send);

                        this.sensors.recordBytesSent(channel.id(), send.size());

                    }

                }

                /* cancel any defunct sockets */

                if (!key.isValid())

                    close(channel, true);

            } catch (Exception e) {

                String desc = channel.socketDescription();

                if (e instanceof IOException)

                    log.debug("Connection with {} disconnected", desc, e);

                else

                    log.warn("Unexpected error from {}; closing connection", desc, e);

                close(channel, true);

            } finally {

                maybeRecordTimePerConnection(channel, channelStartTimeNanos);

            }

        }

    }

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

/**

 * Queue up the given request for sending. Requests can only be sent on ready connections.

 * @param request The request

 * @param now The current timestamp

 */

void send(ClientRequest request, long now);

/**

 * Do actual reads and writes from sockets.

 *

 * @param timeout The maximum amount of time to wait for responses in ms, must be non-negative. The implementation

 *                is free to use a lower value if appropriate (common reasons for this are a lower request or

 *                metadata update timeout)

 * @param now The current time in ms

 * @throws IllegalStateException If a request is sent to an unready node

 */

List<ClientResponse> poll(long timeout, long now);

1

2

3

4

5

6

7

8

public final class ClientRequest {

    private final String destination;

    private final AbstractRequest.Builder<?> requestBuilder;

    private final int correlationId;

    private final String clientId;

    private final long createdTimeMs;

    private final boolean expectResponse;

    private final RequestCompletionHandler callback;

1

2

3

4

5

6

7

8

9

public class ClientResponse {

    private final RequestHeader requestHeader;

    private final RequestCompletionHandler callback;

    private final String destination;

    private final long receivedTimeMs;

    private final long latencyMs;

    private final boolean disconnected;

    private final UnsupportedVersionException versionMismatch;

    private final AbstractResponse responseBody;

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

private void doSend(ClientRequest clientRequest, boolean isInternalRequest, long now, AbstractRequest request) {

    String nodeId = clientRequest.destination();

    ...

    Send send = request.toSend(nodeId, header);

    InFlightRequest inFlightRequest = new InFlightRequest(

            header,

            clientRequest.createdTimeMs(),

            clientRequest.destination(),

            clientRequest.callback(),

            clientRequest.expectResponse(),

            isInternalRequest,

            request,

            send,

            now);

    this.inFlightRequests.add(inFlightRequest);

    selector.send(inFlightRequest.send);

}

1

2

3

4

5

6

7

8

9

10

11

12

13

14

public void send(Send send) {

    String connectionId = send.destination();

    if (closingChannels.containsKey(connectionId))

        this.failedSends.add(connectionId);

    else {

        KafkaChannel channel = channelOrFail(connectionId, false);

        try {

            channel.setSend(send);

        } catch (CancelledKeyException e) {

            this.failedSends.add(connectionId);

            close(channel, false);

        }

    }

}

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

public List<ClientResponse> poll(long timeout, long now) {

    if (!abortedSends.isEmpty()) {

        // If there are aborted sends because of unsupported version exceptions or disconnects,

        // handle them immediately without waiting for Selector#poll.

        List<ClientResponse> responses = new ArrayList<>();

        handleAbortedSends(responses);

        completeResponses(responses);

        return responses;

    }

    long metadataTimeout = metadataUpdater.maybeUpdate(now);

    try {

        this.selector.poll(Utils.min(timeout, metadataTimeout, requestTimeoutMs));

    } catch (IOException e) {

        log.error("Unexpected error during I/O", e);

    }

    // process completed actions

    long updatedNow = this.time.milliseconds();

    List<ClientResponse> responses = new ArrayList<>();

    handleCompletedSends(responses, updatedNow);

    handleCompletedReceives(responses, updatedNow);

    handleDisconnections(responses, updatedNow);

    handleConnections();

    handleInitiateApiVersionRequests(updatedNow);

    handleTimedOutRequests(responses, updatedNow);

    completeResponses(responses);

    return responses;

}