You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "lucasbru (via GitHub)" <gi...@apache.org> on 2025/04/23 14:31:29 UTC
[PR] KAFKA-19190: Handle shutdown application correctly [kafka]
lucasbru opened a new pull request, #19544:
URL: https://github.com/apache/kafka/pull/19544
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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19190: Handle shutdown application correctly [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #19544:
URL: https://github.com/apache/kafka/pull/19544#discussion_r2057847775
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -1079,8 +1081,9 @@ public void shutdownToError() {
shutdownErrorHook.run();
}
- public void sendShutdownRequest(final AssignorError assignorError) {
- assignmentErrorCode.set(assignorError.code());
+ public void sendShutdownRequest() {
+ assignmentErrorCode.set(AssignorError.SHUTDOWN_REQUESTED.code());
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19190: Handle shutdown application correctly [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru merged PR #19544:
URL: https://github.com/apache/kafka/pull/19544
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19190: Handle shutdown application correctly [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on PR #19544:
URL: https://github.com/apache/kafka/pull/19544#issuecomment-2824527450
PTAL @aliehsaeedii
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19190: Handle shutdown application correctly [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #19544:
URL: https://github.com/apache/kafka/pull/19544#discussion_r2057842172
##########
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 " +
Review Comment:
This is actually an intentional fix. The error message was "handleropted" before
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19190: Handle shutdown application correctly [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #19544:
URL: https://github.com/apache/kafka/pull/19544#discussion_r2057854972
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java:
##########
@@ -521,11 +521,14 @@ private void onSuccessResponse(final StreamsGroupHeartbeatResponse response, fin
}
List<StreamsGroupHeartbeatResponseData.Status> statuses = data.status();
- if (statuses != null && !statuses.isEmpty()) {
- String statusDetails = statuses.stream()
- .map(status -> "(" + status.statusCode() + ") " + status.statusDetail())
- .collect(Collectors.joining(", "));
- logger.warn("Membership is in the following statuses: {}", statusDetails);
+ if (statuses != null) {
+ streamsRebalanceData.setStatuses(statuses);
+ if (!statuses.isEmpty()) {
+ String statusDetails = statuses.stream()
+ .map(status -> "(" + status.statusCode() + ") " + status.statusDetail())
+ .collect(Collectors.joining(", "));
+ logger.warn("Membership is in the following statuses: {}", statusDetails);
Review Comment:
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`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19190: Handle shutdown application correctly [kafka]
Posted by "cadonna (via GitHub)" <gi...@apache.org>.
cadonna commented on code in PR #19544:
URL: https://github.com/apache/kafka/pull/19544#discussion_r2056345054
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceData.java:
##########
@@ -346,11 +349,24 @@ public Map<HostInfo, List<TopicPartition>> partitionsByHost() {
return partitionsByHost.get();
}
+ /** For the current stream thread to request a shutdown of all applications. */
public void requestShutdown() {
shutdownRequested.set(true);
}
+ /** True of the current stream thread requested a shutdown of all applications. */
Review Comment:
```suggestion
/** True if the current stream thread requested a shutdown of all Streams clients belonging to the same application. */
```
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceData.java:
##########
@@ -346,11 +349,24 @@ public Map<HostInfo, List<TopicPartition>> partitionsByHost() {
return partitionsByHost.get();
}
+ /** For the current stream thread to request a shutdown of all applications. */
Review Comment:
```suggestion
/** For the current stream thread to request a shutdown of all Streams clients belonging to the same applications. */
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19190: Handle shutdown application correctly [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck commented on code in PR #19544:
URL: https://github.com/apache/kafka/pull/19544#discussion_r2056910733
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsGroupHeartbeatRequestManager.java:
##########
@@ -521,11 +521,14 @@ private void onSuccessResponse(final StreamsGroupHeartbeatResponse response, fin
}
List<StreamsGroupHeartbeatResponseData.Status> statuses = data.status();
- if (statuses != null && !statuses.isEmpty()) {
- String statusDetails = statuses.stream()
- .map(status -> "(" + status.statusCode() + ") " + status.statusDetail())
- .collect(Collectors.joining(", "));
- logger.warn("Membership is in the following statuses: {}", statusDetails);
+ if (statuses != null) {
+ streamsRebalanceData.setStatuses(statuses);
+ if (!statuses.isEmpty()) {
+ String statusDetails = statuses.stream()
+ .map(status -> "(" + status.statusCode() + ") " + status.statusDetail())
+ .collect(Collectors.joining(", "));
+ logger.warn("Membership is in the following statuses: {}", statusDetails);
Review Comment:
Why a warn? Seems more like an `INFO`
##########
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 " +
Review Comment:
```suggestion
"Encountered the following exception during processing and the registered exception handler" +
```
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -1079,8 +1081,9 @@ public void shutdownToError() {
shutdownErrorHook.run();
}
- public void sendShutdownRequest(final AssignorError assignorError) {
- assignmentErrorCode.set(assignorError.code());
+ public void sendShutdownRequest() {
+ assignmentErrorCode.set(AssignorError.SHUTDOWN_REQUESTED.code());
Review Comment:
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?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19190: Handle shutdown application correctly [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #19544:
URL: https://github.com/apache/kafka/pull/19544#discussion_r2057669349
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceData.java:
##########
@@ -300,6 +301,8 @@ public String toString() {
private final AtomicBoolean shutdownRequested = new AtomicBoolean(false);
+ private final AtomicReference<List<StreamsGroupHeartbeatResponseData.Status>> statuses = new AtomicReference<>(Collections.emptyList());
Review Comment:
```suggestion
private final AtomicReference<List<StreamsGroupHeartbeatResponseData.Status>> statuses = new AtomicReference<>(List.of());
```
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceData.java:
##########
@@ -346,11 +349,24 @@ public Map<HostInfo, List<TopicPartition>> partitionsByHost() {
return partitionsByHost.get();
}
+ /** For the current stream thread to request a shutdown of all Streams clients belonging to the same applications. */
Review Comment:
The comment of `shutdownRequested` uses singular. It would probably be better to unify it.
```suggestion
/** For the current stream thread to request a shutdown of all Streams clients belonging to the same application. */
```
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/StreamsRebalanceDataTest.java:
##########
@@ -420,4 +420,21 @@ public void streamsRebalanceDataShouldBeConstructedWithShutDownRequestedSetFalse
assertFalse(streamsRebalanceData.shutdownRequested());
}
+
+ @Test
+ public void streamsRebalanceDataShouldBeConstructedWithEmptyStatuses() {
+ final UUID processId = UUID.randomUUID();
+ final Optional<StreamsRebalanceData.HostInfo> endpoint = Optional.of(new StreamsRebalanceData.HostInfo("localhost", 9090));
+ final Map<String, StreamsRebalanceData.Subtopology> subtopologies = new HashMap<>();
Review Comment:
```suggestion
final Map<String, StreamsRebalanceData.Subtopology> subtopologies = Map.of();
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19190: Handle shutdown application correctly [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on PR #19544:
URL: https://github.com/apache/kafka/pull/19544#issuecomment-2824855191
> Thanks for the PR, @lucasbru !
>
> I had only minor comments.
Thanks for the review, @cadonna ! Ready for re-review.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org