Skip to content

KAFKA-7956 In ShutdownableThread, immediately complete the shutdown if the thread has not been started #6218

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
junrao merged 9 commits into from
Feb 26, 2019
Merged

KAFKA-7956 In ShutdownableThread, immediately complete the shutdown if the thread has not been started #6218

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
d20095e
If the thread is not alive, complete shutdown when
gardnervickers Feb 1, 2019
292d279
Go through the regular shutdown logic, just avoid awaiting the
gardnervickers Feb 1, 2019
4dfef44
move the `Thread.isAlive` check in ShutdownableThread to
gardnervickers Feb 6, 2019
445a3af
Allow blocking on ShutdownableThread.awaitShutdown() regardless
gardnervickers Feb 7, 2019
a2c0f91
Don't block in `awaitShutdown()` if the thread is not started.
gardnervickers Feb 13, 2019
7c6ac0d
Set ShutdownableThread started status in `start()` method to ensure
gardnervickers Feb 18, 2019
92f71e4
Remove start() override.
gardnervickers Feb 19, 2019
4a20222
Adapt ControllerEventManager to the new ShutdownableThread behavior.
gardnervickers Feb 25, 2019
54826d1
Add comment explaining why the ShutdownEventThread handler does
gardnervickers Feb 26, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion core/src/main/scala/kafka/controller/ControllerEventManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,7 @@ class ControllerEventManager(controllerId: Int, rateAndTimeMetrics: Map[Controll
def start(): Unit = thread.start()

def close(): Unit = {
thread.initiateShutdown()
clearAndPut(KafkaController.ShutdownEventThread)
thread.awaitShutdown()
}
Expand All @@ -83,7 +84,7 @@ class ControllerEventManager(controllerId: Int, rateAndTimeMetrics: Map[Controll

override def doWork(): Unit = {
queue.take() match {
case KafkaController.ShutdownEventThread => initiateShutdown()
case KafkaController.ShutdownEventThread => // The shutting down of the thread has been initiated at this point. Ignore this event.
case controllerEvent =>
_state = controllerEvent.state

Expand Down
13 changes: 10 additions & 3 deletions core/src/main/scala/kafka/utils/ShutdownableThread.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,8 @@ abstract class ShutdownableThread(val name: String, val isInterruptible: Boolean
this.logIdent = "[" + name + "]: "
private val shutdownInitiated = new CountDownLatch(1)
private val shutdownComplete = new CountDownLatch(1)

@volatile private var isStarted: Boolean = false

def shutdown(): Unit = {
initiateShutdown()
awaitShutdown()
Expand All @@ -54,8 +55,13 @@ abstract class ShutdownableThread(val name: String, val isInterruptible: Boolean
* After calling initiateShutdown(), use this API to wait until the shutdown is complete
*/
def awaitShutdown(): Unit = {
shutdownComplete.await()
info("Shutdown completed")
if (shutdownInitiated.getCount != 0)
throw new IllegalStateException("initiateShutdown() was not called before awaitShutdown()")
else {
if (isStarted)
shutdownComplete.await()
info("Shutdown completed")
}
}

/**
Expand All @@ -76,6 +82,7 @@ abstract class ShutdownableThread(val name: String, val isInterruptible: Boolean
def doWork(): Unit

override def run(): Unit = {
isStarted = true

@junrao junrao Feb 19, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hmm, do we need to set isStarted here again since it's already set in start()?

@gardnervickers gardnervickers Feb 19, 2019 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I was trying to cover the case where run() is called directly, without start(), either directly or by an executor implementation.

@junrao junrao Feb 19, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In that case, could we just set it in run() and not in start()?

@gardnervickers gardnervickers Feb 19, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I added in the start() case because there could be a delay between start() getting called and run() getting called in the new thread, but I think this sends up producing inconsistent behavior depending on how the ShutdownableThread is started.

In the current state, if start() is used to start the ShutdownableThread, we're guaranteed to always correctly block in subsequent calls to awaitShutdown() since isStarted is set from the caller thread. If the ShutdownableThread is submitted to an executor, run() can be called at any time in the future, which makes it impossible for the caller thread to know if awaitShutdown() will block as we expect.

We end up with different guarantees based on how the class is run, which is not optimal. I think the only one we can guarantee across both start() and an Executor starting the thread is that the caller thread won't know when isStarted is true, and as a result, when it can rely on the blocking behavior of awaitShutdown.

info("Starting")
try {
while (isRunning)
Expand Down