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

Changes from 5 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/utils/ShutdownableThread.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,8 @@ 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()
if (this.isAlive)
shutdownComplete.await()

@junrao junrao Feb 13, 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.

There are 2 ways to shut down a ShutdownableThread: (1) Just call shutdown(), which blocks until the thread completes. (2) First call initiateShutdown() and then call awaitShutdown(). The reason for this style is to allow the caller to add additional logic between the 2 calls. For example, if the thread can block in a selector, one can wake up the selector between the 2 calls. If the thread wasn't started or died unexpectedly, it's reasonable not to block the shutdown in either case.

Also, just calling awaitShutdown() w/o calling initiateShutdown() first is unexpected and not supported. To protect against this, perhaps we can add a check to make sure that shutdownInitiated is already 0 when awaitShutdown() is called. Otherwise, we can throw an IllegalStateException.

@gardnervickers gardnervickers Feb 18, 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.

Thanks! I believe we can cover both cases here if we check in awaitShutdown() that the thread has been started before blocking? Perhaps I'm misunderstanding though.

I'll add a check that shutdownInitiated == 0.

info("Shutdown completed")
}

Expand Down