add unit tests

Author: varkey98

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

@@ -15,7 +15,6 @@ public class MetricsInterceptor implements ConsumerInterceptor<Object, Object> {
 
   @Override
   public ConsumerRecords<Object, Object> onConsume(ConsumerRecords<Object, Object> records) {
-    System.out.println("Hello world! 2.0");
     for (ConsumerRecord<Object, Object> record : records) {
       timeLagCounter.increment(System.currentTimeMillis() - record.timestamp());
       numRecordsCounter.increment();
@@ -35,7 +34,6 @@ public void onCommit(Map offsets) {
 
   @Override
   public void configure(Map<String, ?> configs) {
-    System.out.println("Hello world! 3.0");
     this.timeLagCounter = PlatformMetricsRegistry.getMeterRegistry().counter(TIME_LAG_COUNTER_NAME);
     this.numRecordsCounter =
         PlatformMetricsRegistry.getMeterRegistry().counter(NUM_RECORDS_COUNTER_NAME);

kafka-streams-framework/src/test/java/org/hypertrace/core/kafkastreams/framework/interceptors/metrics/MetricsInterceptorTest.java

@@ -1,30 +1,48 @@
 package org.hypertrace.core.kafkastreams.framework.interceptors.metrics;
 
-import static org.mockito.Mockito.mock;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
 
-import io.micrometer.core.instrument.Counter;
+import com.google.common.collect.Maps;
+import com.typesafe.config.ConfigFactory;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.common.TopicPartition;
+import org.hypertrace.core.serviceframework.metrics.PlatformMetricsRegistry;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
 public class MetricsInterceptorTest {
 
-  private Counter timeLagCounter;
-  private Counter numRecordsCounter;
-  private MetricsInterceptor interceptor;
-
   @BeforeEach
   void setup() {
-    timeLagCounter = mock(Counter.class);
-    numRecordsCounter = mock(Counter.class);
+    Map<String, Object> config = new HashMap<>();
+    config.put("reporter.names", List.of("testing"));
+
+    PlatformMetricsRegistry.initMetricsRegistry("test", ConfigFactory.parseMap(config));
   }
 
   @Test
-  void shouldIncrementCounters() {
-    //    Record<Object, Object> record =
-    //        new Record<>("key", "value", System.currentTimeMillis() - 50000);
-    //    interceptor.process(record);
-    //
-    //    verify(numRecordsCounter, times(1)).increment();
-    //    verify(timeLagCounter, times(1)).increment(anyDouble());
+  public void testOnConsume() {
+    TopicPartition tp = new TopicPartition("test-topic", 0);
+    List<ConsumerRecord<Object, Object>> records =
+        List.of(
+            new ConsumerRecord<>("test-topic", 0, 0, "k1", "v1"),
+            new ConsumerRecord<>("test-topic", 0, 1, "k2", null));
+
+    Map<TopicPartition, List<ConsumerRecord<Object, Object>>> map = Map.of(tp, records);
+    ConsumerRecords<Object, Object> input = new ConsumerRecords<>(map);
+    MetricsInterceptor interceptor = new MetricsInterceptor();
+    interceptor.configure(Maps.newTreeMap());
+    ConsumerRecords<Object, Object> output = interceptor.onConsume(input);
+    assertEquals(2, output.count());
+
+    assertEquals(
+        2.0, PlatformMetricsRegistry.getMeterRegistry().counter("kafka_records_count").count());
+    assertNotEquals(
+        0, PlatformMetricsRegistry.getMeterRegistry().counter("kafka_records_time_lag").count());
   }
 }