Skip to content

KAFKA-19190: Handle shutdown application correctly #19544

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

Jump to bottom
Merged
merged 7 commits into from
Apr 25, 2025
Merged

KAFKA-19190: Handle shutdown application correctly #19544

merged 7 commits into from
Apr 25, 2025

Conversation

lucasbru
Copy link
Member

@lucasbru lucasbru commented Apr 23, 2025 •
edited by github-actions bot
Loading

If the streams rebalance protocol is enabled in
StreamsUncaughtExceptionHandlerIntegrationTest, the streams application
does not shut down correctly upon error.

There are two causes for this - Sometimes, the SHUTDOWN_APPLICATION
code only sent with the leave heartbeat, but that is not handled broker
side. - The SHUTDOWN_APPLICATION code wasn't properly handled
client-side at all

Reviewers: Bruno Cadonna [email protected], Bill Bejeck
[email protected], PoAn Yang [email protected]

@lucasbru lucasbru requested a review from mjsax April 23, 2025 14:31
@lucasbru lucasbru force-pushed the shutdown_application branch from 3a0c62c to 1e8b847 Compare April 23, 2025 14:33
@lucasbru
Copy link
Member Author

PTAL @aliehsaeedii

@lucasbru lucasbru requested a review from bbejeck April 23, 2025 14:34
cadonna
cadonna reviewed Apr 23, 2025
View reviewed changes
Copy link
Member

@cadonna cadonna left a comment

Choose a reason for hiding this comment

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

Thanks for the PR, @lucasbru !

I had only minor comments.

lucasbru and others added 2 commits April 23, 2025 18:12
@lucasbru @cadonna
Update clients/src/main/java/org/apache/kafka/clients/consumer/intern…
240466a 
…als/StreamsRebalanceData.java

Co-authored-by: Bruno Cadonna <[email protected]>
@lucasbru @cadonna
Update clients/src/main/java/org/apache/kafka/clients/consumer/intern…
12fd1c8 
…als/StreamsRebalanceData.java

Co-authored-by: Bruno Cadonna <[email protected]>
@lucasbru
Copy link
Member Author

Thanks for the PR, @lucasbru !

I had only minor comments.

Thanks for the review, @cadonna ! Ready for re-review.

bbejeck
bbejeck approved these changes Apr 23, 2025
View reviewed changes
Copy link
Member

@bbejeck bbejeck left a comment

Choose a reason for hiding this comment

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

Thanks for the PR @lucasbru - overall LGTM with a couple of minor comments

...in/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java
String statusDetails = statuses.stream()
.map(status -> "(" + status.statusCode() + ") " + status.statusDetail())
.collect(Collectors.joining(", "));
logger.warn("Membership is in the following statuses: {}", statusDetails);
Copy link
Member

Choose a reason for hiding this comment

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

Why a warn? Seems more like an INFO

Copy link
Member Author

Choose a reason for hiding this comment

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

It's a good suggestion, that I was thinking about as well. The thing is, right now the streams application keeps running when a source topic is missing, and this membership log is the only client-side thing that is pointing out why no processing is happening. So I would like to keep it at WARN for now. It was intended in the KIP to be like this, but I discussed with Matthias as well, that we want to fail in these cases instead - so if a source topic doesn't exist, internal topics aren't created within a certain time, or topics are malpartitioned, we actually want to throw an exception (inside StreamThread). This PR is preparing the change by propagating the status to the StreamThread.

Lets make this an INFO log, once we treat the statuses more explicitly inside StreamThread.

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
@@ -518,7 +517,7 @@ private void handleStreamsUncaughtException(final Throwable throwable,
break;
case SHUTDOWN_CLIENT:
log.error(
"Encountered the following exception during processing and the registered exception handler" +
"Encountered the following exception during processing and the registered exception handler " +
Copy link
Member

Choose a reason for hiding this comment

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

Suggested change
"Encountered the following exception during processing and the registered exception handler " +
"Encountered the following exception during processing and the registered exception handler" +

Copy link
Member Author

Choose a reason for hiding this comment

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

This is actually an intentional fix. The error message was "handleropted" before

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
public void sendShutdownRequest(final AssignorError assignorError) {
assignmentErrorCode.set(assignorError.code());
public void sendShutdownRequest() {
assignmentErrorCode.set(AssignorError.SHUTDOWN_REQUESTED.code());
Copy link
Member

Choose a reason for hiding this comment

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

Does this mean a shutdown is always considered an error? What about cases where users proactively shutdown? Or is this only executed in error cases?

Copy link
Member Author

Choose a reason for hiding this comment

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

I do not intend to change the behavior of Kafka Streams here. I believe, the SHUTDOWN_REQUESTED propagated through rebalances is always handled as an error, yes.

The change in this line doesn't change anything functionally, since we'd always pass the same assignorError to this method.

FrankYang0529
FrankYang0529 approved these changes Apr 24, 2025
View reviewed changes
Copy link
Member

@FrankYang0529 FrankYang0529 left a comment

Choose a reason for hiding this comment

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

Overall LGTM. Leave some minor comments.

lucasbru and others added 3 commits April 24, 2025 10:27
@lucasbru @FrankYang0529
Update clients/src/test/java/org/apache/kafka/clients/consumer/intern…
cda36b6 
…als/StreamsRebalanceDataTest.java

Co-authored-by: PoAn Yang <[email protected]>
@lucasbru @FrankYang0529
Update clients/src/main/java/org/apache/kafka/clients/consumer/intern…
fe0f1c9 
…als/StreamsRebalanceData.java

Co-authored-by: PoAn Yang <[email protected]>
@lucasbru @FrankYang0529
Update clients/src/main/java/org/apache/kafka/clients/consumer/intern…
14394f0 
…als/StreamsRebalanceData.java

Co-authored-by: PoAn Yang <[email protected]>
@lucasbru lucasbru merged commit 732ed06 into apache:trunk Apr 25, 2025
24 checks passed
@frankvicky frankvicky mentioned this pull request Apr 25, 2025
Closed
@lucasbru lucasbru added the KIP-1071 PRs related to KIP-1071 label Apr 30, 2025
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Reviewers

@cadonna cadonna cadonna left review comments

@bbejeck bbejeck bbejeck approved these changes

@FrankYang0529 FrankYang0529 FrankYang0529 approved these changes

@mjsax mjsax Awaiting requested review from mjsax

Assignees
No one assigned
Labels
clients consumer KIP-1071 PRs related to KIP-1071 streams
Projects
None yet
Milestone
No milestone
Development

Successfully merging this pull request may close these issues.
4 participants
@lucasbru @bbejeck @cadonna @FrankYang0529