fix: UncaughtExceptionHandler not being set for Persistent Queries (#…

…4087)
confluentinc · Dec 9, 2019 · e193a2a · e193a2a
1 parent 1577801
commit e193a2a
Show file tree

Hide file tree

Showing 2 changed files with 2 additions and 3 deletions.
diff --git a/ksql-engine/src/main/java/io/confluent/ksql/query/KafkaStreamsBuilderImpl.java b/ksql-engine/src/main/java/io/confluent/ksql/query/KafkaStreamsBuilderImpl.java
@@ -15,6 +15,7 @@
 
 package io.confluent.ksql.query;
 
+import io.confluent.ksql.util.KafkaStreamsUncaughtExceptionHandler;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Properties;
@@ -40,6 +41,7 @@ public BuildResult buildKafkaStreams(
     props.putAll(conf);
     final Topology topology = builder.build(props);
     final KafkaStreams kafkaStreams = new KafkaStreams(topology, props, clientSupplier);
+    kafkaStreams.setUncaughtExceptionHandler(new KafkaStreamsUncaughtExceptionHandler());
     return new BuildResult(topology, kafkaStreams);
   }
 }
diff --git a/ksql-engine/src/main/java/io/confluent/ksql/query/QueryExecutor.java b/ksql-engine/src/main/java/io/confluent/ksql/query/QueryExecutor.java
@@ -43,7 +43,6 @@
 import io.confluent.ksql.serde.GenericKeySerDe;
 import io.confluent.ksql.serde.KeyFormat;
 import io.confluent.ksql.services.ServiceContext;
-import io.confluent.ksql.util.KafkaStreamsUncaughtExceptionHandler;
 import io.confluent.ksql.util.KsqlConfig;
 import io.confluent.ksql.util.KsqlConstants;
 import io.confluent.ksql.util.KsqlException;
@@ -163,8 +162,6 @@ public TransientQueryMetadata buildTransientQuery(
     final BuildResult built =
         kafkaStreamsBuilder.buildKafkaStreams(streamsBuilder, streamsProperties);
 
-    built.kafkaStreams.setUncaughtExceptionHandler(new KafkaStreamsUncaughtExceptionHandler());
-
     final LogicalSchema transientSchema = buildTransientQuerySchema(schema);
 
     return new TransientQueryMetadata(