fix working

Author: varkey98

kafka-streams-framework/src/main/java/org/hypertrace/core/kafkastreams/framework/KafkaStreamsApp.java

@@ -104,13 +104,10 @@ protected void doInit() {
       streamsConfig = mergeProperties(getBaseStreamsConfig(), getJobStreamsConfig(getAppConfig()));
       // build topologies
       Map<String, KStream<?, ?>> sourceStreams = new HashMap<>();
-      StreamsBuilder streamsBuilder = new StreamsBuilder();
-
       // initialize MetricsInterceptorFactory
       MetricsInterceptorFactory metricsInterceptorFactory = new MetricsInterceptorFactory();
-      streamsBuilder =
-          new StreamsBuilderWithInterceptor(
-              streamsBuilder, List.of(metricsInterceptorFactory::create));
+      StreamsBuilder streamsBuilder =
+          new StreamsBuilderWithInterceptor(List.of(metricsInterceptorFactory::create));
 
       streamsBuilder = buildTopology(streamsConfig, streamsBuilder, sourceStreams);
       this.topology = streamsBuilder.build();

kafka-streams-framework/src/main/java/org/hypertrace/core/kafkastreams/framework/interceptors/StreamsBuilderWithInterceptor.java

@@ -4,39 +4,30 @@
 import java.util.List;
 import java.util.regex.Pattern;
 import org.apache.kafka.streams.StreamsBuilder;
-import org.apache.kafka.streams.Topology;
 import org.apache.kafka.streams.kstream.Consumed;
 import org.apache.kafka.streams.kstream.KStream;
 import org.apache.kafka.streams.processor.api.ProcessorSupplier;
 
 public class StreamsBuilderWithInterceptor extends StreamsBuilder {
 
-  private final StreamsBuilder delegate;
   private final List<ProcessorSupplier> processorSuppliers;
 
-  public StreamsBuilderWithInterceptor(
-      StreamsBuilder delegate, List<ProcessorSupplier> processorSuppliers) {
-    this.delegate = delegate;
+  public StreamsBuilderWithInterceptor(List<ProcessorSupplier> processorSuppliers) {
     this.processorSuppliers = processorSuppliers;
   }
 
-  @Override
-  public synchronized Topology build() {
-    return delegate.build();
-  }
-
   @Override
   public synchronized <K, V> KStream<K, V> stream(
       final Pattern topicPattern, final Consumed<K, V> consumed) {
-    KStream<K, V> stream = delegate.stream(topicPattern, consumed);
+    KStream<K, V> stream = super.stream(topicPattern, consumed);
     processorSuppliers.forEach(stream::process);
     return stream;
   }
 
   @Override
   public synchronized <K, V> KStream<K, V> stream(
       final Collection<String> topics, final Consumed<K, V> consumed) {
-    KStream<K, V> stream = delegate.stream(topics, consumed);
+    KStream<K, V> stream = super.stream(topics, consumed);
     processorSuppliers.forEach(stream::process);
     return stream;
   }