fix: register state listener after restarting runtime (#9032)

* fix: register state listener after restarting runtime
confluentinc · Apr 14, 2022 · 1f7a09a · 1f7a09a
1 parent 3721a1e
commit 1f7a09a
Showing 1 changed file with 1 addition and 0 deletions.
diff --git a/ksqldb-engine/src/main/java/io/confluent/ksql/util/SharedKafkaStreamsRuntimeImpl.java b/ksqldb-engine/src/main/java/io/confluent/ksql/util/SharedKafkaStreamsRuntimeImpl.java
@@ -258,6 +258,7 @@ public void restartStreamsRuntime() {
     final KafkaStreamsNamedTopologyWrapper kafkaStreamsNamedTopologyWrapper = kafkaStreamsBuilder
         .buildNamedTopologyWrapper(streamsProperties);
     kafkaStreams = kafkaStreamsNamedTopologyWrapper;
+    kafkaStreams.setStateListener(stateListener());
     for (final NamedTopology topology : liveTopologies) {
       final BinPackedPersistentQueryMetadataImpl query = collocatedQueries
           .get(new QueryId(topology.name()));