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 all commits
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
2 changes: 1 addition & 1 deletion 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 @@ -719,7 +719,7 @@ void runOnce() {
// Should only proceed when the thread is still running after #pollRequests(), because no external state mutation
// could affect the task manager state beyond this point within #runOnce().
if (!isRunning()) {
log.debug("Thread state is already {}, skipping the run once call after poll request", state);
log.info("Thread state is already {}, skipping the run once call after poll request", state);
return;
}

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