ableegoldman commented on a change in pull request #10544:
URL: https://github.com/apache/kafka/pull/10544#discussion_r617027950
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -471,6 +472,10 @@ private void replaceStreamThread(final Throwable
throwable) {
private void handleStreamsUncaughtException(final Throwable throwable,
final
StreamsUncaughtExceptionHandler streamsUncaughtExceptionHandler) {
+ if (throwable instanceof NamedTopologyStreamsException) {
Review comment:
Agreed
##########
File path: streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
##########
@@ -471,6 +472,10 @@ private void replaceStreamThread(final Throwable
throwable) {
private void handleStreamsUncaughtException(final Throwable throwable,
final
StreamsUncaughtExceptionHandler streamsUncaughtExceptionHandler) {
+ if (throwable instanceof NamedTopologyStreamsException) {
+ String name = ((NamedTopologyStreamsException)
throwable).getTopologyName();
+ ((StreamThread)
Thread.currentThread()).deprioritizeNamedTopology(name);
Review comment:
I think it's fine to lose some state after a restart, presumably we'll
just start up with all queries assumed to be good and then re-de-prioritize any
that are still failing. We can iterate on this and let ksql persist this info
in the command topic and give Streams hints when starting up that a query may
still be failing, but just tracking it in-memory seems good enough for a first
pass
--
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
