add real counters

Author: varkey98

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

@@ -46,6 +46,8 @@
 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;
@@ -103,8 +105,13 @@ protected void doInit() {
       // build topologies
       Map<String, KStream<?, ?>> sourceStreams = new HashMap<>();
       StreamsBuilder streamsBuilder = new StreamsBuilder();
+
+      // initialize MetricsInterceptorFactory
+      MetricsInterceptorFactory metricsInterceptorFactory = new MetricsInterceptorFactory();
       streamsBuilder =
-          new StreamsBuilderWithInterceptor(streamsBuilder, List.of(MetricsProcessor::new));
+          new StreamsBuilderWithInterceptor(
+              streamsBuilder, List.of(metricsInterceptorFactory::create));
+
       streamsBuilder = buildTopology(streamsConfig, streamsBuilder, sourceStreams);
       this.topology = streamsBuilder.build();
 

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

@@ -1,4 +1,4 @@
-package org.hypertrace.core.kafkastreams.framework;
+package org.hypertrace.core.kafkastreams.framework.interceptors;
 
 import java.util.Collection;
 import java.util.List;

…ework/StreamsBuilderWithInterceptor.java …ptors/StreamsBuilderWithInterceptor.javakafka-streams-framework/src/main/java/org/hypertrace/core/kafkastreams/framework/StreamsBuilderWithInterceptor.java renamed to kafka-streams-framework/src/main/java/org/hypertrace/core/kafkastreams/framework/interceptors/StreamsBuilderWithInterceptor.java kafka-streams-framework/src/main/java/org/hypertrace/core/kafkastreams/framework/StreamsBuilderWithInterceptor.java renamed to kafka-streams-framework/src/main/java/org/hypertrace/core/kafkastreams/framework/interceptors/StreamsBuilderWithInterceptor.java

@@ -0,0 +1,22 @@
+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;
+
+public class MetricsInterceptor implements Processor {
+
+  private final Counter timeLagCounter;
+  private final Counter numRecordsCounter;
+
+  MetricsInterceptor(Counter numRecordsCounter, Counter timeLagCounter) {
+    this.numRecordsCounter = numRecordsCounter;
+    this.timeLagCounter = timeLagCounter;
+  }
+
+  @Override
+  public void process(Record record) {
+    timeLagCounter.increment(System.currentTimeMillis() - record.timestamp());
+    numRecordsCounter.increment();
+  }
+}

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

@@ -0,0 +1,24 @@
+package org.hypertrace.core.kafkastreams.framework.interceptors.metrics;
+
+import io.micrometer.core.instrument.Counter;
+import java.util.Collections;
+import org.hypertrace.core.serviceframework.metrics.PlatformMetricsRegistry;
+
+public class MetricsInterceptorFactory {
+  private static final String TIME_LAG_COUNTER_NAME = "time_lag";
+  private static final String NUM_RECORDS_COUNTER_NAME = "num_records";
+
+  private final Counter timeLagCounter;
+  private final Counter numRecordsCounter;
+
+  public MetricsInterceptorFactory() {
+    this.numRecordsCounter =
+        PlatformMetricsRegistry.registerCounter(NUM_RECORDS_COUNTER_NAME, Collections.emptyMap());
+    this.timeLagCounter =
+        PlatformMetricsRegistry.registerCounter(TIME_LAG_COUNTER_NAME, Collections.emptyMap());
+  }
+
+  public MetricsInterceptor create() {
+    return new MetricsInterceptor(numRecordsCounter, timeLagCounter);
+  }
+}