Skip to content

KAFKA-12462: proceed with task revocation in case of thread in PENDING_SHUTDOWN #10311

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
ableegoldman merged 2 commits into from
Mar 13, 2021
Merged

KAFKA-12462: proceed with task revocation in case of thread in PENDING_SHUTDOWN #10311

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
5721096
fix in StreamsRebalanceListener, exit early in StreamThread main loop…
ableegoldman Mar 12, 2021
513ef98
remove duplicated exit
ableegoldman Mar 13, 2021
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
7 changes: 7 additions & 0 deletions streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -714,6 +714,13 @@ void runOnce() {

final long pollLatency = pollPhase();

// Optimization to skip the rest of the processing loop in case the thread was requested to shut down during
// the poll phase
Comment thread
ableegoldman marked this conversation as resolved.
Outdated
if (!isRunning()) {
log.info("Exiting the processing loop early since StreamThread state is {}", state);
return;
}

// Shutdown hook could potentially be triggered and transit the thread state to PENDING_SHUTDOWN during #pollRequests().
// The task manager internal states could be uninitialized if the state transition happens during #onPartitionsAssigned().
// Should only proceed when the thread is still running after #pollRequests(), because no external state mutation
Expand Down
4 changes: 3 additions & 1 deletion .../src/main/java/org/apache/kafka/streams/processor/internals/StreamsRebalanceListener.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,7 +87,9 @@ public void onPartitionsRevoked(final Collection<TopicPartition> partitions) {
taskManager.activeTaskIds(),
taskManager.standbyTaskIds());

if (streamThread.setState(State.PARTITIONS_REVOKED) != null && !partitions.isEmpty()) {
// We need to still invoke handleRevocation if the thread has been told to shut down, but we shouldn't ever
// transition away from PENDING_SHUTDOWN once it's been initiated (to anything other than DEAD)
if ((streamThread.setState(State.PARTITIONS_REVOKED) != null || streamThread.state() == State.PENDING_SHUTDOWN) && !partitions.isEmpty()) {

@wcarlson5 wcarlson5 Mar 13, 2021

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.

do we need to be concerned about the oder these execute?

@ableegoldman ableegoldman Mar 13, 2021

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

I think this is the correct order (assuming you mean the order of streamThread.setState(State.PARTITIONS_REVOKED) != null relative to streamThread.state() == State.PENDING_SHUTDOWN?) -- if the thread is not in PENDING_SHUTDOWN when it reaches this line, the first condition should return true, which is what we want even if it does get transitioned to PENDING_SHUTDOWN immediately after the transition to PARTITIONS_REVOKED.

final long start = time.milliseconds();
try {
taskManager.handleRevocation(partitions);
Expand Down