add new interceptor framework

Author: varkey98

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

@@ -103,6 +103,8 @@ protected void doInit() {
       // build topologies
       Map<String, KStream<?, ?>> sourceStreams = new HashMap<>();
       StreamsBuilder streamsBuilder = new StreamsBuilder();
+      streamsBuilder =
+          new StreamsBuilderWithInterceptor(streamsBuilder, List.of(MetricsProcessor::new));
       streamsBuilder = buildTopology(streamsConfig, streamsBuilder, sourceStreams);
       this.topology = streamsBuilder.build();
 

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

@@ -0,0 +1,16 @@
+package org.hypertrace.core.kafkastreams.framework;
+
+import java.time.Instant;
+import org.apache.kafka.streams.processor.api.Processor;
+import org.apache.kafka.streams.processor.api.Record;
+
+public class MetricsProcessor implements Processor {
+
+  @Override
+  public void process(Record record) {
+    System.out.println("Hello world");
+    System.out.println(record.timestamp());
+    Instant timestamp = Instant.ofEpochMilli(record.timestamp());
+    System.out.println(timestamp); // e.g., 2026-03-10T12:17:00Z
+  }
+}

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

@@ -0,0 +1,43 @@
+package org.hypertrace.core.kafkastreams.framework;
+
+import java.util.Collection;
+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;
+    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);
+    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);
+    processorSuppliers.forEach(stream::process);
+    return stream;
+  }
+}