[KAFKA-9903] kafka ShutdownableThread judge thread isRuning status has some bug - ASF JIRA

XML

Word

Printable

JSON

Details

Type: Bug
Status: Open
Priority: Major
Resolution: Unresolved
Affects Version/s: 2.3.1
Fix Version/s: None
Component/s: core
Labels:
None

Flags:

Important

Description

1.bug

override def run(): Unit = {
  isStarted = true
  info("Starting")
  try {
    while (isRunning)
      doWork()
  } catch {
    case e: FatalExitError =>
      shutdownInitiated.countDown()
      shutdownComplete.countDown()
      info("Stopped")
      Exit.exit(e.statusCode())
    case e: Throwable =>
      if (isRunning)
        error("Error due to", e)
  } finally {
    shutdownInitiated.countDown()
    shutdownComplete.countDown()
  }
  info("Stopped")
}

def isRunning: Boolean = {
  shutdownInitiated.getCount() != 0
}

1.when replicaThread has exception which is not fatalExitError, the thread will exit,and run finally logic(countdown the shutdownComplete conutdownLatch),but shutdownInitiated is not be countdown.

2.with 1, shutdownInitiated is just not countdown, its value is 1, isRunning logic just judge thread isRuning through shutdownInitiated != 0, so through this method to judge thread status is wrong.

3.isRunning method is used in shutdownIdleFetcherThreads, processFetchRequest, controller request send and oher else, maybe cause thread can't be remove and something can not be done

2.bugfix

Just like the following code，countdown shutdownInitiated in finally logic

override def run(): Unit = {
  isStarted = true
  info("Starting")
  try {
    while (isRunning)
      doWork()
  } catch {
    case e: FatalExitError =>
      shutdownInitiated.countDown()
      shutdownComplete.countDown()
      info("Stopped")
      Exit.exit(e.statusCode())
    case e: Throwable =>
      if (isRunning)
        error("Error due to", e)
  } finally {
    shutdownInitiated.countDown()
    shutdownComplete.countDown()
  }
  info("Stopped")
}