update to latest kafka streams framework (#456)

* update to latest kafka streams framework * vuln fix
hypertrace · Feb 22, 2024 · d0e409f · d0e409f
1 parent aca9577
commit d0e409f
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 7 deletions.
diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml
@@ -4,8 +4,8 @@ hypertrace-attribute-service = "0.14.38"
 hypertrace-config-service = "0.1.60"
 hypertrace-grpc-utils = "0.12.6"
 hypertrace-serviceFramework = "0.1.68"
-hypertrace-kafkaStreams = "0.4.4"
-hypertrace-view-generator = "0.4.21"
+hypertrace-kafkaStreams = "0.4.7"
+hypertrace-view-generator = "0.4.24"
 grpc = "1.57.2"
 
 [libraries]

diff --git a/.../src/main/java/org/hypertrace/traceenricher/enrichment/clients/DefaultClientRegistry.java b/.../src/main/java/org/hypertrace/traceenricher/enrichment/clients/DefaultClientRegistry.java
@@ -21,6 +21,7 @@
 import org.hypertrace.core.grpcutils.client.GrpcChannelConfig;
 import org.hypertrace.core.grpcutils.client.GrpcChannelRegistry;
 import org.hypertrace.core.grpcutils.client.RequestContextClientCallCredsProviderFactory;
+import org.hypertrace.core.kafka.event.listener.KafkaConsumerUtils;
 import org.hypertrace.core.kafka.event.listener.KafkaLiveEventListener;
 import org.hypertrace.entity.change.event.v1.EntityChangeEventKey;
 import org.hypertrace.entity.change.event.v1.EntityChangeEventValue;
@@ -99,7 +100,7 @@ public DefaultClientRegistry(
             EntityServiceClientConfig.from(config).getCacheConfig(),
             cacheLoaderExecutor);
     this.entityChangeEventListener =
-        getEntityChangeEventConsumer(config, edsCacheClient::updateBasedOnChangeEvent);
+        getEntityChangeEventListener(config, edsCacheClient::updateBasedOnChangeEvent);
     this.entityDataClient = EntityDataClient.builder(this.entityServiceChannel).build();
     this.entityCache = new EntityCache(this.edsCacheClient, cacheLoaderExecutor);
     this.entityAccessor =
@@ -197,7 +198,7 @@ protected Channel buildChannel(String host, int port, GrpcChannelConfig grpcChan
   }
 
   private static Optional<KafkaLiveEventListener<EntityChangeEventKey, EntityChangeEventValue>>
-      getEntityChangeEventConsumer(
+      getEntityChangeEventListener(
           Config clientsConfig, BiConsumer<EntityChangeEventKey, EntityChangeEventValue> callback) {
     if (clientsConfig.hasPath(ENTITY_CHANGE_EVENTS_CONSUMER_ENABLED_KEY)
         && clientsConfig.getBoolean(ENTITY_CHANGE_EVENTS_CONSUMER_ENABLED_KEY)) {
@@ -206,14 +207,17 @@ protected Channel buildChannel(String host, int port, GrpcChannelConfig grpcChan
           Collections.singletonMap(
               "schema.registry.url",
               clientsConfig.getString(ENTITY_CHANGE_EVENTS_SCHEMA_REGISTRY_URL_KEY));
+      Config kafkaConfig = clientsConfig.getConfig(ENTITY_CHANGE_EVENTS_CONFIG_KEY);
       return Optional.of(
           new KafkaLiveEventListener.Builder<EntityChangeEventKey, EntityChangeEventValue>()
               .registerCallback(callback)
               .build(
                   consumerName,
-                  clientsConfig.getConfig(ENTITY_CHANGE_EVENTS_CONFIG_KEY),
-                  getEntityChangeEventKeyDeser(deserConfig),
-                  getEntityChangeEventValueDeser(deserConfig)));
+                  kafkaConfig,
+                  KafkaConsumerUtils.getKafkaConsumer(
+                      kafkaConfig,
+                      getEntityChangeEventKeyDeser(deserConfig),
+                      getEntityChangeEventValueDeser(deserConfig))));
     }
     return Optional.empty();
   }