Added intercepted channel registry in kafka streams app for its implementers (#55)

Author: singhalprerana

kafka-streams-framework/build.gradle.kts

@@ -14,6 +14,7 @@ dependencies {
   api(project(":kafka-streams-serdes"))
   api("org.apache.kafka:kafka-streams:7.2.1-ccs")
   api("io.confluent:kafka-streams-avro-serde:7.2.1")
+  api("org.hypertrace.core.grpcutils:grpc-client-utils:0.11.2")
 
   implementation("org.apache.avro:avro:1.11.1")
   implementation("org.apache.kafka:kafka-clients:7.2.1-ccs")

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

@@ -1,5 +1,7 @@
 package org.hypertrace.core.kafkastreams.framework;
 
+import static io.grpc.Deadline.after;
+import static java.util.concurrent.TimeUnit.SECONDS;
 import static org.apache.kafka.clients.consumer.ConsumerConfig.AUTO_OFFSET_RESET_CONFIG;
 import static org.apache.kafka.clients.producer.ProducerConfig.ACKS_CONFIG;
 import static org.apache.kafka.clients.producer.ProducerConfig.BATCH_SIZE_CONFIG;
@@ -23,6 +25,7 @@
 import io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde;
 import io.micrometer.core.instrument.Tag;
 import io.micrometer.core.instrument.Tags;
+import io.micrometer.core.instrument.binder.grpc.MetricCollectingClientInterceptor;
 import io.micrometer.core.instrument.binder.kafka.KafkaStreamsMetrics;
 import java.time.Duration;
 import java.util.ArrayList;
@@ -39,6 +42,8 @@
 import org.apache.kafka.streams.Topology;
 import org.apache.kafka.streams.errors.LogAndContinueExceptionHandler;
 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.listeners.LoggingStateListener;
 import org.hypertrace.core.kafkastreams.framework.listeners.LoggingStateRestoreListener;
 import org.hypertrace.core.kafkastreams.framework.rocksdb.BoundedMemoryConfigSetter;
@@ -58,6 +63,9 @@ public abstract class KafkaStreamsApp extends PlatformService {
   public static final String PRE_CREATE_TOPICS = "precreate.topics";
   public static final String KAFKA_STREAMS_CONFIG_KEY = "kafka.streams.config";
   private static final Logger logger = LoggerFactory.getLogger(KafkaStreamsApp.class);
+
+  private final GrpcChannelRegistry grpcChannelRegistry;
+
   protected KafkaStreams app;
   private KafkaStreamsMetrics metrics;
 
@@ -67,6 +75,17 @@ public abstract class KafkaStreamsApp extends PlatformService {
 
   public KafkaStreamsApp(ConfigClient configClient) {
     super(configClient);
+    this.grpcChannelRegistry =
+        new GrpcChannelRegistry(
+            GrpcRegistryConfig.builder()
+                .defaultInterceptor(
+                    new MetricCollectingClientInterceptor(
+                        PlatformMetricsRegistry.getMeterRegistry()))
+                .build());
+  }
+
+  protected GrpcChannelRegistry getGrpcChannelRegistry() {
+    return grpcChannelRegistry;
   }
 
   @Override
@@ -147,6 +166,7 @@ protected void doStop() {
       metrics.close();
     }
     app.close(Duration.ofSeconds(30));
+    grpcChannelRegistry.shutdown(after(10, SECONDS));
   }
 
   /**