update interceptors

Author: varkey98

kafka-bom/build.gradle.kts

@@ -29,9 +29,6 @@ dependencies {
       because("[https://nvd.nist.gov/vuln/detail/CVE-2025-12183] in org.lz4:lz4-java:1.8.0")
       because("CVE-2025-12183 is fixed in 1.8.1")
     }
-    api("org.eclipse.jetty:jetty-bom:11.0.26") {
-      because("CVE-2025-5115 is fixed in 11.0.26")
-    }
 
     api("io.confluent:kafka-streams-avro-serde:$confluentVersion")
     api("io.confluent:kafka-protobuf-serializer:$confluentVersion")
@@ -40,6 +37,6 @@ dependencies {
     api("org.apache.kafka:kafka-clients:$confluentCcsVersion")
     api("org.apache.kafka:kafka-streams:$confluentCcsVersion")
     api("org.apache.kafka:kafka-streams-test-utils:$confluentCcsVersion")
-    api("org.apache.avro:avro:1.12.0")
+    api("org.apache.avro:avro:1.12.1")
   }
 }

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

@@ -46,8 +46,6 @@
 import org.apache.kafka.streams.kstream.KStream;
 import org.hypertrace.core.grpcutils.client.GrpcChannelRegistry;
 import org.hypertrace.core.grpcutils.client.GrpcRegistryConfig;
-import org.hypertrace.core.kafkastreams.framework.interceptors.StreamsBuilderWithInterceptor;
-import org.hypertrace.core.kafkastreams.framework.interceptors.metrics.MetricsInterceptorFactory;
 import org.hypertrace.core.kafkastreams.framework.listeners.LoggingStateListener;
 import org.hypertrace.core.kafkastreams.framework.listeners.LoggingStateRestoreListener;
 import org.hypertrace.core.kafkastreams.framework.rocksdb.BoundedMemoryConfigSetter;
@@ -104,11 +102,8 @@ protected void doInit() {
       streamsConfig = mergeProperties(getBaseStreamsConfig(), getJobStreamsConfig(getAppConfig()));
       // build topologies
       Map<String, KStream<?, ?>> sourceStreams = new HashMap<>();
-      // initialize MetricsInterceptorFactory
-      MetricsInterceptorFactory metricsInterceptorFactory = new MetricsInterceptorFactory();
-      StreamsBuilder streamsBuilder =
-          new StreamsBuilderWithInterceptor(List.of(metricsInterceptorFactory::create));
 
+      StreamsBuilder streamsBuilder = new StreamsBuilder();
       streamsBuilder = buildTopology(streamsConfig, streamsBuilder, sourceStreams);
       this.topology = streamsBuilder.build();
 

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

@@ -1,10 +1,12 @@
 package org.hypertrace.core.kafkastreams.framework.interceptors.metrics;
 
 import io.micrometer.core.instrument.Counter;
-import org.apache.kafka.streams.processor.api.Processor;
-import org.apache.kafka.streams.processor.api.Record;
+import java.util.Map;
+import org.apache.kafka.clients.consumer.ConsumerInterceptor;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
 
-public class MetricsInterceptor implements Processor {
+public class MetricsInterceptor implements ConsumerInterceptor<Object, Object> {
 
   private final Counter timeLagCounter;
   private final Counter numRecordsCounter;
@@ -15,8 +17,26 @@ public class MetricsInterceptor implements Processor {
   }
 
   @Override
-  public void process(Record record) {
-    timeLagCounter.increment(System.currentTimeMillis() - record.timestamp());
-    numRecordsCounter.increment();
+  public ConsumerRecords<Object, Object> onConsume(ConsumerRecords<Object, Object> records) {
+    for (ConsumerRecord<Object, Object> record : records) {
+      timeLagCounter.increment(System.currentTimeMillis() - record.timestamp());
+      numRecordsCounter.increment();
+    }
+    return records;
+  }
+
+  @Override
+  public void close() {
+    // no-op
+  }
+
+  @Override
+  public void onCommit(Map offsets) {
+    // no-op
+  }
+
+  @Override
+  public void configure(Map<String, ?> configs) {
+    // no-op
   }
 }