[jira] [Commented] (KAFKA-16768) SocketServer leaks accepted SocketChannel instances due to race condition

["Christo Lolov (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/KAFKA-16768?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18062171#comment-18062171 ] 

Christo Lolov commented on KAFKA-16768:
---------------------------------------

Hello [~cyh2012aa] and [~jainamshah102]! I am happy to review efforts on this ticket. For my understanding, did you consider a second draining in Processor#close? Something like:


{color:#0033b3}def {color}{color:#00627a}close{color}(): Unit = {
  {color:#0033b3}try {color}{
    beginShutdown()
    {color:#871094}thread{color}.join()
    {color:#0033b3}if {color}({color:#000000}!{color}{color:#871094}started{color}.get) {
      {color:#000000}Utils{color}.swallow({color:#0033b3}this{color}.{color:#871094}logger{color}.{color:#000000}underlying{color}, {color:#000000}Level{color}.{color:#871094}ERROR{color}, () => closeAll())
    }
{color:#8c8c8c}    closeStraggleConnections() // <-- Second pass at draining connections
{color}{color:#8c8c8c}  {color}} {color:#0033b3}finally {color}{
    {color:#871094}metricsGroup{color}.removeMetric({color:#067d17}"IdlePercent"{color}, {color:#000000}Map{color}({color:#067d17}"networkProcessor" {color}-> {color:#000000}id{color}.toString).asJava)
    {color:#000000}metrics{color}.removeMetric({color:#871094}expiredConnectionsKilledCountMetricName{color})
  }
}

> SocketServer leaks accepted SocketChannel instances due to race condition
> -------------------------------------------------------------------------
>
>                 Key: KAFKA-16768
>                 URL: https://issues.apache.org/jira/browse/KAFKA-16768
>             Project: Kafka
>          Issue Type: Bug
>          Components: core
>    Affects Versions: 3.8.0
>            Reporter: Greg Harris
>            Assignee: Chang-Yu Huang
>            Priority: Major
>              Labels: newbie
>
> The SocketServer has threads for Acceptors and Processors. These threads communicate via Processor#accept/Processor#configureNewConnections and the `newConnections` queue.
> During shutdown, the Acceptor and Processors are each stopped by setting shouldRun to false, and then shutdown proceeds asynchronously in all instances together. This leads to a race condition where an Acceptor accepts a SocketChannel and queues it to a Processor, but that Processor instance has already started shutting down and has already drained the newConnections queue.
> KAFKA-16765 is an analogous bug in NioEchoServer, which uses a completely different implementation but has the same flaw.
> An example execution order that includes this leak:
> 1. Acceptor#accept() is called, and a new SocketChannel is accepted.
> 2. Acceptor#assignNewConnection() begins
> 3. Acceptor#close() is called, which sets shouldRun to false in the Acceptor and attached Processor instances
> 4. Processor#run() checks the shouldRun variable, and exits the loop
> 5. Processor#closeAll() executes, and drains the `newConnections` variable
> 6. Processor#run() returns and the Processor thread terminates
> 7. Acceptor#assignNewConnection() calls Processor#accept(), which adds the SocketChannel to `newConnections`
> 8. Acceptor#assignNewConnection() returns
> 9. Acceptor#run() checks the shouldRun variable and exits the loop, and the Acceptor thread terminates.
> 10. Acceptor#close() joins all of the terminated threads, and returns
> At the end of this sequence, there are still open SocketChannel instances in newConnections, which are then considered leaked.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)