Wighted group partitioner (#51)

Author: laxmanchekka

kafka-streams-framework/build.gradle.kts

@@ -11,18 +11,14 @@ tasks.test {
 }
 
 dependencies {
-  annotationProcessor("org.projectlombok:lombok:1.18.24")
-  compileOnly("org.projectlombok:lombok:1.18.24")
-
   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")
 
-  implementation("com.google.guava:guava:31.1-jre")
   implementation("org.apache.avro:avro:1.11.1")
   implementation("org.apache.kafka:kafka-clients:7.2.1-ccs")
-  implementation("org.hypertrace.core.serviceframework:platform-metrics:0.1.39")
-  implementation("org.hypertrace.core.serviceframework:platform-service-framework:0.1.39")
+  implementation("org.hypertrace.core.serviceframework:platform-metrics:0.1.47")
+  implementation("org.hypertrace.core.serviceframework:platform-service-framework:0.1.47")
   implementation("org.apache.commons:commons-lang3:3.12.0")
 
   testImplementation("org.apache.kafka:kafka-streams-test-utils:7.2.1-ccs")

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

@@ -6,6 +6,7 @@
 import static org.apache.kafka.clients.producer.ProducerConfig.COMPRESSION_TYPE_CONFIG;
 import static org.apache.kafka.clients.producer.ProducerConfig.LINGER_MS_CONFIG;
 import static org.apache.kafka.common.config.TopicConfig.RETENTION_MS_CONFIG;
+import static org.apache.kafka.streams.StreamsConfig.APPLICATION_ID_CONFIG;
 import static org.apache.kafka.streams.StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG;
 import static org.apache.kafka.streams.StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG;
 import static org.apache.kafka.streams.StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG;
@@ -20,6 +21,8 @@
 import com.google.common.collect.Streams;
 import com.typesafe.config.Config;
 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.kafka.KafkaStreamsMetrics;
 import java.time.Duration;
 import java.util.ArrayList;
@@ -90,7 +93,11 @@ protected void doInit() {
       app = new KafkaStreams(topology, streamsConfigProps);
 
       // export kafka streams metrics
-      metrics = new KafkaStreamsMetrics(app);
+      metrics =
+          new KafkaStreamsMetrics(
+              app,
+              Tags.of(
+                  Tag.of("kstreams.app", streamsConfigProps.getProperty(APPLICATION_ID_CONFIG))));
       metrics.bindTo(PlatformMetricsRegistry.getMeterRegistry());
 
       // useful for resetting local state - during testing or any other scenarios where

kafka-streams-partitioners/avro-partitioners/build.gradle.kts

@@ -0,0 +1,34 @@
+plugins {
+  `java-library`
+  jacoco
+  id("org.hypertrace.avro-plugin")
+  id("org.hypertrace.publish-plugin")
+  id("org.hypertrace.jacoco-report-plugin")
+}
+
+tasks.test {
+  useJUnitPlatform()
+}
+
+dependencies {
+  annotationProcessor("org.projectlombok:lombok:1.18.24")
+  compileOnly("org.projectlombok:lombok:1.18.24")
+
+  implementation("com.google.guava:guava:31.1-jre")
+  implementation("org.apache.avro:avro:1.11.1")
+  implementation("com.typesafe:config:1.4.2")
+  implementation("org.apache.kafka:kafka-clients:7.2.1-ccs")
+  implementation("org.apache.kafka:kafka-streams:7.2.1-ccs")
+  implementation("org.slf4j:slf4j-api:1.7.36")
+
+  testImplementation("org.junit.jupiter:junit-jupiter:5.8.2")
+  testImplementation("org.junit-pioneer:junit-pioneer:1.7.1")
+  testImplementation("org.mockito:mockito-core:4.5.1")
+  testRuntimeOnly("org.apache.logging.log4j:log4j-slf4j-impl:2.17.2")
+}
+
+// Disabling compatibility check for the test avro definitions.
+tasks.named<org.hypertrace.gradle.avro.CheckAvroCompatibility>("avroCompatibilityCheck") {
+  enabled = false
+  setAgainstFiles(null)
+}

…rtitioner/AvroFieldValuePartitioner.java …rtitioner/AvroFieldValuePartitioner.javakafka-streams-framework/src/main/java/org/hypertrace/core/kafkastreams/framework/partitioner/AvroFieldValuePartitioner.java renamed to kafka-streams-partitioners/avro-partitioners/src/main/java/org/hypertrace/core/kafkastreams/framework/partitioner/AvroFieldValuePartitioner.java kafka-streams-framework/src/main/java/org/hypertrace/core/kafkastreams/framework/partitioner/AvroFieldValuePartitioner.java renamed to kafka-streams-partitioners/avro-partitioners/src/main/java/org/hypertrace/core/kafkastreams/framework/partitioner/AvroFieldValuePartitioner.java

@@ -0,0 +1,14 @@
+@namespace("org.hypertrace.core.kafkastreams.framework.partitioner")
+protocol TestRecordProtocol {
+  record TestCustomerRecord {
+    string customer_id;
+    string span_id;
+    string trace_id;
+  }
+
+  record TestTenantRecord {
+    string tenant_id;
+    string span_id;
+    string trace_id;
+  }
+}