ableegoldman commented on a change in pull request #10387:
URL: https://github.com/apache/kafka/pull/10387#discussion_r602525060
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -492,22 +492,18 @@ private void handleStreamsUncaughtException(final
Throwable throwable,
closeToError();
break;
case SHUTDOWN_APPLICATION:
+ if (getNumLiveStreamThreads() <= 1) {
Review comment:
You can use `StreamsBuilder#addGlobalStore` to inject an error, this
method has a required parameter for a Processor which you're supposed to use to
insert the data into the store (tbh no idea why you have to implement this
yourself, it's caused a LOT of problems since it's unclear what you can/can't
do with this -- but that's a whole other can of worms).
Adding a thread is probably a fine workaround for now, it clearly answered
the question, but imo we should just go ahead and use the
`StreamsBuilder#addGlobalStore` + injected error to decouple the test from the
way we currently happen to handle this. Then we won't have to rewrite it if we
ever change something.
Regarding the IllegalSTateException: wouldn't the thread hit this
`IllegalStateException` before it got the chance to send the shutdown signal in
a rebalance? IIRC the `subscribe` that's throwing is at the very beginning of
the StreamThread loop. And unfortunately I think this may actually result in an
endless cycle -- the new thread hits an unexpected exception and invokes the
handler, SHUTDOWN_APPLICATION notices that this is the last StreamThread so it
starts a new one, and so on.
--
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
