ableegoldman commented on a change in pull request #9487:
URL: https://github.com/apache/kafka/pull/9487#discussion_r523302976
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStreamThread.java
##########
@@ -311,6 +314,8 @@ public void run() {
"Updating global state failed. You can restart KafkaStreams to
recover from this error.",
recoverableException
);
+ } catch (final Exception e) {
+ this.streamsUncaughtExceptionHandler.accept(e);
Review comment:
Ah ok I thought we executed this cleanup logic in the
GlobalStreamThread's `shutdown` method but now I see that's not true. Sorry for
the confusion there.
I do see some minor outstanding issues here, mainly around the state
diagram. Let's say the user opts to `SHUTDOWN_CLIENT` in the new handler: the
intended semantics are to end up in `NOT_RUNNING`
But I think what would happen is that from the global thread we would
immediately call `KafkaStreams#close` , which kicks off a shutdown thread to
wait for all threads to join and then sets the state to `NOT_RUNNING`. Then
when the handler returns, it would transition the global thread to
`PENDING_SHUTDOWN` and then finally to `DEAD`. And during the transition to
`DEAD`, we would actually end up transitioning the KafkaStreams instance to
`ERROR`, rather than `NOT_RUNNING` as intended. So probably, we just need to
update the `onChange` method in KafkaStreams.
This also reminds me of another thing, we need to update the FSM diagram and
allowed transitions in KafkaStreams to reflect the new semantics we decided on
for ERROR (which IIRC is basically just to make it a terminal state). Does that
sound right to you?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
