add unit tests

Author: varkey98

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

@@ -16,7 +16,6 @@ public class MetricsInterceptor implements Processor {
 
   @Override
   public void process(Record record) {
-    System.out.println("Inside custom processor");
     timeLagCounter.increment(System.currentTimeMillis() - record.timestamp());
     numRecordsCounter.increment();
   }

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

@@ -5,8 +5,8 @@
 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 static final String TIME_LAG_COUNTER_NAME = "kafka_records_time_lag";
+  private static final String NUM_RECORDS_COUNTER_NAME = "kafka_records_count";
 
   private final Counter timeLagCounter;
   private final Counter numRecordsCounter;

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

@@ -0,0 +1,20 @@
+package org.hypertrace.core.kafkastreams.framework.interceptors;
+
+import java.util.List;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.processor.api.Processor;
+import org.apache.kafka.streams.processor.api.Record;
+
+public class CachingInterceptor implements Processor<String, String, Void, Void> {
+
+  private List<KeyValue<String, String>> pairs;
+
+  public CachingInterceptor(List<KeyValue<String, String>> pairs) {
+    this.pairs = pairs;
+  }
+
+  @Override
+  public void process(Record<String, String> record) {
+    pairs.add(new KeyValue<>(record.key(), record.value()));
+  }
+}

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

@@ -0,0 +1,21 @@
+package org.hypertrace.core.kafkastreams.framework.interceptors;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.kafka.streams.KeyValue;
+
+public class CachingInterceptorFactory {
+  private List<KeyValue<String, String>> pairs;
+
+  CachingInterceptorFactory() {
+    this.pairs = new ArrayList<>();
+  }
+
+  CachingInterceptor create() {
+    return new CachingInterceptor(pairs);
+  }
+
+  List<KeyValue<String, String>> getCache() {
+    return pairs;
+  }
+}

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

@@ -0,0 +1,88 @@
+package org.hypertrace.core.kafkastreams.framework.interceptors;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.util.List;
+import java.util.Properties;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.streams.TestInputTopic;
+import org.apache.kafka.streams.TestOutputTopic;
+import org.apache.kafka.streams.Topology;
+import org.apache.kafka.streams.TopologyTestDriver;
+import org.apache.kafka.streams.kstream.KStream;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+public class StreamsBuilderWithInterceptorTest {
+
+  private static final String INPUT_TOPIC = "input";
+  private static final String OUTPUT_TOPIC = "output";
+
+  private StreamsBuilder streamsBuilder;
+  private Topology topology;
+  private TopologyTestDriver td;
+  private TestInputTopic<String, String> inputTopic;
+  private TestOutputTopic<String, String> outputTopic;
+
+  private Properties streamsConfig;
+
+  CachingInterceptorFactory factory;
+
+  @BeforeEach
+  public void setup() {
+
+    factory = new CachingInterceptorFactory();
+    streamsBuilder = new StreamsBuilderWithInterceptor(List.of(factory::create));
+    KStream<String, String> stream = streamsBuilder.stream(INPUT_TOPIC);
+    stream.to(OUTPUT_TOPIC);
+    topology = streamsBuilder.build();
+
+    streamsConfig = new Properties();
+    streamsConfig.put("application.id", "test-interceptor");
+    streamsConfig.put("bootstrap.servers", "dummy:1234");
+    streamsConfig.put("default.key.serde", Serdes.StringSerde.class.getName());
+    streamsConfig.put("default.value.serde", Serdes.StringSerde.class.getName());
+    td = new TopologyTestDriver(topology, streamsConfig);
+  }
+
+  @AfterEach
+  public void tearDown() {
+    td.close();
+  }
+
+  @Test
+  public void interceptorTest() {
+    inputTopic =
+        td.createInputTopic(
+            INPUT_TOPIC, Serdes.String().serializer(), Serdes.String().serializer());
+    outputTopic =
+        td.createOutputTopic(
+            OUTPUT_TOPIC, Serdes.String().deserializer(), Serdes.String().deserializer());
+
+    assertThat(outputTopic.isEmpty(), is(true));
+
+    inputTopic.pipeInput("fooooo", "barrrrr");
+    assertThat(outputTopic.readValue(), equalTo("barrrrr"));
+    assertThat(outputTopic.isEmpty(), is(true));
+
+    inputTopic.pipeInput("foo", "bar");
+    assertThat(outputTopic.readValue(), equalTo("bar"));
+    assertThat(outputTopic.isEmpty(), is(true));
+
+    List<KeyValue<String, String>> cache = factory.getCache();
+    assertThat(cache.size(), is(2));
+    KeyValue<String, String> pair = cache.get(0);
+    assertEquals("fooooo", pair.key);
+    assertEquals("barrrrr", pair.value);
+
+    pair = cache.get(1);
+    assertEquals("foo", pair.key);
+    assertEquals("bar", pair.value);
+  }
+}

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

@@ -0,0 +1,35 @@
+package org.hypertrace.core.kafkastreams.framework.interceptors.metrics;
+
+import static org.mockito.ArgumentMatchers.anyDouble;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+import io.micrometer.core.instrument.Counter;
+import org.apache.kafka.streams.processor.api.Record;
+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);
+    interceptor = new MetricsInterceptor(numRecordsCounter, timeLagCounter);
+  }
+
+  @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());
+  }
+}